"""资产负债表取数层重分类（R1~R12，调表不调账）测试 — 2026-09-16 规则来源：《CMA 资产负债表重分类规则表 v1》（任总 2026-09-16 批准） 被测：`app/services/report_reclass.py` + `/api/cma/reports/balance-sheet*` 端点 覆盖： T1 R1~R10 触发与转正（含互为镜像的 R3/R4、多规则叠加） T2 模板未列报目标项（R7/R8 应收/应付票据）→ 虚拟行列报 T3 不变量：Δ资产=Δ负债、权益不变、勾稽差额不变、无负数（规则范围内） T4 幂等：同参数重复分析结果一致（不写库） T5 多租户隔离：entity1/entity2 独立，不串账套 T6 取数口径：basis=end（期末余额，累计）vs basis=period（本期发生额） T7 无凭证明细 → 示例数据路径不受影响（不报错、不重分类） T8 R11/R12 与模板待核项如实暴露（blocked / template_gap） T9 规则范围外的反方向余额（货币资金/存货）如实标记为未覆盖，不误报"已解决" T10 底稿导出（JSON + Excel）可用 """ from datetime import date import pytest 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 VoucherDetail, Subject from app.services import report_reclass as rr BASE = "/api/cma/reports" PERIOD = "2026-06" PREV = "2026-05" _n = {"v": 0} def _v(db: Session, entity_id: int, period: str, lines, carry_forward: int = 0): """插入一张凭证的若干行（lines: [(subject_code, debit, credit)]）""" _n["v"] += 1 y, m = period.split("-") for code, d, c in lines: db.add(VoucherDetail( entity_id=entity_id, voucher_no=f"T{_n['v']:05d}", voucher_date=date(int(y), int(m), 15), subject_code=code, debit_amount=d, credit_amount=c, period=period, summary="test", carry_forward=carry_forward, )) db.commit() def seed_reverse_items(db: Session, entity_id: int = 1, period: str = PERIOD): """构造含 R1~R10 反向余额的**平衡**账套（每张凭证借贷相等）。 期望触发（原始余额 NSB<0）： R2 2203 预收账款 借方 100,000 → 应收账款 R3 2202 应付账款 借方 30,000 → 预付账款 R4 1123 预付账款 贷方 50,000 → 应付账款 R5 1221 其他应收款 贷方 12,000 → 其他应付款 R6 2241 其他应付款 借方 7,000 → 其他应收款 R9 2221 应交税费 借方 25,000 → 其他流动资产 R10 2211 应付职工薪酬 借方 9,000 → 其他应收款 """ _v(db, entity_id, period, [("1002", 1_000_000, 0), ("4001", 0, 1_000_000)]) # 实收资本 _v(db, entity_id, period, [("1002", 400_000, 0), ("2203", 0, 400_000)]) # 预收 _v(db, entity_id, period, [("2203", 500_000, 0), ("1002", 0, 500_000)]) # 预收反方向 _v(db, entity_id, period, [("1123", 80_000, 0), ("1002", 0, 80_000)]) # 预付 _v(db, entity_id, period, [("1002", 130_000, 0), ("1123", 0, 130_000)]) # 预付反方向 _v(db, entity_id, period, [("2202", 30_000, 0), ("1002", 0, 30_000)]) # 应付借方 _v(db, entity_id, period, [("1002", 12_000, 0), ("1221", 0, 12_000)]) # 其他应收反方向 _v(db, entity_id, period, [("2241", 7_000, 0), ("1002", 0, 7_000)]) # 其他应付借方 _v(db, entity_id, period, [("2221", 25_000, 0), ("1002", 0, 25_000)]) # 应交税费留抵 _v(db, entity_id, period, [("2211", 9_000, 0), ("1002", 0, 9_000)]) # 应付职工薪酬多付 # 规则未覆盖的反方向余额（货币资金）：库存现金贷方 1,500,000 → 货币资金为负 _v(db, entity_id, period, [("1122", 1_500_000, 0), ("1001", 0, 1_500_000)]) def seed_receivable_side(db: Session, entity_id: int = 1, period: str = PREV): """R1 + R7/R8（模板未列报的票据类）""" _v(db, entity_id, period, [("1002", 300_000, 0), ("1122", 0, 300_000)]) # 回款超应收 _v(db, entity_id, period, [("1002", 60_000, 0), ("1121", 0, 60_000)]) # 应收票据反方向 _v(db, entity_id, period, [("2201", 20_000, 0), ("1002", 0, 20_000)]) # 应付票据借方 def _analyze(db, period=PERIOD, entity_id=1, basis="end"): from app.api.reports import BALANCE_SHEET_SECTIONS return rr.analyze(db, entity_id, period, basis=basis, template_sections=BALANCE_SHEET_SECTIONS) class TestReclassEngine: def test_t1_triggers_and_reversal(self, db: Session): seed_reverse_items(db) p = _analyze(db) got = {e["rule"]: e for e in p["entries"]} assert set(got) == {"R2", "R3", "R4", "R5", "R6", "R9", "R10"}, f"实际触发: {sorted(got)}" assert got["R2"]["reclass_amount"] == pytest.approx(100_000.0) assert got["R2"]["target_item"] == "应收账款" and got["R2"]["target_in_template"] is True assert got["R3"]["reclass_amount"] == pytest.approx(30_000.0) assert got["R4"]["reclass_amount"] == pytest.approx(50_000.0) # 互为镜像的 R3/R4 各自按**原始余额**触发，不互相影响，两侧都被转正 assert p["after"]["2202"] == pytest.approx(50_000.0) assert p["after"]["1123"] == pytest.approx(30_000.0) assert p["after"]["2203"] == pytest.approx(0.0) assert p["after"]["1221"] == pytest.approx(16_000.0) # R5 转出 + R6/R10 转入 assert p["after"]["2241"] == pytest.approx(12_000.0) assert p["after"]["OCA_OTHER"] == pytest.approx(25_000.0) # R9 → 其他流动资产 assert p["after"]["2211"] == pytest.approx(0.0) # 全部转正额 = 重分类总额 assert sum(e["reclass_amount"] for e in p["entries"]) == pytest.approx(233_000.0) def test_t2_template_gap_virtual_lines(self, db: Session): seed_receivable_side(db) p = _analyze(db, period=PREV) got = {e["rule"]: e for e in p["entries"]} assert set(got) == {"R1", "R7", "R8"} assert got["R1"]["target_item"] == "预收账款" # 小企业准则无「合同负债」 assert got["R7"]["target_item"] == "应付票据" assert got["R7"]["target_in_template"] is False # 模板未列报 → 虚拟行 assert p["after"]["1121"] == pytest.approx(20_000.0) assert p["after"]["2201"] == pytest.approx(60_000.0) assert p["after"]["1122"] == pytest.approx(0.0) assert p["after"]["2203"] == pytest.approx(300_000.0) assert any(g["rule"] == "R7" and g["status"] == "template_gap" for g in p["template_gaps"]) def test_t3_invariants(self, db: Session): seed_reverse_items(db) p = _analyze(db) inv = p["invariants"] assert inv["delta_assets_equals_delta_liabilities"] is True assert inv["assets_delta"] == pytest.approx(233_000.0) assert inv["liabilities_delta"] == pytest.approx(233_000.0) assert inv["equity_unchanged"] is True assert inv["residual_unchanged"] is True assert inv["identity_holds_before"] is True and inv["identity_holds_after"] is True # 跨方向重分类必然改变总资产 —— 不隐藏这一点 assert inv["total_assets_unchanged"] is False assert inv["reclass_total"] == pytest.approx(233_000.0) # 规则范围内的负数项目重分类后清零 assert inv["no_negative_in_scope_after"] is True assert p["negatives"]["after_in_scope"] == [] # 规则未覆盖的货币资金负数如实标记（不误报已解决） out = [n for n in p["negatives"]["after"] if not n["in_reclass_scope"]] assert any(n["name"] == "货币资金" for n in out) def test_t4_idempotent(self, db: Session): seed_reverse_items(db) a = _analyze(db) b = _analyze(db) assert [e["rule"] for e in a["entries"]] == [e["rule"] for e in b["entries"]] assert a["after"] == b["after"] assert a["invariants"]["reclass_total"] == b["invariants"]["reclass_total"] def test_t5_entity_isolation(self, db: Session): seed_reverse_items(db, entity_id=1) _v(db, 2, PERIOD, [("1002", 5_000, 0), ("4001", 0, 5_000)]) p1 = _analyze(db, entity_id=1) p2 = _analyze(db, entity_id=2) assert p1["invariants"]["entries_count"] == 7 assert p2["invariants"]["entries_count"] == 0 # entity2 无反方向余额 assert p2["after"]["2202"] == pytest.approx(0.0) assert p2["invariants"]["assets_after"] == pytest.approx(5_000.0) # entity3 完全无数据 → 不重分类、不报错 p3 = _analyze(db, entity_id=3) assert p3["has_voucher_data"] is False and p3["entries"] == [] def test_t6_basis_semantics(self, db: Session): # 2026-05 预收贷方 400,000（正常）；2026-06 预收借方 500,000 _v(db, 1, PREV, [("1002", 400_000, 0), ("2203", 0, 400_000)]) _v(db, 1, PERIOD, [("2203", 500_000, 0), ("1002", 0, 500_000)]) end = _analyze(db, period=PERIOD, basis="end") # 期末余额 = 累计 → -100,000 per = _analyze(db, period=PERIOD, basis="period") # 本期发生额 → -500,000 assert end["basis"] == "end" and per["basis"] == "period" assert end["raw"]["2203"] == pytest.approx(-100_000.0) assert per["raw"]["2203"] == pytest.approx(-500_000.0) # 两个口径各自自洽（都转为正数），但转正额不同 → 报表列报必须用期末余额口径 assert [e["rule"] for e in end["entries"]] == ["R2"] assert [e["rule"] for e in per["entries"]] == ["R2"] assert end["entries"][0]["reclass_amount"] == pytest.approx(100_000.0) assert per["entries"][0]["reclass_amount"] == pytest.approx(500_000.0) assert per["after"]["2203"] == pytest.approx(0.0) def test_t8_blocked_rules_exposed(self, db: Session): seed_reverse_items(db) p = _analyze(db) gaps = {g["rule"]: g for g in p["template_gaps"]} assert gaps["R11"]["status"] == "blocked" and gaps["R12"]["status"] == "blocked" assert gaps["R9"]["status"] == "template_gap" # 阻塞规则（R11/R12）绝不落地 triggered = {e["rule"] for e in p["entries"]} assert "R11" not in triggered and "R12" not in triggered # 模板未列报的目标项（R7/R8/R9）落地时在报表内按虚拟行列报 for rule in ("R7", "R8", "R9"): assert gaps[rule]["in_report"] is True def test_t10_working_paper_rows(self, db: Session): seed_reverse_items(db) p = _analyze(db) rows = rr.entry_rows(p) assert len(rows) == len(p["entries"]) == 7 for row in rows: assert row["处理"] == "调表不调账" and row["转正额(元)"] > 0 assert row["来源科目"] and row["目标报表项目"] diag = {r["项目"]: r for r in rr.diagnostic_rows(p)} assert diag["Δ资产 = Δ负债"]["值"] == "PASS" assert diag["权益不变"]["值"] == "PASS" assert diag["规则范围内负数项目(重分类后)"]["值"] == 0 class TestReclassEndpoints: def _auth(self, client, db): create_test_user(db) return auth_header(get_token_for_user(client)) def test_t7_no_voucher_data_demo_path(self, client: TestClient, db: Session): h = self._auth(client, db) resp = client.get(f"{BASE}/balance-sheet?period={PERIOD}", headers=h) assert resp.status_code == 200 data = resp.json() assert len(data["sections"]) == 5 assert data["reclass"]["entries"] == [] assert data["reclass"]["applied"] is False assert data["all_items_have_data"] is False assert data["totals"]["balanced"] is True def test_t7b_reclass_applied_on_real_data(self, client: TestClient, db: Session): seed_reverse_items(db) h = self._auth(client, db) data = client.get(f"{BASE}/balance-sheet?period={PERIOD}&basis=end", headers=h).json() assert data["basis"] == "end" assert data["reclass"]["applied"] is True assert data["reclass"]["invariants"]["entries_count"] == 7 assert data["reclass"]["invariants"]["no_negative_in_scope_after"] is True # 有凭证明细 → 不再混入示例数据（展示口径与落库口径同源） assert all(not ln["is_demo"] for sec in data["sections"] for ln in sec["lines"]) # 明细行带原值/调整额，可追溯 lines = {ln["item_key"]: ln for sec in data["sections"] for ln in sec["lines"]} assert lines["2203"]["raw_value"] == pytest.approx(-100_000.0) assert lines["2203"]["reclass_adjustment"] == pytest.approx(100_000.0) assert lines["2203"]["end_value"] == pytest.approx(0.0) # 不变量同步暴露给前端/底稿 inv = data["reclass"]["invariants"] assert inv["delta_assets_equals_delta_liabilities"] and inv["equity_unchanged"] assert data["reclass"]["working_paper_url"] def test_t2b_virtual_line_rendered_in_section(self, client: TestClient, db: Session): seed_receivable_side(db) create_test_user(db) h = auth_header(get_token_for_user(client)) data = client.get(f"{BASE}/balance-sheet?period={PREV}", headers=h).json() ca = next(s for s in data["sections"] if s["key"] == "current_assets") cl = next(s for s in data["sections"] if s["key"] == "current_liabilities") assert any(ln["item_key"] == "1121" and ln["is_reclass_line"] for ln in ca["lines"]) assert any(ln["item_key"] == "2201" and ln["is_reclass_line"] for ln in cl["lines"]) def test_t10b_working_paper_endpoint(self, client: TestClient, db: Session): seed_reverse_items(db) create_test_user(db) h = auth_header(get_token_for_user(client)) wp = client.get(f"{BASE}/balance-sheet/reclass-working-paper?period={PERIOD}", headers=h) assert wp.status_code == 200 body = wp.json() assert body["accounting_method"].startswith("调表不调账") assert body["rule_version"] == rr.RULE_VERSION assert len(body["entries"]) == 7 assert body["invariants"]["reclass_total"] == pytest.approx(233_000.0) assert len(body["rules"]) == 12 # R1~R12 全部登记 xl = client.get(f"{BASE}/balance-sheet/reclass-working-paper/export?period={PERIOD}", headers=h) assert xl.status_code == 200 assert "spreadsheetml" in xl.headers["content-type"] assert len(xl.content) > 4000 # 非空 Excel def test_t5b_entity_isolation_endpoint(self, client: TestClient, db: Session): seed_reverse_items(db, entity_id=1) _v(db, 2, PERIOD, [("1002", 5_000, 0), ("4001", 0, 5_000)]) create_test_user(db) h = auth_header(get_token_for_user(client)) # 登录账套 = entity1 d1 = client.get(f"{BASE}/balance-sheet?period={PERIOD}", headers=h).json() assert d1["reclass"]["invariants"]["entries_count"] == 7 # 显式声明另一账套 → 403（多租户边界） assert client.get(f"{BASE}/balance-sheet?period={PERIOD}&entity_id=2", headers=h).status_code == 403 # 无凭证明细的账套（切到 2 需另一 token，这里用内部调用验证引擎层隔离） p2 = _analyze(db, entity_id=2) assert p2["invariants"]["entries_count"] == 0 class TestUnmappedDiagnosis: def test_unmapped_subjects_and_gap_explained(self, db: Session): """模板未列报科目 → 出现在诊断里，且勾稽差额与之一致""" _v(db, 1, PERIOD, [("1002", 1_000, 0), ("4001", 0, 1_000)]) _v(db, 1, PERIOD, [("1801", 200, 0), ("1002", 0, 200)]) # 长期待摊费用（新补的行，不再未列报） _v(db, 1, PERIOD, [("1901", 300, 0), ("1002", 0, 300)]) # 待处理财产损溢（模板未列报） db.add(Subject(subject_code="1901", subject_name="待处理财产损溢", category="asset")) db.add(Subject(subject_code="190102", subject_name="待处理流动资产损溢", category="detail")) db.commit() p = _analyze(db) codes = [u["subject_code"] for u in p["unmapped_subjects"]] assert "1901" in codes unmapped_sum = sum(u["normal_side_balance"] for u in p["unmapped_subjects"]) assert p["invariants"]["residual_after"] == pytest.approx(-unmapped_sum)