from __future__ import annotations import json from datetime import UTC, date, datetime from decimal import Decimal from types import SimpleNamespace import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.core.agent_enums import ( AgentAssetDomain, AgentAssetStatus, AgentAssetType, AgentReviewStatus, ) from app.db.base import Base from app.models.agent_asset import AgentAsset from app.models.employee import Employee from app.models.financial_record import ExpenseClaim, ExpenseClaimItem from app.schemas.agent_asset import ( AgentAssetReviewCreate, AgentAssetRiskRuleGenerateRequest, AgentAssetRiskRuleReportRequest, AgentAssetRiskRuleSampleTestRequest, AgentAssetRiskRuleScenarioTestRequest, AgentAssetRiskRuleSimulationRequest, ) from app.services.agent_asset_rule_library import AgentAssetRuleLibraryManager from app.services.agent_asset_spreadsheet import RISK_RULES_LIBRARY from app.services.agent_assets import AgentAssetService from app.services.agent_foundation_risk_rules import AgentFoundationRiskRuleMixin from app.services.expense_claim_platform_risk import ExpenseClaimPlatformRiskMixin from app.services.risk_rule_flow_diagram import ( RiskRuleFlowDiagramRenderer, RiskRuleFlowDiagramSpec, ) from app.services.risk_rule_generation import RiskRuleGenerationService from app.services.risk_rule_generation_jobs import RiskRuleGenerationJobService from app.services.risk_rule_manifest_normalizer import normalize_risk_rule_manifest from app.services.risk_rule_scoring import calculate_risk_rule_score, risk_level_from_score from app.services.risk_rule_template_executor import RiskRuleTemplateExecutor class NullRuntimeChatService: def complete(self, *args, **kwargs) -> None: return None class TravelRouteSemanticRuntimeChatService: def complete(self, *args, **kwargs) -> str: return json.dumps( { "name": "差旅票据路线一致性校验", "description": "交通票或住宿票据城市需要与申报行程形成一致关系。", "template_key": "field_compare_v1", "semantic_type": "travel_route_city_consistency", "field_keys": [ "attachment.route_cities", "attachment.hotel_city", "claim.location", "item.item_location", "employee.location", "claim.reason", "item.item_reason", ], "condition_summary": ( "A=交通票行程城市∪住宿发票城市，B=申报目的地∪明细发生地点，" "C=员工常驻地；A与B无交集且无合理说明，或A出现B∪C之外城市时命中。" ), "keywords": [], "exception_keywords": ["绕行", "跨城办事", "临时改签"], "flow": { "start": "差旅报销提交", "evidence": "读取票据城市、申报地点、明细地点和报销事由", "decision": "票据城市是否覆盖申报行程，是否出现额外中转城市", "pass": "票据城市与申报行程一致", "fail": "票据城市与申报行程不一致，进入复核", }, }, ensure_ascii=False, ) def build_session() -> Session: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(bind=engine) session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False) return session_factory() def test_generate_risk_rule_asset_creates_draft_json_rule(tmp_path) -> None: with build_session() as db: service = RiskRuleGenerationService( db, rule_library_manager=AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules"), runtime_chat_service=NullRuntimeChatService(), ) asset_id = service.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="travel", rule_title="差旅住宿城市一致性校验", risk_level="high", natural_language="住宿城市必须出现在本次差旅行程城市中，否则提示高风险。", ), actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None assert asset.name == "差旅住宿城市一致性校验" assert asset.status == AgentAssetStatus.DRAFT.value assert asset.config_json["detail_mode"] == "json_risk" assert asset.config_json["evaluator"] == "template_rule" assert asset.config_json["expense_category"] == "travel" assert asset.config_json["risk_category"] == "差旅费" assert asset.config_json["business_stage"] == "reimbursement" assert asset.config_json["business_stage_label"] == "费用报销" assert asset.scenario_json == ["差旅费"] assert asset.current_version == "v0.1.0" file_name = asset.config_json["rule_document"]["file_name"] rule_path = tmp_path / "rules" / RISK_RULES_LIBRARY / file_name payload = json.loads(rule_path.read_text(encoding="utf-8")) assert payload["rule_code"] == asset.code assert payload["name"] == "差旅住宿城市一致性校验" assert payload["applies_to"]["expense_categories"] == ["travel"] assert payload["applies_to"]["business_stages"] == ["reimbursement"] assert payload["risk_category"] == "差旅费" assert payload["metadata"]["expense_category"] == "travel" assert payload["metadata"]["business_stage"] == "reimbursement" assert payload["metadata"]["business_stage_label"] == "费用报销" assert payload["metadata"]["rule_title"] == "差旅住宿城市一致性校验" assert isinstance(payload["metadata"]["risk_score"], int) assert payload["metadata"]["risk_level"] == payload["outcomes"]["fail"]["severity"] assert asset.config_json["risk_score"] == payload["metadata"]["risk_score"] assert payload["outcomes"]["fail"]["severity"] == "high" assert payload["template_key"] == "field_compare_v1" assert payload["metadata"]["natural_language"].startswith("住宿城市") assert payload["inputs"]["fields"] assert payload["flow_diagram_svg"].startswith(" None: assert risk_level_from_score(30) == "low" assert risk_level_from_score(31) == "medium" assert risk_level_from_score(61) == "high" assert risk_level_from_score(81) == "critical" result = calculate_risk_rule_score( natural_language="同一发票号码重复报销时禁止提交并进入审计复核。", draft={ "template_key": "composite_rule_v1", "field_keys": ["attachment.invoice_no", "claim.amount"], "conditions": [{"id": "duplicate_invoice", "operator": "overlap"}], "risk_scoring_evidence": { "impact_level": "critical", "violation_certainty": "critical", "evidence_strength": "high", "exception_dependence": "medium", "control_action": "block", "business_sensitivity": "critical", }, }, fields=[], expense_category="travel", expense_category_label="差旅费", requires_attachment=True, ) assert result["score"] >= 81 assert result["level"] == "critical" assert result["level_label"] == "极高风险" def test_generate_expense_application_risk_rule_marks_business_stage(tmp_path) -> None: with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") service = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=NullRuntimeChatService(), ) asset_id = service.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, business_stage="expense_application", expense_category="travel", rule_title="差旅申请预算余额校验", natural_language="费用申请时，若差旅申请金额超过可用预算余额，则提示风险并要求补充审批说明。", ), actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None assert asset.config_json["business_stage"] == "expense_application" assert asset.config_json["business_stage_label"] == "费用申请" payload = manager.read_rule_library_json( library=RISK_RULES_LIBRARY, file_name=asset.config_json["rule_document"]["file_name"], ) assert payload["applies_to"]["business_stages"] == ["expense_application"] assert payload["metadata"]["business_stage_label"] == "费用申请" assert payload["params"]["business_stage_label"] == "费用申请" def test_risk_score_model_keeps_explicit_low_control_rules_low() -> None: field_keys = ["attachment.invoice_no", "attachment.goods_name", "claim.reason"] result = calculate_risk_rule_score( natural_language=( "差旅报销时，票据已上传但发票号码或商品服务名称缺失，" "且报销事由、人员和部门能够说明费用归属，则标记为低风险，" "仅提醒补齐票据要素。" ), draft={ "template_key": "field_required_v1", "field_keys": field_keys, "condition_summary": "票据要素缺失但归属清晰时提醒补齐。", }, fields=[SimpleNamespace(key=key, source=key.split(".", 1)[0]) for key in field_keys], expense_category="travel", expense_category_label="差旅费", requires_attachment=True, ) assert result["score"] <= 30 assert result["level"] == "low" assert result["calibration"]["rules"][0]["name"] == "explicit_low_control_cap" def test_risk_score_model_ignores_negated_hard_risk_words_for_low_rules() -> None: result = calculate_risk_rule_score( natural_language=( "差旅费报销提交时，若缺少申报目的地、明细地点或明细事由，" "但暂未发现票据城市冲突、金额异常或重复报销迹象，则标记为低风险，" "提示经办人补齐基础差旅信息后继续提交。" ), draft={ "template_key": "field_required_v1", "field_keys": ["claim.location", "item.item_location", "item.item_reason"], "condition_summary": "基础差旅字段缺失但暂无硬风险迹象时提示补齐。", }, fields=[], expense_category="travel", expense_category_label="差旅费", requires_attachment=False, ) assert result["score"] <= 30 assert result["level"] == "low" def test_risk_score_model_does_not_cap_hard_risk_signals() -> None: result = calculate_risk_rule_score( natural_language=( "差旅报销时，交通票或住宿票据城市均无法与申报目的地一致，" "且没有绕行、跨城办事或改签说明，则标记为高风险，要求补充说明或退回修改。" ), draft={ "template_key": "composite_rule_v1", "field_keys": ["claim.destination_city", "attachment.route_cities"], "conditions": [{"id": "city_mismatch", "operator": "not_overlap"}], }, fields=[], expense_category="travel", expense_category_label="差旅费", requires_attachment=True, ) assert result["score"] >= 61 assert result["level"] == "high" assert not result["calibration"]["rules"] def test_set_risk_rule_level_rejects_manual_override(tmp_path) -> None: with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") generator = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=NullRuntimeChatService(), ) asset_id = generator.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="travel", rule_title="差旅住宿城市一致性校验", risk_level="high", natural_language="住宿城市必须出现在本次差旅行程城市中，否则提示高风险。", ), actor="pytest", ) asset_service = AgentAssetService(db) asset_service.rule_library_manager = manager with pytest.raises(ValueError, match="评分模型"): asset_service.set_risk_rule_level( asset_id, risk_level="low", actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None assert asset.config_json["severity"] != "low" def test_enqueue_risk_rule_generation_creates_visible_generating_asset(tmp_path) -> None: with build_session() as db: service = RiskRuleGenerationJobService( db, rule_library_manager=AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules"), runtime_chat_service=NullRuntimeChatService(), ) request = AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="travel", rule_title="差旅城市一致性校验", risk_level="high", natural_language="住宿城市必须出现在本次差旅行程城市中，否则提示高风险。", ) asset_id = service.enqueue_rule_asset_generation(request, actor="pytest") asset = db.get(AgentAsset, asset_id) assert asset is not None assert asset.status == AgentAssetStatus.GENERATING.value assert asset.owner == "pytest" assert asset.name == "差旅城市一致性校验" assert asset.config_json["generation_status"] == "generating" assert asset.config_json["expense_category_label"] == "差旅费" assert asset.current_version is None detail = AgentAssetService(db).get_asset(asset_id) assert detail is not None assert detail.status == AgentAssetStatus.GENERATING.value assert detail.latest_test_summary is None service.complete_rule_asset_generation(asset_id, request, actor="pytest") db.refresh(asset) assert asset.status == AgentAssetStatus.DRAFT.value assert asset.working_version == "v0.1.0" assert asset.config_json["generation_status"] == "completed" assert asset.config_json["expense_category_label"] == "差旅费" assert asset.scenario_json == ["差旅费"] def test_platform_risk_sync_skips_natural_language_drafts() -> None: assert AgentFoundationRiskRuleMixin._is_user_generated_risk_manifest( { "rule_code": "risk.expense.travel.generated_20260525123000000000", "metadata": { "stability": "generated_draft", "source_ref": "自然语言风险规则", }, } ) assert not AgentFoundationRiskRuleMixin._is_user_generated_risk_manifest( { "rule_code": "risk.travel.destination_location_mismatch", "metadata": {"source_ref": "平台内置风险规则"}, } ) def test_stale_demo_risk_rules_are_marked_deprecated() -> None: class FoundationRiskSyncProbe(AgentFoundationRiskRuleMixin): def __init__(self, db: Session) -> None: self.db = db with build_session() as db: stale_asset = AgentAsset( asset_type=AgentAssetType.RULE.value, code="risk.standard.training_per_capita_over_limit", name="培训费人均超标准", domain=AgentAssetDomain.EXPENSE.value, owner="风控与审计部", reviewer="顾承宇", status=AgentAssetStatus.ACTIVE.value, config_json={ "enabled": True, "tag": "风险规则", "source_ref": "费用管控 Demo 风险规则库", }, ) kept_asset = AgentAsset( asset_type=AgentAssetType.RULE.value, code="risk.standard.software_contract_missing", name="软件服务费缺少合同", domain=AgentAssetDomain.EXPENSE.value, owner="风控与审计部", reviewer="顾承宇", status=AgentAssetStatus.ACTIVE.value, config_json={ "enabled": True, "tag": "风险规则", "source_ref": "费用管控 Demo 风险规则库", }, ) db.add_all([stale_asset, kept_asset]) db.flush() FoundationRiskSyncProbe(db)._hide_stale_demo_risk_rules( {"risk.standard.software_contract_missing"} ) assert stale_asset.status == AgentAssetStatus.DISABLED.value assert stale_asset.config_json["tag"] == "废弃风险规则" assert stale_asset.config_json["enabled"] is False assert kept_asset.status == AgentAssetStatus.ACTIVE.value assert kept_asset.config_json["tag"] == "风险规则" def test_platform_risk_applies_to_chinese_expense_type_labels() -> None: class PlatformRiskProbe(ExpenseClaimPlatformRiskMixin): pass claim = ExpenseClaim( claim_no="TEST-MARKETING-RISK", employee_name="测试员工", department_name="市场部", expense_type="市场推广费", reason="品牌投放活动", amount=Decimal("20000.00"), currency="CNY", invoice_count=1, occurred_at=datetime.now(UTC), status="draft", ) manifest = { "applies_to": { "domains": ["expense"], "expense_types": ["marketing"], } } assert PlatformRiskProbe()._risk_manifest_applies_to_claim( manifest, claim=claim, contexts=[], ) manifest["applies_to"]["expense_types"] = ["software"] assert not PlatformRiskProbe()._risk_manifest_applies_to_claim( manifest, claim=claim, contexts=[], ) def test_risk_rule_flow_diagram_uses_risk_level_palette() -> None: renderer = RiskRuleFlowDiagramRenderer() def render(severity: str, label: str) -> str: return renderer.render( RiskRuleFlowDiagramSpec( title="测试规则", domain_label="报销", severity=severity, severity_label=label, fields=(), start="业务单据提交", evidence="读取规则字段", decision="判断是否命中风险", basis="根据规则字段判断", pass_text="未命中风险，继续流转", fail_text=f"命中{label}，进入复核", ) ) assert "#2563eb" in render("low", "低风险") assert "#f97316" in render("medium", "中风险") high_svg = render("high", "高风险") assert "#dc2626" in high_svg assert high_svg.count("#dc2626") >= 1 assert "#10a37f" not in high_svg def test_risk_rule_simulation_extracts_ticket_route_cities() -> None: with build_session() as db: service = AgentAssetService(db) value = service._find_attachment_field_value( "attachment.route_cities", "行程城市", [ { "document_fields": [ {"key": "route", "label": "行程路线", "value": "上海虹桥-武汉"} ], "ocr_text": "G123 上海虹桥 至 武汉 二等座", "summary": "高铁票 上海虹桥-武汉", } ], ) assert value == ["上海", "武汉"] def test_current_keyword_city_consistency_rule_hits_ticket_city_mismatch() -> None: manifest = { "template_key": "keyword_match_v1", "params": { "template_key": "keyword_match_v1", "field_keys": [ "attachment.hotel_city", "claim.location", "attachment.route_cities", "item.item_location", ], "search_fields": [ "attachment.hotel_city", "claim.location", "attachment.route_cities", "item.item_location", ], "natural_language": ( "差旅报销时，交通票或住宿票据中的城市必须与申报目的地一致；" "未说明绕行、跨城或改签原因时标记风险。" ), "condition_summary": "检查住宿城市、申报地点、行程城市是否一致", "keywords": ["绕行", "跨城", "改签", "变更"], }, "outcomes": {"fail": {"severity": "medium"}}, } claim = ExpenseClaim( claim_no="TEST-CURRENT-RISK", employee_name="测试员工", department_name="测试部门", expense_type="差旅费", reason="去北京出差3天", location="北京", amount=Decimal("320.00"), currency="CNY", invoice_count=1, occurred_at=datetime.now(UTC), status="draft", ) claim.items = [ ExpenseClaimItem( item_date=date.today(), item_type="交通费", item_reason="去北京出差3天", item_location="北京", item_amount=Decimal("320.00"), ) ] result = RiskRuleTemplateExecutor().evaluate( manifest, claim=claim, contexts=[ { "document_info": { "route_cities": ["武汉", "上海"], "fields": [ {"key": "route_cities", "label": "行程城市", "value": ["武汉", "上海"]} ], }, "ocr_text": "武汉 到 上海", } ], ) assert result is not None assert result["evidence"]["city_consistency"]["attachment_values"] == ["武汉", "上海"] assert result["evidence"]["city_consistency"]["reference_values"] == ["北京"] def test_generate_complex_travel_route_rule_uses_formula_not_keyword_match(tmp_path) -> None: text = ( "差旅报销时，先检查是否已上传交通票据、住宿票据或其他能识别城市的附件；" "再读取申报目的地、明细发生地点、交通票行程城市和住宿发票城市。" "若交通票或住宿票据中的城市均无法与申报目的地、明细地点形成一致关系，" "且报销事由中没有说明绕行、跨城办事或临时改签原因，则标记为高风险，" "要求补充行程说明或退回修改。" ) with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") service = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=TravelRouteSemanticRuntimeChatService(), ) asset_id = service.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="travel", rule_title="差旅票据路线一致性校验", risk_level="high", natural_language=text, requires_attachment=True, ), actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None payload = manager.read_rule_library_json( library=RISK_RULES_LIBRARY, file_name=asset.config_json["rule_document"]["file_name"], ) assert payload["template_key"] == "field_compare_v1" assert payload["semantic_type"] == "travel_route_city_consistency" assert payload["params"]["semantic_type"] == "travel_route_city_consistency" assert payload["params"]["keywords"] == [] assert payload["params"]["exception_keywords"][:3] == ["绕行", "跨城办事", "跨城"] assert "A=交通票行程城市" in payload["params"]["condition_summary"] assert "风险关键词" not in payload["params"]["condition_summary"] assert "employee.location" in payload["params"]["field_keys"] assert "route_anomaly_policy" in payload["params"] def test_legacy_city_route_keyword_manifest_is_normalized_before_display_and_execution() -> None: manifest = { "schema_version": "2.0", "rule_code": "risk.expense.travel.legacy_city_keyword", "name": "差旅票据路线一致性校验", "description": "差旅报销时读取申报目的地、明细发生地点、交通票行程城市和住宿发票城市。", "evaluator": "template_rule", "template_key": "keyword_match_v1", "risk_category": "差旅费", "inputs": { "fields": [ {"key": "attachment.hotel_city", "label": "住宿城市"}, {"key": "claim.location", "label": "申报地点"}, {"key": "attachment.route_cities", "label": "行程城市"}, ] }, "params": { "template_key": "keyword_match_v1", "field_keys": [ "attachment.hotel_city", "claim.location", "attachment.route_cities", ], "search_fields": [ "attachment.hotel_city", "claim.location", "attachment.route_cities", ], "keywords": ["绕行", "跨城办事", "临时改签"], "condition_summary": "检查住宿城市、申报地点、行程城市是否出现规则描述中的风险关键词", "natural_language": ( "差旅报销时，先检查是否已上传交通票据、住宿票据或其他能识别城市的附件；" "再读取申报目的地、明细发生地点、交通票行程城市和住宿发票城市。" "若交通票或住宿票据中的城市均无法与申报目的地、明细地点形成一致关系，" "且报销事由中没有说明绕行、跨城办事或临时改签原因，则标记为高风险。" ), }, "outcomes": {"fail": {"severity": "high"}}, "metadata": { "condition_summary": "检查住宿城市、申报地点、行程城市是否出现规则描述中的风险关键词", "flow": { "decision": "检查住宿城市、申报地点、行程城市是否出现规则描述中的风险关键词" }, }, "flow_diagram_svg": ( '' "检查住宿城市、申报地点、行程城市是否出现规则描述中的风险关键词" "" ), } normalized = normalize_risk_rule_manifest(manifest) assert normalized["template_key"] == "field_compare_v1" assert normalized["semantic_type"] == "travel_route_city_consistency" assert normalized["params"]["keywords"] == [] assert "风险关键词" not in normalized["params"]["condition_summary"] assert "风险关键词" not in normalized["metadata"]["flow"]["decision"] assert "风险关键词" not in normalized["flow_diagram_svg"] claim = ExpenseClaim( claim_no="TEST-LEGACY-NORMALIZER", employee_name="测试员工", department_name="测试部门", expense_type="差旅费", reason="去上海办事", location="上海", amount=Decimal("520.00"), currency="CNY", invoice_count=1, occurred_at=datetime.now(UTC), status="draft", ) claim.employee = Employee( employee_no="TEST-EMPLOYEE", name="测试员工", email="legacy-route-risk@example.com", location="武汉", ) claim.items = [ ExpenseClaimItem( item_date=date.today(), item_type="交通费", item_reason="去上海办事", item_location="上海", item_amount=Decimal("520.00"), ) ] result = RiskRuleTemplateExecutor().evaluate( normalized, claim=claim, contexts=[{"document_info": {"route_cities": ["上海", "北京", "武汉"]}}], ) assert result is not None assert result["evidence"]["city_consistency"]["unexpected_route_cities"] == ["北京"] def test_travel_route_rule_hits_unexpected_intermediate_city_even_when_returning_home() -> None: manifest = { "template_key": "field_compare_v1", "params": { "template_key": "field_compare_v1", "semantic_type": "travel_route_city_consistency", "field_keys": [ "attachment.route_cities", "claim.location", "item.item_location", "employee.location", "claim.reason", ], "attachment_city_fields": ["attachment.route_cities"], "reference_city_fields": ["claim.location", "item.item_location"], "home_city_fields": ["employee.location"], "exception_fields": ["claim.reason"], "exception_keywords": ["绕行", "跨城办事", "临时改签"], "condition_summary": ( "A=票据路线城市，B=申报城市，C=员工常驻地，" "A中出现B∪C之外城市则命中。" ), }, "outcomes": {"fail": {"severity": "high"}}, } claim = ExpenseClaim( claim_no="TEST-ROUTE-ANOMALY", employee_name="测试员工", department_name="测试部门", expense_type="差旅费", reason="去上海办事", location="上海", amount=Decimal("520.00"), currency="CNY", invoice_count=1, occurred_at=datetime.now(UTC), status="draft", ) claim.employee = Employee( employee_no="TEST-EMPLOYEE", name="测试员工", email="route-risk@example.com", location="武汉", ) claim.items = [ ExpenseClaimItem( item_date=date.today(), item_type="交通费", item_reason="去上海办事", item_location="上海", item_amount=Decimal("520.00"), ) ] result = RiskRuleTemplateExecutor().evaluate( manifest, claim=claim, contexts=[ { "document_info": { "route_cities": ["上海", "北京", "武汉"], "fields": [ { "key": "route_cities", "label": "行程城市", "value": ["上海", "北京", "武汉"], } ], }, "ocr_text": "上海 到 北京 到 武汉", } ], ) assert result is not None evidence = result["evidence"]["city_consistency"] assert evidence["reference_values"] == ["上海"] assert evidence["home_values"] == ["武汉"] assert evidence["unexpected_route_cities"] == ["北京"] def test_simulation_uses_current_rule_manifest_for_ticket_city_mismatch(tmp_path) -> None: with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") generator = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=NullRuntimeChatService(), ) asset_id = generator.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="travel", rule_title="当前差旅票据城市一致性规则", risk_level="medium", natural_language="差旅报销时，交通票或住宿票据中的城市必须与申报目的地一致；未说明绕行、跨城或改签原因时标记风险。", requires_attachment=True, ), actor="pytest", ) service = AgentAssetService(db) service.rule_library_manager = manager asset = db.get(AgentAsset, asset_id) assert asset is not None manifest = manager.read_rule_library_json( library=RISK_RULES_LIBRARY, file_name=asset.config_json["rule_document"]["file_name"], ) manifest["template_key"] = "keyword_match_v1" manifest["params"]["template_key"] = "keyword_match_v1" manifest["params"]["keywords"] = ["绕行", "跨城", "改签", "变更"] manifest["params"]["search_fields"] = [ "attachment.hotel_city", "claim.location", "attachment.route_cities", "item.item_location", ] manifest["params"]["field_keys"] = manifest["params"]["search_fields"] manager.write_rule_library_json( library=RISK_RULES_LIBRARY, file_name=asset.config_json["rule_document"]["file_name"], payload=manifest, ) simulation = service.simulate_risk_rule_message( asset_id, AgentAssetRiskRuleSimulationRequest( message="去北京出差3天", attachments=[ { "name": "train-ticket.pdf", "content_type": "application/pdf", "ocr_text": "武汉 到 上海", "summary": "高铁票 武汉-上海", "document_fields": [ {"key": "route", "label": "行程路线", "value": "武汉-上海"} ], } ], ), ) assert simulation.ready is True assert simulation.hit is True assert simulation.field_values["claim.location"] == "北京" assert simulation.field_values["attachment.route_cities"] == ["武汉", "上海"] def test_risk_rule_requires_test_report_before_review_and_publish(tmp_path) -> None: with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") generator = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=NullRuntimeChatService(), ) asset_id = generator.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, risk_level="high", natural_language="酒店发票城市必须与行程城市一致，不一致时标记高风险。", ), actor="pytest", ) service = AgentAssetService(db) service.rule_library_manager = manager asset = db.get(AgentAsset, asset_id) assert asset is not None try: service.create_review( asset_id, AgentAssetReviewCreate( version=asset.working_version or "v0.1.0", reviewer="manager", review_status=AgentReviewStatus.PENDING, review_note="送审", ), actor="pytest", ) except PermissionError as exc: assert "测试通过" in str(exc) else: raise AssertionError("未测试通过的风险规则不应允许提交审核") simulation = service.simulate_risk_rule_message( asset_id, AgentAssetRiskRuleSimulationRequest( message="我想仿真一张酒店报销单，酒店发票城市上海，申报目的地北京，金额580元。", ), ) assert simulation.execution_mode == "risk_rule_simulation" assert simulation.ready is True assert simulation.hit is True assert simulation.severity == "high" assert "不创建业务单据" in simulation.summary assert service.get_latest_risk_rule_test_summary(asset_id).sample is None blocked_simulation = service.simulate_risk_rule_message( asset_id, AgentAssetRiskRuleSimulationRequest( message="请识别上传单据是否命中风险规则。", attachments=[{"name": "hotel-invoice.pdf", "content_type": "application/pdf"}], ), ) assert blocked_simulation.ready is False assert blocked_simulation.stage == "needs_recognition" assert blocked_simulation.hit is False assert "尚未完成识别" in blocked_simulation.summary db.add( ExpenseClaim( claim_no="TEST-CLAIM-001", employee_name="张三", department_name="财务部", expense_type="住宿费", reason="北京出差住宿", location="北京", amount=Decimal("300.00"), currency="CNY", invoice_count=0, occurred_at=datetime.now(UTC), created_at=datetime.now(UTC), status="draft", ) ) db.commit() sample = service.run_risk_rule_sample_test( asset_id, AgentAssetRiskRuleSampleTestRequest(), actor="pytest", ) assert sample.passed is True scenario = service.run_risk_rule_scenario_test( asset_id, AgentAssetRiskRuleScenarioTestRequest(intent="用最近30天的住宿报销单试运行"), actor="pytest", ) assert scenario.passed is True assert scenario.result_json["total_count"] == 1 report = service.confirm_risk_rule_test_report( asset_id, AgentAssetRiskRuleReportRequest(confirm_passed=True), actor="pytest", ) assert report.passed is True review = service.create_review( asset_id, AgentAssetReviewCreate( version=asset.working_version or "v0.1.0", reviewer="manager", review_status=AgentReviewStatus.PENDING, review_note="送审", ), actor="pytest", ) assert review.review_status == AgentReviewStatus.PENDING.value published = service.publish_risk_rule(asset_id, actor="manager") assert published.status == AgentAssetStatus.ACTIVE.value assert published.published_version == asset.working_version disabled = service.set_risk_rule_enabled( asset_id, enabled=False, actor="manager", ) assert disabled.status == AgentAssetStatus.DISABLED.value assert disabled.published_version == asset.working_version assert disabled.config_json["enabled"] is False assert disabled.config_json["last_operation"]["action"] == "offline" rule_document = disabled.config_json["rule_document"] manifest = manager.read_rule_library_json( library=RISK_RULES_LIBRARY, file_name=rule_document["file_name"], ) assert manifest["enabled"] is False enabled = service.set_risk_rule_enabled(asset_id, enabled=True, actor="manager") assert enabled.status == AgentAssetStatus.ACTIVE.value assert enabled.config_json["enabled"] is True assert enabled.config_json["last_operation"]["action"] == "online" attachment_required_id = generator.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, risk_level="medium", natural_language="发票号码不能为空，缺失时进入中风险复核。", requires_attachment=True, ), actor="pytest", ) attachment_required_asset = db.get(AgentAsset, attachment_required_id) assert attachment_required_asset is not None assert attachment_required_asset.config_json["requires_attachment"] is True attachment_rule_document = attachment_required_asset.config_json["rule_document"] attachment_manifest = manager.read_rule_library_json( library=RISK_RULES_LIBRARY, file_name=attachment_rule_document["file_name"], ) assert attachment_manifest["requires_attachment"] is True no_attachment_simulation = service.simulate_risk_rule_message( attachment_required_id, AgentAssetRiskRuleSimulationRequest(message="请测试这条规则。"), ) assert no_attachment_simulation.ready is False assert no_attachment_simulation.stage == "needs_attachment" attachment_only_simulation = service.simulate_risk_rule_message( attachment_required_id, AgentAssetRiskRuleSimulationRequest( message="请识别上传单据是否命中风险规则。", attachments=[ { "name": "invoice.pdf", "content_type": "application/pdf", "document_fields": [ {"key": "invoice_no", "label": "发票号码", "value": "INV-001"} ], } ], ), ) assert attachment_only_simulation.ready is False assert attachment_only_simulation.stage == "needs_test_intent" def test_delete_unpublished_risk_rule_removes_asset_and_json_file(tmp_path) -> None: with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") asset_id = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=NullRuntimeChatService(), ).generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, risk_level="medium", natural_language="报销事由不能为空，缺失时进入中风险复核。", ), actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None file_name = asset.config_json["rule_document"]["file_name"] rule_path = tmp_path / "rules" / RISK_RULES_LIBRARY / file_name assert rule_path.exists() service = AgentAssetService(db) service.rule_library_manager = manager service.delete_unpublished_asset(asset_id, actor="pytest") assert db.get(AgentAsset, asset_id) is None assert not rule_path.exists()