109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from app.services.risk_rule_flow_diagram import (
|
||
|
|
RiskRuleFlowDiagramField,
|
||
|
|
RiskRuleFlowDiagramRenderer,
|
||
|
|
build_risk_rule_flow_diagram_spec,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
FIELDS = (
|
||
|
|
RiskRuleFlowDiagramField(key="claim.amount", label="申请金额"),
|
||
|
|
RiskRuleFlowDiagramField(key="budget.remaining_amount", label="可用预算"),
|
||
|
|
RiskRuleFlowDiagramField(key="claim.reason", label="申请事由"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_flow_diagram_spec_prefers_flow_model_nodes() -> None:
|
||
|
|
spec = build_risk_rule_flow_diagram_spec(
|
||
|
|
{"name": "预算余额校验", "metadata": {"condition_summary": "legacy summary"}},
|
||
|
|
fields=FIELDS,
|
||
|
|
domain_label="差旅费",
|
||
|
|
severity="high",
|
||
|
|
severity_label="高风险",
|
||
|
|
flow_model={
|
||
|
|
"nodes": [
|
||
|
|
{"id": "start", "type": "start", "title": "开始", "description": "费用申请提交"},
|
||
|
|
{
|
||
|
|
"id": "evidence",
|
||
|
|
"type": "evidence",
|
||
|
|
"title": "字段事实",
|
||
|
|
"description": "读取申请金额与可用预算",
|
||
|
|
"fields": ["claim.amount", "budget.remaining_amount"],
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"id": "amount_exceeds_budget",
|
||
|
|
"type": "decision",
|
||
|
|
"title": "金额超过预算",
|
||
|
|
"description": "申请金额大于可用预算余额",
|
||
|
|
},
|
||
|
|
{"id": "pass", "type": "pass", "description": "预算充足，继续流转"},
|
||
|
|
{"id": "hit", "type": "risk", "description": "进入预算复核"},
|
||
|
|
],
|
||
|
|
"metadata": {"hit_logic": "amount_exceeds_budget"},
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert spec.start == "费用申请提交"
|
||
|
|
assert spec.fact_lines[:2] == (
|
||
|
|
"A=申请金额[claim.amount]",
|
||
|
|
"B=可用预算[budget.remaining_amount]",
|
||
|
|
)
|
||
|
|
assert spec.condition_lines == ("金额超过预算: 申请金额大于可用预算余额",)
|
||
|
|
assert spec.hit_logic == "amount_exceeds_budget"
|
||
|
|
svg = RiskRuleFlowDiagramRenderer().render(spec)
|
||
|
|
assert "金额超过预算" in svg
|
||
|
|
assert "#dc2626" in svg
|
||
|
|
|
||
|
|
|
||
|
|
def test_flow_diagram_spec_falls_back_to_dsl_when_flow_model_missing() -> None:
|
||
|
|
spec = build_risk_rule_flow_diagram_spec(
|
||
|
|
{
|
||
|
|
"name": "重复发票校验",
|
||
|
|
"params": {
|
||
|
|
"conditions": [
|
||
|
|
{
|
||
|
|
"id": "same_invoice_no_repeated",
|
||
|
|
"operator": "duplicate_value",
|
||
|
|
"fields": ["claim.reason"],
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"hit_logic": {"all": ["same_invoice_no_repeated"]},
|
||
|
|
},
|
||
|
|
"metadata": {"condition_summary": "发票号重复时命中"},
|
||
|
|
},
|
||
|
|
fields=FIELDS,
|
||
|
|
domain_label="通用",
|
||
|
|
severity="medium",
|
||
|
|
severity_label="中风险",
|
||
|
|
flow_model={},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert spec.condition_lines == ("same_invoice_no_repeated: 申请事由 出现重复值",)
|
||
|
|
assert spec.hit_logic == "same_invoice_no_repeated"
|
||
|
|
assert "发票号重复时命中" in spec.basis
|
||
|
|
|
||
|
|
|
||
|
|
def test_flow_diagram_spec_compresses_too_many_decision_nodes() -> None:
|
||
|
|
nodes = [{"id": "start", "type": "start", "description": "提交单据"}]
|
||
|
|
nodes.extend(
|
||
|
|
{
|
||
|
|
"id": f"condition_{index}",
|
||
|
|
"type": "decision",
|
||
|
|
"title": f"判断{index}",
|
||
|
|
"description": f"第{index}个判断条件",
|
||
|
|
}
|
||
|
|
for index in range(1, 7)
|
||
|
|
)
|
||
|
|
spec = build_risk_rule_flow_diagram_spec(
|
||
|
|
{"name": "复杂规则"},
|
||
|
|
fields=FIELDS,
|
||
|
|
domain_label="通用",
|
||
|
|
severity="low",
|
||
|
|
severity_label="低风险",
|
||
|
|
flow_model={"nodes": nodes},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert len(spec.condition_lines) == 4
|
||
|
|
assert "另有 2 个判断节点" in spec.condition_lines[-1]
|