e124e4bbcb
- 重构报销状态注册表、审批流路由与平台风险标记 - 完善管家意图规划器与模型计划构建器全链路 - 新增 OCR Worker 脚本、数据库会话管理与通知状态 - 优化文档中心、日志视图、预算中心与员工管理交互 - 增强工作台摘要、图标资源与全局主题样式 - 补充审批路由、状态注册、OCR 服务与管家规划器测试覆盖
293 lines
9.8 KiB
Python
293 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.core.config import SERVER_DIR
|
|
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
|
|
from app.services.risk_rule_dsl_examples import (
|
|
get_risk_rule_dsl_example,
|
|
list_risk_rule_dsl_examples,
|
|
)
|
|
from app.services.risk_rule_dsl_validator import validate_risk_rule_draft
|
|
from app.services.risk_rule_generation_interpreter import COMPOSITE_RULE_TEMPLATE_KEY
|
|
from app.services.risk_rule_generation_ontology import FIELD_ONTOLOGY
|
|
from app.services.risk_rule_template_executor import RiskRuleTemplateExecutor
|
|
|
|
|
|
def test_dsl_examples_pass_validator() -> None:
|
|
examples = list_risk_rule_dsl_examples()
|
|
|
|
assert {example["code"] for example in examples} == {
|
|
"travel_city_mismatch",
|
|
"lodging_date_outside_range",
|
|
"budget_threshold",
|
|
"duplicate_invoice",
|
|
"entertainment_per_capita_over_limit",
|
|
}
|
|
for example in examples:
|
|
manifest = example["manifest"]
|
|
normalized = validate_risk_rule_draft(
|
|
manifest["params"],
|
|
fields=list(FIELD_ONTOLOGY),
|
|
natural_language=example["natural_language"],
|
|
)
|
|
|
|
assert normalized["template_key"] == COMPOSITE_RULE_TEMPLATE_KEY
|
|
assert normalized["dsl_validation"]["status"] == "passed"
|
|
assert normalized["conditions"]
|
|
assert not normalized.get("keywords")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("code", "hit_contexts", "pass_contexts"),
|
|
[
|
|
(
|
|
"travel_city_mismatch",
|
|
[{"document_info": {"route_cities": ["武汉", "上海"]}}],
|
|
[{"document_info": {"route_cities": ["武汉", "北京"]}}],
|
|
),
|
|
(
|
|
"lodging_date_outside_range",
|
|
[{"document_info": {"stay_start_date": "2026-05-08", "stay_end_date": "2026-05-13"}}],
|
|
[{"document_info": {"stay_start_date": "2026-05-10", "stay_end_date": "2026-05-12"}}],
|
|
),
|
|
(
|
|
"budget_threshold",
|
|
[{"budget_context": {"remaining_amount": "1000.00"}}],
|
|
[{"budget_context": {"remaining_amount": "2000.00"}}],
|
|
),
|
|
(
|
|
"duplicate_invoice",
|
|
[
|
|
{"document_info": {"invoice_no": "INV-20260530-001"}},
|
|
{"document_info": {"invoice_no": "INV-20260530-001"}},
|
|
],
|
|
[
|
|
{"document_info": {"invoice_no": "INV-20260530-001"}},
|
|
{"document_info": {"invoice_no": "INV-20260530-002"}},
|
|
],
|
|
),
|
|
(
|
|
"entertainment_per_capita_over_limit",
|
|
[],
|
|
[],
|
|
),
|
|
],
|
|
)
|
|
def test_dsl_examples_execute_hit_and_pass(
|
|
code: str,
|
|
hit_contexts: list[dict],
|
|
pass_contexts: list[dict],
|
|
) -> None:
|
|
example = get_risk_rule_dsl_example(code)
|
|
assert example is not None
|
|
executor = RiskRuleTemplateExecutor()
|
|
claim = _claim(amount=Decimal("1200.00"))
|
|
if code == "entertainment_per_capita_over_limit":
|
|
claim.expense_type = "业务招待费"
|
|
claim.reason = "客户接待餐费"
|
|
claim.attendee_count = 2
|
|
claim.per_capita_amount = Decimal("600.00")
|
|
|
|
hit_result = executor.evaluate(example["manifest"], claim=claim, contexts=hit_contexts)
|
|
if code == "entertainment_per_capita_over_limit":
|
|
claim.per_capita_amount = Decimal("400.00")
|
|
pass_result = executor.evaluate(example["manifest"], claim=claim, contexts=pass_contexts)
|
|
|
|
assert hit_result is not None
|
|
assert pass_result is None
|
|
|
|
|
|
def test_duplicate_invoice_example_reports_duplicate_evidence() -> None:
|
|
example = get_risk_rule_dsl_example("duplicate_invoice")
|
|
assert example is not None
|
|
|
|
result = RiskRuleTemplateExecutor().evaluate(
|
|
example["manifest"],
|
|
claim=_claim(),
|
|
contexts=[
|
|
{"document_info": {"invoice_no": "INV-DUP-001"}},
|
|
{"document_info": {"invoice_no": "INV-DUP-001"}},
|
|
],
|
|
)
|
|
|
|
assert result is not None
|
|
condition = result["evidence"]["conditions"][0]
|
|
assert condition["operator"] == "duplicate_value"
|
|
assert condition["duplicates"] == ["inv-dup-001"]
|
|
|
|
|
|
def test_date_rule_uses_application_month_before_ticket_item_date() -> None:
|
|
claim = _claim()
|
|
claim.trip_start_date = None
|
|
claim.trip_end_date = None
|
|
claim.occurred_at = datetime(2026, 2, 20, tzinfo=UTC)
|
|
claim.items[0].item_date = date(2026, 2, 20)
|
|
claim.risk_flags_json = [
|
|
{
|
|
"source": "application_handoff",
|
|
"application_detail": {
|
|
"application_time": "6月",
|
|
},
|
|
}
|
|
]
|
|
|
|
manifest = {
|
|
"template_key": COMPOSITE_RULE_TEMPLATE_KEY,
|
|
"params": {
|
|
"template_key": COMPOSITE_RULE_TEMPLATE_KEY,
|
|
"conditions": [
|
|
{
|
|
"id": "ticket_date_outside_trip",
|
|
"operator": "date_outside_range",
|
|
"date_fields": ["item.item_date", "attachment.issue_date"],
|
|
"range_start_fields": ["claim.trip_start_date"],
|
|
"range_end_fields": ["claim.trip_end_date"],
|
|
"tolerance_days": 0,
|
|
}
|
|
],
|
|
"hit_logic": "ticket_date_outside_trip",
|
|
"condition_summary": "ticket date outside trip window",
|
|
},
|
|
}
|
|
|
|
result = RiskRuleTemplateExecutor().evaluate(
|
|
manifest,
|
|
claim=claim,
|
|
contexts=[{"document_info": {"issue_date": "2026-02-20"}}],
|
|
)
|
|
|
|
assert result is not None
|
|
condition = result["evidence"]["conditions"][0]
|
|
assert condition["range_start"] == "2026-06-01"
|
|
assert condition["range_end"] == "2026-06-30"
|
|
assert condition["outside_dates"] == ["2026-02-20"]
|
|
|
|
|
|
def test_application_context_values_are_available_to_composite_rules() -> None:
|
|
claim = _claim(amount=Decimal("3000.00"))
|
|
claim.risk_flags_json = [
|
|
{
|
|
"source": "application_link",
|
|
"application_claim_id": "application-ctx-1",
|
|
"application_claim_no": "AP-202606-CTX",
|
|
"application_detail": {
|
|
"application_amount": "3000",
|
|
"application_expense_type": "office",
|
|
},
|
|
}
|
|
]
|
|
manifest = {
|
|
"template_key": COMPOSITE_RULE_TEMPLATE_KEY,
|
|
"params": {
|
|
"template_key": COMPOSITE_RULE_TEMPLATE_KEY,
|
|
"conditions": [
|
|
{
|
|
"id": "application_present",
|
|
"operator": "exists_any",
|
|
"fields": ["application.id", "application.claim_no"],
|
|
}
|
|
],
|
|
"hit_logic": "application_present",
|
|
"condition_summary": "application exists",
|
|
},
|
|
}
|
|
|
|
result = RiskRuleTemplateExecutor().evaluate(manifest, claim=claim, contexts=[])
|
|
|
|
assert result is not None
|
|
condition = result["evidence"]["conditions"][0]
|
|
assert condition["values"] == ["application-ctx-1", "AP-202606-CTX"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("file_name", "expense_type", "amount"),
|
|
[
|
|
("risk.application.meal_high_value_without_preapproval.json", "meal", Decimal("501.00")),
|
|
("risk.application.office_bulk_without_purchase.json", "office", Decimal("2001.00")),
|
|
("risk.application.large_expense_without_preapproval.json", "software", Decimal("2001.00")),
|
|
],
|
|
)
|
|
def test_preapproval_amount_rules_hit_without_linked_application(
|
|
file_name: str,
|
|
expense_type: str,
|
|
amount: Decimal,
|
|
) -> None:
|
|
claim = _claim(amount=amount)
|
|
claim.expense_type = expense_type
|
|
manifest = _load_rule_manifest(file_name)
|
|
|
|
result = RiskRuleTemplateExecutor().evaluate(manifest, claim=claim, contexts=[])
|
|
|
|
assert result is not None
|
|
assert result["evidence"]["condition_results"]["amount_exceeds_preapproval_threshold"] is True
|
|
assert result["evidence"]["condition_results"]["application_present"] is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("file_name", "expense_type", "amount"),
|
|
[
|
|
("risk.application.meal_high_value_without_preapproval.json", "entertainment", Decimal("800.00")),
|
|
("risk.application.office_bulk_without_purchase.json", "office", Decimal("2600.00")),
|
|
("risk.application.large_expense_without_preapproval.json", "software", Decimal("2600.00")),
|
|
],
|
|
)
|
|
def test_preapproval_amount_rules_skip_when_application_is_linked(
|
|
file_name: str,
|
|
expense_type: str,
|
|
amount: Decimal,
|
|
) -> None:
|
|
claim = _claim(amount=amount)
|
|
claim.expense_type = expense_type
|
|
claim.risk_flags_json = [
|
|
{
|
|
"source": "application_link",
|
|
"application_claim_id": "application-linked-ok",
|
|
"application_claim_no": "AP-202606-OK",
|
|
}
|
|
]
|
|
manifest = _load_rule_manifest(file_name)
|
|
|
|
result = RiskRuleTemplateExecutor().evaluate(manifest, claim=claim, contexts=[])
|
|
|
|
assert result is None
|
|
|
|
|
|
def _claim(*, amount: Decimal = Decimal("1000.00")) -> ExpenseClaim:
|
|
claim = ExpenseClaim(
|
|
claim_no="TEST-RISK-RULE-DSL",
|
|
employee_name="测试员工",
|
|
department_name="测试部门",
|
|
expense_type="差旅费",
|
|
reason="北京出差项目支持",
|
|
location="北京",
|
|
amount=amount,
|
|
currency="CNY",
|
|
invoice_count=2,
|
|
occurred_at=datetime(2026, 5, 10, tzinfo=UTC),
|
|
status="draft",
|
|
)
|
|
claim.trip_start_date = date(2026, 5, 10)
|
|
claim.trip_end_date = date(2026, 5, 12)
|
|
claim.items = [
|
|
ExpenseClaimItem(
|
|
item_date=date(2026, 5, 10),
|
|
item_type="travel",
|
|
item_reason="北京出差",
|
|
item_location="北京",
|
|
item_amount=amount,
|
|
invoice_id="INV-ITEM-001",
|
|
)
|
|
]
|
|
return claim
|
|
|
|
|
|
def _load_rule_manifest(file_name: str) -> dict:
|
|
path = Path(SERVER_DIR) / "rules" / "risk-rules" / file_name
|
|
return json.loads(path.read_text(encoding="utf-8"))
|