"""T06-a 方案维度解析测试：AC-29 / AC-44。 维度是 Skills/Rules 的元数据，不新增"维度卡"（14 §5）。本文件锁定三件事： - 同 ID 不同定义必须拒绝合并（不让模型或程序临时挑一个） - 相同定义只出现一次，引用被合并而不是重复计数 - 呈现判定覆盖 0 / 2 / 6 / 9 维与异刻度，且**不丢失任何维度** """ import tempfile import unittest from pathlib import Path from services import assistant_store as store from services import component_registry as registry from services import review_plan as plan SCALE = "evidence-stage" LEVELS = [ {"value": 0, "label": "未提供", "criterion": "材料未提供"}, {"value": 1, "label": "仅有说明", "criterion": "仅客户经理口头或文字说明"}, {"value": 2, "label": "材料待核", "criterion": "提供了材料但未交叉核对"}, {"value": 3, "label": "交叉核验", "criterion": "两个独立来源一致"}, {"value": 4, "label": "人工复核", "criterion": "评审员已人工核实"}, ] def spec(dimension_id="d1", label="回款依据", order=0, scale_id=SCALE, scale_version="v1", levels=None, required=False, refs=None): return {"id": dimension_id, "label": label, "order": order, "scale_id": scale_id, "scale_version": scale_version, "levels": levels or LEVELS, "component_refs": refs or [], "required": required} def many(count, **over): return [spec(f"d{index}", f"维度{index}", index, **over) for index in range(count)] class DeduplicationTests(unittest.TestCase): def test_identical_definitions_merge_into_one(self): # AC-29：相同定义只出现一次 merged = plan.resolve_dimensions([ spec("d1", refs=[{"id": "skill.a", "version": "1"}]), spec("d1", refs=[{"id": "skill.b", "version": "1"}]), ]) self.assertEqual(len(merged), 1) self.assertEqual(sorted(ref["id"] for ref in merged[0]["component_refs"]), ["skill.a", "skill.b"]) def test_same_id_different_label_is_a_conflict(self): with self.assertRaises(ValueError) as ctx: plan.resolve_dimensions([spec("d1"), spec("d1", label="另一个名字")]) self.assertIn("不一致定义", str(ctx.exception)) self.assertIn("不得由程序或模型临时选择其一", str(ctx.exception)) def test_same_id_different_scale_id_is_a_conflict(self): with self.assertRaises(ValueError): plan.resolve_dimensions([spec("d1"), spec("d1", scale_id="other-scale")]) def test_same_scale_id_different_version_is_a_conflict(self): # 刻度改版后语义可能变化，必须视为不同定义 with self.assertRaises(ValueError): plan.resolve_dimensions([spec("d1"), spec("d1", scale_version="v2")]) def test_same_id_different_levels_is_a_conflict(self): changed = [dict(level) for level in LEVELS] changed[3] = {"value": 3, "label": "两来源一致", "criterion": "改了判据"} with self.assertRaises(ValueError): plan.resolve_dimensions([spec("d1"), spec("d1", levels=changed)]) def test_required_flag_is_unioned_on_merge(self): merged = plan.resolve_dimensions([spec("d1"), spec("d1", required=True)]) self.assertTrue(merged[0]["required"]) def test_order_is_stable_and_sorted(self): merged = plan.resolve_dimensions([spec("db", order=5), spec("da", order=1)]) self.assertEqual([item["id"] for item in merged], ["da", "db"]) def test_invalid_dimension_is_rejected(self): broken = spec() broken["levels"] = [{"value": 0, "label": "只有一个层级", "criterion": "c"}] with self.assertRaises(ValueError) as ctx: plan.resolve_dimensions([broken]) self.assertIn("维度不合法", str(ctx.exception)) class PresentationTests(unittest.TestCase): def check(self, dimensions, hard=()): result = plan.presentation(dimensions, hard) plan.assert_nothing_dropped(result, dimensions) # 不丢项是每次都要过的硬断言 return result def test_empty_plan_says_so_without_placeholder_axes(self): # UI Spec：空维度显示"尚未配置"，不补六轴占位 result = self.check([]) self.assertEqual(result["layout"], "empty") self.assertIn("尚未配置核查维度", result["reason"]) def test_two_dimensions_use_list_not_radar(self): result = self.check(many(2)) self.assertEqual(result["layout"], "list") self.assertEqual(result["radar_ids"], []) def test_six_dimensions_use_radar_plus_equivalent_list(self): result = self.check(many(6)) self.assertEqual(result["layout"], "radar") self.assertEqual(len(result["radar_ids"]), 6) self.assertIn("等价文字列表", result["reason"]) def test_nine_dimensions_are_grouped_not_radar(self): result = self.check(many(9)) self.assertEqual(result["layout"], "grouped") self.assertEqual(result["radar_ids"], []) self.assertEqual(len(result["listed_ids"]), 9, "全部维度仍可查看") def test_three_and_eight_are_the_radar_boundaries(self): self.assertEqual(self.check(many(3))["layout"], "radar") self.assertEqual(self.check(many(8))["layout"], "radar") def test_mixed_scales_force_faceted_display(self): # 异刻度不强行归一化 dimensions = [spec("d1"), spec("d2", "负债率", 1, scale_id="ratio-scale")] result = self.check(dimensions) self.assertEqual(result["layout"], "faceted") self.assertEqual(result["radar_ids"], []) def test_hard_blocks_are_listed_separately_and_never_in_the_radar(self): # 14 §5：硬阻断单列，不被平均数或面积覆盖 dimensions = many(6) result = self.check(dimensions, hard={"d0"}) self.assertEqual(result["hard_block_ids"], ["d0"]) self.assertNotIn("d0", result["radar_ids"]) self.assertEqual(len(result["radar_ids"]), 5) self.assertEqual(result["layout"], "radar") def test_hard_blocks_are_separated_even_when_faceted(self): dimensions = [spec("d1"), spec("d2", "负债率", 1, scale_id="ratio-scale")] result = self.check(dimensions, hard={"d1"}) self.assertEqual(result["hard_block_ids"], ["d1"]) self.assertIn("d2", result["listed_ids"]) def test_dropping_a_dimension_would_be_caught(self): # 守卫本身必须真的会报错，否则"不丢项"只是口号 with self.assertRaises(ValueError): plan.assert_nothing_dropped( {"radar_ids": ["d0"], "listed_ids": [], "hard_block_ids": []}, many(2)) def test_unknown_scale_versions_never_share_a_radar(self): dimensions = [spec("d1"), spec("d2", "同刻度不同版本", 1, scale_version="v2")] self.assertEqual(self.check(dimensions)["layout"], "faceted") class PlanAssemblyTests(unittest.TestCase): """AC-28 / E-19：只有全部依赖有效且同域才装配；否则拒绝并指明是哪一项。""" def setUp(self): self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.db_path = Path(self._tmp.name) / "plan.sqlite3" store.initialize(self.db_path) with store.transaction(self.db_path) as db: registry.seed(db) # 内置演示组件，含 corporate.default 方案 def tearDown(self): self._tmp.cleanup() def assemble(self, plan_id="corporate.default"): with store.transaction(self.db_path) as db: return plan.assemble(db, plan_id) def test_published_plan_assembles_and_freezes(self): frozen = self.assemble() self.assertEqual(frozen["plan"]["id"], "corporate.default") self.assertEqual({s["kind"] for s in frozen["skills"]}, {"skill"}) self.assertEqual({r["kind"] for r in frozen["rules"]}, {"rule_set"}) self.assertTrue(frozen["content_hash"]) # 原生方法文本在装配时即固化，运行中改动无法影响本次 model_skills = [s for s in frozen["skills"] if s.get("native_skills")] self.assertTrue(model_skills) self.assertTrue(all(s["methods"] for s in model_skills)) def test_unpublished_plan_is_rejected_with_a_clear_reason(self): with self.assertRaises(ValueError) as ctx: self.assemble("corporate.does-not-exist") self.assertIn("评审方案不可用", str(ctx.exception)) def test_draft_dependency_is_named_in_the_error(self): # 造一个未发布的 skill 并被方案引用 → 必须指出是哪一项 with store.transaction(self.db_path) as db: plan_body = registry.resolve(db, "corporate.default") skill_id = plan_body["skills"][0] # 移除该 skill 的发布指针，模拟"草稿/已撤回" db.execute("DELETE FROM assistant_objects WHERE kind='release' AND id=?", (skill_id,)) with self.assertRaises(ValueError) as ctx: self.assemble() message = str(ctx.exception) self.assertIn("不可用依赖", message) self.assertIn(skill_id, message, "必须指明是哪个依赖不可用") self.assertIn("尚未发布或已撤回", message) def test_cross_domain_dependency_is_rejected(self): # 演示与正式组件不能混用（freeze 的既有保证） with store.transaction(self.db_path) as db: production = dict(registry.resolve(db, "corporate.analysis"), scope="production") production.pop("revision", None) store.put(db, "component_version", production, immutable=True) store.put(db, "release", {"id": "corporate.analysis", "version_id": production["id"], "scope": "production"}) body = registry.resolve(db, "corporate.default") body["skills"] = ["corporate.analysis"] + [s for s in body["skills"] if s != "corporate.analysis"] with self.assertRaises(ValueError): with store.transaction(self.db_path) as db: plan.check_dependencies(db, body) registry.freeze(db, "corporate.default") def test_frozen_hash_changes_when_a_dependency_version_changes(self): first = self.assemble() with store.transaction(self.db_path) as db: upgraded = dict(registry.resolve(db, "corporate.cashflow"), version="9.9.9", name="现金流完整性（改版）") upgraded.pop("revision", None) store.put(db, "component_version", upgraded, immutable=True) store.put(db, "release", {"id": "corporate.cashflow", "version_id": upgraded["id"], "scope": upgraded["scope"]}) second = self.assemble() self.assertNotEqual(first["content_hash"], second["content_hash"], "换版本后装配结果必须变化，证明版本确实被冻结进结果") def test_assembled_dimensions_are_resolved_and_presented(self): with store.transaction(self.db_path) as db: body = dict(registry.resolve(db, "corporate.default"), dimensions=[spec("d1"), spec("d2", "负债率", 1), spec("d3", "设备", 2)]) body.pop("revision", None) store.put(db, "component_version", body, immutable=True) store.put(db, "release", {"id": "corporate.default", "version_id": body["id"], "scope": body["scope"]}) frozen = plan.assemble(db, "corporate.default") self.assertEqual(len(frozen["dimensions"]), 3) self.assertEqual(frozen["presentation"]["layout"], "radar") def test_conflicting_dimensions_are_rejected_at_assembly(self): with store.transaction(self.db_path) as db: body = dict(registry.resolve(db, "corporate.default"), dimensions=[spec("d1"), spec("d1", label="同名不同义")]) body.pop("revision", None) store.put(db, "component_version", body, immutable=True) store.put(db, "release", {"id": "corporate.default", "version_id": body["id"], "scope": body["scope"]}) with self.assertRaises(ValueError) as ctx: plan.assemble(db, "corporate.default") self.assertIn("不一致定义", str(ctx.exception)) class PlanGovernanceTests(unittest.TestCase): """T06-c：方案治理复用既有入口；不得出现业务 Tools 卡（ADR-11）。""" def setUp(self): self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.db_path = Path(self._tmp.name) / "governance.sqlite3" store.initialize(self.db_path) with store.transaction(self.db_path) as db: registry.seed(db) def tearDown(self): self._tmp.cleanup() def plan_body(self, **over): body = {"id": "corporate.custom", "kind": "plan", "name": "自定义方案", "version": "1.0.0", "scope": "demo", "skills": ["corporate.cashflow"], "rules": ["demo.coverage"], "template": "template.ai-pre-review", "dimensions": [spec("d1", order=0), spec("d2", "负债率", 1), spec("d3", "设备", 2)]} body.update(over) return body def propose(self, body=None, actor="demo-manager"): with store.transaction(self.db_path) as db: return plan.propose_plan(db, body or self.plan_body(), actor, "DEMO-A") def test_plan_proposal_creates_a_draft_only(self): draft = self.propose() self.assertEqual(draft["status"], "draft") with store.transaction(self.db_path) as db: published = [r for r in store.listing(db, "component_version") if r.get("component_id") == "corporate.custom"] self.assertEqual(published, [], "提议方案不得直接生效") def test_business_tools_card_is_rejected(self): # ADR-11：不开放业务 Tools 卡，方案不得自带工具配置面 for key in ("tools", "tool_cards", "allowed_tools", "permissions"): with self.subTest(key=key): with self.assertRaises(ValueError) as ctx: self.propose(self.plan_body(**{key: ["terminal"]})) self.assertIn("不开放业务 Tools 卡", str(ctx.exception)) def test_dimensions_are_deduplicated_at_proposal(self): draft = self.propose(self.plan_body(dimensions=[spec("d1"), spec("d1")])) with store.transaction(self.db_path) as db: row = store.get(db, "component_draft", draft["id"]) self.assertEqual(len(row["dimensions"]), 1, "相同定义只应保留一份") def test_conflicting_dimensions_are_rejected_at_proposal(self): with self.assertRaises(ValueError) as ctx: self.propose(self.plan_body(dimensions=[spec("d1"), spec("d1", scale_version="v2")])) self.assertIn("不一致定义", str(ctx.exception)) def test_self_approval_is_rejected_by_existing_governance(self): draft = self.propose() with store.transaction(self.db_path) as db: with self.assertRaises(PermissionError) as ctx: registry.release(db, draft["id"], "demo-manager") self.assertIn("独立审核人", str(ctx.exception)) def test_only_plans_can_be_proposed_through_this_entry(self): with self.assertRaises(ValueError): self.propose(self.plan_body(kind="skill")) class CriticalCheckTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.db_path = Path(self._tmp.name) / "checks.sqlite3" store.initialize(self.db_path) with store.transaction(self.db_path) as db: registry.seed(db) self.frozen = plan.assemble(db, "corporate.default") def tearDown(self): self._tmp.cleanup() def test_real_codes_are_accepted(self): codes = sorted(plan._plan_codes(self.frozen)) self.assertTrue(codes) plan.validate_critical_checks(self.frozen, codes[:1]) def test_invented_critical_check_is_rejected(self): with self.assertRaises(ValueError) as ctx: plan.validate_critical_checks(self.frozen, ["not.a.real.code"]) self.assertIn("不属于本方案", str(ctx.exception)) def test_empty_critical_list_is_allowed(self): plan.validate_critical_checks(self.frozen, []) if __name__ == "__main__": unittest.main()