"""T09 推进沙盘测试：AC-11/12/30/31/44（BR-11/29/30、E-21/22）。 覆盖两条主线： 1. 情景是独立对象——基线副本随行，真实运行与它的快照/概率字段全程不变。 2. 演示值查表且有版本——组合不是单项加和，越域直接拒绝，说明文字固定。 """ import copy import unittest from contracts.prescreen import ErrorCode from contracts.simulation import Simulation from services import judgment_provider as jp from services import prescreen, prescreen_policy from services import prescreen_sandbox as sandbox DIMENSIONS = [ {"id": "DC-CASH", "component_refs": [{"id": "corporate.cashflow"}]}, {"id": "DC-TITLE", "component_refs": [{"id": "corporate.title"}]}, {"id": "DC-DEAL", "component_refs": [{"id": "corporate.deal"}]}, ] FINDINGS = [ {"code": "corporate.cashflow", "result": "unknown", "stage": 1, "evidence_ids": ["EV-1"]}, {"code": "corporate.title", "result": "unknown", "stage": 1, "evidence_ids": ["EV-2"]}, {"code": "corporate.deal", "result": "checked", "stage": 2, "evidence_ids": ["EV-3"]}, ] def frozen_plan(): return { "plan": {"id": "corporate.default", "version": "1.0.0"}, "skills": [{"id": "corporate.cashflow", "version": "1.0.0"}], "rules": [], "template": {"id": "template.ai-pre-review", "version": "1.0.0"}, "content_hash": "abc123", "dimensions": DIMENSIONS, } def run_view(data_domain="demo"): """用真实装配入口产出一条运行视图，避免测试自造第二份形状。""" decision = prescreen_policy.decide(FINDINGS) decision["findings"] = FINDINGS decision["missing_inputs"] = ["近六个月银行流水", "设备权属材料"] return prescreen.assemble( {"id": "run-9", "project_id": "DEMO-026", "created_at": "2026-09-30T01:00:00+00:00", "input_snapshot_id": "snap-9", "revision": 3}, frozen_plan(), decision, jp.resolve_config(False), data_domain=data_domain) class BaselineIsolationTests(unittest.TestCase): """AC-11：模拟不修改原运行，也不动它的快照。""" def test_run_view_is_not_mutated_by_simulation(self): view = run_view() before = copy.deepcopy(view) sandbox.simulate(view, ["assume_bank_flow", "assume_title"]) self.assertEqual(view, before, "情景推演不得改写真实运行对象") def test_simulation_carries_its_own_baseline_copy(self): view = run_view() result = sandbox.simulate(view, ["assume_bank_flow"]) baseline = result["baseline_probability_view"] self.assertEqual(baseline["demo_estimate"]["value"], 0.42) self.assertIsNone(baseline["real_estimate"]) self.assertIsNone(result["probability_view"]["real_estimate"], "演示情景的真实估计必须为空") sandbox.assert_baseline_untouched(view, result) def test_real_run_probability_stays_empty(self): view = run_view() self.assertIsNone(view["probability"]) self.assertIsNone(view["probability_view"]["real_estimate"]) self.assertIsNone(view["probability_view"].get("demo_estimate"), "真实运行不得携带演示值") class DemoTableTests(unittest.TestCase): """AC-30/12：演示值必须有表版本与固定说明，且组合不是加和。""" def test_table_matches_the_agreed_prototype_values(self): values = {key: row["value"] for key, row in sandbox.DEMO_SCENARIOS.items()} self.assertEqual(values, {"00": 42, "10": 58, "01": 51, "11": 64}) def test_combination_is_not_the_sum_of_single_deltas(self): sandbox.assert_not_additive() baseline = sandbox.DEMO_SCENARIOS["00"]["value"] single = (sandbox.DEMO_SCENARIOS["10"]["value"] - baseline) \ + (sandbox.DEMO_SCENARIOS["01"]["value"] - baseline) self.assertEqual(baseline + single, 67) self.assertEqual(sandbox.DEMO_SCENARIOS["11"]["value"], 64, "组合值必须来自查表，不是 42+16+9") def test_every_demo_estimate_is_labeled_and_versioned(self): result = sandbox.simulate(run_view(), ["assume_title"]) demo = result["probability_view"]["demo_estimate"] self.assertEqual(demo["source"], "synthetic_demo") self.assertEqual(demo["scenario_table_version"], sandbox.DEMO_TABLE_VERSION) self.assertEqual(demo["scenario_key"], "01") self.assertIn("人工设定", demo["label"]) def test_production_domain_is_refused_before_any_result(self): with self.assertRaises(sandbox.DemoEstimateForbidden): sandbox.simulate(run_view(data_domain="production"), ["assume_bank_flow"]) def test_contract_itself_refuses_demo_estimate_outside_demo_domain(self): result = sandbox.simulate(run_view(), ["assume_bank_flow"]) broken = dict(result, data_domain="production") with self.assertRaises(ValueError): Simulation(**broken) self.assertEqual(ErrorCode.DEMO_ESTIMATE_FORBIDDEN.value, "DEMO_ESTIMATE_FORBIDDEN") class AssumptionGuardTests(unittest.TestCase): """AC-31：冲突假设拒绝；组合摘要随选择变化。""" def test_unregistered_assumption_is_rejected(self): with self.assertRaises(sandbox.AssumptionConflict): sandbox.simulate(run_view(), ["assume_everything"]) def test_assumption_for_missing_dimension_is_rejected(self): view = run_view() view["dimension_results"] = [item for item in view["dimension_results"] if item["id"] != "DC-TITLE"] with self.assertRaises(sandbox.AssumptionConflict) as ctx: sandbox.simulate(view, ["assume_title"]) self.assertIn("DC-TITLE", str(ctx.exception)) def test_selection_hash_follows_the_full_combination(self): view = run_view() first = sandbox.simulate(view, ["assume_bank_flow"]) again = sandbox.simulate(view, ["assume_bank_flow"]) other = sandbox.simulate(view, ["assume_bank_flow", "assume_title"]) self.assertEqual(first["selection_hash"], again["selection_hash"]) self.assertNotEqual(first["selection_hash"], other["selection_hash"]) def test_new_revision_changes_the_selection_hash(self): view = run_view() self.assertNotEqual(sandbox.selection_hash("run-9", 3, ["assume_bank_flow"]), sandbox.selection_hash("run-9", 4, ["assume_bank_flow"])) class DimensionOverlayTests(unittest.TestCase): """AC-44/13-P-17：只改被假设的维度，其余维度原样保留并标明未变。""" def test_only_assumed_dimensions_change(self): result = sandbox.simulate(run_view(), ["assume_bank_flow"]) rows = {item["id"]: item for item in result["dimensions"]} self.assertEqual(rows["DC-CASH"]["baseline_stage"], 1) self.assertEqual(rows["DC-CASH"]["scenario_stage"], 3) self.assertTrue(rows["DC-CASH"]["assumed"]) self.assertEqual(rows["DC-TITLE"]["scenario_stage"], rows["DC-TITLE"]["baseline_stage"]) self.assertFalse(rows["DC-TITLE"]["assumed"]) def test_unchanged_dimensions_are_still_listed(self): result = sandbox.simulate(run_view(), ["assume_bank_flow"]) body = Simulation(**result) self.assertEqual(sorted(body.unchanged_dimensions()), ["DC-DEAL", "DC-TITLE"]) def test_no_assumption_returns_the_baseline_scenario(self): result = sandbox.simulate(run_view(), []) self.assertEqual(result["probability_view"]["demo_estimate"]["value"], 0.42) self.assertEqual(result["probability_view"]["demo_estimate"]["scenario_key"], "00") self.assertTrue(all(not item["assumed"] for item in result["dimensions"])) class FailureKeepsBaselineTests(unittest.TestCase): """AC-31：失败保留基线并标未完成，不得回退成演示表默认值。""" def test_failed_simulation_keeps_baseline_and_marks_incomplete(self): result = sandbox.failed(run_view(), ["assume_bank_flow"], "情景计算未完成，已保留基线与原有结论", ErrorCode.JUDGMENT_UNAVAILABLE) self.assertEqual(result["status"], "failed") self.assertEqual(result["error_code"], ErrorCode.JUDGMENT_UNAVAILABLE) self.assertIsNone(result["probability_view"], "失败不得显示任何情景数值") self.assertEqual(result["baseline_probability_view"]["demo_estimate"]["value"], 0.42) self.assertEqual(result["remaining_inputs"], ["近六个月银行流水", "设备权属材料"]) def test_contract_refuses_optimistic_values_while_computing(self): result = sandbox.simulate(run_view(), ["assume_bank_flow"]) pending = dict(result, status="computing") with self.assertRaises(ValueError): Simulation(**pending) if __name__ == "__main__": unittest.main()