selfrelease
74e35fed96
- 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景
101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
"""配置驱动场景（B 快捷通道）测试：原语库、YAML 加载与校验、ConfigScenario 扫描。
|
|
|
|
纯逻辑，无需数据库。验证「新增 YAML → 自动注册 → 可扫描产出线索草稿」链路。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.scenarios.config import (
|
|
ScenarioConfig,
|
|
load_yaml_scenarios,
|
|
)
|
|
from app.scenarios.primitives import cliff_drop, ratio_threshold, threshold_edge
|
|
from app.scenarios.registry import SCENARIO_REGISTRY, get_scenario
|
|
|
|
|
|
# ---------- 原语库 ----------
|
|
|
|
def test_threshold_edge_primitive_hits():
|
|
rows = [{"refund_amount": 450000, "customer": f"C{i}"} for i in range(5)]
|
|
res = threshold_edge(
|
|
rows,
|
|
{"amount_field": "refund_amount", "threshold": 500000, "key_field": "customer"},
|
|
)
|
|
assert res.hit
|
|
assert res.score > 0
|
|
assert res.evidence["near_count"] == 5
|
|
assert len(res.subjects["keys"]) == 5
|
|
|
|
|
|
def test_threshold_edge_primitive_no_hit_when_spread():
|
|
rows = [{"refund_amount": 100000}, {"refund_amount": 900000}]
|
|
res = threshold_edge(rows, {"amount_field": "refund_amount", "threshold": 500000})
|
|
assert not res.hit
|
|
assert res.score == 0.0
|
|
|
|
|
|
def test_threshold_edge_requires_positive_threshold():
|
|
with pytest.raises(ValueError):
|
|
threshold_edge([], {"amount_field": "x", "threshold": 0})
|
|
|
|
|
|
def test_cliff_drop_primitive():
|
|
rows = [{"m": 0, "v": 1.0}, {"m": 1, "v": 0.95}, {"m": 2, "v": 0.1}]
|
|
res = cliff_drop(rows, {"index_field": "m", "value_field": "v"})
|
|
assert res.hit
|
|
assert res.evidence["cliff_at"] == 2
|
|
|
|
|
|
def test_ratio_threshold_primitive():
|
|
rows = [{"z": 0.9}, {"z": 0.95}]
|
|
res = ratio_threshold(
|
|
rows, {"value_field": "z", "agg": "mean", "op": ">=", "threshold": 0.8}
|
|
)
|
|
assert res.hit
|
|
assert res.score == 0.8
|
|
|
|
|
|
# ---------- YAML 配置校验 ----------
|
|
|
|
def test_scenario_config_validation_error():
|
|
with pytest.raises(ValidationError):
|
|
ScenarioConfig(code="R99") # 缺 title/risk_domain/label/detector
|
|
|
|
|
|
def test_load_yaml_scenarios_registers_r10():
|
|
load_yaml_scenarios() # 加载内置 rules/*.yaml（含 R10 示例）
|
|
assert "R10" in SCENARIO_REGISTRY
|
|
cls = get_scenario("R10")
|
|
assert cls is not None
|
|
assert cls.label == "退款拆分"
|
|
assert cls.risk_domain == "收入"
|
|
|
|
|
|
# ---------- ConfigScenario 扫描 ----------
|
|
|
|
def test_config_scenario_scan_produces_clue_draft():
|
|
load_yaml_scenarios()
|
|
cls = get_scenario("R10")
|
|
rows = [{"refund_amount": 460000 + i * 5000, "customer": f"政企{i}"} for i in range(6)]
|
|
outcome = cls(rows=rows).scan(None)
|
|
assert outcome.scanned_count == 6
|
|
assert len(outcome.drafts) == 1
|
|
draft = outcome.drafts[0]
|
|
assert draft.score > 0
|
|
assert "退款" in draft.rationale # rationale_template 生效
|
|
assert "keys" in draft.subjects
|
|
|
|
|
|
def test_config_scenario_no_draft_when_clean():
|
|
load_yaml_scenarios()
|
|
cls = get_scenario("R10")
|
|
rows = [
|
|
{"refund_amount": 100000, "customer": "A"},
|
|
{"refund_amount": 2_000_000, "customer": "B"},
|
|
]
|
|
outcome = cls(rows=rows).scan(None)
|
|
assert outcome.drafts == []
|