qiaoqichaoandClaude Code
1bc17bb9bc
T06 第三块，收尾。方案提案复用既有组件治理（registry.draft/release）， 不新建发布路径；自批拒绝等既有保证有测试锁定。 - propose_plan()：维度在提案时即去重与冲突拦截（T06-a 的解析前移）， 保证草稿阶段就挡住"同 ID 不同定义" - **不得出现业务 Tools 卡**（ADR-11）：方案自带 tools/tool_cards/ allowed_tools/permissions 一律拒绝，否则业务侧可绕过工具白名单； 工具只能由后台治理 - validate_critical_checks()：关键核查项必须真由本方案的 Skill/规则产生， 凭空声明的"关键项"会让 BR-06 的硬性阻断语义失去可追溯性 T06 三块完成，卡片验收对照：AC-28（装配与依赖拒绝）· AC-29（维度去重与刻度 冲突）· AC-44（0/2/6/9 维与异刻度的呈现判定）均有测试；"没有 Tools 卡" 有守卫。 测试：test_review_plan.py 增至 34 项。全量 349 项 OK（340 → 349）； verify.py 10/10 PASS。 Co-Authored-By: Claude Code <noreply@anthropic.com>
348 lines
16 KiB
Python
348 lines
16 KiB
Python
"""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()
|