selfrelease
74e35fed96
- 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""场景插件框架测试：注册表、场景 scan() 产出、/scenarios API（纯逻辑，无需数据库）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.scenarios.base import ScanOutcome
|
|
from app.scenarios.churn_fraud import ChurnFraudScenario, CohortPoint
|
|
from app.scenarios.registry import SCENARIO_REGISTRY, get_scenario, scenario_meta
|
|
from app.scenarios.split_contract import ContractRecord, SplitContractScenario
|
|
|
|
|
|
def test_registry_contains_builtin_scenarios():
|
|
# 导入 app.scenarios.* 已通过自动发现触发 @register_scenario 注册 R8/R9
|
|
assert "R8" in SCENARIO_REGISTRY
|
|
assert "R9" in SCENARIO_REGISTRY
|
|
assert get_scenario("R8") is SplitContractScenario
|
|
assert get_scenario("R9") is ChurnFraudScenario
|
|
|
|
|
|
def test_scenario_meta_fields():
|
|
meta = {m["code"]: m for m in scenario_meta()}
|
|
assert meta["R8"]["label"] == "政企拆单"
|
|
assert meta["R8"]["risk_domain"] == "收入"
|
|
assert meta["R9"]["label"] == "养卡骗补"
|
|
assert set(meta["R8"]) == {"code", "title", "label", "risk_domain"}
|
|
|
|
|
|
def test_split_scenario_scan_hits():
|
|
contracts = [ContractRecord(f"C{i}", f"CUST{i}", 850000) for i in range(8)]
|
|
scenario = SplitContractScenario(
|
|
contracts, approval_threshold=1_000_000, shared_controller=True
|
|
)
|
|
outcome = scenario.scan(None)
|
|
assert isinstance(outcome, ScanOutcome)
|
|
assert outcome.scanned_count == 8
|
|
assert len(outcome.drafts) == 1
|
|
draft = outcome.drafts[0]
|
|
assert draft.score >= scenario.score_threshold
|
|
assert draft.amount_involved > 0
|
|
assert draft.evidence["shared_controller"] is True
|
|
|
|
|
|
def test_split_scenario_no_draft_when_clean():
|
|
contracts = [ContractRecord("C1", "A", 100000), ContractRecord("C2", "B", 3_000_000)]
|
|
outcome = SplitContractScenario(contracts, approval_threshold=1_000_000).scan(None)
|
|
assert outcome.drafts == []
|
|
assert outcome.scanned_count == 2
|
|
|
|
|
|
def test_churn_scenario_scan_produces_draft():
|
|
curve = [CohortPoint(0, 1.0), CohortPoint(1, 0.95), CohortPoint(2, 0.1)]
|
|
scenario = ChurnFraudScenario(
|
|
retention_curve=curve,
|
|
commission_paid=300000,
|
|
active_ratio=0.05,
|
|
zero_usage_ratio=0.9,
|
|
channel_key="CH-001",
|
|
)
|
|
outcome = scenario.scan(None)
|
|
assert outcome.scanned_count == 3
|
|
assert len(outcome.drafts) == 1
|
|
draft = outcome.drafts[0]
|
|
assert draft.subjects["channel"] == "CH-001"
|
|
assert draft.score >= scenario.score_threshold
|
|
|
|
|
|
def test_scenarios_api_endpoint():
|
|
client = TestClient(app)
|
|
resp = client.get("/scenarios")
|
|
assert resp.status_code == 200
|
|
codes = {item["code"] for item in resp.json()}
|
|
assert {"R8", "R9"} <= codes
|