"""纠偏第二批 B-c · 卡片选入方案并影响核查（BR-49 / AC-52）。 审计（audit.json P1-3）确认：知识卡片此前没有任何调用方，"卡片影响核查"只存在于 孤立的单元测试里。本测试走**真实执行链**： 提炼/人工卡片 → 隔离评测 → 独立审核发布 → 装配方案 → 发布方案 → service.enqueue → DurableWorker.process_next() → 运行结果 用**同项目两次运行对照**证明卡片确实改变了核查结果：基座方案里没有该卡片， 新方案里有；两次运行都能读出各自的 `component_versions` 与核查项依据。 不用注入记录；临时库；不触碰真实数据根。 """ import os import tempfile import unittest from pathlib import Path from unittest.mock import patch from fastapi import FastAPI from fastapi.testclient import TestClient from api import prescreen as prescreen_api from api import project_assistant as api from api import risk_agent_config as catalog from services import assistant_store as store from services import knowledge_cards from services.assistant_worker import DurableWorker from services.project_assistant import ProjectAssistantService PROJECT = "DEMO-B" BASE = "/api/xingbang/risk-agent/project-assistant" PRESCREEN = BASE + "/prescreen" MANAGER = {"X-Xingbang-Demo-Role": "manager"} REVIEWER = {"X-Xingbang-Demo-Role": "reviewer"} CARD_ID = "skill.history.repayment" CARD_VERSION = "1.0.0" CARD_CODE = "history.repayment.dependency" # 自定义 Skill 的发布门禁要求独立测试样本（component_evaluation 的既有规则） TEST_CASES = [ {"name": "缺证样本", "values": {"financial.operating_cashflow": None}, "expected_outputs": {}, "expected_checks": {CARD_CODE: "unknown"}}, {"name": "已提供样本", "values": {"financial.operating_cashflow": "1200"}, "expected_outputs": {}, "expected_checks": {CARD_CODE: "pass"}}, ] class CardToRunTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.path = Path(self.temp.name) self.service = ProjectAssistantService(self.path / "assistant.sqlite3") self.patches = [ patch.object(api, "service", lambda: self.service), patch.object(prescreen_api, "_service", lambda: self.service), patch.object(catalog, "DB_PATH", self.path / "catalog.sqlite3"), patch.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0", "XINGBANG_LOCAL_MODEL_URL": "", "XINGBANG_DEMO_ROLE_LOGIN": "1"}), ] for item in self.patches: item.start() self.worker = DurableWorker(self.service) app = FastAPI() app.include_router(api.router) app.include_router(prescreen_api.router) self.client = TestClient(app) def tearDown(self): self.worker.stop() self.client.close() for item in reversed(self.patches): item.stop() self.temp.cleanup() # ------------------------------------------------------------ 工具 def call(self, method, path, body=None, headers=MANAGER, root=PRESCREEN): return self.client.request(method, root + path, json=body, headers=headers) def propose_card(self): with self.service.connect() as db: return knowledge_cards.propose(db, { "id": CARD_ID, "kind": "skill", "name": "核查回款集中依赖", "engine": "required", "inputs": ["financial.operating_cashflow"], "code": CARD_CODE, "title": "核查回款集中依赖", "test_cases": TEST_CASES, }, "demo-manager", PROJECT, origin="history_derived", source_event_ids=["event-1"]) def publish(self, path, headers=REVIEWER): evaluated = self.call("POST", path, {"action": "evaluate"}, headers) self.assertEqual(evaluated.status_code, 200, evaluated.text) return self.call("POST", path, {"action": "publish"}, headers) def assemble_plan(self, skills, name="含历史经验卡的预审方案"): return self.call("POST", f"/projects/{PROJECT}/review-plans", { "name": name, "skills": skills, "rules": ["demo.coverage"], "template": "template.ai-pre-review", "budget": {"timeout_seconds": 180, "max_tool_calls": 18}, "audience": ["manager"], "run_kind": "pre_review", }) def run_with(self, plan_id): queued = self.call("POST", f"/projects/{PROJECT}/runs", {"requirement": "核对回款与现金流", "plan_id": plan_id}, root=BASE) self.assertIn(queued.status_code, (200, 202), queued.text) run_id = queued.json()["id"] self.assertTrue(self.worker.process_next(), "队列里应当有一次待执行运行") with self.service.connect() as db: return store.get(db, "run", run_id) @staticmethod def checks_of(run, code): return [check for step in run["steps"] for check in step.get("findings", []) if check.get("code") == code] # ------------------------------------------------------------ AC-52 def test_published_card_changes_the_run_and_is_traceable(self): # 1) 历史提炼候选 → 评测 → 独立审核发布 card = self.propose_card() published = self.publish(f"/knowledge-candidates/{CARD_ID}/transitions") self.assertEqual(published.status_code, 200, published.text) self.assertEqual(published.json()["version"], CARD_VERSION) # 2) 基座方案：卡片不在方案里 → 运行里不应出现该核查项 baseline = self.run_with("corporate.pre-review") self.assertEqual(self.checks_of(baseline, CARD_CODE), [], "卡片没进方案时不得凭空产生该核查项") # 3) 把已发布卡片选入方案并发布 plan = self.assemble_plan( ["corporate.consistency", "corporate.cashflow", "corporate.assets", "corporate.coverage", CARD_ID]) self.assertEqual(plan.status_code, 201, plan.text) plan_id = plan.json()["component_id"] self.assertEqual(self.publish(f"/review-plans/{plan_id}/transitions").status_code, 200) # 4) 用新方案真实运行 run = self.run_with(plan_id) refs = {(item["id"], item["version"]) for item in run["component_versions"]} self.assertIn((CARD_ID, CARD_VERSION), refs, "运行必须冻结并记录卡片版本") checks = self.checks_of(run, CARD_CODE) self.assertEqual(len(checks), 1, "已发布卡片必须真的产生核查项") step = next(step for step in run["steps"] if step.get("skill", {}).get("id") == CARD_ID) self.assertEqual(step["status"] in ("completed", "partial"), True) self.assertEqual(step["findings"][0]["basis_ids"], [f"{CARD_ID}@{CARD_VERSION}"], "核查项必须能追溯到卡片版本") self.assertEqual(step["findings"][0]["fields"], ["financial.operating_cashflow"]) # 5) 旧运行仍可读，且没有被新方案改写 self.assertEqual(self.checks_of(baseline, CARD_CODE), []) with self.service.connect() as db: self.assertNotEqual(store.get(db, "run", baseline["id"])["status"], "") def test_draft_card_cannot_be_selected_into_a_plan(self): """草稿卡片不得进入新运行（BR-27 / AC-28）——装配阶段就要拒。""" self.propose_card() # 只提议，不发布 plan = self.assemble_plan(["corporate.coverage", CARD_ID]) self.assertEqual(plan.status_code, 201, plan.text) # 发布时依赖解析拒绝未发布组件 evaluated = self.call("POST", f"/review-plans/{plan.json()['component_id']}/transitions", {"action": "evaluate"}, REVIEWER) self.assertEqual(evaluated.status_code, 200, evaluated.text) published = self.call("POST", f"/review-plans/{plan.json()['component_id']}/transitions", {"action": "publish"}, REVIEWER) self.assertEqual(published.status_code, 409, published.text) def test_missing_dependency_fails_evaluation_instead_of_being_dropped(self): """方案引用不存在的组件 → 评测不通过、发布被拒，不静默丢掉该项。""" self.propose_card() plan = self.assemble_plan(["corporate.coverage", "skill.not.exists"]) self.assertEqual(plan.status_code, 201, plan.text) plan_ref = plan.json()["component_id"] evaluated = self.call("POST", f"/review-plans/{plan_ref}/transitions", {"action": "evaluate"}, REVIEWER) self.assertEqual(evaluated.status_code, 200, evaluated.text) self.assertFalse(evaluated.json()["passed"], "依赖缺失必须让评测不通过") published = self.call("POST", f"/review-plans/{plan_ref}/transitions", {"action": "publish"}, REVIEWER) self.assertEqual(published.status_code, 409, published.text) def test_plan_rejects_tool_configuration_surface(self): """ADR-11：方案不得自带工具配置面，业务侧不能借方案扩权。""" response = self.call("POST", f"/projects/{PROJECT}/review-plans", { "name": "越权方案", "skills": ["corporate.coverage"], "template": "template.ai-pre-review", "allowed_tools": ["execute_sql"], }) self.assertEqual(response.status_code, 422) self.assertIn("allowed_tools", response.text) def test_transitioning_an_unknown_plan_is_404(self): """不存在的方案必须是 404，不是 500（归属查询在 run() 内完成）。""" response = self.call("POST", "/review-plans/plan.not.exists/transitions", {"action": "evaluate"}, REVIEWER) self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["error"]["code"], "NOT_FOUND") def test_listed_plans_show_the_selected_cards(self): self.propose_card() self.publish(f"/knowledge-candidates/{CARD_ID}/transitions") plan = self.assemble_plan(["corporate.coverage", CARD_ID]) listed = self.call("GET", f"/projects/{PROJECT}/review-plans").json()["items"] mine = next(item for item in listed if item["id"] == plan.json()["component_id"]) self.assertIn(CARD_ID, mine["skills"], "方案列表必须显示选入的卡片") self.assertEqual(mine["status"], "draft") if __name__ == "__main__": unittest.main()