Files
mini-openclaw-v2/backend/test_knowledge_cards.py
T
qiaoqichaoandClaude Code d18c147076 feat(v2): T05-c 运行补充说明只能形成草稿(AC-26)
运行时的"本次预审补充说明"只影响本次运行(run_instruction 落在运行对象上,
属 T07 的 ScreeningRun 字段)。要变成持久知识只有一条路:显式提议为草稿,
且走正常审核。本批实现卡片侧的那一半。

新增 knowledge_cards.draft_from_run_instruction(),三条约束:
1. 只产草稿——运行中说过的话绝不直接生效(有测试断言无已发布版本产生)
2. **不得携带工具/权限/引擎字段**:allowed_tools / tools / permissions / engine /
   run_instruction 出现即拒,扩大工具授权必须走独立提议与审核
3. 记录 derived_from_run,便于回溯知识来自哪次运行
另:运行指令不能声明为 history_derived(历史提炼须提供来源事件或证据),
hard 门禁对运行衍生的卡片同样生效。

未完成:运行对象上的 run_instruction 字段与"只影响本次运行"的服务端实现属 T07,
本批只交付卡片侧,不宣称 AC-26 已整体闭环。

测试:test_knowledge_cards.py 增至 25 项(新增 5 项)。
全量 315 项 OK(310 → 315);verify.py 10/10 PASS。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-29 18:51:12 +08:00

267 lines
13 KiB
Python

"""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))
class RunInstructionScopeTests(KnowledgeCardTestCase):
"""AC-26:运行补充说明只影响当前运行;持久化只能形成草稿,且不得扩大工具权限。"""
def draft_from_run(self, body, actor="demo-manager"):
with store.transaction(self.db_path) as db:
return cards.draft_from_run_instruction(db, "DEMO-A", "run-1", body, actor)
def test_instruction_becomes_a_draft_never_a_published_card(self):
draft = self.draft_from_run(rule_body(id="corporate.from_run"))
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.from_run"]
self.assertEqual(published, [], "运行中说过的话不得直接成为生效知识")
def test_instruction_cannot_widen_tool_permissions(self):
for field in ("allowed_tools", "tools", "permissions", "engine", "run_instruction"):
with self.subTest(field=field):
body = rule_body(id="corporate.tool_grab")
body[field] = ["terminal"]
with self.assertRaises(ValueError) as ctx:
self.draft_from_run(body)
self.assertIn("不能持久化工具权限", str(ctx.exception))
def test_instruction_cannot_claim_history_derived_origin(self):
with self.assertRaises(ValueError) as ctx:
self.draft_from_run(rule_body(id="corporate.fake_history", origin="history_derived"))
self.assertIn("不能声明为历史提炼", str(ctx.exception))
def test_draft_records_which_run_it_came_from(self):
mock = self.draft_from_run(rule_body(id="corporate.traced"))
with store.transaction(self.db_path) as db:
row = store.get(db, "component_draft", mock["id"])
self.assertEqual(row["derived_from_run"], "run-1")
self.assertEqual(row["proposed_via"], "run_instruction")
def test_hard_gate_still_applies_to_run_derived_cards(self):
with self.assertRaises(ValueError) as ctx:
self.draft_from_run(rule_body(id="corporate.run_hard", effect="hard"))
self.assertIn("正式制度或授权依据", str(ctx.exception))
if __name__ == "__main__":
unittest.main()