fix(cash): select cash-balance sync targets by explicit kpi_code whitelist

The voucher-import linkage branch picked write targets by kpi_name LIKE '%cash%' and wrote the monetary-fund ledger balance (yuan) into derived KPIs (F_OP_CFLOW / F_FCF / F_BEP_REVENUE_CASH / F_BEP_COVER_CASH). Replace with an explicit kpi_code whitelist plus a unit/caliber gate that refuses writes on mismatch and reports them via kpi_sync_skipped. Read-only equivalence probe (real function + intercepted writes + rollback) before/after: before wrote 4 derived KPIs, after writes only F_CASH_SAFETY. tests: +9 new cases; pytest failure set identical to baseline (15/15 same ids).
2026-09-14 20:37:12 +08:00
parent 8693c8e342
commit 1f7be6b5ae
3 changed files with 292 additions and 18 deletions
@@ -0,0 +1,215 @@
+"""cash.py `_sync_cash_kpis` ① 分支白名单化 + 单位口径校验 单测 — 2026-09-14
+
+任务: fullstack-20260914200226（`_sync_cash_kpis` ① 分支按名称模糊匹配选取 KPI → 口径错配）
+
+缺陷复现（生产 entity 1 只读实测）：原判据
+    kpi_name LIKE '%现金%' OR kpi_name LIKE '%货币资金%' AND status='active'
+命中 4 个派生指标 —— F_OP_CFLOW(经营性现金流) / F_FCF(自由现金流) /
+F_BEP_REVENUE_CASH(现金口径保本销售额) / F_BEP_COVER_CASH(现金口径保本覆盖率)，
+会把「货币资金科目期末余额（元）」写进它们（覆盖率还是 % 单位）且置 data_status='verified'。
+
+本测试锁定修复后的判据（**只认显式白名单 kpi_code**）与单位口径闸门：
+  1. 名称含「现金/货币资金」但未登记白名单的 KPI（含派生/保本/比率类）→ 零写入；
+  2. 白名单命中且 unit 与写入口径一致 → 按登记科目选择器写对值（1001 / 1002 / 合计）；
+  3. unit 与写入口径不符 → 拒绝写入 + WARNING 告警 + 进 skipped；
+  4. status != active / 跨账套（entity_id 隔离）→ 不写；
+  5. 同 (kpi_id, period, source_type=ledger) 幂等：二次调用更新同一行不新增；
+  6. 白名单科目选择器笔误 → 硬报错（不静默退化成合计口径）；
+  7. ② 分支 F_CASH_SAFETY 判据同样只用 kpi_code（名称兜底已移除）+ 单位闸门。
+"""
+import datetime
+import logging
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import pytest  # noqa: E402
+
+from app.models import KPIValue, VoucherDetail  # noqa: E402
+from app.api import cash as CASH  # noqa: E402
+from tests.conftest import create_test_kpi  # noqa: E402
+
+D = datetime.datetime(2026, 8, 31)
+PERIOD = "2026-08"
+
+
+def _add_voucher(db, no, code, debit=0.0, credit=0.0, entity_id=1, period=PERIOD):
+    db.add(VoucherDetail(entity_id=entity_id, voucher_no=str(no), voucher_date=D,
+                         subject_code=code, subject_name="测试科目",
+                         debit_amount=debit, credit_amount=credit,
+                         summary="白名单单测", carry_forward=0, period=period))
+    db.flush()
+
+
+def _rows(db, kpi_id):
+    return db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id).all()
+
+
+def _codes(items):
+    return {i["kpi_code"] for i in items}
+
+
+# ── 1. 派生/保本/比率类：名称命中旧 LIKE 判据也不得被写入 ──────────────────────
+def test_derived_cash_named_kpis_not_written(db):
+    """名称含「现金」的派生指标（旧判据的命中集）→ 零写入（白名单未登记即不写）。"""
+    derived = [
+        create_test_kpi(db, kpi_code="T_F_OP_CFLOW", kpi_name="经营性现金流", unit="万元"),
+        create_test_kpi(db, kpi_code="T_F_FCF", kpi_name="自由现金流(FCF)", unit="万元"),
+        create_test_kpi(db, kpi_code="T_F_BEP_REVENUE_CASH", kpi_name="现金口径保本销售额", unit="万元"),
+        create_test_kpi(db, kpi_code="T_F_BEP_COVER_CASH", kpi_name="现金口径保本覆盖率", unit="%"),
+        create_test_kpi(db, kpi_code="T_EXT_CASH_MISC", kpi_name="货币资金合计(衍生)", unit="万元"),
+    ]
+    _add_voucher(db, 88001, "1001", debit=300.0)
+    _add_voucher(db, 88002, "1002", debit=700.0)
+
+    updates, skipped = CASH._sync_cash_kpis(db, 1, "batch-whitelist-neg", [PERIOD])
+
+    hit = _codes(updates) & {k.kpi_code for k in derived}
+    assert hit == set(), f"派生/保本类被联动写入: {hit}"
+    for k in derived:
+        assert _rows(db, k.id) == [], f"{k.kpi_code} 不应有 kpi_values 行"
+    # 白名单本身不含这些派生指标
+    assert not (_codes(derived) if False else {"T_F_OP_CFLOW"} & set(CASH.CASH_SYNC_KPI_WHITELIST))
+    assert skipped == [], "未被白名单命中 ≠ 拒写项（skipped 只记命中白名单但口径不符的）"
+
+
+# ── 2. 白名单命中：按登记科目选择器写对值 ──────────────────────────────────
+def test_whitelisted_kpis_written_with_registered_selector(db):
+    """EXT_069(库存现金→1001) / EXT_089(银行存款→1002) 分别写各自科目净额。"""
+    k_1001 = create_test_kpi(db, kpi_code="EXT_069", kpi_name="库存现金", unit="元")
+    k_1002 = create_test_kpi(db, kpi_code="EXT_089", kpi_name="银行存款", unit="元")
+    _add_voucher(db, 88101, "1001", debit=1234.5)
+    _add_voucher(db, 88102, "100201", debit=7654.25)
+    _add_voucher(db, 88103, "100201", credit=1000.0)
+
+    updates, skipped = CASH._sync_cash_kpis(db, 1, "batch-wl-pos", [PERIOD])
+
+    got = {u["kpi_code"]: u["value"] for u in updates}
+    assert got["EXT_069"] == 1234.5, got
+    assert got["EXT_089"] == 7654.25 - 1000.0, got        # 1002 科目净额（借-贷）
+    assert skipped == [], skipped
+
+    rows_1001 = _rows(db, k_1001.id)
+    rows_1002 = _rows(db, k_1002.id)
+    assert len(rows_1001) == 1 and rows_1001[0].actual_value == 1234.5
+    assert len(rows_1002) == 1 and rows_1002[0].actual_value == 6654.25
+    for r in rows_1001 + rows_1002:
+        assert r.source_type == "ledger" and r.period == PERIOD
+        assert r.source_batch == "batch-wl-pos"
+
+
+# ── 3. 单位口径不符 → 拒写 + 告警（负例）────────────────────────────────
+def test_unit_mismatch_rejected_and_warned(db, caplog):
+    """unit=万元 的「库存现金」→ 元口径不得写入（防万元/百分比误写元值）。"""
+    k_bad = create_test_kpi(db, kpi_code="EXT_069", kpi_name="库存现金", unit="万元")
+    k_pct = create_test_kpi(db, kpi_code="EXT_089", kpi_name="银行存款", unit="%")
+    _add_voucher(db, 88201, "1001", debit=50000.0)
+    _add_voucher(db, 88202, "1002", debit=20000.0)
+
+    with caplog.at_level(logging.WARNING, logger="cma.cash"):
+        updates, skipped = CASH._sync_cash_kpis(db, 1, "batch-unit-neg", [PERIOD])
+
+    assert "EXT_069" not in _codes(updates) and "EXT_089" not in _codes(updates)
+    assert _rows(db, k_bad.id) == [] and _rows(db, k_pct.id) == []
+    assert _codes(skipped) == {"EXT_069", "EXT_089"}, skipped
+    for s in skipped:
+        assert "unit_mismatch" in s["reason"] and s["reason"]
+    msgs = [r.getMessage() for r in caplog.records]
+    assert any("拒绝联动写KPI EXT_069" in m and "unit_mismatch" in m for m in msgs), msgs
+    assert any("拒绝联动写KPI EXT_089" in m for m in msgs), msgs
+
+
+# ── 4. 非 active / 跨账套 → 不写 ────────────────────────────────────────
+def test_inactive_and_cross_entity_not_written(db):
+    """archived 的 EXT_069 不写；entity 2 的 EXT_069 不被 entity 1 的调用写。"""
+    k_arch = create_test_kpi(db, kpi_code="EXT_069", kpi_name="库存现金", unit="元",
+                             status="archived")
+    k_e2 = create_test_kpi(db, kpi_code="EXT_089", kpi_name="银行存款", unit="元", entity_id=2)
+    _add_voucher(db, 88301, "1001", debit=9999.0)
+    _add_voucher(db, 88302, "1002", debit=8888.0)
+
+    updates, skipped = CASH._sync_cash_kpis(db, 1, "batch-status-neg", [PERIOD])
+
+    assert _codes(updates) & {"EXT_069", "EXT_089"} == set(), updates
+    assert _rows(db, k_arch.id) == [] and _rows(db, k_e2.id) == []
+    assert skipped == [], skipped
+
+
+# ── 5. 幂等：同批次重复调用不新增行 ─────────────────────────────────────
+def test_idempotent_same_period_single_row(db):
+    k = create_test_kpi(db, kpi_code="EXT_069", kpi_name="库存现金", unit="元")
+    _add_voucher(db, 88401, "1001", debit=100.0)
+
+    CASH._sync_cash_kpis(db, 1, "batch-idem-1", [PERIOD])
+    db.flush()
+    CASH._sync_cash_kpis(db, 1, "batch-idem-2", [PERIOD])
+    db.flush()
+
+    rows = [r for r in _rows(db, k.id) if r.source_type == "ledger"]
+    assert len(rows) == 1, rows
+    assert rows[0].actual_value == 100.0 and rows[0].source_batch == "batch-idem-2"
+
+
+# ── 6. 科目选择器笔误 → 硬报错（不静默退化成合计口径）──────────────────────
+def test_invalid_selector_raises(db, monkeypatch):
+    monkeypatch.setitem(CASH.CASH_SYNC_KPI_WHITELIST, "EXT_069", ("9999", "元"))
+    create_test_kpi(db, kpi_code="EXT_069", kpi_name="库存现金", unit="元")
+    _add_voucher(db, 88501, "1001", debit=1.0)
+    with pytest.raises(ValueError):
+        CASH._sync_cash_kpis(db, 1, "batch-selector-neg", [PERIOD])
+
+
+def test_selector_none_means_total_balance(db):
+    """选择器 None = 货币资金合计（1001+1002）——演示「合计口径」的登记形态。"""
+    monkeypatch_free = CASH.CASH_SYNC_KPI_WHITELIST
+    CASH.CASH_SYNC_KPI_WHITELIST = dict(monkeypatch_free, **{"EXT_089": (None, "元")})
+    try:
+        k = create_test_kpi(db, kpi_code="EXT_089", kpi_name="银行存款", unit="元",
+                            data_source="测试-合计口径")
+        _add_voucher(db, 88601, "1001", debit=10.0)
+        _add_voucher(db, 88602, "1002", debit=25.0)
+        updates, _ = CASH._sync_cash_kpis(db, 1, "batch-total", [PERIOD])
+        got = {u["kpi_code"]: u["value"] for u in updates}
+        assert got["EXT_089"] == 35.0, got
+        assert _rows(db, k.id)[0].actual_value == 35.0
+    finally:
+        CASH.CASH_SYNC_KPI_WHITELIST = monkeypatch_free
+
+
+# ── 8. 判据源级断言：①/② 选取不得再出现名称模糊匹配 ─────────────────────
+def test_selection_has_no_fuzzy_name_matching():
+    """源码级锁定：`_sync_cash_kpis` 内不得出现 kpi_name.like / LIKE '%现金%' 之类判据。"""
+    import inspect
+    src = inspect.getsource(CASH._sync_cash_kpis)
+    assert "kpi_name.like" not in src, "选取判据出现名称模糊匹配"
+    assert ".like(" not in src, src
+    assert "KPIDefinition.kpi_code.in_(list(CASH_SYNC_KPI_WHITELIST))" in src
+    assert 'KPIDefinition.kpi_code == "F_CASH_SAFETY"' in src
+
+
+# ── 7. ② 分支：判据只用 kpi_code + 单位闸门 ─────────────────────────────
+def test_safety_kpi_selected_by_code_only_and_unit_guarded(db, caplog):
+    """名称含「安全垫」但 code != F_CASH_SAFETY 的 KPI 不得被 ② 写入；
+    F_CASH_SAFETY 单位非万元 → 拒写 + 告警。"""
+    impostor = create_test_kpi(db, kpi_code="T_SAFETY_LOOKALIKE", kpi_name="现金安全垫(旧)",
+                               unit="万元")
+    _add_voucher(db, 88701, "1001", debit=10000.0)
+
+    # (a) 无 F_CASH_SAFETY → ② 创建自身行，impostor 不参与
+    updates, skipped = CASH._sync_cash_kpis(db, 1, "batch-safety-a", [PERIOD])
+    assert "T_SAFETY_LOOKALIKE" not in _codes(updates)
+    assert _rows(db, impostor.id) == []
+    safety = [u for u in updates if u["kpi_code"] == "F_CASH_SAFETY"]
+    assert len(safety) == 1 and safety[0]["value"] == 1.0, updates   # 10000元 = 1.0万元
+
+    # (b) F_CASH_SAFETY 单位被改成 % → 拒写 + 告警
+    from app.models import KPIDefinition
+    db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_CASH_SAFETY").update(
+        {"unit": "%"}, synchronize_session=False)
+    db.flush()
+    with caplog.at_level(logging.WARNING, logger="cma.cash"):
+        updates2, skipped2 = CASH._sync_cash_kpis(db, 1, "batch-safety-b", [PERIOD])
+    assert "F_CASH_SAFETY" not in _codes(updates2)
+    assert [s["kpi_code"] for s in skipped2] == ["F_CASH_SAFETY"], skipped2
+    assert any("拒绝联动写KPI F_CASH_SAFETY" in r.getMessage() for r in caplog.records)