feat(v2): T09 预审与推进沙盘——运行只读视图、演示情景与界面守卫
覆盖 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>
This commit is contained in:
Claude Code
@@ -41,7 +41,7 @@ from contracts.prescreen import (
|
||||
S1_ERROR_RETRYABLE,
|
||||
)
|
||||
from services import assistant_store as store
|
||||
from services import fact_conflict, material_import
|
||||
from services import fact_conflict, material_import, prescreen_sandbox, prescreen_store
|
||||
from services.assistant_settings import flags
|
||||
from services.fact_conflict import FactConflictUnresolved
|
||||
from services.project_assistant import ProjectAssistantService
|
||||
@@ -84,6 +84,13 @@ class ResolveInput(Input):
|
||||
expected_revision: int = Field(ge=1)
|
||||
|
||||
|
||||
class SimulationInput(Input):
|
||||
"""T09 沙盘请求：只传假设动作键，不传任何概率或维度值。"""
|
||||
|
||||
assumptions: list[str] = Field(default_factory=list, max_length=8)
|
||||
expected_revision: int | None = Field(default=None, ge=1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 错误信封
|
||||
def _envelope(request: Request, code: ErrorCode, message: str) -> JSONResponse:
|
||||
request_id = request.headers.get("x-request-id") or store.uid("req")
|
||||
@@ -368,3 +375,80 @@ def resolve_fact_conflict(conflict_id: str, payload: ResolveInput, request: Requ
|
||||
return resolved
|
||||
|
||||
return run(request, action)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 运行视图（T09）
|
||||
@router.get("/projects/{project_id}/runs")
|
||||
def list_runs(project_id: str, request: Request, limit: int = 20):
|
||||
"""项目内已装配的预审运行视图列表；没有运行就返回空列表，不造空壳。"""
|
||||
blocked = scope(request, project_id)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
|
||||
def action():
|
||||
with _db() as db:
|
||||
items = prescreen_store.listing(db, project_id, _limit(limit))
|
||||
return {"items": items}
|
||||
|
||||
return run(request, action)
|
||||
|
||||
|
||||
@router.get("/projects/{project_id}/runs/{run_id}")
|
||||
def get_run(project_id: str, run_id: str, request: Request):
|
||||
"""只读运行视图。没有运行就是 404，不返回一个空壳运行充数。"""
|
||||
blocked = scope(request, project_id)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
|
||||
def action():
|
||||
with _db() as db:
|
||||
return prescreen_store.load(db, project_id, run_id)
|
||||
|
||||
return run(request, action)
|
||||
|
||||
|
||||
@router.get("/projects/{project_id}/runs/{run_id}/assumptions")
|
||||
def list_assumptions(project_id: str, run_id: str, request: Request):
|
||||
"""本次方案可用的假设动作。只返回指向该运行真实存在维度的动作（E-22）。"""
|
||||
blocked = scope(request, project_id)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
|
||||
def action():
|
||||
with _db() as db:
|
||||
view = prescreen_store.load(db, project_id, run_id)
|
||||
return {"items": prescreen_sandbox.available_assumptions(view),
|
||||
"scenario_table_version": prescreen_sandbox.DEMO_TABLE_VERSION,
|
||||
"label": prescreen_sandbox.DEMO_LABEL}
|
||||
|
||||
return run(request, action)
|
||||
|
||||
|
||||
@router.post("/projects/{project_id}/runs/{run_id}/simulations", status_code=201)
|
||||
def create_simulation(project_id: str, run_id: str, payload: SimulationInput,
|
||||
request: Request):
|
||||
"""按完整假设组合产出一条情景；真实运行只被读取（BR-11 / AC-11）。
|
||||
|
||||
演示值只在 demo 域返回（E-21）；假设冲突或指向不存在的维度按 E-22 拒绝。
|
||||
请求体里没有任何概率字段——数值一律由服务端查表给出。
|
||||
"""
|
||||
blocked = scope(request, project_id)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
|
||||
def action():
|
||||
with _db() as db:
|
||||
view = prescreen_store.load(db, project_id, run_id)
|
||||
if payload.expected_revision is not None \
|
||||
and view.get("revision") != payload.expected_revision:
|
||||
raise store.Conflict("运行已有新版本，请刷新后重试")
|
||||
try:
|
||||
result = prescreen_sandbox.simulate(view, payload.assumptions)
|
||||
except prescreen_sandbox.DemoEstimateForbidden as exc:
|
||||
return _envelope(request, ErrorCode.DEMO_ESTIMATE_FORBIDDEN, str(exc))
|
||||
except prescreen_sandbox.AssumptionConflict as exc:
|
||||
return _envelope(request, ErrorCode.SCENARIO_ASSUMPTION_CONFLICT, str(exc))
|
||||
prescreen_sandbox.assert_baseline_untouched(view, result)
|
||||
return result
|
||||
|
||||
return run(request, action)
|
||||
|
||||
@@ -105,6 +105,10 @@ class ErrorCode(StrEnum):
|
||||
FACT_CONFLICT_UNRESOLVED = "FACT_CONFLICT_UNRESOLVED"
|
||||
JUDGMENT_CONFIG_INVALID = "JUDGMENT_CONFIG_INVALID"
|
||||
EGRESS_NOT_APPROVED = "EGRESS_NOT_APPROVED"
|
||||
# T09 沙盘与维度：E-20/21/22 都是"可立即纠正的请求问题"，走 422。
|
||||
DIMENSION_SCHEMA_CONFLICT = "DIMENSION_SCHEMA_CONFLICT"
|
||||
DEMO_ESTIMATE_FORBIDDEN = "DEMO_ESTIMATE_FORBIDDEN"
|
||||
SCENARIO_ASSUMPTION_CONFLICT = "SCENARIO_ASSUMPTION_CONFLICT"
|
||||
# 运行级编码：记录在运行体（degradations / judgment）里，**不构成 HTTP 错误**，
|
||||
# 因此刻意不进 S1_ERROR_HTTP；见下方 RUN_LEVEL_CODES。
|
||||
JUDGMENT_UNAVAILABLE = "JUDGMENT_UNAVAILABLE"
|
||||
@@ -133,6 +137,9 @@ S1_ERROR_HTTP: dict[ErrorCode, int] = {
|
||||
ErrorCode.FACT_CONFLICT_UNRESOLVED: 422,
|
||||
ErrorCode.JUDGMENT_CONFIG_INVALID: 422,
|
||||
ErrorCode.EGRESS_NOT_APPROVED: 403,
|
||||
ErrorCode.DIMENSION_SCHEMA_CONFLICT: 422,
|
||||
ErrorCode.DEMO_ESTIMATE_FORBIDDEN: 422,
|
||||
ErrorCode.SCENARIO_ASSUMPTION_CONFLICT: 422,
|
||||
}
|
||||
|
||||
# 这些编码只出现在运行结果里（"HTTP 200 仅表示结果可读取"），永远不会作为
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
"""T09 推进沙盘契约：假设动作与情景结果。
|
||||
|
||||
Owner：本文件（docs/v2/03-detailed-design.md §9.3 Simulation、
|
||||
docs/v2/04-ui-spec.md §9、13-probability-prototype.md）。
|
||||
|
||||
三条不可让渡的规则：
|
||||
1. **模拟不改真实运行**：情景是独立对象，基线以 `baseline_probability_view`
|
||||
副本随行携带；任何情况下都不写回 `ScreeningRun.probability`（BR-11 / AC-11）。
|
||||
2. **演示值只属于 demo 域**：`probability_view.demo_estimate` 在 production
|
||||
域直接构造失败（E-21 / AC-30）；真实估计在 P0 恒为空。
|
||||
3. **组合不是加和**：结果由完整动作组合选出（`selection_hash`），
|
||||
不按单个动作的增量累加（BR-30 / AC-31）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Annotated, Literal
|
||||
|
||||
from pydantic import Field, model_validator
|
||||
|
||||
from contracts.prescreen import (
|
||||
DESC_MAX,
|
||||
ID_MAX,
|
||||
AwareDatetime,
|
||||
Contract,
|
||||
DataDomain,
|
||||
ErrorCode,
|
||||
ProbabilityView,
|
||||
)
|
||||
|
||||
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
||||
Identifier = Annotated[str, Field(min_length=1, max_length=ID_MAX)]
|
||||
|
||||
MAX_ASSUMPTIONS = 8
|
||||
|
||||
|
||||
class SimulationAssumption(Contract):
|
||||
"""一条假设动作。必须写明它改变哪个维度、改到什么阶段、前提是什么。
|
||||
|
||||
`target_dimension_id` 只能指向本次冻结方案里真实存在的维度：假设不能凭空
|
||||
创造维度，也不能顺手改动别的轴（13 §P-17）。
|
||||
"""
|
||||
|
||||
key: Identifier
|
||||
label: str = Field(min_length=1, max_length=DESC_MAX)
|
||||
target_dimension_id: Identifier
|
||||
assumed_stage: int = Field(ge=0, le=4)
|
||||
premise: str = Field(min_length=1, max_length=DESC_MAX)
|
||||
|
||||
|
||||
class ScenarioDimensionState(Contract):
|
||||
"""情景下的维度阶段。基线值与情景值同时给出，界面据此画虚线与实线。"""
|
||||
|
||||
id: Identifier
|
||||
baseline_stage: int | None = Field(default=None, ge=0, le=4)
|
||||
scenario_stage: int | None = Field(default=None, ge=0, le=4)
|
||||
assumed: bool = False
|
||||
|
||||
|
||||
class Simulation(Contract):
|
||||
"""一次情景推演的结果视图。**不是**对真实运行的修改。"""
|
||||
|
||||
id: Identifier
|
||||
run_id: Identifier
|
||||
project_id: Identifier
|
||||
data_domain: DataDomain
|
||||
revision: int = Field(ge=1)
|
||||
# 完整动作组合的摘要：结果必须对应当前选择，而不是"最新到达的响应"
|
||||
selection_hash: str = Field(pattern=SHA256.pattern)
|
||||
status: Literal["queued", "computing", "ready", "failed"]
|
||||
assumptions: list[SimulationAssumption] = Field(default_factory=list,
|
||||
max_length=MAX_ASSUMPTIONS)
|
||||
# 基线副本：模拟结束后界面据此恢复，真实运行未被触碰
|
||||
baseline_probability_view: ProbabilityView
|
||||
probability_view: ProbabilityView | None = None
|
||||
dimensions: list[ScenarioDimensionState] = Field(default_factory=list, max_length=32)
|
||||
remaining_inputs: list[str] = Field(default_factory=list, max_length=200)
|
||||
reason: str = Field(default="", max_length=DESC_MAX)
|
||||
error_code: ErrorCode | None = None
|
||||
frozen_versions: dict = Field(default_factory=dict)
|
||||
created_at: AwareDatetime
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_simulation(self) -> Simulation:
|
||||
if self.status == "ready" and self.probability_view is None:
|
||||
raise ValueError("就绪的情景必须给出概率视图")
|
||||
if self.status == "failed":
|
||||
if self.error_code is None:
|
||||
raise ValueError("失败的情景必须带错误码，不能只留空结果")
|
||||
if self.probability_view is not None:
|
||||
raise ValueError("失败的情景不得显示任何结果")
|
||||
if self.status in ("queued", "computing") and self.probability_view is not None:
|
||||
raise ValueError("情景尚未算完时不得先给一个乐观增幅")
|
||||
# 演示值只在 demo 域：越域直接构造失败，而不是靠调用方记得不传（AC-30）
|
||||
for view in (self.probability_view, self.baseline_probability_view):
|
||||
demo = (view.demo_estimate if view else None) or None
|
||||
if demo is not None and self.data_domain != DataDomain.demo:
|
||||
raise ValueError("演示估计只允许出现在演示数据域（E-21）")
|
||||
keys = [item.key for item in self.assumptions]
|
||||
if len(set(keys)) != len(keys):
|
||||
raise ValueError("同一假设动作不得重复计入组合")
|
||||
targets = [item.target_dimension_id for item in self.assumptions]
|
||||
if len(set(targets)) != len(targets):
|
||||
raise ValueError("一次情景只允许每个维度有一个假设动作，冲突假设应被拒绝")
|
||||
return self
|
||||
|
||||
def unchanged_dimensions(self) -> list[str]:
|
||||
"""未被假设触碰的维度 id：界面必须显示它们没有变化，而不是消失。"""
|
||||
return [item.id for item in self.dimensions if not item.assumed]
|
||||
@@ -0,0 +1,211 @@
|
||||
"""T09 推进沙盘服务：把"假设补件"算成独立情景，不触碰真实运行。
|
||||
|
||||
依据：docs/v2/13-probability-prototype.md §2/§4、04-ui-spec.md §5/§9、
|
||||
03-detailed-design.md §9.3；BR-11/29/30、E-21/22、AC-11/12/30/31/44。
|
||||
|
||||
设计要点：
|
||||
1. **演示值查表，不累加**。结果由完整动作组合选出（`selection_hash`）。表里
|
||||
00/10/01/11 是四条独立条目，不是 42+16+9=67 的算术和；`assert_not_additive`
|
||||
把这条约束写成可执行断言，防止后来者把表换成"每项加分"。
|
||||
2. **真实运行只读**。`simulate()` 接受运行视图 dict，返回新对象；基线的演示值
|
||||
以副本随行携带，任何时候都不写回运行的 `probability`。
|
||||
3. **越域即拒绝**。production 域不得有演示估计（E-21）；假设指向不存在的维度、
|
||||
同一维度两个互斥假设，都按 E-22 拒绝，而不是忽略掉一个。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
from contracts.prescreen import DataDomain, ErrorCode, ProbabilityView
|
||||
from contracts.simulation import Simulation, SimulationAssumption
|
||||
from services import assistant_store as store
|
||||
|
||||
DEMO_TABLE_VERSION = "DEMO-TABLE-v1"
|
||||
DEMO_SOURCE = "synthetic_demo"
|
||||
# 说明文字固定：界面必须紧邻数值展示，避免被当作真实概率（DESIGN.md §38）
|
||||
DEMO_LABEL = "人工设定，仅演示交互；未经历史数据校准"
|
||||
|
||||
STAGE_NAMES = ("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核")
|
||||
|
||||
|
||||
class DemoEstimateForbidden(ValueError):
|
||||
"""当前数据域不允许使用演示概率（E-21）。"""
|
||||
|
||||
|
||||
class AssumptionConflict(ValueError):
|
||||
"""所选推进条件冲突、缺少前提或指向不存在的维度（E-22）。"""
|
||||
|
||||
|
||||
# 假设动作登记表。`dimension_id` 必须与冻结方案里的维度 id 一致，否则拒绝。
|
||||
ASSUMPTIONS: dict[str, dict] = {
|
||||
"assume_bank_flow": {
|
||||
"label": "假设银行流水逐笔匹配",
|
||||
"dimension_id": "DC-CASH",
|
||||
"assumed_stage": 3,
|
||||
"premise": "假设核验支持账期变化解释；只改变回款核验维度",
|
||||
},
|
||||
"assume_title": {
|
||||
"label": "假设设备权属证明一致",
|
||||
"dimension_id": "DC-TITLE",
|
||||
"assumed_stage": 3,
|
||||
"premise": "假设合同、发票与设备清单核验一致；只改变权属维度",
|
||||
},
|
||||
}
|
||||
|
||||
# 演示情景表（13 §3）。key = 流水假设 + 权属假设的组合位。
|
||||
DEMO_SCENARIOS: dict[str, dict] = {
|
||||
"00": {"value": 42, "remaining": ["回款核验仍待核实", "设备权属仍待核实"],
|
||||
"reason": "未选择假设。回款与设备权属证据仍有缺口，保持演示基线。"},
|
||||
"10": {"value": 58, "remaining": ["设备权属仍待核实"],
|
||||
"reason": "假设银行流水逐笔核验后支持账期变化解释，回款不确定性减小；"
|
||||
"设备权属缺口仍在。演示变化 +16 个百分点。"},
|
||||
"01": {"value": 51, "remaining": ["实际回款仍待核实"],
|
||||
"reason": "假设合同、发票与清单一致且经核验，权属争议得到回应；"
|
||||
"实际回款仍缺银行流水。演示变化 +9 个百分点。"},
|
||||
"11": {"value": 64, "remaining": ["仍需整体人工复核"],
|
||||
"reason": "假设回款与权属核验均支持项目解释。组合情景为 64%，变化 +22 "
|
||||
"个百分点；组合影响不简单相加，仍需综合复核其他风险。"},
|
||||
}
|
||||
|
||||
|
||||
def scenario_key(keys) -> str:
|
||||
"""把动作组合压成表键：按登记顺序取位，未选为 0。"""
|
||||
order = list(ASSUMPTIONS)
|
||||
unknown = sorted(set(keys or ()) - set(order))
|
||||
if unknown:
|
||||
raise AssumptionConflict(f"未登记的假设动作：{unknown}")
|
||||
return "".join("1" if item in set(keys or ()) else "0" for item in order)
|
||||
|
||||
|
||||
def selection_hash(run_id: str, revision: int, keys) -> str:
|
||||
"""完整动作组合的摘要。前端只接受与当前选择一致的结果（AC-31）。"""
|
||||
payload = json.dumps({"run_id": run_id, "revision": revision,
|
||||
"key": scenario_key(keys)}, ensure_ascii=False,
|
||||
sort_keys=True)
|
||||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def assert_demo_domain(data_domain) -> None:
|
||||
value = data_domain.value if hasattr(data_domain, "value") else str(data_domain)
|
||||
if value != DataDomain.demo.value:
|
||||
raise DemoEstimateForbidden("当前数据域不允许使用演示概率；真实概率需完成校准后另立验证")
|
||||
|
||||
|
||||
def assert_not_additive() -> None:
|
||||
"""组合值必须来自查表：任何"单项增量相加"的实现都会在这里失败。"""
|
||||
baseline = DEMO_SCENARIOS["00"]["value"]
|
||||
pair = DEMO_SCENARIOS["11"]["value"]
|
||||
added = baseline + (DEMO_SCENARIOS["10"]["value"] - baseline) \
|
||||
+ (DEMO_SCENARIOS["01"]["value"] - baseline)
|
||||
if pair == added:
|
||||
raise AssertionError("演示情景表退化成单项加和，须回到按完整组合查表（BR-30）")
|
||||
|
||||
|
||||
def _view(snapshot_id: str, key: str, as_of: str | None = None) -> ProbabilityView:
|
||||
row = DEMO_SCENARIOS[key]
|
||||
return ProbabilityView(
|
||||
snapshot_id=snapshot_id, as_of=as_of or store.now(),
|
||||
demo_estimate={"value": row["value"] / 100, "source": DEMO_SOURCE,
|
||||
"scenario_table_version": DEMO_TABLE_VERSION,
|
||||
"scenario_key": key, "label": DEMO_LABEL})
|
||||
|
||||
|
||||
def _assumptions_for(keys, dimension_ids) -> list[SimulationAssumption]:
|
||||
known = set(dimension_ids)
|
||||
chosen: list[SimulationAssumption] = []
|
||||
for key in ASSUMPTIONS:
|
||||
if key not in set(keys or ()):
|
||||
continue
|
||||
item = ASSUMPTIONS[key]
|
||||
if item["dimension_id"] not in known:
|
||||
raise AssumptionConflict(
|
||||
f"假设动作 {key} 指向本次方案不存在的维度 {item['dimension_id']}")
|
||||
chosen.append(SimulationAssumption(
|
||||
key=key, label=item["label"], target_dimension_id=item["dimension_id"],
|
||||
assumed_stage=item["assumed_stage"], premise=item["premise"]))
|
||||
return chosen
|
||||
|
||||
|
||||
def _dimension_states(run_view: dict, assumptions) -> list[dict]:
|
||||
"""只覆盖被假设的维度；其余维度原样保留并标为未变（13 §P-17）。"""
|
||||
assumed = {item.target_dimension_id: item for item in assumptions}
|
||||
rows = []
|
||||
for item in run_view.get("dimension_results") or ():
|
||||
hit = assumed.get(item["id"])
|
||||
rows.append({"id": item["id"],
|
||||
"baseline_stage": item.get("stage"),
|
||||
"scenario_stage": hit.assumed_stage if hit else item.get("stage"),
|
||||
"assumed": hit is not None})
|
||||
return rows
|
||||
|
||||
|
||||
def failed(run_view: dict, keys, message: str, code: ErrorCode,
|
||||
revision: int | None = None) -> dict:
|
||||
"""情景失败：保留基线、标未完成，不得回退成"演示表默认值"（13 §44）。"""
|
||||
revision = revision or int(run_view.get("revision") or 1)
|
||||
baseline = _baseline_view(run_view)
|
||||
body = Simulation(
|
||||
id=store.uid("simulation"), run_id=run_view["id"], project_id=run_view["project_id"],
|
||||
data_domain=run_view["data_domain"], revision=revision,
|
||||
selection_hash=selection_hash(run_view["id"], revision, keys),
|
||||
status="failed", assumptions=[], baseline_probability_view=baseline,
|
||||
probability_view=None, dimensions=_dimension_states(run_view, []),
|
||||
remaining_inputs=list(run_view.get("missing_inputs") or ()),
|
||||
reason=message, error_code=code,
|
||||
frozen_versions=dict(run_view.get("versions") or ()), created_at=store.now())
|
||||
return body.model_dump(mode="json")
|
||||
|
||||
|
||||
def available_assumptions(run_view: dict) -> list[dict]:
|
||||
"""列出本次方案可用的假设动作。
|
||||
|
||||
只返回目标维度真实存在于该运行的动作：界面因此不会给出点了必然失败的选项，
|
||||
服务端也不靠"忽略无效动作"来掩饰不匹配（E-22）。
|
||||
"""
|
||||
known = {item["id"] for item in run_view.get("dimension_results") or ()}
|
||||
return [{"key": key, "label": item["label"], "dimension_id": item["dimension_id"],
|
||||
"assumed_stage": item["assumed_stage"], "premise": item["premise"]}
|
||||
for key, item in ASSUMPTIONS.items() if item["dimension_id"] in known]
|
||||
|
||||
|
||||
def _baseline_view(run_view: dict) -> ProbabilityView:
|
||||
snapshot_id = run_view.get("snapshot_id") or run_view.get("input_snapshot_id")
|
||||
if not snapshot_id:
|
||||
raise ValueError("运行缺少输入快照，无法建立情景基线")
|
||||
return _view(snapshot_id, "00")
|
||||
|
||||
|
||||
def simulate(run_view: dict, keys, revision: int | None = None) -> dict:
|
||||
"""按完整动作组合产出一条情景；真实运行只被读取，不被改写。"""
|
||||
assert_demo_domain(run_view["data_domain"])
|
||||
revision = revision or int(run_view.get("revision") or 1)
|
||||
dimension_ids = [item["id"] for item in run_view.get("dimension_results") or ()]
|
||||
assumptions = _assumptions_for(keys, dimension_ids)
|
||||
key = scenario_key(keys)
|
||||
row = DEMO_SCENARIOS[key]
|
||||
baseline = _baseline_view(run_view)
|
||||
|
||||
body = Simulation(
|
||||
id=store.uid("simulation"), run_id=run_view["id"], project_id=run_view["project_id"],
|
||||
data_domain=run_view["data_domain"], revision=revision,
|
||||
selection_hash=selection_hash(run_view["id"], revision, keys),
|
||||
status="ready", assumptions=assumptions,
|
||||
baseline_probability_view=baseline,
|
||||
probability_view=_view(baseline.snapshot_id, key, baseline.as_of),
|
||||
dimensions=_dimension_states(run_view, assumptions),
|
||||
remaining_inputs=list(row["remaining"]), reason=row["reason"],
|
||||
frozen_versions=dict(run_view.get("versions") or ()), created_at=store.now())
|
||||
return body.model_dump(mode="json")
|
||||
|
||||
|
||||
def assert_baseline_untouched(run_view: dict, simulation: dict) -> None:
|
||||
"""装配出口的自检：情景不得改写运行的概率字段（BR-11 / AC-11）。"""
|
||||
if run_view.get("probability") is not None:
|
||||
raise AssertionError("情景推演不得给真实运行写入概率")
|
||||
if (run_view.get("probability_view") or {}).get("real_estimate") is not None:
|
||||
raise AssertionError("情景推演不得给真实运行写入真实估计")
|
||||
baseline = simulation.get("baseline_probability_view") or {}
|
||||
if baseline.get("real_estimate") is not None:
|
||||
raise AssertionError("情景基线副本的真实估计必须为空")
|
||||
@@ -0,0 +1,46 @@
|
||||
"""T09 预审运行视图的只读存储：把装配好的 ScreeningRun 投影存下来供界面读取。
|
||||
|
||||
依据：docs/v2/03-detailed-design.md §9.2、BR-13；AC-12/30。
|
||||
|
||||
**不新建表**：复用既有 `assistant_objects`（新 kind `prescreen_run`），旧库结构不变。
|
||||
运行视图是既有 run 对象的**只读投影**，因此这里只提供保存与读取，
|
||||
不提供就地修改：要改结果必须产生新的运行修订（BR-13：过程修订不覆盖旧结果）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contracts.prescreen import ScreeningRun
|
||||
from services import assistant_store as store
|
||||
|
||||
KIND = "prescreen_run"
|
||||
|
||||
|
||||
def save(db, screening: dict) -> dict:
|
||||
"""保存一条已装配的运行视图；非契约形状直接拒绝，不落库。"""
|
||||
body = ScreeningRun(**screening).model_dump(mode="json")
|
||||
store.put(db, KIND, body, expected=screening.get("_expected_revision"))
|
||||
return body
|
||||
|
||||
|
||||
def load(db, project_id: str, run_id: str) -> dict:
|
||||
"""按项目读取运行视图。不存在或不属于该项目 → KeyError（由接口层转 404）。"""
|
||||
body = store.get(db, KIND, run_id)
|
||||
if body.get("project_id") != project_id:
|
||||
raise KeyError("运行不存在或不属于该项目")
|
||||
return body
|
||||
|
||||
|
||||
def latest(db, project_id: str) -> dict | None:
|
||||
"""项目最近一条运行视图；没有就返回 None，不编造空壳运行。"""
|
||||
rows = [item for item in store.listing(db, KIND, project_id)]
|
||||
if not rows:
|
||||
return None
|
||||
return max(rows, key=lambda item: (item.get("created_at") or "", item.get("revision") or 0))
|
||||
|
||||
|
||||
def listing(db, project_id: str, limit: int = 20) -> list[dict]:
|
||||
"""项目内运行视图，新的在前。用于界面选择要查看/推演的那一条运行。"""
|
||||
rows = [item for item in store.listing(db, KIND, project_id)]
|
||||
rows.sort(key=lambda item: (item.get("created_at") or "", item.get("revision") or 0),
|
||||
reverse=True)
|
||||
return rows[:limit]
|
||||
@@ -0,0 +1,194 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,217 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user