217 lines
11 KiB
Python
217 lines
11 KiB
Python
"""纠偏第二批 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"}
|
|
ADMIN = {"X-Xingbang-Demo-Role": "admin"}
|
|
PUBLISHER = {"Authorization": "Bearer fixture-publisher"}
|
|
|
|
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": '[{"id":"config-admin-b","role":"admin","projects":["DEMO-B"],"token":"fixture-publisher"}]', "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=PUBLISHER):
|
|
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",
|
|
}, headers=ADMIN)
|
|
|
|
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"}, PUBLISHER)
|
|
self.assertEqual(evaluated.status_code, 200, evaluated.text)
|
|
published = self.call("POST", f"/review-plans/{plan.json()['component_id']}/transitions",
|
|
{"action": "publish"}, PUBLISHER)
|
|
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"}, PUBLISHER)
|
|
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"}, PUBLISHER)
|
|
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"}, PUBLISHER)
|
|
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()
|