195 lines
8.9 KiB
Python
195 lines
8.9 KiB
Python
 
|
"""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()
|