"""纠偏第二批 B-b · 候选治理与自然语言修订测试（AC-51 / BR-42 / AC-43）。 三条必须机械成立的约束： 1. **自然语言修订必须真的调用模型**——没有"把中文猜成条件"的确定性回退； 模型不可用时拒绝且**卡片一字未改**（反例断言比对的是修订前后的卡片内容）。 2. **只产草稿**：修订后状态仍是 draft，绝不因"改过"而生效。 3. **来源不可洗 + 不自批**：历史卡片修订后仍是 history_derived 且保留来源事件； 发布需要独立审核人且先过隔离评测。 用临时库；不触碰真实数据根。模型用假传输层，走真实 invoke_scoped 路径。 """ import json 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 langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage from langchain_core.outputs import ChatGeneration, ChatResult from api import prescreen as prescreen_api from api import project_assistant as api from api import risk_agent_config as catalog from contracts.prescreen import ErrorCode from services import assistant_store as store from services import knowledge_cards from services.project_assistant import ProjectAssistantService PROJECT = "DEMO-B" PRESCREEN = prescreen_api.router.prefix MANAGER = {"X-Xingbang-Demo-Role": "manager"} REVIEWER = {"X-Xingbang-Demo-Role": "reviewer"} CARD_ID = "skill.history.fixture" class FakeModel(BaseChatModel): """假传输层；`calls` 记录调用次数，用于证明"确实发起了调用"。""" payload: str = "" calls: list = [] @property def _llm_type(self) -> str: return "fake" def _generate(self, messages, stop=None, run_manager=None, **kwargs): FakeModel.calls.append(messages) return ChatResult(generations=[ChatGeneration(message=AIMessage(content=self.payload))]) class GovernanceTestCase(unittest.TestCase): def setUp(self): FakeModel.calls = [] 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() app = FastAPI() app.include_router(api.router) app.include_router(prescreen_api.router) self.client = TestClient(app) with self.service.connect() as db: store.put(db, "project", { "id": PROJECT, "data_domain": "demo", "revision": 1, "facts": {}, "values": {}, "sources": {}, "selected_sources": {}, "document_ids": [], "pending_candidates": [], "data_revision": 1, }) def tearDown(self): self.client.close() for item in reversed(self.patches): item.stop() self.temp.cleanup() # ------------------------------------------------------------ 夹具与工具 def propose_history_card(self, card_id: str = CARD_ID, actor: str = "demo-manager"): with self.service.connect() as db: return knowledge_cards.propose(db, { "id": card_id, "kind": "skill", "name": "核查回款集中依赖", "engine": "required", "inputs": ["financial.operating_cashflow"], "code": "history.fixture", "title": "核查回款集中依赖", "conditions": [{"code": "c1", "field": "collection.dependency", "operator": "eq", "threshold": True}], }, actor, PROJECT, origin="history_derived", source_event_ids=["event-1"]) def add_test_cases(self, card_id: str = CARD_ID, headers=MANAGER): """补独立测试样本——14 §4 的"专家修改 → 测试"步骤，也是发布门禁的前置条件。""" return self.request("PATCH", f"/knowledge-candidates/{card_id}", { "changes": {"test_cases": [ {"name": "缺证样本", "values": {"financial.operating_cashflow": None, "collection.dependency": True}, "expected_outputs": {}, "expected_checks": {"history.fixture": "unknown"}}, {"name": "已提供样本", "values": {"financial.operating_cashflow": "1200", "collection.dependency": True}, "expected_outputs": {}, "expected_checks": {"history.fixture": "pass"}}, ]}, "note": "补独立测试样本以满足评测门禁", }, headers) def card(self, card_id: str = CARD_ID) -> dict: with self.service.connect() as db: draft = knowledge_cards.current_draft(db, card_id) return knowledge_cards.draft_card(db, draft["id"]) def request(self, method, path, body=None, headers=MANAGER): return self.client.request(method, PRESCREEN + path, json=body, headers=headers) def with_fake_model(self, payload): return patch("services.experience_extraction.ChatOpenAI", lambda **kwargs: FakeModel(payload=json.dumps(payload, ensure_ascii=False))) class NaturalLanguageRevisionTests(GovernanceTestCase): def setUp(self): super().setUp() self.propose_history_card() def test_natural_language_revision_actually_calls_the_model(self): answer = {"conditions": {"collection.dependency": False}, "exception_note": "仅适用于设备类租赁", "note": "限定适用范围"} env = {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1", "XINGBANG_LOCAL_MODEL_NAME": "fake-local"} with patch.dict(os.environ, env), self.with_fake_model(answer): response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"requirement": "只适用于设备类租赁，并加上例外"}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(len(FakeModel.calls), 1, "自然语言修订必须真的发起一次模型调用") card = self.card() self.assertEqual(card["conditions"][0]["field"], "collection.dependency") self.assertEqual(card["conditions"][0]["value"], False, "修订应真正落到卡片上") def test_revision_keeps_history_origin_sources_and_draft_status(self): with patch.dict(os.environ, {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}): self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"changes": {"exception_note": "已有到账凭证时不重复核查"}, "note": "补充例外说明"}) card = self.card() self.assertEqual(card["origin"], "history_derived", "人工修订不得洗掉历史来源") self.assertEqual(card["source_event_ids"], ["event-1"]) self.assertEqual(card["status"], "draft", "修订只产草稿，不得因此生效") self.assertGreater(card["revision"], 1, "修订必须产生新 revision") def test_unavailable_model_blocks_natural_language_and_leaves_card_untouched(self): before = self.card() response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"requirement": "只适用于设备类租赁"}) self.assertEqual(response.status_code, 422) self.assertIn("自然语言修改需要已配置模型", response.json()["error"]["message"]) after = self.card() self.assertEqual(after["conditions"], before["conditions"], "拒绝时必须一字未改") self.assertEqual(after["revision"], before["revision"]) def test_model_output_outside_standard_fields_is_rejected(self): before = self.card() with patch.dict(os.environ, {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}), \ self.with_fake_model({"conditions": {"自造字段": True}, "note": "乱改"}): response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"requirement": "随便改"}) self.assertEqual(response.status_code, 422) self.assertIn("必须来自标准字段", response.json()["error"]["message"]) self.assertEqual(self.card()["conditions"], before["conditions"]) def test_revision_requires_a_change_note(self): response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"changes": {"exception_note": "x"}, "note": " "}) self.assertEqual(response.status_code, 422) self.assertIn("变更说明", response.json()["error"]["message"]) def test_revision_rejects_fields_outside_the_editable_set(self): response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"changes": {"scope": "production"}, "note": "改数据域"}) self.assertEqual(response.status_code, 422) self.assertIn("不支持的修订字段", response.json()["error"]["message"]) class PublicationTests(GovernanceTestCase): def setUp(self): super().setUp() self.propose_history_card() def publish(self, headers=REVIEWER): return self.request("POST", f"/knowledge-candidates/{CARD_ID}/transitions", {"action": "publish"}, headers) def test_publish_requires_an_independent_evaluation_first(self): response = self.publish() self.assertEqual(response.status_code, 409) self.assertIn("评测", response.json()["error"]["message"]) def test_manager_cannot_evaluate_or_publish(self): """角色门禁与既有 /components 入口一致：客户经理不得自评自批。""" evaluate = self.request("POST", f"/knowledge-candidates/{CARD_ID}/transitions", {"action": "evaluate"}) self.assertEqual(evaluate.status_code, 403) def test_self_approval_is_rejected(self): """AC-03/AC-43：提议人不能给自己的候选发布版本。""" drafted = self.propose_history_card("skill.history.selfmade", actor="demo-reviewer") evaluated = self.request("POST", f"/knowledge-candidates/{drafted['component_id']}/transitions", {"action": "evaluate"}, REVIEWER) self.assertEqual(evaluated.status_code, 200, evaluated.text) published = self.request("POST", f"/knowledge-candidates/{drafted['component_id']}/transitions", {"action": "publish"}, REVIEWER) self.assertEqual(published.status_code, 403, "同一人不得既提议又发布") def test_revision_after_publication_bumps_version_and_keeps_lineage(self): self.assertEqual(self.add_test_cases().status_code, 200) self.request("POST", f"/knowledge-candidates/{CARD_ID}/transitions", {"action": "evaluate"}, REVIEWER) published = self.publish() self.assertEqual(published.status_code, 200, published.text) self.assertEqual(published.json()["version"], "1.0.0") revised = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}", {"changes": {"exception_note": "补充例外"}, "note": "上会前修订"}, REVIEWER) self.assertEqual(revised.status_code, 200, revised.text) card = self.card() self.assertEqual(card["version"], "1.0.1", "已发布版本不可覆盖，必须升版本") self.assertEqual(card["parent_version"], {"id": CARD_ID, "version": "1.0.0"}) self.assertEqual(card["status"], "draft") class UnknownObjectTests(GovernanceTestCase): """不存在的卡片必须是 404，不是 500。 这是回归守卫：归属查询若放在 run() 之外，KeyError 会绕过错误信封映射变成 500。 """ def test_revising_an_unknown_candidate_is_404(self): response = self.request("PATCH", "/knowledge-candidates/skill.not.exists", {"changes": {"exception_note": "x"}, "note": "改"}) self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["error"]["code"], ErrorCode.NOT_FOUND.value) def test_transitioning_an_unknown_candidate_is_404(self): response = self.request("POST", "/knowledge-candidates/skill.not.exists/transitions", {"action": "evaluate"}, REVIEWER) self.assertEqual(response.status_code, 404) class ManualCandidateTests(GovernanceTestCase): def test_manual_candidate_is_created_as_draft(self): response = self.request("POST", f"/projects/{PROJECT}/knowledge-candidates", { "kind": "skill", "title": "人工新增：核对租赁物发票", "content": {"engine": "required", "inputs": ["lease.total_value"], "code": "manual.invoice"}, }) self.assertEqual(response.status_code, 201, response.text) card = self.card(response.json()["component_id"]) self.assertEqual(card["origin"], "manual") self.assertEqual(card["status"], "draft") self.assertEqual(card["source_event_ids"], []) def test_unknown_origin_is_rejected(self): response = self.request("POST", f"/projects/{PROJECT}/knowledge-candidates", { "kind": "skill", "title": "x", "origin": "模型自创", "content": {"engine": "required", "inputs": ["lease.total_value"], "code": "manual.x"}, }) self.assertEqual(response.status_code, 422) self.assertIn("来源取值须为", response.json()["error"]["message"]) def test_candidate_list_exposes_origin_and_status(self): self.propose_history_card() listed = self.request("GET", f"/projects/{PROJECT}/knowledge-candidates").json()["items"] self.assertEqual(len(listed), 1) self.assertEqual(listed[0]["origin"], "history_derived") self.assertEqual(listed[0]["status"], "draft") if __name__ == "__main__": unittest.main()