"""T07-c 预审运行装配测试：AC-13/14/45 与冻结语义。 AC-13（新证据 → 新运行，旧运行留存）与 AC-14（幂等键）由既有 enqueue 实现， 本文件**验证它确实成立**而不是重新实现；其余断言针对装配与冻结。 """ import tempfile import unittest from pathlib import Path from unittest.mock import patch from services import assistant_store as store from services import component_registry as registry from services import judgment_provider as jp from services import prescreen, prescreen_policy from services.project_assistant import ProjectAssistantService def frozen_plan(dimensions=None): return { "plan": {"id": "corporate.default", "version": "1.0.0"}, "skills": [{"id": "corporate.cashflow", "version": "1.0.0", "code": "financial.cashflow.missing"}], "rules": [{"id": "demo.coverage", "version": "1.0.0", "conditions": [{"code": "lease.coverage.below_demo_threshold"}]}], "template": {"id": "template.ai-pre-review", "version": "1.0.0"}, "content_hash": "abc123", "dimensions": dimensions or [], } def run_body(**over): body = {"id": "run-1", "project_id": "DEMO-B", "created_at": store.now(), "input_snapshot_id": "snap-1", "requirement": "重点核对现金流"} body.update(over) return body class AssemblyTests(unittest.TestCase): def setUp(self): self.config = jp.resolve_config(False) def assemble(self, decision=None, plan=None, status="completed"): decision = decision if decision is not None else prescreen_policy.decide([]) return prescreen.assemble(run_body(), plan or frozen_plan(), decision, self.config, status=status) def test_assembled_run_is_contract_valid(self): screening = self.assemble() self.assertEqual(screening["id"], "run-1") self.assertEqual(screening["plan_ref"], {"id": "corporate.default", "version": "1.0.0"}) self.assertEqual(screening["snapshot_id"], "snap-1") def test_probability_is_null_and_reason_recorded(self): screening = self.assemble() self.assertIsNone(screening["probability"]) self.assertEqual(screening["probability_reason"], "not_calibrated") self.assertIsNone(screening["probability_view"]["real_estimate"]) self.assertEqual(screening["probability_view"]["reason"], "not_calibrated") def test_demo_estimate_must_be_labelled_and_cannot_become_real(self): view = prescreen.probability_view("snap-1", demo_estimate={ "value": 0.42, "source": "synthetic_demo", "scenario_table_version": "v1", "scenario_key": "k", "label": "人工设定、未经真实历史校准"}) self.assertEqual(view["demo_estimate"]["value"], 0.42) self.assertIsNone(view["real_estimate"]) for bad in ({"value": 0.42, "source": "model", "label": "x"}, # 来源不符 {"value": 0.42, "source": "synthetic_demo"}, # 缺说明 {"value": 1.5, "source": "synthetic_demo", "label": "x"}): # 越界 with self.subTest(bad=bad): with self.assertRaises(Exception): prescreen.probability_view("snap-1", demo_estimate=bad) def test_judgment_config_is_frozen_into_the_run(self): screening = self.assemble() self.assertEqual(screening["judgment_config"]["provider"], "disabled") self.assertFalse(screening["judgment_config"]["enabled"]) self.assertTrue(screening["judgment_config"]["disabled_reason"]) def test_versions_block_records_plan_and_hash(self): screening = self.assemble() self.assertEqual(screening["versions"]["plan"]["id"], "corporate.default") self.assertEqual(screening["versions"]["plan_hash"], "abc123") self.assertIsNone(screening["versions"]["questions"], "题库版本未接入时留空，不编造") def test_hard_block_is_carried_into_the_run(self): decision = prescreen_policy.decide( [{"code": "lease.coverage.below_demo_threshold", "result": "issue", "basis_version": "demo.coverage@1.0.0"}], hard_rule_codes=["lease.coverage.below_demo_threshold"]) screening = self.assemble(decision) self.assertEqual(screening["recommendation"], "stop") def test_failed_run_must_carry_degradations(self): with self.assertRaises(ValueError) as ctx: self.assemble(prescreen_policy.decide([]), status="failed") self.assertIn("降级记录", str(ctx.exception)) def test_unfinished_run_cannot_carry_a_recommendation(self): decision = prescreen_policy.decide([]) with self.assertRaises(ValueError) as ctx: self.assemble(decision, status="running") self.assertIn("未结束时不得给出建议", str(ctx.exception)) def test_missing_snapshot_is_refused(self): with self.assertRaises(ValueError) as ctx: prescreen.assemble(run_body(input_snapshot_id=None), frozen_plan(), prescreen_policy.decide([]), self.config) self.assertIn("输入快照", str(ctx.exception)) def test_junk_versions_are_not_invented(self): block = prescreen.versions_block({"plan": {"id": "p", "version": "1"}}) self.assertIsNone(block["model"], "主审模型未配置时留空，不编造版本") self.assertIsNone(block["questions"]) class DimensionResultTests(unittest.TestCase): def test_unknown_when_no_component_reports(self): rows = prescreen.dimension_results( frozen_plan([{"id": "d1", "component_refs": [{"id": "corporate.cashflow"}]}]), []) self.assertEqual(rows[0]["state"], "unknown") self.assertIsNone(rows[0]["stage"]) def test_issue_marks_the_dimension_blocked(self): rows = prescreen.dimension_results( frozen_plan([{"id": "d1", "component_refs": [{"id": "corporate.cashflow"}]}]), [{"code": "corporate.cashflow", "result": "issue", "stage": 2, "evidence_ids": ["ev-1"]}]) self.assertEqual(rows[0]["state"], "blocked") self.assertEqual(rows[0]["stage"], 2) self.assertEqual(rows[0]["basis_ids"], ["ev-1"]) def test_pass_marks_checked_and_unknown_beats_pass(self): plan = frozen_plan([{"id": "d1", "component_refs": [{"id": "a"}, {"id": "b"}]}]) rows = prescreen.dimension_results(plan, [{"code": "a", "result": "pass"}]) self.assertEqual(rows[0]["state"], "unknown", "部分无结果时不得算已核查") rows = prescreen.dimension_results(plan, [{"code": "a", "result": "pass"}, {"code": "b", "result": "pass"}]) self.assertEqual(rows[0]["state"], "checked") def test_dimension_evidence_is_traceable(self): rows = prescreen.dimension_results( frozen_plan([{"id": "d1", "component_refs": [{"id": "a"}]}]), [{"code": "a", "result": "pass", "evidence_ids": ["ev-9"]}]) self.assertIn("ev-9", rows[0]["basis_ids"]) class FrozenVersionsGuardTests(unittest.TestCase): def test_run_without_a_plan_ref_is_refused(self): with self.assertRaises(ValueError) as ctx: prescreen.assert_versions_frozen({"plan_ref": {"id": "", "version": ""}, "judgment_config": {"enabled": False}}) self.assertIn("未冻结评审方案版本", str(ctx.exception)) def test_enabled_judgment_without_a_frozen_revision_is_refused(self): with self.assertRaises(ValueError) as ctx: prescreen.assert_versions_frozen({ "plan_ref": {"id": "p", "version": "1"}, "judgment_config": {"enabled": True, "profile_id": None, "profile_revision": None}}) self.assertIn("必须冻结配置 ID 与修订", str(ctx.exception)) class ExistingRunGuaranteesTests(unittest.TestCase): """AC-13/14 由既有 enqueue 提供——这里验证其成立，不重新实现。""" def setUp(self): self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.path = Path(self._tmp.name) self.service = ProjectAssistantService(self.path / "assistant.sqlite3") def tearDown(self): self._tmp.cleanup() def identity(self): return {"id": "demo-manager", "role": "manager", "projects": ["*"], "preview": False} def test_new_evidence_produces_a_new_snapshot_and_keeps_the_old_run(self): # AC-13：新证据版本 → 新快照 + 新运行，旧运行留存 first = self.service.get_project("DEMO-B") run_one = self.service.enqueue("DEMO-B", self.identity(), "第一轮") self.service.supplement("DEMO-B", "demo-manager", {"collection_evidence": "已补充到账凭证"}, [], self.service.get_project("DEMO-B")["revision"], "补件说明") run_two = self.service.enqueue("DEMO-B", self.identity(), "第二轮") self.assertNotEqual(run_one["input_snapshot_id"], run_two["input_snapshot_id"]) with self.service.connect() as db: snapshots = store.listing(db, "input_snapshot", "DEMO-B") runs = store.listing(db, "run", "DEMO-B") self.assertGreaterEqual(len(snapshots), 2, "两次运行各自冻结快照") self.assertIn(run_one["id"], {item["id"] for item in runs}, "旧运行必须留存") self.assertNotEqual(first["revision"], self.service.get_project("DEMO-B")["revision"]) def test_same_idempotency_key_replays_the_same_run(self): # AC-14：同键同载荷返回同一运行 one = self.service.enqueue("DEMO-B", self.identity(), "重复要求", idempotency_key="k1") two = self.service.enqueue("DEMO-B", self.identity(), "重复要求", idempotency_key="k1") self.assertEqual(one["id"], two["id"]) def test_same_key_with_different_payload_conflicts(self): self.service.enqueue("DEMO-B", self.identity(), "要求甲", idempotency_key="k2") with self.assertRaises(store.Conflict): self.service.enqueue("DEMO-B", self.identity(), "要求乙", idempotency_key="k2") def test_stale_revision_is_rejected(self): # AC-13 的另一半：依据旧 revision 更新会冲突 project = self.service.get_project("DEMO-B") self.service.supplement("DEMO-B", "demo-manager", {"collection_evidence": "先改一次"}, [], project["revision"], "说明") with self.assertRaises(store.Conflict): self.service.supplement("DEMO-B", "demo-manager", {"collection_evidence": "再用旧版本改"}, [], project["revision"], "说明") if __name__ == "__main__": unittest.main()