feat(proforma): 预编报表(预算版三张报表)P2 — 预算vs实际vs差异,独立/proforma端点,复用budget_plans+报表模板,无预算行显式标注

This commit is contained in:
Hermes CI Fix
2026-08-30 23:13:32 +08:00
parent db0f7aa591
commit 27d8269667
3 changed files with 711 additions and 3 deletions
+212
View File
@@ -0,0 +1,212 @@
"""
预编报表(预算版三张报表)测试 — P2 2026-08-30
覆盖:
1. 三张预算版报表接口 200
2. has_budget 标注正确(有预算行 true / 无预算映射行 false
3. 差异计算与 budget-execution 一致(同 KPI 同 period 对比)
4. 无预算行显式标注(budget_source=none
5. 比率型KPI单独标注(ratio_kpi=true,不计算金额差异)
"""
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, get_token_for_user, auth_header
from app.models import KPIDefinition, KPIValue, BudgetPlan
def _create_kpi(db: Session, code: str, name: str, entity_id: int = 1, frequency: str = "monthly", target_value=None) -> KPIDefinition:
kpi = KPIDefinition(
entity_id=entity_id,
kpi_code=code,
kpi_name=name,
dimension="finance",
category="financial_report",
formula="-",
data_source="测试",
data_owner="财务部",
frequency=frequency,
unit="",
target_value=target_value,
kpi_level="operational",
status="active",
)
db.add(kpi)
db.commit()
db.refresh(kpi)
return kpi
def _create_budget(db: Session, kpi_id: int, period: str, value: float, version: str = "v1.0") -> BudgetPlan:
plan = BudgetPlan(
entity_id=1,
kpi_id=kpi_id,
period=period,
budget_value=value,
budget_year=int(period.split("-")[0]),
budget_month=int(period.split("-")[1]),
version=version,
status="active",
)
db.add(plan)
db.commit()
db.refresh(plan)
return plan
def _create_actual(db: Session, kpi_id: int, period: str, value: float) -> KPIValue:
v = KPIValue(kpi_id=kpi_id, period=period, actual_value=value, source_type="manual")
db.add(v)
db.commit()
db.refresh(v)
return v
class TestProformaProfitStatement:
BASE = "/api/cma/reports/proforma/profit-statement"
def _setup(self, db: Session):
"""F_REVENUE: 预算150 / 实际123.45F_OP_CFLOW: 预算20"""
create_test_user(db)
rev = _create_kpi(db, "F_REVENUE", "营业收入", frequency="quarterly", target_value=1200)
op = _create_kpi(db, "F_OP_CFLOW", "经营性现金流")
_create_budget(db, rev.id, "2026-08", 150.0)
_create_budget(db, op.id, "2026-08", 20.0)
_create_actual(db, rev.id, "2026-08", 123.45)
def test_returns_200_and_budget_mapping(self, client: TestClient, db: Session):
self._setup(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?period=2026-08", headers=auth_header(token))
assert resp.status_code == 200
data = resp.json()
assert data["period"] == "2026-08"
assert data["budget_version"] == "v1.0"
# 营业收入(6001) → F_REVENUE:预算150 实际123.45 差异-26.55/-17.7%
rev_line = None
for block in data["blocks"]:
for item in block["items"]:
if item["code"] == "6001":
rev_line = item
assert rev_line is not None, "利润表应含营业收入(6001)行"
assert rev_line["has_budget"] is True
assert rev_line["mapped_kpi_code"] == "F_REVENUE"
assert rev_line["budget_source"] == "budget_plan"
assert rev_line["budget_value"] == 150.0
assert rev_line["actual_value"] == 123.45
assert rev_line["deviation_amount"] == -26.55
assert rev_line["deviation_rate"] == -17.7
def test_no_budget_line_explicit(self, client: TestClient, db: Session):
"""无预算映射的行(如 6402 其他业务成本)显式 has_budget=false"""
self._setup(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?period=2026-08", headers=auth_header(token))
assert resp.status_code == 200
data = resp.json()
line = None
for block in data["blocks"]:
for item in block["items"]:
if item["code"] == "6402":
line = item
assert line is not None
assert line["has_budget"] is False
assert line["budget_source"] == "none"
assert line["budget_value"] is None
def test_deviation_matches_budget_execution(self, client: TestClient, db: Session):
"""同 KPI 同 periodproforma 差异与 budget-execution 一致"""
self._setup(db)
token = get_token_for_user(client)
# budget-execution 里的 F_REVENUE
be = client.get("/api/cma/reports/budget-execution?period=2026-08", headers=auth_header(token))
assert be.status_code == 200
be_item = next(i for i in be.json()["items"] if i["kpi_code"] == "F_REVENUE")
# proforma 利润表 6001 行
pf = client.get(f"{self.BASE}?period=2026-08", headers=auth_header(token))
pf_item = None
for block in pf.json()["blocks"]:
for item in block["items"]:
if item["code"] == "6001":
pf_item = item
assert pf_item is not None
assert pf_item["actual_value"] == be_item["actual_value"]
assert pf_item["budget_value"] == be_item["budget_value"]
assert pf_item["deviation_amount"] == be_item["deviation_amount"]
assert pf_item["deviation_rate"] == be_item["deviation_rate"]
class TestProformaBalanceSheet:
BASE = "/api/cma/reports/proforma/balance-sheet"
def _setup(self, db: Session):
create_test_user(db)
op = _create_kpi(db, "F_OP_CFLOW", "经营性现金流")
ar = _create_kpi(db, "F_AR_DAYS", "应收账款周转天数", frequency="monthly", target_value=5)
_create_budget(db, op.id, "2026-08", 20.0)
_create_budget(db, ar.id, "2026-08", 5.0)
def test_returns_200_and_mappings(self, client: TestClient, db: Session):
self._setup(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?period=2026-08", headers=auth_header(token))
assert resp.status_code == 200
data = resp.json()
assert data["budget_version"] == "v1.0"
lines = {}
for sec in data["sections"]:
for ln in sec["lines"]:
lines[ln["name"]] = ln
# 货币资金 → F_OP_CFLOW(有预算)
assert lines["货币资金"]["has_budget"] is True
assert lines["货币资金"]["mapped_kpi_code"] == "F_OP_CFLOW"
assert lines["货币资金"]["budget_value"] == 20.0
# 应收账款 → F_AR_DAYS(比率型,单独标注,不计算金额差异)
assert lines["应收账款"]["has_budget"] is True
assert lines["应收账款"]["ratio_kpi"] is True
assert lines["应收账款"]["mapped_kpi_code"] == "F_AR_DAYS"
assert lines["应收账款"]["deviation_amount"] is None
assert lines["应收账款"]["note"] is not None
# 无映射行(存货 1405)显式无预算
assert lines["存货"]["has_budget"] is False
assert lines["存货"]["budget_source"] == "none"
class TestProformaCashFlow:
BASE = "/api/cma/reports/proforma/cash-flow"
def _setup(self, db: Session):
create_test_user(db)
rev = _create_kpi(db, "F_REVENUE", "营业收入", frequency="quarterly")
op = _create_kpi(db, "F_OP_CFLOW", "经营性现金流")
_create_budget(db, rev.id, "2026-08", 150.0)
_create_budget(db, op.id, "2026-08", 20.0)
_create_actual(db, rev.id, "2026-08", 123.45)
def test_returns_200_and_mappings(self, client: TestClient, db: Session):
self._setup(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?period=2026-08", headers=auth_header(token))
assert resp.status_code == 200
data = resp.json()
assert data["budget_version"] == "v1.0"
lines = {}
sections = {s["key"]: s for s in data["sections"]}
for sc in data["sections"]:
for ln in sc["lines"]:
lines[ln["code"]] = ln
# CF01 → F_REVENUE(有预算)
assert lines["CF01"]["has_budget"] is True
assert lines["CF01"]["mapped_kpi_code"] == "F_REVENUE"
assert lines["CF01"]["budget_value"] == 150.0
# CF02 无映射 → 显式无预算
assert lines["CF02"]["has_budget"] is False
assert lines["CF02"]["budget_source"] == "none"
# 经营净额 → F_OP_CFLOW 预算
assert sections["operating"]["net_budget"] == 20.0
assert sections["operating"]["has_budget"] is True