X-Financial/server/tests/test_risk_rule_dsl_validator.py

from __future__ import annotations

from datetime import UTC, datetime
from decimal import Decimal

from app.models.financial_record import ExpenseClaim
from app.services.risk_rule_dsl_validator import validate_risk_rule_draft
from app.services.risk_rule_generation_ontology import RiskRuleField
from app.services.risk_rule_template_executor import RiskRuleTemplateExecutor


FIELDS = [
    RiskRuleField("claim.location", "申报地点", "text", "claim", ("目的地", "城市")),
    RiskRuleField("attachment.hotel_city", "住宿城市", "text", "attachment", ("酒店城市",)),
    RiskRuleField("attachment.route_cities", "行程城市", "list", "attachment", ("交通票城市",)),
    RiskRuleField("claim.amount", "申报金额", "number", "claim", ("金额",)),
    RiskRuleField("budget.remaining_amount", "预算可用余额", "number", "budget", ("预算余额",)),
    RiskRuleField("claim.reason", "报销事由", "text", "claim", ("事由",)),
    RiskRuleField("attachment.ocr_text", "票据全文", "text", "attachment", ("OCR",)),
]


def test_validator_rewrites_city_keyword_rule_to_structured_compare() -> None:
    draft = {
        "template_key": "keyword_match_v1",
        "field_keys": ["attachment.hotel_city", "attachment.route_cities", "claim.location"],
        "keywords": ["绕行", "跨城办事"],
        "condition_summary": "检查住宿城市、申报地点、行程城市是否出现规则描述中的风险关键词",
    }

    normalized = validate_risk_rule_draft(
        draft,
        fields=FIELDS,
        natural_language="差旅报销时，住宿或交通票据城市必须与申报目的地一致，未说明绕行时进入复核。",
    )

    assert normalized["template_key"] == "field_compare_v1"
    assert normalized["semantic_type"] == "travel_route_city_consistency"
    assert normalized["keywords"] == []
    assert "city_rule_normalized_to_structured_compare" in normalized["dsl_validation"]["issues"]


def test_validator_rewrites_budget_keyword_rule_to_numeric_compare() -> None:
    draft = {
        "template_key": "keyword_match_v1",
        "field_keys": ["claim.amount", "budget.remaining_amount", "claim.reason"],
        "keywords": ["超预算"],
        "condition_summary": "检查金额字段是否出现预算风险关键词",
    }

    normalized = validate_risk_rule_draft(
        draft,
        fields=FIELDS,
        natural_language="费用申请时，若申报金额超过预算可用余额，则提示风险并要求补充审批说明。",
    )

    assert normalized["template_key"] == "composite_rule_v1"
    assert normalized["keywords"] == []
    assert normalized["conditions"][0]["operator"] == "numeric_compare"
    assert normalized["conditions"][0]["left_fields"] == ["claim.amount"]
    assert normalized["conditions"][0]["right_fields"] == ["budget.remaining_amount"]
    assert "风险关键词" not in normalized["condition_summary"]


def test_validator_builds_numeric_condition_for_empty_composite_fallback() -> None:
    normalized = validate_risk_rule_draft(
        {"template_key": "composite_rule_v1", "field_keys": ["claim.amount", "budget.remaining_amount"]},
        fields=FIELDS,
        natural_language="费用申请时，若申报金额超过预算可用余额，则提示风险。",
    )

    assert normalized["template_key"] == "composite_rule_v1"
    assert normalized["conditions"][0]["operator"] == "numeric_compare"
    assert normalized["hit_logic"] == {"all": ["amount_exceeds_budget"]}
    assert "empty_composite_rule_built_from_structured_fields" in normalized["dsl_validation"]["issues"]


def test_numeric_compare_condition_executes_against_budget_context() -> None:
    manifest = {
        "template_key": "composite_rule_v1",
        "params": {
            "template_key": "composite_rule_v1",
            "conditions": [
                {
                    "id": "amount_exceeds_budget",
                    "operator": "numeric_compare",
                    "left_fields": ["claim.amount"],
                    "right_fields": ["budget.remaining_amount"],
                    "compare": "gt",
                }
            ],
            "hit_logic": {"all": ["amount_exceeds_budget"]},
            "message_template": "申报金额超过预算可用余额。",
        },
    }
    claim = ExpenseClaim(
        claim_no="TEST-BUDGET-RISK",
        employee_name="测试员工",
        department_name="测试部门",
        expense_type="差旅费",
        reason="北京出差",
        location="北京",
        amount=Decimal("1200.00"),
        currency="CNY",
        invoice_count=0,
        occurred_at=datetime(2026, 5, 30, tzinfo=UTC),
        status="draft",
    )

    result = RiskRuleTemplateExecutor().evaluate(
        manifest,
        claim=claim,
        contexts=[{"budget_context": {"remaining_amount": "1000.00"}}],
    )
    assert result is not None
    assert result["message"] == "申报金额超过预算可用余额。"
    assert result["evidence"]["condition_results"]["amount_exceeds_budget"] is True

    claim.amount = Decimal("800.00")
    assert RiskRuleTemplateExecutor().evaluate(
        manifest,
        claim=claim,
        contexts=[{"budget_context": {"remaining_amount": "1000.00"}}],
    ) is None