feat(v2): T05-b 知识卡片与硬规则依据门禁
T05 第二步:候选与治理。设计明令"复用卡片治理、不新建第二套发布库",
故本模块**只做投影**,状态机与发布完全委托 component_registry。
新增 contracts:Origin / GovernanceStatus / RuleEffect / Condition /
RuleContent / KnowledgeCard。Condition 只允许字段+有限算子,**不执行表达式**。
RuleContent 在契约层就要求 hard 必须带 policy_ref 或 authorization_ref 且附依据证据。
新增 services/knowledge_cards.py:
- propose():投影字段(origin/谱系/范围/条件)后委托 registry.draft(),**只产草稿**
- publish():发布时**再查一次** hard 门禁——草稿可能在提议后被改,不能只信提议时的校验
- effective():按数据域/有效期/范围过滤已发布版本(BR-04)
既有治理的保证原样生效,有测试锁定:自批被 release() 拒绝、无评测不能发布、
已发布需新版本号、提议不产生已发布版本。
三处必须点明的边界与映射:
1. **同键不同义**:本设计的 data_domain ↔ 注册表的 scope(demo/production),
而本设计的 scope 是 project/team。投影层显式互换并存为 card_scope,绝不混用;
调用方私传 data_domain 会被拒
2. **条件形状不同**:注册表条件是 {code,field,operator,threshold,title,suggestion},
本设计是 {field,operator,value}。做**翻译**而非强加;算子只映射注册表实际
会产出的 5 个(gt/lt/eq/gte/lte),出现其它取值即报错而非当作已支持
3. **内置种子组件是全局的**,没有 project_id/edited_by,**不是项目知识卡片**。
投影时明确报错,避免把内置 Skill 误当项目卡片
测试:新增 test_knowledge_cards.py 20 项(门禁 6 + 来源谱系 6 + 治理复用 5 + 有效过滤 3)。
全量 310 项 OK(290 → 310);verify.py 10/10 PASS。
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,224 @@
|
||||
"""T05-b 知识卡片治理测试:AC-03 / AC-25 / AC-27 / AC-43。
|
||||
|
||||
重点证明两件事:
|
||||
1. **没有第二套发布库**:发布路径完全走既有 component_registry,因此自批拒绝、
|
||||
评测门禁、已发布不可覆盖这些既有保证对新卡片同样生效。
|
||||
2. **硬规则有正式依据门槛**:只有历史频次的候选不得发布为 hard(E-18)。
|
||||
"""
|
||||
|
||||
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 knowledge_cards as cards
|
||||
|
||||
|
||||
def rule_body(**over):
|
||||
body = {
|
||||
"id": "corporate.test_rule", "kind": "rule_set", "name": "测试规则",
|
||||
"scope": "demo", "version": "1.0.0",
|
||||
# 注册表规则条件的形状:operator ∈ (gt,lt,eq,gte,lte) 且 threshold 可解析为数值,
|
||||
# title/suggestion 是展示字段(不进设计契约)
|
||||
"conditions": [{"field": "metrics.coverage", "operator": "lt", "threshold": "1.2",
|
||||
"title": "覆盖不足", "suggestion": "请补充担保措施"}],
|
||||
"effect": "advisory", "on_match": "flag",
|
||||
}
|
||||
body.update(over)
|
||||
return body
|
||||
|
||||
|
||||
class KnowledgeCardTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
||||
self.db_path = Path(self._tmp.name) / "cards.sqlite3"
|
||||
store.initialize(self.db_path)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def propose(self, body=None, origin="manual", **kwargs):
|
||||
with store.transaction(self.db_path) as db:
|
||||
return cards.propose(db, body or rule_body(), "demo-manager", "DEMO-A",
|
||||
origin=origin, **kwargs)
|
||||
|
||||
def publish(self, draft_id, actor, preview=False):
|
||||
with store.transaction(self.db_path) as db:
|
||||
return cards.publish(db, draft_id, actor, preview)
|
||||
|
||||
|
||||
class HardRuleGateTests(KnowledgeCardTestCase):
|
||||
def test_frequency_only_candidate_cannot_be_hard(self):
|
||||
# AC-27:只有历史频次,没有正式依据 → 拒绝 hard
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.propose(rule_body(effect="hard"))
|
||||
self.assertIn("正式制度或授权依据", str(ctx.exception))
|
||||
self.assertIn("仅可保存为建议草稿", str(ctx.exception))
|
||||
|
||||
def test_frequency_only_candidate_can_stay_advisory(self):
|
||||
draft = self.propose(rule_body(effect="advisory"))
|
||||
self.assertEqual(draft["status"], "draft")
|
||||
self.assertEqual(draft["effect"], "advisory")
|
||||
|
||||
def test_hard_with_policy_reference_is_accepted(self):
|
||||
draft = self.propose(rule_body(
|
||||
effect="hard", on_match="block",
|
||||
policy_ref={"id": "policy.credit-2025", "version": "v3"},
|
||||
basis_evidence_ids=["ev-1"]))
|
||||
self.assertEqual(draft["effect"], "hard")
|
||||
|
||||
def test_hard_requires_basis_evidence_too(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.propose(rule_body(effect="hard",
|
||||
policy_ref={"id": "p", "version": "v1"}))
|
||||
self.assertIn("依据证据", str(ctx.exception))
|
||||
|
||||
def test_gate_is_rechecked_at_publish_time(self):
|
||||
# 草稿在提议后被改成 hard(模拟旁路),发布时必须再查一次
|
||||
draft = self.propose(rule_body(effect="advisory"))
|
||||
with store.transaction(self.db_path) as db:
|
||||
row = store.get(db, "component_draft", draft["id"])
|
||||
row["effect"] = "hard"
|
||||
store.put(db, "component_draft", row)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.publish(draft["id"], "demo-reviewer")
|
||||
self.assertIn("正式制度或授权依据", str(ctx.exception))
|
||||
|
||||
def test_unknown_effect_is_rejected(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self.propose(rule_body(effect="mandatory"))
|
||||
|
||||
def test_conditions_are_validated(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self.propose(rule_body(conditions=[{"field": "f", "operator": "present"}]))
|
||||
|
||||
|
||||
class OriginAndLineageTests(KnowledgeCardTestCase):
|
||||
def test_history_derived_requires_sources(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.propose(rule_body(), origin="history_derived")
|
||||
self.assertIn("必须保留来源", str(ctx.exception))
|
||||
|
||||
def test_history_derived_keeps_lineage_after_manual_edit(self):
|
||||
# AC-25:历史提炼卡被人工编辑后仍是历史来源,并保留父版本
|
||||
draft = self.propose(rule_body(id="corporate.from_history"), origin="history_derived",
|
||||
source_event_ids=["event-1"])
|
||||
edited = self.propose(rule_body(id="corporate.from_history", version="1.0.1"),
|
||||
origin="history_derived", source_event_ids=["event-1"],
|
||||
parent_version={"id": "corporate.from_history", "version": "1.0.0"})
|
||||
with store.transaction(self.db_path) as db:
|
||||
card = cards.draft_card(db, edited["id"])
|
||||
self.assertEqual(card["origin"], "history_derived")
|
||||
self.assertEqual(card["parent_version"], {"id": "corporate.from_history", "version": "1.0.0"})
|
||||
self.assertEqual(card["source_event_ids"], ["event-1"])
|
||||
self.assertNotEqual(draft["id"], edited["id"])
|
||||
|
||||
def test_manual_card_has_manual_origin_and_no_history_sources(self):
|
||||
with store.transaction(self.db_path) as db:
|
||||
card = cards.draft_card(db, self.propose(rule_body(id="corporate.by_hand"))["id"])
|
||||
self.assertEqual(card["origin"], "manual")
|
||||
self.assertEqual(card["source_event_ids"], [])
|
||||
|
||||
def test_data_domain_maps_to_registry_scope_not_card_scope(self):
|
||||
# 两个 scope 语义不同:data_domain 落注册表 scope,卡片范围落 card_scope
|
||||
with store.transaction(self.db_path) as db:
|
||||
card = cards.draft_card(db, self.propose(
|
||||
rule_body(id="corporate.scoped", scope="production"),
|
||||
card_scope="team", team_id="team-1")["id"])
|
||||
self.assertEqual(card["data_domain"], "production")
|
||||
self.assertEqual(card["card_scope"], "team")
|
||||
self.assertEqual(card["team_id"], "team-1")
|
||||
|
||||
def test_team_scope_requires_team_id(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self.propose(rule_body(), card_scope="team")
|
||||
|
||||
def test_callers_cannot_smuggle_data_domain_key(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.propose(rule_body(data_domain="production"))
|
||||
self.assertIn("不要另传 data_domain", str(ctx.exception))
|
||||
|
||||
|
||||
class GovernanceReuseTests(KnowledgeCardTestCase):
|
||||
def test_self_approval_is_rejected_by_existing_governance(self):
|
||||
# AC-03 / AC-43:提议人不得自批——沿用既有 release() 的守卫
|
||||
draft = self.propose()
|
||||
with self.assertRaises(PermissionError) as ctx:
|
||||
self.publish(draft["id"], "demo-manager")
|
||||
self.assertIn("独立审核人", str(ctx.exception))
|
||||
|
||||
def test_publish_still_requires_a_passing_evaluation(self):
|
||||
# 既有评测门禁对新卡片同样生效:没有评测就不能发布
|
||||
draft = self.propose()
|
||||
with store.transaction(self.db_path) as db:
|
||||
with self.assertRaises(store.Conflict) as ctx:
|
||||
cards.publish(db, draft["id"], "demo-reviewer")
|
||||
self.assertIn("评测", str(ctx.exception))
|
||||
|
||||
def test_reproposing_the_same_version_updates_the_same_draft(self):
|
||||
# 草稿可反复修订:同 id@version 仍是同一草稿,不会产生第二份
|
||||
first = self.propose(rule_body(id="corporate.versioned"))
|
||||
second = self.propose(rule_body(id="corporate.versioned", name="改过的名字"))
|
||||
self.assertEqual(first["id"], second["id"])
|
||||
self.assertEqual(second["name"], "改过的名字")
|
||||
with store.transaction(self.db_path) as db:
|
||||
self.assertEqual(len([r for r in store.listing(db, "component_draft")
|
||||
if r.get("component_id") == "corporate.versioned"]), 1)
|
||||
|
||||
def test_published_version_cannot_be_redrafted_at_the_same_version(self):
|
||||
# AC-43:已发布编辑必须用新版本号,否则被既有治理拒绝
|
||||
self.propose(rule_body(id="corporate.versioned"))
|
||||
with store.transaction(self.db_path) as db:
|
||||
draft = store.get(db, "component_draft", "corporate.versioned@1.0.0")
|
||||
row = dict(draft, status="published", approved_by="demo-reviewer",
|
||||
component_id="corporate.versioned")
|
||||
row.pop("revision", None)
|
||||
store.put(db, "component_version", row, immutable=True)
|
||||
with self.assertRaises(store.Conflict) as ctx:
|
||||
self.propose(rule_body(id="corporate.versioned"))
|
||||
self.assertIn("请使用新版本号", str(ctx.exception))
|
||||
# 换版本号即可继续演进
|
||||
bumped = self.propose(rule_body(id="corporate.versioned", version="1.0.1"))
|
||||
self.assertTrue(bumped["id"].endswith("@1.0.1"))
|
||||
|
||||
def test_propose_never_publishes_directly(self):
|
||||
draft = self.propose()
|
||||
self.assertEqual(draft["status"], "draft")
|
||||
with store.transaction(self.db_path) as db:
|
||||
published = [row for row in store.listing(db, "component_version")
|
||||
if row.get("component_id") == "corporate.test_rule"]
|
||||
self.assertEqual(published, [], "提议不得直接产生已发布版本")
|
||||
|
||||
|
||||
class EffectiveFilterTests(KnowledgeCardTestCase):
|
||||
def test_only_published_and_matching_domain_are_returned(self):
|
||||
self.propose(rule_body(id="corporate.demo_only", scope="demo"))
|
||||
self.propose(rule_body(id="corporate.prod_only", scope="production"))
|
||||
with store.transaction(self.db_path) as db:
|
||||
# 未发布 → 不返回
|
||||
self.assertEqual(cards.effective(db, "demo"), [])
|
||||
# 手工把草稿标记为已发布以验证域过滤(发布路径本身由上面几个类覆盖)
|
||||
for draft in store.listing(db, "component_draft"):
|
||||
row = dict(draft, status="published", approved_by="demo-reviewer",
|
||||
component_id=draft.get("component_id"))
|
||||
row.pop("revision", None)
|
||||
store.put(db, "component_version", row, immutable=True)
|
||||
demo = cards.effective(db, "demo")
|
||||
production = cards.effective(db, "production")
|
||||
self.assertTrue(demo)
|
||||
self.assertTrue(all(card["data_domain"] == "demo" for card in demo))
|
||||
self.assertTrue(all(card["data_domain"] == "production" for card in production))
|
||||
|
||||
def test_global_seed_components_are_not_mistakable_for_project_cards(self):
|
||||
"""内置种子组件是全局配置,不是项目知识卡片——必须明确报错而不是含糊通过。"""
|
||||
for item in registry.seed_components():
|
||||
body = registry.normalize(dict(item, status="published"), "system-demo")
|
||||
with self.subTest(component=item["id"]):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
cards._project(body)
|
||||
self.assertIn("不是项目知识卡片", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user