wecom-fullstack
0824efc8ed
ci/woodpecker/push/main Pipeline failed
依据：任总 2026-09-29 08:49「P1和P2批」放行 P1-cma-dq-rule-config-approval-20260923 ①③
（派单件 0-P1-project-20260929-qrc-write-entry-and-demo-exclusion-fullstack）
三处尾项（本单）：
1) 前端可编辑入口（补拍板件 ① 未被上轮覆盖的部分）
- frontend/src/views/DataQuality.vue：规则配置 Tab 由「只读」改为
ceo/it 可编辑、finance/business 只读（角色口径与 router/index.ts:45 及
系统设置页一致）；新增「操作/编辑」列（平台级项仅 it 可点、演示账套禁用）；
编辑走 MyDialog（业务弹窗铁律），必填变更理由（>=10 字），取值区间/枚举
由后端返回字段呈现，越界/非法一律如实回显后端 detail，前端不自造校验通过。
2) scope.excluded_entities 接线 + 出厂默认 [3,4,5,6]（拍板件 ③-2「排除」）
- app/dq_rule_config.py：默认由 [] 改为 [3,4,5,6]；新增单一事实源
EXCLUDED_ENTITIES_KEY / EXCLUDED_ENTITIES_FIELD、normalize_entity_ids()、
excluded_entities_from_resolution()、excluded_entity_ids()、
is_excluded_from_business_caliber()。
- app/api/data_quality.py：新增 _excluded_clause()（仅全账套口径 eid=0 生效，
COALESCE(col,1) NOT IN (...) 参数化，NULL 归属不吞行）与 _business_excluded()；
七条规则的明细查询与 _detail_count 全部叠加排除子句；_plan_id_set /
_pending_alert_plan_links 支持 excluded（播报口径同源）。
- 响应新增 business_caliber{caliber,exclusion_applied,excluded_entities,included}
与 scope_excluded_entities；单账套查询保持展示、仅标注不计入经营口径。
- PUT /rule-config：平台级项改写全账套模板行（entity_id=0），审计 detail 与响应
增 target_entity_id；GET 增 scope_note 与 write_contract{roles,platform_roles,
min_reason_len,entity_config_entities,bot_channel_write,endpoint}。
3) 修复两处实现缺陷（本单测试新增用例的 RED 驱动）
- excluded_entities_from_resolution() 取键与 resolve()["scope"] 的归组形态不一致
（配了 scope. 全名去取已归组后的短名字典）⇒ 恒返回空集，接线静默失效；
改为短名优先、全名兼容（EXCLUDED_ENTITIES_FIELD）。
- upsert_param() 全局项 before 取值多一层 rule_key 查找 ⇒ before 恒 None；
改为全局项直接以 param_key 为键。
4) 测试：tests/test_dq_rule_config_20260928.py 3 处既有断言按新口径更新 +
新增 E 类 8 例（排除默认/单一事实源、全账套与单账套两种口径面、A/B 聚合差异
只来自被排除账套、明细计数同源、平台级项改名单即时生效、平台级写落模板行）。
不改：判定表达式本体 / 级别归属 / 评分函数 / 规则集（DAMA §4 锁死 10 类）、
unit_check.amount_threshold（保持 10000）、virtual_pollution 关键词表与来源白名单。
边界：本提交只含 4 文件（backend/app/dq_rule_config.py、
backend/app/api/data_quality.py、backend/tests/test_dq_rule_config_20260928.py、
frontend/src/views/DataQuality.vue）；未裹挟他人在制件
（audit_log.py / migrate_subjects_entity_id_20260923.py / package.json）。
证据：pending-tasks/_evidence/ask5-tail-20260929/
（pytest 全量 1 failed/1516 passed + 失败项归因 A/B、vue-tsc rc=0、
ab-exclusion-pre-write.json 只读 A/B 原始输出、gate dry-run 等）
688 lines
39 KiB
Python
688 lines
39 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""CMA 财务七规则可配置化（ASK-20260928-05）回归套件 —— 规格 / 三层解析 / 引擎接线 / 端点。
|
||
|
||
任务：任总 2026-09-28 13:38「放行 4 和 5」→ 最小实现 = 新表 `quality_rule_config` + 端点 + 只读 UI，
|
||
外加本单唯一的行为变更（去掉 `sync` 污染判定）与「口径用清单默认值」。
|
||
|
||
判据分四组：
|
||
A 规格面：37 规则级参数 + 2 全局项；默认值 == 清单 §7；边界「拒绝 / 钳制」两类；锁死项 400。
|
||
B 引擎接线（**真函数** `_run_governance_checks` + 夹具库，非重写逻辑）：
|
||
· 默认配置 = 原硬编码口径（零行为差异，除 sync）；
|
||
· **sync A/B 三态**：默认(0) / 恢复旧关键词表(命中) / 白名单豁免(0) —— 证明配置真的接线；
|
||
· 阈值、条数下限、开关、扣分上限系数、比较方向、明细条数逐项可配；
|
||
· `/rule-detail` total 与新口径一致性（两处同源，不漂移）。
|
||
C 端点面：GET 只读四角色可见；PUT 仅 ceo/it（平台级仅 it）；Bot 通道禁写；演示账套拒写；
|
||
change_reason 必填；锁死/未知参数 400；越界拒绝或钳制；写行与 `operation_logs` 同事务。
|
||
D 韧性：配置表缺失（DDL 未落地）时读路径回落内置默认、不 500（部署顺序解耦）。
|
||
"""
|
||
import json
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app import dq_rule_config as dqcfg
|
||
from app.api.data_quality import _run_governance_checks
|
||
from app.models import (
|
||
KPIDefinition, KPIValue, KPIAlert, CashPlan, BudgetPlan, QualityRuleConfig,
|
||
OperationLog, Entity,
|
||
)
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
REASON = "回归用例：调整治理规则阈值以验证配置化接线（≥10 字）"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# 夹具
|
||
# ══════════════════════════════════════════════════════════════════
|
||
def _mk_kpi(db: Session, **kw) -> KPIDefinition:
|
||
defaults = {
|
||
"entity_id": 1, "kpi_code": "KPI_X", "kpi_name": "夹具KPI", "status": "active",
|
||
"target_value": 100.0, "target_yearly": 100.0, "formula": "x",
|
||
"data_source": "test", "data_owner": "财务部", "unit": "元",
|
||
}
|
||
defaults.update(kw)
|
||
k = KPIDefinition(**defaults)
|
||
db.add(k)
|
||
db.commit()
|
||
db.refresh(k)
|
||
return k
|
||
|
||
|
||
def _mk_plan(db: Session, **kw) -> CashPlan:
|
||
defaults = {
|
||
"entity_id": 1, "plan_type": "receive", "amount": 10.0, "source": "manual",
|
||
"status": "pending", "plan_date": datetime(2026, 8, 1),
|
||
}
|
||
defaults.update(kw)
|
||
p = CashPlan(**defaults)
|
||
db.add(p)
|
||
db.commit()
|
||
db.refresh(p)
|
||
return p
|
||
|
||
|
||
def _mk_row(db: Session, rule_key: str, param_key: str, value, **kw) -> QualityRuleConfig:
|
||
"""直接落/更新一行配置（模拟「已存在配置」；端点写路径另有用例）。"""
|
||
entity_id = kw.pop("entity_id", 1)
|
||
exist = (db.query(QualityRuleConfig)
|
||
.filter_by(entity_id=entity_id, rule_key=rule_key, param_key=param_key).first())
|
||
if exist is not None:
|
||
exist.param_value = json.dumps(value, ensure_ascii=False)
|
||
exist.value_type = dqcfg.value_type_of(rule_key, param_key)
|
||
db.commit()
|
||
db.refresh(exist)
|
||
return exist
|
||
row = QualityRuleConfig(
|
||
entity_id=entity_id, rule_key=rule_key, param_key=param_key,
|
||
param_value=json.dumps(value, ensure_ascii=False),
|
||
value_type=dqcfg.value_type_of(rule_key, param_key),
|
||
open_level=dqcfg.open_level_of(rule_key, param_key),
|
||
is_default=kw.pop("is_default", False), updated_by=kw.pop("updated_by", "fixture"),
|
||
**kw,
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
db.refresh(row)
|
||
return row
|
||
|
||
|
||
def _counts(db: Session, eid: int = 0):
|
||
"""跑一次真函数，返回 (按规则聚合的 issues, 完整结果)。"""
|
||
r = _run_governance_checks(db, eid)
|
||
return {i["rule"]: i for i in r["issues"]}, r
|
||
|
||
|
||
def _vp_fixture(db: Session) -> None:
|
||
"""规则4 夹具：5 条正常同步来源（酣客形态）+ 2 条测试来源 + 1 条 manual + 1 条博海来源。"""
|
||
for i in range(5):
|
||
_mk_plan(db, id=100 + i, source="zxt-sync", amount=10.0 + i)
|
||
for i in range(2):
|
||
_mk_plan(db, id=200 + i, source="test_import", amount=20.0 + i)
|
||
_mk_plan(db, id=300, source="manual", amount=30.0)
|
||
_mk_plan(db, id=301, source="bohai_ar_20260630", amount=40.0)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# A 规格面
|
||
# ══════════════════════════════════════════════════════════════════
|
||
class TestSpecSurface:
|
||
def test_param_count_is_37_rule_level_plus_2_global(self):
|
||
"""清单 §10.2：37 个规则级参数 + 2 个全局项；且 (rule, param) 无重复。"""
|
||
assert dqcfg.count_rule_params() == 37
|
||
assert len(dqcfg.GLOBAL_SPEC) == 2
|
||
pairs = dqcfg.all_param_pairs()
|
||
assert len(pairs) == 39
|
||
assert len(set(pairs)) == 39
|
||
per_rule = {r: len(dqcfg.SPEC[r]) for r in dqcfg.RULE_KEYS}
|
||
assert per_rule == {"unit_check": 5, "dup_alert": 5, "orphan_check": 4,
|
||
"virtual_pollution": 9, "entity_check": 4,
|
||
"kpi_completeness": 4, "reconciliation": 6}
|
||
|
||
def test_defaults_match_inventory_table(self):
|
||
"""清单 §7 建议默认值逐项（口径「用清单默认值」）；唯一行为变更 = 关键词表无 sync。"""
|
||
vp = dqcfg.SPEC["virtual_pollution"]
|
||
assert vp["suspect_keywords"]["default"] == ["test", "虚拟", "demo"]
|
||
assert "sync" not in vp["suspect_keywords"]["default"]
|
||
assert vp["source_whitelist"]["default"] == [
|
||
"zxt-sync", "bohai_ar", "bohai_ar_20260630", "bohai_prepay_20260630",
|
||
"bohai_ap_20260630", "receivables_migrate"]
|
||
assert vp["manual_source_value"]["default"] == "manual"
|
||
assert vp["manual_cap"]["default"] == 10 and vp["manual_factor"]["default"] == 1
|
||
assert dqcfg.SPEC["unit_check"]["amount_threshold"]["default"] == 10000
|
||
assert dqcfg.SPEC["dup_alert"]["min_count"]["default"] == 2
|
||
assert dqcfg.SPEC["reconciliation"]["diff_threshold_pct"]["default"] == 20
|
||
assert dqcfg.SPEC["reconciliation"]["compare_mode"]["default"] == "abs"
|
||
for rule in dqcfg.RULE_KEYS:
|
||
caps = dqcfg.SPEC[rule]
|
||
assert caps["enabled"]["default"] is True
|
||
assert caps["detail_limit"]["default"] == 10
|
||
assert caps["deduct_factor"]["default"] == (3 if rule in
|
||
("unit_check", "dup_alert", "orphan_check", "virtual_pollution", "entity_check") else 1)
|
||
assert dqcfg.GLOBAL_SPEC["detail_limit"]["default"] == 10
|
||
# F6 排除名单出厂默认 = [3,4,5,6]（2026-09-29 尾项，拍板件 ③-2「排除」；DAMA §6.4）
|
||
assert dqcfg.GLOBAL_SPEC["scope.excluded_entities"]["default"] == [3, 4, 5, 6]
|
||
|
||
def test_locked_items_cover_inventory_ten(self):
|
||
"""清单 §4 十类锁死面完整登记（少一类即视为私自扩面）。"""
|
||
assert len(dqcfg.LOCKED_ITEMS) == 10
|
||
keys = {it["key"] for it in dqcfg.LOCKED_ITEMS}
|
||
assert {"expression", "level", "score_fn", "entity_scope", "rule_set",
|
||
"match_mode", "completeness_caliber", "null_entity_policy",
|
||
"bot_write", "weight_range"} == keys
|
||
|
||
@pytest.mark.parametrize("rule,param,value,expect_ok", [
|
||
("unit_check", "amount_threshold", 5, False), # 越界 → 拒绝
|
||
("unit_check", "amount_threshold", 1000, True),
|
||
("dup_alert", "min_count", 1, False), # 1 无意义 → 拒绝
|
||
("dup_alert", "min_count", 3, True),
|
||
("reconciliation", "diff_threshold_pct", 500, False), # 越界 → 拒绝
|
||
("reconciliation", "compare_mode", "both", False), # 非枚举 → 拒绝
|
||
("unit_check", "enabled", "yes-please", False), # 非布尔 → 拒绝
|
||
("virtual_pollution", "suspect_keywords", ["test", ""], False), # 空元素 → 拒绝
|
||
("virtual_pollution", "suspect_keywords", ["te%st"], False), # 通配符 → 拒绝
|
||
("virtual_pollution", "suspect_keywords", ["a" * 33], False), # 元素超长 → 拒绝
|
||
("virtual_pollution", "manual_source_value", "", False), # 空串 → 拒绝
|
||
])
|
||
def test_validate_reject_cases(self, rule, param, value, expect_ok):
|
||
r = dqcfg.validate_param(rule, param, value)
|
||
assert r["ok"] is expect_ok
|
||
if not expect_ok:
|
||
assert r["error"]
|
||
|
||
def test_validate_clamp_cases(self):
|
||
"""权重/明细条数越界 → 钳制到端点 + 告警（清单 §3 的「钳制 + 告警」）。"""
|
||
r = dqcfg.validate_param("kpi_completeness", "deduct_cap", 999)
|
||
assert r["ok"] and r["value"] == 50 and r["warning"]
|
||
r = dqcfg.validate_param("unit_check", "deduct_factor", -3)
|
||
assert r["ok"] and r["value"] == 0 and r["warning"]
|
||
r = dqcfg.validate_param("unit_check", "detail_limit", 0)
|
||
assert r["ok"] and r["value"] == 1 and r["warning"]
|
||
|
||
def test_locked_and_unknown_params_rejected(self):
|
||
for param in ("level", "expression", "score_fn", "regex", "exempt_kpi_codes"):
|
||
r = dqcfg.validate_param("virtual_pollution", param, "x")
|
||
assert not r["ok"]
|
||
assert "锁死" in r["error"]
|
||
r = dqcfg.validate_param("unit_check", "no_such_param", 1)
|
||
assert not r["ok"] and "未知参数" in r["error"]
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# B 引擎接线（真函数 + 夹具库）
|
||
# ══════════════════════════════════════════════════════════════════
|
||
class TestEngineWiring:
|
||
def test_default_zero_behavior_change_except_sync(self, db):
|
||
"""默认配置 = 原硬编码口径：阈值 10000 / 条数 2 / 差异 20%；sync 来源不再算污染。"""
|
||
_vp_fixture(db)
|
||
_mk_plan(db, id=400, amount=50000.0, source="bohai_ar") # 单位校验命中（>10000）
|
||
k = _mk_kpi(db, kpi_code="K1", target_yearly=100.0)
|
||
db.add(KPIValue(kpi_id=k.id, entity_id=1, period="2026-06", actual_value=50.0))
|
||
for m in range(1, 13):
|
||
db.add(BudgetPlan(kpi_id=k.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"))
|
||
_mk_kpi(db, kpi_code="K2", target_yearly=10.0) # 无值 → KPI完整性
|
||
db.commit()
|
||
|
||
issues, r = _counts(db, 1)
|
||
assert issues["unit_check"]["count"] == 1
|
||
assert issues["virtual_pollution"]["count"] == 2 # 只剩 test_import 两条
|
||
assert issues["virtual_pollution"]["manual_count"] == 1
|
||
assert issues["kpi_completeness"]["count"] == 1
|
||
assert issues["reconciliation"]["count"] == 1
|
||
assert r["config_version"] and len(r["config_version"]) == 12
|
||
assert r["disabled"] == []
|
||
assert r["config_source"]["table_ready"] is True
|
||
assert r["scope_excluded_entities"] == [3, 4, 5, 6]
|
||
assert r["excluded_from_business_caliber"] is False # eid=1 不在名单
|
||
assert r["business_caliber"]["caliber"] == "entity"
|
||
|
||
def test_sync_ab_three_states(self, db):
|
||
"""A/B：默认(0 命中) → 恢复旧关键词表+空白名单(7 命中) → 只留白名单(2 命中)。
|
||
|
||
三态各跑一次**同一个** `_run_governance_checks`（不是重写逻辑），
|
||
证明「配置真的驱动引擎」+「去掉 sync 确实消除了 100% 假阳性」。
|
||
"""
|
||
_vp_fixture(db)
|
||
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["virtual_pollution"]["count"] == 2, "默认：zxt-sync 不再误判"
|
||
|
||
legacy = _mk_row(db, "virtual_pollution", "suspect_keywords",
|
||
["test", "sync", "虚拟", "demo"]) # 改前口径
|
||
_mk_row(db, "virtual_pollution", "source_whitelist", []) # 改前：无白名单
|
||
issues, r = _counts(db, 1)
|
||
assert issues["virtual_pollution"]["count"] == 7, "恢复旧口径 → zxt-sync 5 条重新被误判"
|
||
# 扣分 = min(15, 7×3) + manual 附加 min(10, 1×1) = 16（夹具里有 1 条 manual）
|
||
assert issues["virtual_pollution"]["deducted"] == 16
|
||
|
||
db.query(QualityRuleConfig).filter(QualityRuleConfig.rule_key == "virtual_pollution").delete()
|
||
_mk_row(db, "virtual_pollution", "source_whitelist", ["zxt-sync"])
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["virtual_pollution"]["count"] == 2, "白名单豁免优先于关键词（豁免即不算污染）"
|
||
db.delete(legacy)
|
||
db.commit()
|
||
|
||
def test_threshold_and_switch_and_caps(self, db):
|
||
_mk_plan(db, id=500, amount=150.0)
|
||
_mk_plan(db, id=501, amount=50000.0)
|
||
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["unit_check"]["count"] == 1 # 默认 10000
|
||
|
||
_mk_row(db, "unit_check", "amount_threshold", 100) # 下调阈值 → 2 条
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["unit_check"]["count"] == 2
|
||
|
||
_mk_row(db, "unit_check", "deduct_cap", 0) # 只报不计分
|
||
issues, r = _counts(db, 1)
|
||
assert issues["unit_check"]["count"] == 2 and issues["unit_check"]["deducted"] == 0
|
||
|
||
_mk_row(db, "unit_check", "enabled", False) # 关开关 → 零命中零扣分
|
||
issues, r = _counts(db, 1)
|
||
assert issues["unit_check"]["count"] == 0 and issues["unit_check"]["deducted"] == 0
|
||
assert issues["unit_check"]["enabled"] is False
|
||
assert "unit_check" in r["disabled"] and "unit_check" not in r["passed"]
|
||
assert r["total_rules"] == 7 # 规则集锁死 7 条
|
||
|
||
def test_dup_alert_min_count(self, db):
|
||
k = _mk_kpi(db, kpi_code="K9", target_yearly=10.0)
|
||
_mk_plan(db, id=600, source="manual")
|
||
for _ in range(2): # plan#600 两条 pending
|
||
db.add(KPIAlert(kpi_id=k.id, alert_type="cash_plan", status="pending",
|
||
alert_message="x", suggestion=json.dumps({"plan_id": 600})))
|
||
db.commit()
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["dup_alert"]["count"] == 1 # 默认 ≥2 即重复
|
||
|
||
_mk_row(db, "dup_alert", "min_count", 3)
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["dup_alert"]["count"] == 0 # 阈值抬到 3 → 不再重复
|
||
|
||
def test_reconciliation_threshold_and_direction(self, db):
|
||
k = _mk_kpi(db, kpi_code="K7", target_yearly=100.0)
|
||
for m in range(1, 13):
|
||
db.add(BudgetPlan(kpi_id=k.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")) # 合计 200 → +100%
|
||
k2 = _mk_kpi(db, kpi_code="K8", target_yearly=100.0)
|
||
for m in range(1, 13):
|
||
db.add(BudgetPlan(kpi_id=k2.id, entity_id=1, period=f"2026-{m:02d}",
|
||
budget_value=50.0 / 12, budget_year=2026, budget_month=m,
|
||
version="v1.0", status="active")) # 合计 50 → -50%
|
||
db.commit()
|
||
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["reconciliation"]["count"] == 2 # 默认 abs 双侧 >20%
|
||
|
||
_mk_row(db, "reconciliation", "diff_threshold_pct", 150)
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["reconciliation"]["count"] == 0 # 阈值抬高
|
||
|
||
_mk_row(db, "reconciliation", "diff_threshold_pct", 20)
|
||
_mk_row(db, "reconciliation", "compare_mode", "upper_only")
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["reconciliation"]["count"] == 1 # 只看「超目标」→ 低于目标的 K8 不算
|
||
|
||
_mk_row(db, "reconciliation", "compare_mode", "lower_only")
|
||
issues, _ = _counts(db, 1)
|
||
assert issues["reconciliation"]["count"] == 1 # 只看「低于目标」→ K8
|
||
|
||
def test_detail_limit_configurable(self, db):
|
||
for i in range(4):
|
||
_mk_plan(db, id=700 + i, amount=50000.0)
|
||
issues, _ = _counts(db, 1)
|
||
assert len([d for d in issues["unit_check"]["detail"] if not d.startswith("…")]) == 4
|
||
|
||
_mk_row(db, "unit_check", "detail_limit", 2)
|
||
issues, _ = _counts(db, 1)
|
||
rows = [d for d in issues["unit_check"]["detail"] if not d.startswith("…")]
|
||
assert len(rows) == 2 and any(d.startswith("…") for d in issues["unit_check"]["detail"])
|
||
|
||
def test_entity_config_isolation_no_cross_entity_fallback(self, db):
|
||
"""账套隔离：账套 2 的行不得被账套 1 读到（禁止跨账套回退）。"""
|
||
_mk_plan(db, id=800, entity_id=1, amount=50000.0)
|
||
_mk_plan(db, id=801, entity_id=2, amount=50000.0)
|
||
_mk_row(db, "unit_check", "amount_threshold", 1000000, entity_id=2)
|
||
|
||
issues1, _ = _counts(db, 1)
|
||
issues2, _ = _counts(db, 2)
|
||
assert issues1["unit_check"]["count"] == 1 # 账套1 用默认 10000
|
||
assert issues2["unit_check"]["count"] == 0 # 账套2 用自己那行 1000000
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# C 端点面
|
||
# ══════════════════════════════════════════════════════════════════
|
||
def _ceo_token(db: Session, client: TestClient) -> str:
|
||
create_test_user(db, username="cfgceo", role="ceo")
|
||
return get_token_for_user(client, "cfgceo")
|
||
|
||
|
||
class TestConfigEndpoints:
|
||
def test_get_rule_config_readonly_shape(self, db, client):
|
||
token = _ceo_token(db, client)
|
||
r = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["entity_id"] == 1
|
||
assert body["entity_config_enabled"] is True
|
||
assert body["config_entities"] == [1, 2]
|
||
assert body["rule_param_count"] == 37
|
||
assert set(body["rules"].keys()) == set(dqcfg.RULE_KEYS)
|
||
vp = body["rules"]["virtual_pollution"]["params"]
|
||
assert vp["suspect_keywords"]["value"] == ["test", "虚拟", "demo"]
|
||
assert vp["suspect_keywords"]["source"] == "builtin"
|
||
assert vp["suspect_keywords"]["open_level"] == "entity"
|
||
assert body["global_params"]["detail_limit"]["value"] == 10
|
||
assert body["scope"]["excluded_entities"] == [3, 4, 5, 6] # F6 出厂默认（尾项 2026-09-29）
|
||
assert len(body["locked"]) == 10
|
||
assert body["readonly"] is True
|
||
# 写契约（前端编辑入口的判据来源，与 PUT 的 require_role 同口径）
|
||
wc = body["write_contract"]
|
||
assert wc["roles"] == ["ceo", "it"] and wc["platform_roles"] == ["it"]
|
||
assert wc["min_reason_len"] == dqcfg.MIN_REASON_LEN == 10
|
||
assert wc["bot_channel_write"] is False
|
||
assert wc["entity_config_entities"] == [1, 2]
|
||
assert wc["endpoint"] == "PUT /api/cma/data-quality/rule-config"
|
||
assert "已接线" in body["scope_note"]
|
||
# 前端呈现所需的类型/区间提示（仅提示，非校验通过依据）
|
||
th = body["rules"]["unit_check"]["params"]["amount_threshold"]
|
||
assert th["value_type"] == "int" and th["bounds"] == [100, 1000000]
|
||
assert th["overflow"] == "reject"
|
||
assert body["global_params"]["scope.excluded_entities"]["value"] == [3, 4, 5, 6]
|
||
assert body["global_params"]["scope.excluded_entities"]["open_level"] == "platform"
|
||
|
||
def test_get_rule_config_finance_readonly_allowed(self, db, client):
|
||
create_test_user(db, username="cfgfin", role="finance")
|
||
token = get_token_for_user(client, "cfgfin")
|
||
assert client.get("/api/cma/data-quality/rule-config",
|
||
headers=auth_header(token)).status_code == 200
|
||
|
||
def test_put_requires_ceo_or_it(self, db, client):
|
||
create_test_user(db, username="cfgfin2", role="finance")
|
||
token = get_token_for_user(client, "cfgfin2")
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 5000, "change_reason": REASON})
|
||
assert r.status_code == 403, r.text
|
||
|
||
def test_put_writes_row_and_audit_same_txn(self, db, client):
|
||
token = _ceo_token(db, client)
|
||
before_version = client.get("/api/cma/data-quality/rule-config",
|
||
headers=auth_header(token)).json()["config_version"]
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 5000, "change_reason": REASON})
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["success"] and body["before"] == 10000 and body["after"] == 5000
|
||
assert body["warning"] is None
|
||
|
||
row = db.query(QualityRuleConfig).filter_by(
|
||
entity_id=1, rule_key="unit_check", param_key="amount_threshold").one()
|
||
assert json.loads(row.param_value) == 5000 and row.updated_by == "cfgceo"
|
||
|
||
log = db.query(OperationLog).filter_by(action="update_dq_rule_config").one()
|
||
assert log.user_id is not None # 清单 §5.2：user_id 必写
|
||
assert log.target_type == "dq_rule_config" and log.target_id == row.id
|
||
assert log.detail["entity_id"] == 1 and log.detail["before"] == 10000
|
||
assert log.detail["after"] == 5000 and log.detail["reason"] == REASON
|
||
|
||
# 生效回读：GET 显示来源=账套覆盖；配置指纹随之变化
|
||
g = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token)).json()
|
||
entry = g["rules"]["unit_check"]["params"]["amount_threshold"]
|
||
assert entry["value"] == 5000 and entry["source"] == "entity"
|
||
assert g["config_version"] == body["config_version"] != before_version
|
||
|
||
def test_put_locked_unknown_and_reason_rejected(self, db, client):
|
||
token = _ceo_token(db, client)
|
||
base = {"rule_key": "unit_check", "param_key": "level", "value": "warning",
|
||
"change_reason": REASON}
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), json=base)
|
||
assert r.status_code == 400 and "锁死" in r.json()["detail"]
|
||
|
||
base["param_key"] = "nope"
|
||
assert client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json=base).status_code == 400
|
||
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 5000, "change_reason": "太短"})
|
||
assert r.status_code == 400 and "change_reason" in r.json()["detail"]
|
||
|
||
def test_put_boundary_reject_and_clamp(self, db, client):
|
||
token = _ceo_token(db, client)
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 5, "change_reason": REASON})
|
||
assert r.status_code == 400 and "越界" in r.json()["detail"]
|
||
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "unit_check", "param_key": "deduct_cap",
|
||
"value": 999, "change_reason": REASON})
|
||
assert r.status_code == 200 and r.json()["after"] == 50 and r.json()["warning"]
|
||
|
||
def test_platform_param_requires_it(self, db, client):
|
||
ceo = _ceo_token(db, client)
|
||
payload = {"rule_key": "unit_check", "param_key": "detail_limit", "value": 25,
|
||
"change_reason": REASON}
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(ceo), json=payload)
|
||
assert r.status_code == 403, r.text
|
||
assert "平台级" in r.json()["detail"]
|
||
|
||
create_test_user(db, username="cma", role="it")
|
||
it = get_token_for_user(client, "cma")
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it), json=payload)
|
||
assert r.status_code == 200 and r.json()["after"] == 25
|
||
|
||
def test_bot_channel_write_forbidden(self, db, client):
|
||
token = _ceo_token(db, client)
|
||
headers = {**auth_header(token), "X-BOT-KEY": "dummy-bot-key"}
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=headers,
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 5000, "change_reason": REASON})
|
||
assert r.status_code == 403 and "Bot" in r.json()["detail"]
|
||
|
||
def test_demo_entity_write_rejected(self, db, client):
|
||
"""口径②：演示/测试账套只读、不开配置（写 400，读仍 200）。"""
|
||
db.add(Entity(id=6, name="演示企业", short_name="演示", status="demo"))
|
||
db.commit()
|
||
u = create_test_user(db, username="cfgdemo", role="ceo")
|
||
from app.models import UserEntity
|
||
db.add(UserEntity(user_id=u.id, entity_id=1, granted_by=1))
|
||
db.add(UserEntity(user_id=u.id, entity_id=6, granted_by=1))
|
||
db.commit()
|
||
|
||
# 登录 entity 6（该账号已显式授权 1/6）
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "cfgdemo", "password": "test-fixture-pass", "entity_id": 6})
|
||
assert resp.status_code == 200, resp.text
|
||
tok6 = resp.json()["token"]
|
||
|
||
assert client.get("/api/cma/data-quality/rule-config",
|
||
headers=auth_header(tok6)).status_code == 200
|
||
assert client.get("/api/cma/data-quality/rule-config",
|
||
headers=auth_header(tok6)).json()["entity_config_enabled"] is False
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(tok6),
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 5000, "change_reason": REASON})
|
||
assert r.status_code == 400 and "不开放规则配置" in r.json()["detail"]
|
||
|
||
def test_history_endpoint(self, db, client):
|
||
token = _ceo_token(db, client)
|
||
client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "dup_alert", "param_key": "min_count", "value": 3,
|
||
"change_reason": REASON})
|
||
r = client.get("/api/cma/data-quality/rule-config/history", headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
items = r.json()["items"]
|
||
assert items and items[0]["rule_key"] == "dup_alert"
|
||
assert items[0]["before"] == 2 and items[0]["after"] == 3
|
||
assert items[0]["reason"] == REASON and items[0]["user_id"] is not None
|
||
|
||
def test_rule_detail_total_matches_governance_check(self, db, client):
|
||
"""两处口径同源：改了阈值后 `/rule-detail` total == `governance-check` count。"""
|
||
_mk_plan(db, entity_id=1, id=900, amount=150.0)
|
||
token = _ceo_token(db, client)
|
||
client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "unit_check", "param_key": "amount_threshold",
|
||
"value": 100, "change_reason": REASON})
|
||
gov = client.get("/api/cma/data-quality/governance-check",
|
||
headers=auth_header(token)).json()
|
||
count = {i["rule"]: i["count"] for i in gov["issues"]}["unit_check"]
|
||
detail = client.get("/api/cma/data-quality/rule-detail?rule=unit_check",
|
||
headers=auth_header(token)).json()
|
||
assert count == 1 and detail["total"] == count
|
||
assert detail["items"][0]["key_fields"]["threshold"] == ">100（万元口径可疑）"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# D 韧性：DDL 未落地时不 500
|
||
# ══════════════════════════════════════════════════════════════════
|
||
class TestTableMissingTolerance:
|
||
def test_read_falls_back_to_builtin_when_table_absent(self, db, client):
|
||
from tests import conftest as cf
|
||
|
||
token = _ceo_token(db, client)
|
||
_mk_plan(db, entity_id=1, id=950, amount=50000.0)
|
||
QualityRuleConfig.__table__.drop(bind=cf.TEST_ENGINE)
|
||
try:
|
||
assert dqcfg.table_ready(db) is False
|
||
r = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["table_ready"] is False
|
||
assert r.json()["rules"]["unit_check"]["params"]["amount_threshold"]["value"] == 10000
|
||
|
||
gov = client.get("/api/cma/data-quality/governance-check",
|
||
headers=auth_header(token))
|
||
assert gov.status_code == 200, gov.text
|
||
assert {i["rule"]: i["count"] for i in gov.json()["issues"]}["unit_check"] == 1
|
||
finally:
|
||
QualityRuleConfig.__table__.create(bind=cf.TEST_ENGINE)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# E 经营口径排除（F6 / ASK-05 ③-2 接线，2026-09-29 尾项）
|
||
# 口径：名单只对**全账套口径（eid=0）**生效 —— 被排除账套的行不计入评分/明细；
|
||
# 单账套查询（eid!=0）保持展示，仅由响应字段标注「不计入经营口径」。
|
||
# ══════════════════════════════════════════════════════════════════
|
||
GLOBAL_KEY = dqcfg.GLOBAL_RULE_KEY # "global"
|
||
|
||
|
||
def _gov(db: Session, eid: int):
|
||
"""跑真函数，返回 (各规则 count 字典, 完整结果)。"""
|
||
r = _run_governance_checks(db, eid)
|
||
return {i["rule"]: i["count"] for i in r["issues"]}, r
|
||
|
||
|
||
def _seed_caliber_rows(db: Session) -> None:
|
||
"""账套1（计入经营口径）+ 账套6（演示企业，被排除）各一条 unit_check 命中行。"""
|
||
db.add(Entity(id=6, name="演示企业", short_name="演示", status="demo"))
|
||
db.commit()
|
||
_mk_plan(db, entity_id=1, id=8101, amount=50000.0, source="batch_import")
|
||
_mk_plan(db, entity_id=6, id=8102, amount=80000.0, source="batch_import")
|
||
|
||
|
||
class TestBusinessCaliberExclusion:
|
||
def test_default_is_346_and_single_source(self, db):
|
||
"""出厂默认 [3,4,5,6]；编号规范化；判据函数与解析结果同源。"""
|
||
assert dqcfg.GLOBAL_SPEC["scope.excluded_entities"]["default"] == [3, 4, 5, 6]
|
||
assert dqcfg.excluded_entity_ids(db, 0) == [3, 4, 5, 6]
|
||
assert dqcfg.normalize_entity_ids([6, "5", 5, None, "x", -1, 3, 4]) == [6, 5, 3, 4]
|
||
assert dqcfg.is_excluded_from_business_caliber(6, [3, 4, 5, 6]) is True
|
||
assert dqcfg.is_excluded_from_business_caliber(1, [3, 4, 5, 6]) is False
|
||
assert dqcfg.is_excluded_from_business_caliber(0, [3, 4, 5, 6]) is False # 全账套标识本身不排除
|
||
|
||
def test_caliber_surface_full_scope(self, db):
|
||
"""全账套口径（eid=0）：exclusion_applied=True，excluded_from_business_caliber=False。"""
|
||
_seed_caliber_rows(db)
|
||
counts, r = _gov(db, 0)
|
||
assert r["scope_excluded_entities"] == [3, 4, 5, 6]
|
||
assert r["business_caliber"] == {
|
||
"caliber": "business", "exclusion_applied": True,
|
||
"excluded_entities": [3, 4, 5, 6], "included": True,
|
||
}
|
||
assert r["excluded_from_business_caliber"] is False
|
||
assert counts["unit_check"] == 1 # 演示账套那条（80000）已被排除
|
||
|
||
def test_caliber_surface_single_entity_keeps_display(self, db):
|
||
"""单账套口径（eid=6）：仍可查看自己的结果，仅标注不计入经营口径。"""
|
||
_seed_caliber_rows(db)
|
||
counts, r = _gov(db, 6)
|
||
assert counts["unit_check"] == 1 # 展示保留
|
||
assert r["business_caliber"]["caliber"] == "entity"
|
||
assert r["business_caliber"]["exclusion_applied"] is False
|
||
assert r["business_caliber"]["included"] is False
|
||
assert r["excluded_from_business_caliber"] is True # 标注=不计入
|
||
|
||
def test_ab_aggregate_only_excluded_entity_rows_drop(self, db):
|
||
"""A/B：名单 [] （关）vs [3,4,5,6]（开）—— 差异**只能**来自被排除账套的行。"""
|
||
_seed_caliber_rows(db)
|
||
_mk_row(db, GLOBAL_KEY, dqcfg.EXCLUDED_ENTITIES_KEY, [], entity_id=0)
|
||
ca, ra = _gov(db, 0)
|
||
assert ra["business_caliber"]["exclusion_applied"] is False
|
||
assert ca["unit_check"] == 2
|
||
|
||
_mk_row(db, GLOBAL_KEY, dqcfg.EXCLUDED_ENTITIES_KEY, [3, 4, 5, 6], entity_id=0)
|
||
cb, rb = _gov(db, 0)
|
||
assert rb["business_caliber"]["exclusion_applied"] is True
|
||
assert cb["unit_check"] == 1
|
||
assert ra["score"] < rb["score"] # 排除后扣分减少 ⇒ 分数上升
|
||
for rule in ca: # 其余规则逐项不变（无越界影响）
|
||
if rule != "unit_check":
|
||
assert ca[rule] == cb[rule], rule
|
||
|
||
def test_ab_detail_count_same_source(self, db):
|
||
"""明细总数与评分同源：`_detail_count` 在同名单下的读数与 issues.count 一致。"""
|
||
from app.api.data_quality import _detail_count
|
||
_seed_caliber_rows(db)
|
||
with_exc = _detail_count(db, "unit_check", 0, excluded=[3, 4, 5, 6])
|
||
without = _detail_count(db, "unit_check", 0, excluded=[])
|
||
assert without == 2 and with_exc == 1
|
||
counts, _r = _gov(db, 0)
|
||
assert counts["unit_check"] == with_exc
|
||
# 单账套口径不受名单影响
|
||
assert _detail_count(db, "unit_check", 6, excluded=[3, 4, 5, 6]) == 1
|
||
|
||
def test_platform_write_of_excluded_list_changes_caliber(self, db, client):
|
||
"""平台级项（仅 it）：写向全账套模板行；改名单纯读即改变经营口径（零代码回退）。"""
|
||
_seed_caliber_rows(db)
|
||
create_test_user(db, username="cmait2", role="it")
|
||
it = get_token_for_user(client, "cmait2")
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it),
|
||
json={"rule_key": GLOBAL_KEY, "param_key": dqcfg.EXCLUDED_ENTITIES_KEY,
|
||
"value": [], "change_reason": REASON})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["target_entity_id"] == 0 and r.json()["after"] == []
|
||
row = db.query(QualityRuleConfig).filter_by(
|
||
entity_id=0, rule_key=GLOBAL_KEY, param_key=dqcfg.EXCLUDED_ENTITIES_KEY).one()
|
||
assert row.updated_by == "cmait2"
|
||
counts, res = _gov(db, 0)
|
||
assert res["business_caliber"]["exclusion_applied"] is False and counts["unit_check"] == 2
|
||
# 恢复原型（幂等：同键原位更新，不新增行）
|
||
back = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it),
|
||
json={"rule_key": GLOBAL_KEY, "param_key": dqcfg.EXCLUDED_ENTITIES_KEY,
|
||
"value": [3, 4, 5, 6], "change_reason": REASON})
|
||
assert back.status_code == 200 and back.json()["before"] == []
|
||
counts2, res2 = _gov(db, 0)
|
||
assert res2["business_caliber"]["exclusion_applied"] is True and counts2["unit_check"] == 1
|
||
assert db.query(QualityRuleConfig).filter_by(
|
||
entity_id=0, rule_key=GLOBAL_KEY, param_key=dqcfg.EXCLUDED_ENTITIES_KEY).count() == 1
|
||
|
||
|
||
class TestPlatformWriteTarget:
|
||
"""平台级项落全账套模板行（entity_id=0）——「一处写、全账套生效」。"""
|
||
|
||
def test_it_writes_template_row_and_audit_carries_target(self, db, client):
|
||
create_test_user(db, username="cmait3", role="it")
|
||
it = get_token_for_user(client, "cmait3")
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it),
|
||
json={"rule_key": "unit_check", "param_key": "detail_limit", "value": 25,
|
||
"change_reason": REASON})
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["entity_id"] == 1 and body["target_entity_id"] == 0
|
||
assert "模板行" in body["note"]
|
||
assert db.query(QualityRuleConfig).filter_by(
|
||
entity_id=0, rule_key="unit_check", param_key="detail_limit").count() == 1
|
||
assert db.query(QualityRuleConfig).filter_by(
|
||
entity_id=1, rule_key="unit_check", param_key="detail_limit").count() == 0
|
||
log = db.query(OperationLog).filter_by(action="update_dq_rule_config").one()
|
||
assert log.detail["entity_id"] == 1 and log.detail["target_entity_id"] == 0
|
||
assert log.detail["open_level"] == "platform"
|
||
|
||
def test_entity_level_param_still_writes_own_row(self, db, client):
|
||
"""账套级项不受模板改写影响：仍写本账套行，target_entity_id == 本账套。"""
|
||
token = _ceo_token(db, client)
|
||
r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
|
||
json={"rule_key": "dup_alert", "param_key": "min_count", "value": 3,
|
||
"change_reason": REASON})
|
||
assert r.status_code == 200 and r.json()["target_entity_id"] == 1
|
||
assert db.query(QualityRuleConfig).filter_by(
|
||
entity_id=1, rule_key="dup_alert", param_key="min_count").count() == 1
|
||
assert db.query(QualityRuleConfig).filter_by(
|
||
entity_id=0, rule_key="dup_alert", param_key="min_count").count() == 0
|