Hermes CI Fix
5b920df8a0
- auto-decompose批量聚合只取年度行(period=YYYY-00)排除月度行，修复重复点击执行分解年底预算滚雪球 - 同一KPI多个version年度行时只取一行(优先请求version)，避免多版本叠加总额虚高 - 新增tests/test_data_quality.py: 财务七规则(governance-check)3用例全过
142 lines
5.5 KiB
Python
142 lines
5.5 KiB
Python
"""DAMA数据治理规则检查（财务七规则）测试 — 2026-08-30
|
||
|
||
覆盖 data_quality.py 的 _run_governance_checks（governance-check 端点核心逻辑）：
|
||
7条规则：unit_check/dup_alert/orphan_check/virtual_pollution/entity_check/kpi_completeness/reconciliation
|
||
+ 评分规则（error 扣 min(15,count*3)，warning 扣 min(10,count*1)，规则4 manual 附加扣）
|
||
|
||
验收活数据对照（生产环境实测）：reconciliation=3、kpi_completeness=53、其余0。
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from app.models import (
|
||
KPIDefinition, KPIValue, KPIAlert, CashPlan, BudgetPlan,
|
||
)
|
||
from app.api.data_quality import _run_governance_checks
|
||
|
||
|
||
def _mk_kpi(db, **kw):
|
||
defaults = {
|
||
"entity_id": 1, "kpi_code": "TEST_001", "kpi_name": "测试KPI",
|
||
"status": "active", "target_value": 100.0, "target_yearly": 100.0,
|
||
"formula": "x", "data_source": "test", "data_owner": "财务部",
|
||
"unit": "元", "kpi_level": "operational",
|
||
}
|
||
defaults.update(kw)
|
||
k = KPIDefinition(**defaults)
|
||
db.add(k)
|
||
db.commit()
|
||
db.refresh(k)
|
||
return k
|
||
|
||
|
||
def _mk_plan(db, **kw):
|
||
defaults = {
|
||
"entity_id": 1, "plan_type": "receive", "amount": 10.0,
|
||
"plan_date": datetime(2026, 8, 1), "source": "manual", "status": "pending",
|
||
}
|
||
defaults.update(kw)
|
||
p = CashPlan(**defaults)
|
||
db.add(p)
|
||
db.commit()
|
||
db.refresh(p)
|
||
return p
|
||
|
||
|
||
def _mk_alert(db, kpi_id, plan_id, **kw):
|
||
defaults = {
|
||
"kpi_id": kpi_id, "alert_type": "cash_plan", "status": "pending",
|
||
"alert_message": "应收预警",
|
||
"suggestion": json.dumps({"plan_id": plan_id}),
|
||
}
|
||
defaults.update(kw)
|
||
a = KPIAlert(**defaults)
|
||
db.add(a)
|
||
db.commit()
|
||
db.refresh(a)
|
||
return a
|
||
|
||
|
||
class TestGovernanceSevenRules:
|
||
def test_clean_db_all_pass(self, db):
|
||
"""空库：7条规则全部通过，score=100"""
|
||
r = _run_governance_checks(db, 0)
|
||
assert r["total_rules"] == 7
|
||
assert r["score"] == 100
|
||
assert r["total_deduct"] == 0
|
||
assert len(r["passed"]) == 7
|
||
for item in r["issues"]:
|
||
assert item["count"] == 0
|
||
|
||
def test_all_rules_hit(self, db):
|
||
"""构造数据触发全部7条规则"""
|
||
# KPI-1: active + 有值 + 年度目标100 但预算月度合计200（勾稽差异100%）
|
||
k1 = _mk_kpi(db, kpi_code="KPI_001", kpi_name="勾稽KPI", target_yearly=100.0)
|
||
db.add(KPIValue(kpi_id=k1.id, entity_id=1, period="2026-06", actual_value=50.0))
|
||
for m in range(1, 13):
|
||
db.add(BudgetPlan(kpi_id=k1.id, entity_id=1, period=f"2026-{m:02d}",
|
||
budget_value=200.0 / 12, budget_year=2026, budget_month=m,
|
||
version="v1.0", status="active"))
|
||
db.commit()
|
||
|
||
# KPI-2: active 无任何值（KPI完整性命中）
|
||
_mk_kpi(db, kpi_code="KPI_002", kpi_name="无值KPI", target_yearly=10.0)
|
||
|
||
# KPI-3: 值实体=2 ≠ 定义实体=1（实体归属命中）
|
||
k3 = _mk_kpi(db, kpi_code="KPI_003", kpi_name="实体错乱KPI", target_yearly=10.0)
|
||
db.add(KPIValue(kpi_id=k3.id, entity_id=2, period="2026-06", actual_value=5.0))
|
||
db.commit()
|
||
|
||
# 单位校验：amount=50000 > 10000（单位错乱命中）
|
||
_mk_plan(db, id=1, amount=50000.0)
|
||
|
||
# 虚拟污染：source=test_import（error）+ source=manual（待人工确认）
|
||
_mk_plan(db, id=2, amount=100.0, source="test_import")
|
||
_mk_plan(db, id=3, amount=100.0, source="manual")
|
||
|
||
# 孤儿预警：plan_id=999 不存在（孤儿命中）
|
||
_mk_alert(db, k1.id, 999)
|
||
|
||
# 重复预警：plan_id=5 两条 pending（重复命中）
|
||
_mk_plan(db, id=5, amount=100.0)
|
||
_mk_alert(db, k1.id, 5)
|
||
_mk_alert(db, k1.id, 5)
|
||
|
||
r = _run_governance_checks(db, 0)
|
||
by_rule = {i["rule"]: i for i in r["issues"]}
|
||
|
||
assert by_rule["unit_check"]["count"] == 1
|
||
assert by_rule["dup_alert"]["count"] == 1
|
||
assert by_rule["orphan_check"]["count"] == 1
|
||
assert by_rule["virtual_pollution"]["count"] == 1
|
||
assert by_rule["virtual_pollution"]["manual_count"] == 3 # plan#1/#3/#5 默认manual
|
||
assert by_rule["entity_check"]["count"] == 1
|
||
assert by_rule["kpi_completeness"]["count"] == 1
|
||
assert by_rule["reconciliation"]["count"] == 1
|
||
|
||
# 评分：5条error × min(15,3)=3 → 15；2条warning × 1 → 2；manual附加 min(10,3)=3 → 总扣20
|
||
assert r["total_deduct"] == 20
|
||
assert r["score"] == 80
|
||
assert len(r["passed"]) == 0
|
||
|
||
def test_entity_scoped(self, db):
|
||
"""entity_id 限定：只检查该实体数据"""
|
||
k1 = _mk_kpi(db, kpi_code="KPI_001", kpi_name="实体1KPI", entity_id=1, target_yearly=10.0)
|
||
k2 = _mk_kpi(db, kpi_code="KPI_002", kpi_name="实体2KPI", entity_id=2, target_yearly=10.0)
|
||
db.add(KPIValue(kpi_id=k2.id, entity_id=2, period="2026-06", actual_value=5.0))
|
||
db.commit()
|
||
|
||
r1 = _run_governance_checks(db, 1)
|
||
comp1 = {i["rule"]: i["count"] for i in r1["issues"]}
|
||
assert comp1["kpi_completeness"] == 1 # 实体1下KPI_001无值
|
||
assert comp1["entity_check"] == 0 # 实体1下无实体错乱
|
||
|
||
r2 = _run_governance_checks(db, 2)
|
||
comp2 = {i["rule"]: i["count"] for i in r2["issues"]}
|
||
assert comp2["kpi_completeness"] == 0 # KPI_002有值
|
||
assert comp2["entity_check"] == 0 # 值实体=2与定义实体=2一致
|
||
assert comp2["reconciliation"] == 0 # 无预算行不参与勾稽
|