# -*- 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