qiaoqichaoandClaude Code
22f8a37799
覆盖 AC-11/12/29/30/31/42/44（BR-11/29/30/41/43、E-20/21/22）。 后端： - contracts/simulation.py：Simulation/SimulationAssumption/ScenarioDimensionState。 契约层直接禁止 production 域携带演示估计（E-21）；ready 必须有结果、 failed 必须带错误码且不得显示结果、计算中不得预告乐观增幅。 - services/prescreen_sandbox.py：演示情景表（DEMO-TABLE-v1）与推演入口。 组合查表不是单项加和（assert_not_additive 把这条写成可执行断言）； 只改被假设的维度，其余维度原样保留；失败保留基线并标未完成。 - services/prescreen_store.py：运行视图只读存储，复用 assistant_objects 新 kind prescreen_run，不新建表；只提供保存与读取，要改结果必须产生新修订。 - api/prescreen.py：新增 3 个只读端点（运行列表、运行视图、可用假设动作） 与 1 个情景端点；请求体里没有任何概率字段。越域返回 E-21、冲突假设 E-22、 过期修订 409。 - contracts/prescreen.py：补 E-20/21/22 三个错误码与其 HTTP 映射。 前端（挂进既有工作台「材料与问题」页，不新建路由、不新增样式表）： - src/lib/prescreen-run.ts：客户端契约、呈现判定（同刻度 3–8 维才画雷达）、 演示值域判定、来源与数据域徽标（文字＋图标＋样式，去色仍可区分）。 - src/components/xingbang/PrescreenPanel.tsx：主审意见、演示值（紧邻固定说明）、 维度雷达与等价文字列表、维度依据抽屉、假设勾选即时推演。 旧请求晚到或结果与当前选择不一致即丢弃；请求失败保留基线并标未完成。 - scripts/test_prescreen_ui.cjs：界面源码守卫，已注册进 scripts/verify.py。 未做：运行装配结果的写入方仍是执行链路（T10 范围），当前由测试写入； 沙盘尚未接入真实执行，界面数值全部为服务端演示表值。 验证：backend 全量 unittest 469 项 OK；前端 npm run build 通过。 Co-Authored-By: Claude Code <noreply@anthropic.com>
218 lines
9.8 KiB
Python
218 lines
9.8 KiB
Python
"""T09 沙盘接口层测试：运行只读视图与情景端点（AC-11/12/30/31）。
|
|
|
|
与 test_prescreen_api.py 同一套路：只挂 prescreen 路由到独立 app，库指向临时文件，
|
|
不触碰真实库、不启动服务。
|
|
"""
|
|
|
|
import copy
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.testclient import TestClient
|
|
|
|
from api import prescreen as api
|
|
from services import assistant_store as store
|
|
from services import judgment_provider as jp
|
|
from contracts.prescreen import ScreeningRun
|
|
from services import prescreen, prescreen_policy, prescreen_store
|
|
from services.project_assistant import ProjectAssistantService
|
|
|
|
HEADERS = {"X-Xingbang-Demo-Role": "manager"}
|
|
PROJECT = "DEMO-B"
|
|
|
|
DIMENSIONS = [
|
|
{"id": "DC-CASH", "component_refs": [{"id": "corporate.cashflow"}]},
|
|
{"id": "DC-TITLE", "component_refs": [{"id": "corporate.title"}]},
|
|
]
|
|
FINDINGS = [
|
|
{"code": "corporate.cashflow", "result": "unknown", "stage": 1,
|
|
"evidence_ids": ["EV-1"]},
|
|
{"code": "corporate.title", "result": "unknown", "stage": 1,
|
|
"evidence_ids": ["EV-2"]},
|
|
]
|
|
|
|
|
|
def screening(data_domain="demo"):
|
|
plan = {
|
|
"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,
|
|
}
|
|
decision = prescreen_policy.decide(FINDINGS)
|
|
decision["findings"] = FINDINGS
|
|
decision["missing_inputs"] = ["近六个月银行流水", "设备权属材料"]
|
|
return prescreen.assemble(
|
|
{"id": "run-1", "project_id": PROJECT, "created_at": store.now(),
|
|
"input_snapshot_id": "snap-1", "revision": 3},
|
|
plan, decision, jp.resolve_config(False), data_domain=data_domain)
|
|
|
|
|
|
class SandboxApiTests(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.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0",
|
|
"XINGBANG_LOCAL_MODEL_URL": ""}),
|
|
]
|
|
for p in self.patches:
|
|
p.start()
|
|
app = FastAPI()
|
|
app.add_exception_handler(RequestValidationError, api.validation_handler)
|
|
app.include_router(api.router)
|
|
self.client = TestClient(app)
|
|
self.base = "/api/xingbang/risk-agent/project-assistant/prescreen"
|
|
self.store_run()
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
for p in reversed(self.patches):
|
|
p.stop()
|
|
self.temp.cleanup()
|
|
|
|
def store_run(self, data_domain="demo"):
|
|
self.view = screening(data_domain=data_domain)
|
|
with store.transaction(self.service.db_path) as db:
|
|
prescreen_store.save(db, self.view)
|
|
return self.view
|
|
|
|
def post(self, path, body=None):
|
|
return self.client.post(self.base + path, json=body if body is not None else {},
|
|
headers=HEADERS)
|
|
|
|
def get(self, path):
|
|
return self.client.get(self.base + path, headers=HEADERS)
|
|
|
|
def error(self, response):
|
|
body = response.json()
|
|
self.assertIn("error", body, body)
|
|
return body["error"]
|
|
|
|
# ------------------------------------------------------------ 运行只读视图
|
|
def test_run_view_is_readable(self):
|
|
response = self.get(f"/projects/{PROJECT}/runs/run-1")
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["id"], "run-1")
|
|
self.assertIsNone(body["probability"])
|
|
self.assertIsNone(body["probability_view"]["real_estimate"])
|
|
self.assertIsNone(body["probability_view"]["demo_estimate"],
|
|
"真实运行视图不得携带演示值")
|
|
|
|
def test_unknown_run_is_404_without_leaking_existence(self):
|
|
response = self.get(f"/projects/{PROJECT}/runs/run-missing")
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(self.error(response)["code"], "NOT_FOUND")
|
|
|
|
def test_run_of_another_project_is_404(self):
|
|
response = self.get("/projects/DEMO-C/runs/run-1")
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_foreign_project_is_404(self):
|
|
response = self.get("/projects/OTHER-9/runs/run-1")
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_run_listing_is_newest_first(self):
|
|
with store.transaction(self.service.db_path) as db:
|
|
older = dict(screening(), id="run-old", revision=1,
|
|
created_at="2026-09-29T01:00:00+00:00")
|
|
prescreen_store.save(db, ScreeningRun(**older).model_dump(mode="json"))
|
|
response = self.get(f"/projects/{PROJECT}/runs")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual([item["id"] for item in response.json()["items"]],
|
|
["run-1", "run-old"])
|
|
|
|
def test_available_assumptions_only_reference_existing_dimensions(self):
|
|
response = self.get(f"/projects/{PROJECT}/runs/run-1/assumptions")
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(sorted(item["key"] for item in body["items"]),
|
|
["assume_bank_flow", "assume_title"])
|
|
self.assertEqual(body["scenario_table_version"], "DEMO-TABLE-v1")
|
|
self.assertIn("人工设定", body["label"])
|
|
|
|
trimmed = dict(screening())
|
|
trimmed["dimension_results"] = [item for item in trimmed["dimension_results"]
|
|
if item["id"] != "DC-TITLE"]
|
|
with store.transaction(self.service.db_path) as db:
|
|
prescreen_store.save(db, ScreeningRun(**trimmed).model_dump(mode="json"))
|
|
trimmed_response = self.get(f"/projects/{PROJECT}/runs/run-1/assumptions")
|
|
self.assertEqual([item["key"] for item in trimmed_response.json()["items"]],
|
|
["assume_bank_flow"])
|
|
|
|
# ------------------------------------------------------------ 情景
|
|
def test_simulation_returns_demo_estimate_only(self):
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": ["assume_bank_flow"]})
|
|
self.assertEqual(response.status_code, 201)
|
|
body = response.json()
|
|
demo = body["probability_view"]["demo_estimate"]
|
|
self.assertEqual(demo["value"], 0.58)
|
|
self.assertEqual(demo["scenario_table_version"], "DEMO-TABLE-v1")
|
|
self.assertEqual(demo["scenario_key"], "10")
|
|
self.assertIn("人工设定", demo["label"])
|
|
self.assertIsNone(body["probability_view"]["real_estimate"])
|
|
self.assertEqual(body["baseline_probability_view"]["demo_estimate"]["value"], 0.42)
|
|
rows = {item["id"]: item for item in body["dimensions"]}
|
|
self.assertEqual(rows["DC-CASH"]["scenario_stage"], 3)
|
|
self.assertEqual(rows["DC-TITLE"]["scenario_stage"],
|
|
rows["DC-TITLE"]["baseline_stage"])
|
|
|
|
def test_combination_uses_the_table_not_the_sum(self):
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": ["assume_bank_flow", "assume_title"]})
|
|
self.assertEqual(response.json()["probability_view"]["demo_estimate"]["value"], 0.64)
|
|
|
|
def test_empty_assumptions_return_the_baseline_scenario(self):
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations", {})
|
|
self.assertEqual(response.status_code, 201)
|
|
demo = response.json()["probability_view"]["demo_estimate"]
|
|
self.assertEqual(demo["scenario_key"], "00")
|
|
self.assertEqual(demo["value"], 0.42)
|
|
|
|
def test_simulation_does_not_touch_the_real_run(self):
|
|
before = copy.deepcopy(self.get(f"/projects/{PROJECT}/runs/run-1").json())
|
|
self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": ["assume_title"]})
|
|
after = self.get(f"/projects/{PROJECT}/runs/run-1").json()
|
|
self.assertEqual(before, after, "情景推演不得改动真实运行视图")
|
|
self.assertIsNone(after["probability"])
|
|
self.assertIsNone(after["probability_view"]["demo_estimate"])
|
|
|
|
def test_conflicting_assumption_is_rejected(self):
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": ["assume_everything"]})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertEqual(self.error(response)["code"], "SCENARIO_ASSUMPTION_CONFLICT")
|
|
|
|
def test_production_domain_cannot_use_demo_estimates(self):
|
|
self.store_run(data_domain="production")
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": ["assume_bank_flow"]})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertEqual(self.error(response)["code"], "DEMO_ESTIMATE_FORBIDDEN")
|
|
|
|
def test_stale_revision_is_a_conflict(self):
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": [], "expected_revision": 2})
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(self.error(response)["code"], "REVISION_CONFLICT")
|
|
|
|
def test_unknown_field_in_request_is_rejected(self):
|
|
response = self.post(f"/projects/{PROJECT}/runs/run-1/simulations",
|
|
{"assumptions": [], "probability": 0.9})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertEqual(self.error(response)["code"], "INVALID_REQUEST")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|