from __future__ import annotations from datetime import UTC, date, datetime, timedelta from decimal import Decimal import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.api.deps import CurrentUserContext from app.db.base import Base from app.models.employee import Employee from app.models.financial_record import ExpenseClaim, ExpenseClaimItem from app.models.organization import OrganizationUnit from app.schemas.ontology import OntologyParseRequest from app.schemas.ocr import OcrRecognizeBatchRead, OcrRecognizeDocumentRead from app.schemas.reimbursement import ExpenseClaimItemCreate, ExpenseClaimItemUpdate, ExpenseClaimUpdate from app.services.agent_conversations import AgentConversationService from app.services.expense_claims import ExpenseClaimService from app.services.ontology import SemanticOntologyService from app.services.ocr import OcrService def build_claim(*, expense_type: str, location: str) -> ExpenseClaim: claim = ExpenseClaim( id="claim-1", claim_no="EXP-202605-001", employee_id="emp-1", employee_name="张三", department_id="dept-1", department_name="市场部", project_code=None, expense_type=expense_type, reason="费用报销", location=location, amount=Decimal("88.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 13, tzinfo=UTC), submitted_at=None, status="draft", approval_stage="待提交", risk_flags_json=[], ) claim.items = [ ExpenseClaimItem( id="item-1", claim_id="claim-1", item_date=date(2026, 5, 13), item_type=expense_type, item_reason="费用报销", item_location=location, item_amount=Decimal("88.00"), invoice_id="invoice-1", ) ] return claim 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 _count_claims(db: Session) -> int: return int(db.query(ExpenseClaim).count()) def test_validate_claim_for_submission_allows_office_claim_without_location() -> None: service = ExpenseClaimService.__new__(ExpenseClaimService) claim = build_claim(expense_type="office", location="待补充") issues = service._validate_claim_for_submission(claim) assert "业务地点未完善" not in issues assert not any("缺少地点" in item for item in issues) def test_validate_claim_for_submission_allows_transport_claim_without_location() -> None: service = ExpenseClaimService.__new__(ExpenseClaimService) claim = build_claim(expense_type="transport", location="待补充") issues = service._validate_claim_for_submission(claim) assert "业务地点未完善" not in issues assert not any("缺少地点" in item for item in issues) def test_validate_claim_for_submission_still_requires_location_for_travel_claim() -> None: service = ExpenseClaimService.__new__(ExpenseClaimService) claim = build_claim(expense_type="travel", location="待补充") issues = service._validate_claim_for_submission(claim) assert "业务地点未完善" in issues assert any("缺少地点" in item for item in issues) def test_save_or_submit_preview_does_not_create_claim_without_explicit_action() -> None: user_id = "preview-only@example.com" message = "业务发生时间:2026-03-04，打车去客户现场，交通费32元，请帮我看看怎么报" with build_session() as db: employee = Employee( employee_no="E5100", name="预览员工", email=user_id, ) db.add(employee) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query=message, user_id=user_id, ) ) before_count = _count_claims(db) result = ExpenseClaimService(db).save_or_submit_from_ontology( run_id=ontology.run_id, user_id=user_id, message=message, ontology=ontology, context_json={ "name": "预览员工", "user_input_text": message, }, ) assert result["preview_only"] is True assert result["status"] == "preview" assert "尚未保存为草稿" in result["message"] assert _count_claims(db) == before_count def test_save_or_submit_persists_claim_only_after_save_draft_action() -> None: user_id = "save-draft-explicit@example.com" message = "业务发生时间:2026-03-04，打车去客户现场，交通费32元，请帮我看看怎么报" with build_session() as db: employee = Employee( employee_no="E5101", name="保存员工", email=user_id, ) db.add(employee) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query=message, user_id=user_id, ) ) before_count = _count_claims(db) result = ExpenseClaimService(db).save_or_submit_from_ontology( run_id=ontology.run_id, user_id=user_id, message=message, ontology=ontology, context_json={ "name": "保存员工", "user_input_text": message, "review_action": "save_draft", }, ) assert result["draft_only"] is True assert result["claim_id"] assert result["status"] == "draft" assert _count_claims(db) == before_count + 1 def test_unsaved_conversation_expires_after_retention_but_saved_conversation_stays() -> None: with build_session() as db: service = AgentConversationService(db) unsaved = service.get_or_create_conversation( conversation_id="conv-unsaved-expire", user_id="expire@example.com", source="user_message", context_json={"session_type": "expense"}, ) saved = service.get_or_create_conversation( conversation_id="conv-saved-keep", user_id="expire@example.com", source="user_message", context_json={ "session_type": "expense", "draft_claim_id": "claim-saved", }, ) old_time = datetime.now(UTC) - timedelta(days=4) unsaved.updated_at = old_time saved.updated_at = old_time db.add_all([unsaved, saved]) db.commit() deleted_count = service.prune_expired_conversations(retention_days=3) assert deleted_count == 1 assert service.get_conversation("conv-unsaved-expire") is None assert service.get_conversation("conv-saved-keep") is not None def test_resolve_expense_type_maps_office_supplies_review_value_to_office() -> None: expense_type = ExpenseClaimService._resolve_expense_type( [], context_json={ "review_form_values": { "expense_type": "办公用品" } }, ) assert expense_type == "office" def test_resolve_expense_type_maps_riding_fare_review_value_to_transport() -> None: expense_type = ExpenseClaimService._resolve_expense_type( [], context_json={ "review_form_values": { "expense_type": "乘车费用" } }, ) assert expense_type == "transport" def test_upsert_draft_from_ontology_defers_multi_document_association_choice() -> None: user_id = "zhangsan@example.com" with build_session() as db: employee = Employee( employee_no="E5001", name="张三", email=user_id, ) db.add(employee) db.flush() existing_claim = ExpenseClaim( claim_no="EXP-202605-010", employee_id=employee.id, employee_name="张三", department_name="市场部", project_code=None, expense_type="transport", reason="原有交通报销", location="深圳", amount=Decimal("20.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 13, tzinfo=UTC), status="draft", approval_stage="待提交", risk_flags_json=[], ) existing_claim.items = [ ExpenseClaimItem( claim_id=existing_claim.id, item_date=date(2026, 5, 13), item_type="transport", item_reason="原有交通报销", item_location="深圳", item_amount=Decimal("20.00"), invoice_id="old-trip.png", ) ] db.add(existing_claim) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我上传了两张交通票据，帮我生成报销草稿", user_id=user_id, ) ) service = ExpenseClaimService(db) result = service.upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="我上传了两张交通票据，帮我生成报销草稿", ontology=ontology, context_json={ "name": "张三", "attachment_names": ["didi-trip.png", "parking-ticket.jpg"], "attachment_count": 2, "draft_claim_id": existing_claim.id, "ocr_documents": [ { "filename": "didi-trip.png", "summary": "滴滴出行 支付金额 32 元", "text": "滴滴出行 支付金额 32 元", }, { "filename": "parking-ticket.jpg", "summary": "停车费 合计 18 元", "text": "停车费 合计 18 元", }, ], }, ) db.refresh(existing_claim) assert result["pending_association_decision"] is True assert result["association_candidate_claim_id"] == existing_claim.id assert existing_claim.invoice_count == 1 assert len(existing_claim.items) == 1 assert existing_claim.items[0].invoice_id == "old-trip.png" def test_linked_document_supplement_keeps_existing_claim_expense_type() -> None: user_id = "type-lock@example.com" with build_session() as db: employee = Employee( employee_no="E5010", name="类型锁定员工", email=user_id, ) db.add(employee) db.flush() existing_claim = ExpenseClaim( claim_no="EXP-202605-020", employee_id=employee.id, employee_name="类型锁定员工", department_name="市场部", project_code=None, expense_type="transport", reason="原有交通报销", location="深圳", amount=Decimal("32.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 13, tzinfo=UTC), status="draft", approval_stage="待提交", risk_flags_json=[], ) existing_claim.items = [ ExpenseClaimItem( claim_id=existing_claim.id, item_date=date(2026, 5, 13), item_type="transport", item_reason="原有交通报销", item_location="深圳", item_amount=Decimal("32.00"), invoice_id="old-trip.png", ) ] db.add(existing_claim) db.commit() context_json = { "name": "类型锁定员工", "review_action": "link_to_existing_draft", "draft_claim_id": existing_claim.id, "attachment_names": ["hotel-invoice.pdf"], "attachment_count": 1, "ocr_documents": [ { "filename": "hotel-invoice.pdf", "document_type": "hotel_invoice", "scene_code": "hotel", "scene_label": "住宿票据", "summary": "酒店住宿 发票金额 300 元", "text": "酒店住宿 发票金额 ￥300.00", "document_fields": [ {"key": "amount", "label": "金额", "value": "300"}, {"key": "merchant", "label": "酒店名称", "value": "上海酒店"}, ], } ], } ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="把酒店发票补充到现有草稿", user_id=user_id, context_json=context_json, ) ) ExpenseClaimService(db).upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="把酒店发票补充到现有草稿", ontology=ontology, context_json=context_json, ) db.refresh(existing_claim) assert existing_claim.expense_type == "transport" assert any(item.item_type == "hotel_ticket" for item in existing_claim.items) def test_upsert_draft_from_ontology_keeps_reason_missing_for_attachment_only_upload() -> None: user_id = "wangwu@example.com" with build_session() as db: employee = Employee( employee_no="E5003", name="王五", email=user_id, ) db.add(employee) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我上传了 1 份票据，请结合附件名称给出报销建议并尽量生成草稿。", user_id=user_id, ) ) service = ExpenseClaimService(db) result = service.upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="我上传了 1 份票据，请结合附件名称给出报销建议并尽量生成草稿。\n附件名称：didi-trip.png", ontology=ontology, context_json={ "name": "王五", "user_input_text": "", "attachment_names": ["didi-trip.png"], "attachment_count": 1, "ocr_documents": [ { "filename": "didi-trip.png", "summary": "滴滴出行 支付金额 32 元", "text": "滴滴出行 支付金额 32 元", "document_type": "taxi_receipt", "scene_code": "transport", } ], }, ) claim = db.get(ExpenseClaim, result["claim_id"]) assert claim is not None assert claim.reason == "待补充" def test_upsert_draft_from_ontology_strips_recognized_business_time_from_reason() -> None: user_id = "transport-time@example.com" message = "业务发生时间:2026-03-04，送客户去林萃小区办事，请报销乘车费用" with build_session() as db: employee = Employee( employee_no="E5004", name="赵六", email=user_id, ) db.add(employee) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query=message, user_id=user_id, ) ) result = ExpenseClaimService(db).upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message=message, ontology=ontology, context_json={ "name": "赵六", "user_input_text": message, }, ) claim = db.get(ExpenseClaim, result["claim_id"]) assert claim is not None assert claim.occurred_at.date() == date(2026, 3, 4) assert claim.reason == "送客户去林萃小区办事，请报销乘车费用" assert len(claim.items) == 1 assert claim.items[0].item_date == date(2026, 3, 4) assert claim.items[0].item_reason == "送客户去林萃小区办事，请报销乘车费用" assert "客户单位" not in result["message"] assert "票据附件" not in result["message"] assert "费用明细" not in result["message"] def test_upsert_draft_from_ontology_supports_link_or_create_for_multi_documents() -> None: user_id = "lisi@example.com" with build_session() as db: employee = Employee( employee_no="E5002", name="李四", email=user_id, ) db.add(employee) db.flush() existing_claim = ExpenseClaim( claim_no="EXP-202605-011", employee_id=employee.id, employee_name="李四", department_name="销售部", project_code=None, expense_type="transport", reason="原有交通报销", location="上海", amount=Decimal("20.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 13, tzinfo=UTC), status="draft", approval_stage="待提交", risk_flags_json=[], ) existing_claim.items = [ ExpenseClaimItem( claim_id=existing_claim.id, item_date=date(2026, 5, 13), item_type="transport", item_reason="原有交通报销", item_location="上海", item_amount=Decimal("20.00"), invoice_id="existing.png", ) ] db.add(existing_claim) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我上传了两张交通票据，帮我生成报销草稿", user_id=user_id, ) ) service = ExpenseClaimService(db) context_json = { "name": "李四", "attachment_names": ["didi-trip.png", "parking-ticket.jpg"], "attachment_count": 2, "draft_claim_id": existing_claim.id, "ocr_documents": [ { "filename": "didi-trip.png", "summary": "滴滴出行", "text": "滴滴出行 支付金额 32.50 元", "document_type": "taxi_receipt", "scene_code": "transport", "document_fields": [{"key": "amount", "label": "支付金额", "value": "32.50"}], }, { "filename": "parking-ticket.jpg", "summary": "停车票", "text": "停车费 合计 18 元", "document_type": "parking_toll_receipt", "scene_code": "transport", "document_fields": [{"key": "total_amount", "label": "合计金额", "value": "18"}], }, ], } link_result = service.upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="把这两张票据关联到已有草稿", ontology=ontology, context_json={ **context_json, "review_action": "link_to_existing_draft", }, ) db.refresh(existing_claim) assert link_result["claim_id"] == existing_claim.id assert existing_claim.invoice_count == 3 assert len(existing_claim.items) == 3 assert float(existing_claim.amount) == 70.5 create_result = service.upsert_draft_from_ontology( run_id=f"{ontology.run_id}-new", user_id=user_id, message="单独新建一张报销单", ontology=ontology, context_json={ **context_json, "review_action": "create_new_claim_from_documents", }, ) assert create_result["claim_id"] != existing_claim.id new_claim = db.get(ExpenseClaim, create_result["claim_id"]) assert new_claim is not None assert new_claim.invoice_count == 2 assert len(new_claim.items) == 2 assert float(new_claim.amount) == 50.5 def test_upsert_travel_draft_uses_ticket_item_types_and_auto_allowance() -> None: user_id = "travel-allowance@example.com" with build_session() as db: employee = Employee( employee_no="E5010", name="差旅员工", email=user_id, grade="P4", ) db.add(employee) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我去北京出差 3 天，上传了火车票，帮我生成差旅费报销草稿", user_id=user_id, ) ) result = ExpenseClaimService(db).upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="我去北京出差 3 天，上传了火车票，帮我生成差旅费报销草稿", ontology=ontology, context_json={ "name": "差旅员工", "grade": "P4", "attachment_names": ["train-ticket.png"], "attachment_count": 1, "review_form_values": { "expense_type": "差旅费", "location": "北京", "time_range": "2026-05-13 至 2026-05-15", }, "business_time_context": { "mode": "range", "start_date": "2026-05-13", "end_date": "2026-05-15", "display_value": "2026-05-13 至 2026-05-15", }, "ocr_documents": [ { "filename": "train-ticket.png", "document_type": "train_ticket", "document_type_label": "火车/高铁票", "scene_code": "travel", "scene_label": "差旅费", "summary": "中国铁路电子客票 广州南-北京南 二等座 票价 354 元", "text": "中国铁路电子客票 广州南-北京南 二等座 票价:￥354.00", "document_fields": [ {"key": "amount", "label": "票价", "value": "￥354.00"}, {"key": "route", "label": "行程", "value": "广州南-北京南"}, ], } ], }, ) claim = db.get(ExpenseClaim, result["claim_id"]) assert claim is not None assert claim.expense_type == "travel" assert claim.invoice_count == 1 assert len(claim.items) == 2 train_item = next(item for item in claim.items if item.item_type == "train_ticket") allowance_item = next(item for item in claim.items if item.item_type == "travel_allowance") assert train_item.item_amount == Decimal("354.00") assert train_item.item_reason == "广州南-北京南" assert allowance_item.item_amount == Decimal("300.00") assert allowance_item.invoice_id is None assert allowance_item.is_system_generated is True assert claim.amount == Decimal("654.00") with pytest.raises(ValueError, match="系统自动计算"): ExpenseClaimService(db).update_claim_item( claim_id=claim.id, item_id=allowance_item.id, payload=ExpenseClaimItemUpdate(item_amount=Decimal("1.00")), current_user=CurrentUserContext( username=user_id, name="差旅员工", role_codes=[], is_admin=False, ), ) def test_sync_travel_claim_adds_allowance_from_manual_ticket_dates() -> None: with build_session() as db: employee = Employee( employee_no="E5011", name="手工差旅员工", email="manual-travel-allowance@example.com", grade="P4", ) db.add(employee) db.flush() claim = build_claim(expense_type="travel", location="北京") claim.employee_id = employee.id claim.employee_name = employee.name claim.items[0].item_date = date(2026, 5, 13) claim.items[0].item_type = "train_ticket" claim.items[0].item_reason = "广州南-北京南" claim.items[0].item_location = "北京" claim.items[0].item_amount = Decimal("354.00") claim.items.append( ExpenseClaimItem( claim_id=claim.id, item_date=date(2026, 5, 15), item_type="train_ticket", item_reason="北京南-广州南", item_location="北京", item_amount=Decimal("354.00"), invoice_id="return-train.png", ) ) db.add(claim) db.commit() service = ExpenseClaimService(db) service._sync_claim_from_items(claim) db.commit() db.refresh(claim) allowance_item = next(item for item in claim.items if item.item_type == "travel_allowance") assert allowance_item.item_amount == Decimal("300.00") assert "3天" in allowance_item.item_reason assert allowance_item.invoice_id is None assert claim.amount == Decimal("1008.00") def test_update_claim_item_allows_placeholder_date_reason_and_amount() -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) with build_session() as db: claim = build_claim(expense_type="office", location="深圳") db.add(claim) db.commit() updated = ExpenseClaimService(db).update_claim_item( claim_id=claim.id, item_id=claim.items[0].id, payload=ExpenseClaimItemUpdate( item_reason="", item_location="", item_amount=Decimal("0.00"), ), current_user=current_user, ) assert updated is not None db.refresh(claim) assert claim.items[0].item_date == date(2026, 5, 13) assert claim.items[0].item_reason == "" assert claim.items[0].item_location == "" assert claim.items[0].item_amount == Decimal("0.00") def test_upsert_draft_from_ontology_updates_returned_claim_and_preserves_return_events() -> None: user_id = "returned-owner@example.com" return_flag = { "source": "manual_return", "return_event_id": "return-event-1", "message": "第一次退回：附件缺失。", "reason": "附件缺失。", "return_count": 1, "return_stage": "直属领导审批", "return_stage_key": "direct_manager", "risk_points": ["附件缺失或不清晰"], } with build_session() as db: employee = Employee( employee_no="E5004", name="赵六", email=user_id, ) db.add(employee) db.flush() existing_claim = ExpenseClaim( claim_no="EXP-202605-012", employee_id=employee.id, employee_name="赵六", department_name="市场部", project_code=None, expense_type="transport", reason="原有交通报销", location="上海", amount=Decimal("20.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 13, tzinfo=UTC), status="returned", approval_stage="待提交", risk_flags_json=[return_flag], ) existing_claim.items = [ ExpenseClaimItem( claim_id=existing_claim.id, item_date=date(2026, 5, 13), item_type="transport", item_reason="原有交通报销", item_location="上海", item_amount=Decimal("20.00"), invoice_id="old-trip.png", ) ] db.add(existing_claim) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我补充了交通票据，更新这张退回单据", user_id=user_id, ) ) ontology.risk_flags = ["系统识别：票据金额待人工核对。"] result = ExpenseClaimService(db).upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="我补充了交通票据，更新这张退回单据", ontology=ontology, context_json={ "name": "赵六", "draft_claim_id": existing_claim.id, "attachment_names": ["new-trip.png"], "attachment_count": 1, "ocr_documents": [ { "filename": "new-trip.png", "summary": "滴滴出行 支付金额 32 元", "text": "滴滴出行 支付金额 32 元", "document_type": "taxi_receipt", "scene_code": "transport", } ], }, ) db.refresh(existing_claim) assert result["claim_id"] == existing_claim.id assert existing_claim.status == "draft" assert "系统识别：票据金额待人工核对。" in existing_claim.risk_flags_json manual_returns = [ flag for flag in list(existing_claim.risk_flags_json or []) if isinstance(flag, dict) and flag.get("source") == "manual_return" ] assert manual_returns == [return_flag] def test_generate_claim_no_uses_max_suffix_instead_of_count() -> None: with build_session() as db: db.add_all( [ ExpenseClaim( claim_no="EXP-202605-001", employee_name="张三", department_name="市场部", project_code=None, expense_type="transport", reason="交通报销", location="深圳", amount=Decimal("10.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 10, tzinfo=UTC), status="draft", approval_stage="待提交", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-202605-003", employee_name="李四", department_name="销售部", project_code=None, expense_type="transport", reason="交通报销", location="上海", amount=Decimal("20.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, tzinfo=UTC), status="submitted", approval_stage="审批中", risk_flags_json=[], ), ] ) db.commit() service = ExpenseClaimService(db) assert service._generate_claim_no(datetime(2026, 5, 14, tzinfo=UTC)) == "EXP-202605-004" def test_upsert_draft_from_ontology_retries_claim_no_conflict() -> None: user_id = "zhaoliu-claimno@example.com" with build_session() as db: employee = Employee( employee_no="E5006", name="赵六", email=user_id, ) db.add(employee) db.flush() db.add( ExpenseClaim( claim_no="EXP-202605-004", employee_name="历史单据", department_name="财务部", project_code=None, expense_type="other", reason="历史草稿", location="北京", amount=Decimal("0.00"), currency="CNY", invoice_count=0, occurred_at=datetime(2026, 5, 12, tzinfo=UTC), status="submitted", approval_stage="审批中", risk_flags_json=[], ) ) db.commit() ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="帮我生成报销草稿，我昨天交通费 13.4 元", user_id=user_id, ) ) service = ExpenseClaimService(db) generated_claim_nos = iter(["EXP-202605-004", "EXP-202605-005"]) service._generate_claim_no = lambda occurred_at: next(generated_claim_nos) result = service.upsert_draft_from_ontology( run_id=ontology.run_id, user_id=user_id, message="帮我生成报销草稿，我昨天交通费 13.4 元", ontology=ontology, context_json={ "name": "赵六", "user_input_text": "帮我生成报销草稿，我昨天交通费 13.4 元", }, ) created_claim = db.get(ExpenseClaim, result["claim_id"]) assert created_claim is not None assert created_claim.claim_no == "EXP-202605-005" assert result["claim_no"] == "EXP-202605-005" def test_create_claim_item_adds_blank_draft_row_without_forcing_attachment() -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) with build_session() as db: claim = build_claim(expense_type="office", location="深圳南山") db.add(claim) db.commit() service = ExpenseClaimService(db) updated = service.create_claim_item( claim_id=claim.id, payload=ExpenseClaimItemCreate(), current_user=current_user, ) assert updated is not None assert len(updated.items) == 2 assert updated.amount == Decimal("88.00") assert updated.invoice_count == 1 new_item = next(item for item in updated.items if item.id != "item-1") assert new_item.item_type == "office" assert new_item.item_reason == "" assert new_item.item_location == "" assert new_item.item_amount == Decimal("0.00") assert new_item.invoice_id is None def test_update_claim_reason_only_allows_draft_pending_submission() -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) with build_session() as db: claim = build_claim(expense_type="travel", location="北京") db.add(claim) db.commit() service = ExpenseClaimService(db) updated = service.update_claim( claim_id=claim.id, payload=ExpenseClaimUpdate(reason="去北京客户现场出差，处理项目验收事项"), current_user=current_user, ) assert updated is not None assert updated.reason == "去北京客户现场出差，处理项目验收事项" claim.status = "submitted" claim.submitted_at = datetime(2026, 5, 14, tzinfo=UTC) claim.approval_stage = "直属领导审批" db.commit() with pytest.raises(ValueError, match="草稿待提交"): service.update_claim( claim_id=claim.id, payload=ExpenseClaimUpdate(reason="提交后不能改"), current_user=current_user, ) def test_update_claim_item_reanalyzes_existing_attachment(monkeypatch, tmp_path) -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: return OcrRecognizeBatchRead( total_file_count=1, success_count=1, documents=[ OcrRecognizeDocumentRead( filename="office-note.png", media_type="image/png", text="办公用品发票 金额88元 2026-05-13", summary="识别到办公用品发票，金额 88 元。", avg_score=0.98, line_count=1, page_count=1, warnings=[], ) ], ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: claim = build_claim(expense_type="office", location="深圳南山") claim.invoice_count = 0 claim.items[0].invoice_id = None claim.items[0].item_reason = "办公用品采购" db.add(claim) db.commit() service = ExpenseClaimService(db) service.upload_claim_item_attachment( claim_id=claim.id, item_id=claim.items[0].id, filename="office-note.png", content=b"fake-image-bytes", media_type="image/png", current_user=current_user, ) uploaded_meta = service.get_claim_item_attachment_meta( claim_id=claim.id, item_id=claim.items[0].id, current_user=current_user, ) assert uploaded_meta is not None assert uploaded_meta["preview_kind"] == "image" assert uploaded_meta["preview_url"].endswith( f"/reimbursements/claims/{claim.id}/items/{claim.items[0].id}/attachment/preview" ) assert uploaded_meta["analysis"]["severity"] == "pass" assert uploaded_meta["document_info"]["document_type"] == "office_invoice" assert uploaded_meta["requirement_check"]["matches"] is True updated = service.update_claim_item( claim_id=claim.id, item_id=claim.items[0].id, payload=ExpenseClaimItemUpdate( item_type="transport", item_reason="打车报销", ), current_user=current_user, ) assert updated is not None assert any(flag.get("source") == "attachment_analysis" for flag in updated.risk_flags_json) refreshed_meta = service.get_claim_item_attachment_meta( claim_id=claim.id, item_id=claim.items[0].id, current_user=current_user, ) assert refreshed_meta is not None assert refreshed_meta["analysis"]["severity"] == "high" assert refreshed_meta["requirement_check"]["matches"] is False assert any("附件类型要求" in point for point in refreshed_meta["analysis"]["points"]) def test_upload_train_ticket_attachment_backfills_item_amount(monkeypatch, tmp_path) -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: return OcrRecognizeBatchRead( total_file_count=1, success_count=1, documents=[ OcrRecognizeDocumentRead( filename="train-ticket.png", media_type="image/png", text="中国铁路电子客票 广州南-北京南 二等座 2026-02-20 08:30开 票价:￥354.00", summary="铁路电子客票，2026-02-20 08:30 广州南至北京南，票价 354 元。", avg_score=0.98, line_count=1, page_count=1, document_type="train_ticket", document_type_label="火车/高铁票", scene_code="travel", scene_label="差旅费", document_fields=[ {"key": "invoice_date", "label": "开票日期", "value": "2026-02-18"}, {"key": "trip_date", "label": "行程日期", "value": "2026-02-20 08:30"}, {"key": "fare", "label": "票价", "value": "￥354.00"}, ], ) ], ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: claim = build_claim(expense_type="travel", location="北京") claim.amount = Decimal("0.00") claim.invoice_count = 0 claim.items[0].item_amount = Decimal("0.00") claim.items[0].invoice_id = None db.add(claim) db.commit() service = ExpenseClaimService(db) updated = service.upload_claim_item_attachment( claim_id=claim.id, item_id=claim.items[0].id, filename="train-ticket.png", content=b"fake-image-bytes", media_type="image/png", current_user=current_user, ) assert updated is not None assert updated["item_amount"] == Decimal("354.00") assert updated["item_date"] == "2026-02-20" assert updated["item_type"] == "train_ticket" assert updated["item_reason"] == "广州南-北京南" assert updated["claim_amount"] == Decimal("354.00") db.refresh(claim) assert claim.items[0].item_amount == Decimal("354.00") assert claim.items[0].item_type == "train_ticket" assert claim.items[0].item_date == date(2026, 2, 20) assert claim.items[0].item_reason == "广州南-北京南" assert claim.amount == Decimal("354.00") uploaded_meta = service.get_claim_item_attachment_meta( claim_id=claim.id, item_id=claim.items[0].id, current_user=current_user, ) assert uploaded_meta is not None assert uploaded_meta["document_info"]["document_type"] == "train_ticket" assert any( field["label"] == "列车出发时间" and field["value"] == "2026-02-20 08:30" for field in uploaded_meta["document_info"]["fields"] ) assert any( field["label"] == "开票日期" and field["value"] == "2026-02-18" for field in uploaded_meta["document_info"]["fields"] ) assert any( field["label"] == "票价" and field["value"] == "￥354.00" for field in uploaded_meta["document_info"]["fields"] ) assert not any("用途字段" in point for point in uploaded_meta["analysis"]["points"]) def test_upload_hotel_attachment_audits_date_like_amount(monkeypatch, tmp_path) -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: return OcrRecognizeBatchRead( total_file_count=1, success_count=1, documents=[ OcrRecognizeDocumentRead( filename="hotel-invoice.png", media_type="image/png", text="北京中心酒店 总费用是828元 入住日期 2026-02-20 离店日期 2026-02-21", summary="酒店住宿票据，住宿总费用 828 元。", avg_score=0.96, line_count=1, page_count=1, document_type="hotel_invoice", document_type_label="酒店住宿票据", scene_code="hotel", scene_label="住宿票据", document_fields=[ {"key": "amount", "label": "金额", "value": "2026元"}, {"key": "hotel_name", "label": "酒店", "value": "北京中心酒店"}, {"key": "check_in", "label": "入住日期", "value": "2026-02-20"}, {"key": "check_out", "label": "离店日期", "value": "2026-02-21"}, ], ) ], ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: claim = build_claim(expense_type="hotel", location="北京") claim.amount = Decimal("0.00") claim.invoice_count = 0 claim.items[0].item_type = "hotel" claim.items[0].item_reason = "" claim.items[0].item_amount = Decimal("0.00") claim.items[0].invoice_id = None db.add(claim) db.commit() service = ExpenseClaimService(db) updated = service.upload_claim_item_attachment( claim_id=claim.id, item_id=claim.items[0].id, filename="hotel-invoice.png", content=b"fake-image-bytes", media_type="image/png", current_user=current_user, ) assert updated is not None assert updated["item_type"] == "hotel_ticket" assert updated["item_amount"] == Decimal("828.00") assert updated["claim_amount"] == Decimal("828.00") db.refresh(claim) assert claim.items[0].item_amount == Decimal("828.00") assert claim.amount == Decimal("828.00") uploaded_meta = service.get_claim_item_attachment_meta( claim_id=claim.id, item_id=claim.items[0].id, current_user=current_user, ) assert uploaded_meta is not None assert uploaded_meta["analysis"]["severity"] == "medium" assert any("费用核算" in point and "828.00 元" in point for point in uploaded_meta["analysis"]["points"]) assert not any("2026.00 元与报销金额" in point for point in uploaded_meta["analysis"]["points"]) def test_attachment_analysis_does_not_compare_business_purpose_with_ticket_scene() -> None: with build_session() as db: claim = build_claim(expense_type="travel", location="上海") claim.items[0].item_type = "train_ticket" claim.items[0].item_reason = "2026-02-20 至 2026-02-23，支撑上海电力项目部署" claim.items[0].item_amount = Decimal("354.00") db.add(claim) db.commit() document = OcrRecognizeDocumentRead( filename="train-ticket.png", media_type="image/png", text="中国铁路电子客票 上海虹桥-武汉 二等座 2026-02-20 票价:￥354.00", summary="铁路电子客票，上海虹桥至武汉，票价 354 元。", avg_score=0.98, line_count=1, page_count=1, document_type="train_ticket", document_type_label="火车/高铁票", scene_code="travel", scene_label="差旅票据", document_fields=[ {"key": "amount", "label": "票价", "value": "￥354.00"}, {"key": "date", "label": "日期", "value": "2026-02-20"}, {"key": "route", "label": "行程", "value": "上海虹桥-武汉"}, ], ) analysis = ExpenseClaimService(db)._build_attachment_analysis( document=document, item=claim.items[0], ) assert analysis["severity"] == "medium" assert not any("用途字段" in point for point in analysis["points"]) assert any("行程说明" in point and "起始地-目的地" in point for point in analysis["points"]) def test_attachment_risk_flag_message_uses_specific_points(monkeypatch, tmp_path) -> None: with build_session() as db: claim = build_claim(expense_type="travel", location="上海") claim.items[0].invoice_id = "invoice.png" db.add(claim) db.commit() generic_summary = "当前附件可见部分内容，但金额、用途、日期或附件类型仍有缺失或不一致。" file_path = tmp_path / "invoice.png" file_path.write_bytes(b"fake") service = ExpenseClaimService(db) monkeypatch.setattr(service, "_resolve_attachment_path", lambda storage_key: file_path) monkeypatch.setattr( service, "_read_attachment_meta", lambda path: { "analysis": { "severity": "medium", "label": "中风险", "summary": generic_summary, "points": [ "日期字段：未识别到开票日期或业务发生日期。", "金额字段：附件识别金额 300.00 元与报销金额 88.00 元不一致。", ], } }, ) flags = service._build_claim_attachment_risk_flags([claim.items[0]]) assert len(flags) == 1 assert "日期字段：未识别到开票日期或业务发生日期。" in flags[0]["message"] assert "当前附件可见部分内容" not in flags[0]["message"] assert flags[0]["summary"] == generic_summary assert flags[0]["points"] == [ "日期字段：未识别到开票日期或业务发生日期。", "金额字段：附件识别金额 300.00 元与报销金额 88.00 元不一致。", ] def test_upload_ride_receipt_backfills_item_reason_from_addresses(monkeypatch, tmp_path) -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: return OcrRecognizeBatchRead( total_file_count=1, success_count=1, documents=[ OcrRecognizeDocumentRead( filename="ride-receipt.png", media_type="image/png", text="滴滴出行订单 起点：深圳北站 终点：腾讯滨海大厦 实付金额：42.00元", summary="滴滴出行乘车票据，深圳北站到腾讯滨海大厦，金额 42 元。", avg_score=0.98, line_count=1, page_count=1, document_type="taxi_receipt", document_type_label="出租车/网约车票据", scene_code="transport", scene_label="交通票据", document_fields=[ {"key": "start_location", "label": "起点", "value": "深圳北站"}, {"key": "end_location", "label": "终点", "value": "腾讯滨海大厦"}, {"key": "amount", "label": "实付金额", "value": "42.00元"}, ], ) ], ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: claim = build_claim(expense_type="transport", location="深圳") claim.amount = Decimal("0.00") claim.invoice_count = 0 claim.items[0].item_type = "transport" claim.items[0].item_reason = "打车报销" claim.items[0].item_amount = Decimal("0.00") claim.items[0].invoice_id = None db.add(claim) db.commit() service = ExpenseClaimService(db) updated = service.upload_claim_item_attachment( claim_id=claim.id, item_id=claim.items[0].id, filename="ride-receipt.png", content=b"fake-image-bytes", media_type="image/png", current_user=current_user, ) assert updated is not None db.refresh(claim) assert claim.items[0].item_type == "ride_ticket" assert claim.items[0].item_reason == "深圳北站-腾讯滨海大厦" assert claim.items[0].item_amount == Decimal("42.00") assert claim.amount == Decimal("42.00") def test_delete_claim_item_removes_row_and_attachment_files(monkeypatch, tmp_path) -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: return OcrRecognizeBatchRead( total_file_count=1, success_count=1, documents=[ OcrRecognizeDocumentRead( filename="office-note.png", media_type="image/png", text="办公用品发票 金额88元 2026-05-13", summary="识别到办公用品发票，金额 88 元。", avg_score=0.98, line_count=1, page_count=1, warnings=[], ) ], ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: claim = build_claim(expense_type="office", location="深圳南山") claim.invoice_count = 0 claim.items[0].invoice_id = None claim.items[0].item_reason = "办公用品采购" db.add(claim) db.commit() service = ExpenseClaimService(db) upload_payload = service.upload_claim_item_attachment( claim_id=claim.id, item_id=claim.items[0].id, filename="office-note.png", content=b"fake-image-bytes", media_type="image/png", current_user=current_user, ) assert upload_payload is not None attachment_root = tmp_path / claim.id / claim.items[0].id assert attachment_root.exists() delete_payload = service.delete_claim_item( claim_id=claim.id, item_id=claim.items[0].id, current_user=current_user, ) assert delete_payload is not None assert delete_payload["claim_id"] == claim.id refreshed_claim = service.get_claim(claim.id, current_user) assert refreshed_claim is not None assert refreshed_claim.items == [] assert refreshed_claim.amount == Decimal("0.00") assert refreshed_claim.invoice_count == 0 assert not attachment_root.exists() def test_attachment_preview_resolves_legacy_filename_in_claim_item_directory(monkeypatch, tmp_path) -> None: current_user = CurrentUserContext( username="emp-1", name="张三", role_codes=[], is_admin=False, ) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: claim = build_claim(expense_type="transport", location="上海") claim.items[0].invoice_id = "legacy-ticket.pdf" db.add(claim) db.commit() attachment_dir = tmp_path / claim.id / claim.items[0].id attachment_dir.mkdir(parents=True) file_path = attachment_dir / "legacy-ticket.pdf" file_path.write_bytes(b"legacy-pdf-bytes") (attachment_dir / "legacy-ticket.pdf.meta.json").write_text( '{"file_name":"legacy-ticket.pdf","media_type":"application/pdf","previewable":true}', encoding="utf-8", ) payload = ExpenseClaimService(db).get_claim_item_attachment_preview_content( claim_id=claim.id, item_id=claim.items[0].id, current_user=current_user, ) assert payload is not None resolved_path, media_type, filename = payload assert resolved_path == file_path assert media_type == "application/pdf" assert filename == "legacy-ticket.pdf" def test_submit_claim_runs_ai_review_and_routes_to_direct_manager() -> None: current_user = CurrentUserContext( username="emp-submit@example.com", name="张三", role_codes=[], is_admin=False, ) with build_session() as db: manager = Employee( employee_no="E7000", name="李经理", email="manager@example.com", ) employee = Employee( employee_no="E7001", name="张三", email="emp-submit@example.com", manager=manager, ) claim = build_claim(expense_type="transport", location="上海") claim.employee = employee claim.employee_id = employee.id claim.items[0].invoice_id = "taxi-ticket.png" db.add_all([manager, employee, claim]) db.commit() submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.approval_stage == "直属领导审批" assert submitted.submitted_at is not None def test_submit_claim_allows_returned_claim_to_be_resubmitted() -> None: current_user = CurrentUserContext( username="emp-submit@example.com", name="张三", role_codes=[], is_admin=False, ) with build_session() as db: manager = Employee( employee_no="E7100", name="李经理", email="manager-returned@example.com", ) employee = Employee( employee_no="E7101", name="张三", email="emp-submit@example.com", manager=manager, ) claim = build_claim(expense_type="transport", location="上海") claim.employee = employee claim.employee_id = employee.id claim.status = "returned" claim.approval_stage = "待补充" claim.items[0].invoice_id = "taxi-ticket.png" db.add_all([manager, employee, claim]) db.commit() submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.approval_stage == "直属领导审批" assert submitted.submitted_at is not None def test_submit_claim_backfills_department_from_current_employee() -> None: current_user = CurrentUserContext( username="emp-dept@example.com", name="张三", role_codes=[], is_admin=False, ) with build_session() as db: department = OrganizationUnit( unit_code="D7200", name="销售部", ) manager = Employee( employee_no="E7200", name="李经理", email="manager-dept@example.com", ) employee = Employee( employee_no="E7201", name="张三", email="emp-dept@example.com", organization_unit=department, manager=manager, ) claim = build_claim(expense_type="transport", location="待补充") claim.employee = None claim.employee_id = None claim.employee_name = "张三" claim.department_id = None claim.department_name = "待补充" claim.items[0].item_location = "待补充" db.add_all([department, manager, employee, claim]) db.commit() submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.department_id == department.id assert submitted.department_name == "销售部" assert submitted.approval_stage == "直属领导审批" def test_submit_claim_routes_high_risk_attachment_to_approval_with_review_flag( monkeypatch, tmp_path, ) -> None: current_user = CurrentUserContext( username="emp-risk@example.com", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: return OcrRecognizeBatchRead( total_file_count=1, success_count=1, documents=[ OcrRecognizeDocumentRead( filename="taxi-note.png", media_type="image/png", text="滴滴出行电子发票 金额120元 2026-05-13", summary="识别到交通出行发票，金额 120 元。", avg_score=0.97, line_count=1, page_count=1, warnings=[], ) ], ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: manager = Employee( employee_no="E7100", name="李经理", email="manager2@example.com", ) employee = Employee( employee_no="E7101", name="张三", email="emp-risk@example.com", manager=manager, ) claim = build_claim(expense_type="office", location="深圳南山") claim.employee = employee claim.employee_id = employee.id claim.invoice_count = 0 claim.items[0].invoice_id = None claim.items[0].item_reason = "办公用品采购" db.add_all([manager, employee, claim]) db.commit() service = ExpenseClaimService(db) service.upload_claim_item_attachment( claim_id=claim.id, item_id=claim.items[0].id, filename="taxi-note.png", content=b"fake-image-bytes", media_type="image/png", current_user=current_user, ) submitted = service.submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.approval_stage == "直属领导审批" assert submitted.submitted_at is not None assert any( isinstance(flag, dict) and str(flag.get("source") or "").strip() == "submission_review" for flag in list(submitted.risk_flags_json or []) ) def test_submit_claim_routes_travel_route_mismatch_to_approval_with_review_flag( monkeypatch, tmp_path, ) -> None: current_user = CurrentUserContext( username="emp-travel@example.com", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: documents: list[OcrRecognizeDocumentRead] = [] for filename, _, media_type in files: if filename == "outbound.png": documents.append( OcrRecognizeDocumentRead( filename=filename, media_type=media_type or "image/png", text="电子行程单 2026-05-13 经济舱 武汉-上海 金额 480元 航班号 MU5101", summary="武汉到上海机票", avg_score=0.98, line_count=1, page_count=1, document_type="flight_itinerary", document_type_label="机票/航班行程单", scene_code="travel", scene_label="差旅票据", document_fields=[ {"key": "route", "label": "行程", "value": "武汉-上海"}, {"key": "amount", "label": "金额", "value": "480元"}, {"key": "date", "label": "日期", "value": "2026-05-13"}, ], warnings=[], ) ) elif filename == "onward.png": documents.append( OcrRecognizeDocumentRead( filename=filename, media_type=media_type or "image/png", text="电子行程单 2026-05-14 经济舱 上海-成都 金额 360元 航班号 MU5402", summary="上海到成都机票", avg_score=0.98, line_count=1, page_count=1, document_type="flight_itinerary", document_type_label="机票/航班行程单", scene_code="travel", scene_label="差旅票据", document_fields=[ {"key": "route", "label": "行程", "value": "上海-成都"}, {"key": "amount", "label": "金额", "value": "360元"}, {"key": "date", "label": "日期", "value": "2026-05-14"}, ], warnings=[], ) ) return OcrRecognizeBatchRead( total_file_count=len(files), success_count=len(documents), documents=documents, ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: manager = Employee( employee_no="E7200", name="李经理", email="manager-travel@example.com", ) employee = Employee( employee_no="E7201", name="张三", email="emp-travel@example.com", grade="P4", manager=manager, ) db.add_all([manager, employee]) db.flush() claim = build_claim(expense_type="travel", location="上海") claim.reason = "上海客户现场出差" claim.employee = employee claim.employee_id = employee.id claim.items = [ ExpenseClaimItem( id="travel-item-1", claim_id=claim.id, item_date=date(2026, 5, 13), item_type="travel", item_reason="赴上海客户现场", item_location="上海", item_amount=Decimal("480.00"), invoice_id=None, ), ExpenseClaimItem( id="travel-item-2", claim_id=claim.id, item_date=date(2026, 5, 14), item_type="travel", item_reason="赴上海客户现场", item_location="上海", item_amount=Decimal("360.00"), invoice_id=None, ), ] claim.amount = Decimal("840.00") claim.invoice_count = 0 db.add(claim) db.commit() service = ExpenseClaimService(db) service.upload_claim_item_attachment( claim_id=claim.id, item_id="travel-item-1", filename="outbound.png", content=b"outbound-image", media_type="image/png", current_user=current_user, ) service.upload_claim_item_attachment( claim_id=claim.id, item_id="travel-item-2", filename="onward.png", content=b"onward-image", media_type="image/png", current_user=current_user, ) submitted = service.submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.approval_stage == "直属领导审批" assert any( isinstance(flag, dict) and str(flag.get("source") or "").strip() == "submission_review" and ( "多城市" in str(flag.get("message") or "") or "终点" in str(flag.get("message") or "") ) for flag in list(submitted.risk_flags_json or []) ) def test_submit_claim_routes_hotel_amount_over_travel_policy_to_approval_with_review_flag( monkeypatch, tmp_path, ) -> None: current_user = CurrentUserContext( username="emp-hotel@example.com", name="张三", role_codes=[], is_admin=False, ) def fake_recognize( self, files: list[tuple[str, bytes, str | None]], ) -> OcrRecognizeBatchRead: documents: list[OcrRecognizeDocumentRead] = [] for filename, _, media_type in files: if filename == "beijing-trip.png": documents.append( OcrRecognizeDocumentRead( filename=filename, media_type=media_type or "image/png", text="电子行程单 2026-05-13 经济舱 武汉-北京 金额 520元 航班号 MU6101", summary="武汉到北京机票", avg_score=0.97, line_count=1, page_count=1, document_type="flight_itinerary", document_type_label="机票/航班行程单", scene_code="travel", scene_label="差旅票据", document_fields=[ {"key": "route", "label": "行程", "value": "武汉-北京"}, {"key": "amount", "label": "金额", "value": "520元"}, {"key": "date", "label": "日期", "value": "2026-05-13"}, ], warnings=[], ) ) elif filename == "beijing-hotel.png": documents.append( OcrRecognizeDocumentRead( filename=filename, media_type=media_type or "image/png", text="北京全季酒店 1晚 金额 880元 2026-05-13", summary="北京全季酒店住宿发票", avg_score=0.98, line_count=1, page_count=1, document_type="hotel_invoice", document_type_label="酒店住宿票据", scene_code="hotel", scene_label="住宿票据", document_fields=[ {"key": "merchant_name", "label": "商户", "value": "北京全季酒店"}, {"key": "amount", "label": "金额", "value": "880元"}, {"key": "date", "label": "日期", "value": "2026-05-13"}, ], warnings=[], ) ) return OcrRecognizeBatchRead( total_file_count=len(files), success_count=len(documents), documents=documents, ) monkeypatch.setattr(OcrService, "recognize_files", fake_recognize) monkeypatch.setattr(ExpenseClaimService, "_get_attachment_storage_root", lambda self: tmp_path) with build_session() as db: manager = Employee( employee_no="E7300", name="李经理", email="manager-hotel@example.com", ) employee = Employee( employee_no="E7301", name="张三", email="emp-hotel@example.com", grade="P4", manager=manager, ) db.add_all([manager, employee]) db.flush() claim = build_claim(expense_type="travel", location="北京") claim.reason = "北京客户现场出差" claim.employee = employee claim.employee_id = employee.id claim.items = [ ExpenseClaimItem( id="hotel-trip-item", claim_id=claim.id, item_date=date(2026, 5, 13), item_type="travel", item_reason="赴北京客户现场", item_location="北京", item_amount=Decimal("520.00"), invoice_id=None, ), ExpenseClaimItem( id="hotel-item", claim_id=claim.id, item_date=date(2026, 5, 13), item_type="hotel", item_reason="北京住宿", item_location="北京", item_amount=Decimal("880.00"), invoice_id=None, ), ] claim.amount = Decimal("1400.00") claim.invoice_count = 0 db.add(claim) db.commit() service = ExpenseClaimService(db) service.upload_claim_item_attachment( claim_id=claim.id, item_id="hotel-trip-item", filename="beijing-trip.png", content=b"travel-image", media_type="image/png", current_user=current_user, ) service.upload_claim_item_attachment( claim_id=claim.id, item_id="hotel-item", filename="beijing-hotel.png", content=b"hotel-image", media_type="image/png", current_user=current_user, ) submitted = service.submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.approval_stage == "直属领导审批" assert any( isinstance(flag, dict) and str(flag.get("source") or "").strip() == "submission_review" and "住宿标准" in str(flag.get("message") or "") for flag in list(submitted.risk_flags_json or []) ) def test_list_claims_scopes_to_current_user_id_even_when_names_duplicate() -> None: current_user = CurrentUserContext( username="zhangsan1@example.com", name="张三", role_codes=["manager"], is_admin=False, ) with build_session() as db: employee_a = Employee( employee_no="E2001", name="张三", email="zhangsan1@example.com", ) employee_b = Employee( employee_no="E2002", name="张三", email="zhangsan2@example.com", ) db.add_all([employee_a, employee_b]) db.flush() db.add_all( [ ExpenseClaim( claim_no="EXP-DUP-001", employee_id=employee_a.id, employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="travel", reason="本人报销", location="上海", amount=Decimal("120.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="finance_review", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-DUP-002", employee_id=employee_b.id, employee_name="张三", department_name="销售部", project_code="PRJ-B", expense_type="meal", reason="他人报销", location="杭州", amount=Decimal("300.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC), status="approved", approval_stage="completed", risk_flags_json=[], ), ] ) db.commit() claims = ExpenseClaimService(db).list_claims(current_user) assert len(claims) == 1 assert claims[0].claim_no == "EXP-DUP-001" def test_list_claims_allows_finance_to_view_all_records() -> None: current_user = CurrentUserContext( username="finance@example.com", name="财务", role_codes=["finance"], is_admin=False, ) with build_session() as db: db.add_all( [ ExpenseClaim( claim_no="EXP-FIN-101", employee_name="甲", department_name="A部", project_code="PRJ-A", expense_type="travel", reason="A 报销", location="上海", amount=Decimal("120.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC), status="submitted", approval_stage="finance_review", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-FIN-102", employee_name="乙", department_name="B部", project_code="PRJ-B", expense_type="meal", reason="B 报销", location="杭州", amount=Decimal("300.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, 12, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 11, 13, 0, tzinfo=UTC), status="approved", approval_stage="completed", risk_flags_json=[], ), ] ) db.commit() claims = ExpenseClaimService(db).list_claims(current_user) assert len(claims) == 2 assert {claim.claim_no for claim in claims} == {"EXP-FIN-101", "EXP-FIN-102"} def test_list_claims_allows_executive_to_view_all_records() -> None: current_user = CurrentUserContext( username="executive@example.com", name="高管", role_codes=["executive"], is_admin=False, ) with build_session() as db: db.add_all( [ ExpenseClaim( claim_no="EXP-EXE-101", employee_name="甲", department_name="A部", project_code="PRJ-A", expense_type="travel", reason="A 报销", location="上海", amount=Decimal("120.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-EXE-102", employee_name="乙", department_name="B部", project_code="PRJ-B", expense_type="meal", reason="B 报销", location="杭州", amount=Decimal("300.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, 12, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 11, 13, 0, tzinfo=UTC), status="approved", approval_stage="completed", risk_flags_json=[], ), ] ) db.commit() claims = ExpenseClaimService(db).list_claims(current_user) assert len(claims) == 2 assert {claim.claim_no for claim in claims} == {"EXP-EXE-101", "EXP-EXE-102"} def test_finance_can_return_but_cannot_delete_submitted_claim() -> None: current_user = CurrentUserContext( username="finance@example.com", name="财务", role_codes=["finance"], is_admin=False, ) with build_session() as db: claim = ExpenseClaim( claim_no="EXP-RET-101", employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="travel", reason="差旅报销", location="上海", amount=Decimal("120.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ) db.add(claim) db.commit() claim_id = claim.id service = ExpenseClaimService(db) returned = service.return_claim(claim_id, current_user, reason="资料不完整") assert returned is not None assert returned.status == "returned" assert returned.approval_stage == "待提交" assert any( isinstance(flag, dict) and flag.get("source") == "manual_return" and flag.get("message") == "资料不完整" for flag in returned.risk_flags_json ) with pytest.raises(ValueError, match="只有高级管理人员可以删除"): service.delete_claim(claim_id, current_user) assert db.get(ExpenseClaim, claim_id) is not None def test_executive_can_delete_submitted_claim() -> None: current_user = CurrentUserContext( username="executive-delete@example.com", name="高管", role_codes=["executive"], is_admin=False, ) with build_session() as db: claim = ExpenseClaim( claim_no="EXP-DEL-EXEC-101", employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="travel", reason="差旅报销", location="上海", amount=Decimal("120.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC), status="submitted", approval_stage="财务审批", risk_flags_json=[], ) db.add(claim) db.commit() claim_id = claim.id deleted = ExpenseClaimService(db).delete_claim(claim_id, current_user) assert deleted is not None assert deleted.claim_no == "EXP-DEL-EXEC-101" assert db.get(ExpenseClaim, claim_id) is None def test_direct_manager_can_return_subordinate_claim_to_pending_submission() -> None: current_user = CurrentUserContext( username="manager-return@example.com", name="李经理", role_codes=["manager"], is_admin=False, ) with build_session() as db: manager = Employee( employee_no="E8100", name="李经理", email="manager-return@example.com", ) employee = Employee( employee_no="E8101", name="张三", email="zhangsan-return@example.com", manager=manager, ) db.add_all([manager, employee]) db.flush() claim = ExpenseClaim( claim_no="EXP-RET-201", employee_id=employee.id, employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="transport", reason="交通报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ) db.add(claim) db.commit() claim_id = claim.id returned = ExpenseClaimService(db).return_claim(claim_id, current_user, reason="请补充行程说明") assert returned is not None assert returned.status == "returned" assert returned.approval_stage == "待提交" assert returned.submitted_at is None assert any( isinstance(flag, dict) and flag.get("source") == "manual_return" and flag.get("message") == "请补充行程说明" for flag in returned.risk_flags_json ) def test_direct_manager_can_approve_subordinate_claim_to_finance_review() -> None: current_user = CurrentUserContext( username="manager-approve@example.com", name="李经理", role_codes=["manager"], is_admin=False, ) with build_session() as db: manager = Employee( employee_no="E8110", name="李经理", email="manager-approve@example.com", ) employee = Employee( employee_no="E8111", name="张三", email="zhangsan-approve@example.com", manager=manager, ) db.add_all([manager, employee]) db.flush() claim = ExpenseClaim( claim_no="EXP-APP-201", employee_id=employee.id, employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="transport", reason="交通报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ) db.add(claim) db.commit() claim_id = claim.id approved = ExpenseClaimService(db).approve_claim( claim_id, current_user, opinion="情况属实，同意报销。", ) assert approved is not None assert approved.status == "submitted" assert approved.approval_stage == "财务审批" assert approved.submitted_at is not None assert any( isinstance(flag, dict) and flag.get("source") == "manual_approval" and flag.get("event_type") == "expense_claim_approval" and flag.get("opinion") == "情况属实，同意报销。" and flag.get("previous_approval_stage") == "直属领导审批" and flag.get("next_approval_stage") == "财务审批" for flag in approved.risk_flags_json ) def test_finance_can_approve_claim_to_archive_stage() -> None: current_user = CurrentUserContext( username="finance-approve@example.com", name="财务复核", role_codes=["finance"], is_admin=False, ) with build_session() as db: claim = ExpenseClaim( claim_no="EXP-FIN-APP-201", employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="transport", reason="交通报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="财务审批", risk_flags_json=[], ) db.add(claim) db.commit() claim_id = claim.id approved = ExpenseClaimService(db).approve_claim( claim_id, current_user, opinion="票据与明细一致，同意入账。", ) assert approved is not None assert approved.status == "approved" assert approved.approval_stage == "归档入账" assert any( isinstance(flag, dict) and flag.get("source") == "finance_approval" and flag.get("event_type") == "expense_claim_finance_approval" and flag.get("opinion") == "票据与明细一致，同意入账。" and flag.get("previous_approval_stage") == "财务审批" and flag.get("next_approval_stage") == "归档入账" for flag in approved.risk_flags_json ) def test_return_claim_rejects_already_returned_claim_without_adding_event() -> None: current_user = CurrentUserContext( username="finance-returned@example.com", name="财务", role_codes=["finance"], is_admin=False, ) return_flag = { "source": "manual_return", "return_event_id": "return-event-existing", "message": "请补充附件。", "return_count": 1, "return_stage": "直属领导审批", "return_stage_key": "direct_manager", } with build_session() as db: claim = ExpenseClaim( claim_no="EXP-RET-202", employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="transport", reason="交通报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=None, status="returned", approval_stage="待提交", risk_flags_json=[return_flag], ) db.add(claim) db.commit() claim_id = claim.id with pytest.raises(ValueError, match="无需重复退回"): ExpenseClaimService(db).return_claim(claim_id, current_user, reason="重复退回") db.refresh(claim) manual_returns = [ flag for flag in list(claim.risk_flags_json or []) if isinstance(flag, dict) and flag.get("source") == "manual_return" ] assert manual_returns == [return_flag] def test_return_claim_records_each_return_event_with_stage_reason_and_counts() -> None: current_user = CurrentUserContext( username="finance-return@example.com", name="财务复核", role_codes=["finance"], is_admin=False, ) with build_session() as db: claim = ExpenseClaim( claim_no="EXP-RET-301", employee_name="张三", department_name="市场部", project_code="PRJ-A", expense_type="transport", reason="交通报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ) db.add(claim) db.commit() claim_id = claim.id service = ExpenseClaimService(db) first_returned = service.return_claim( claim_id, current_user, reason="发票金额与明细金额不一致，请重新核对。", reason_codes=["invoice_mismatch", "business_explanation"], ) assert first_returned is not None first_returned.status = "submitted" first_returned.approval_stage = "财务审批" first_returned.submitted_at = datetime(2026, 5, 12, 11, 0, tzinfo=UTC) db.commit() second_returned = service.return_claim( claim_id, current_user, reason="超标说明仍不完整，请补充制度例外依据。", reason_codes=["over_policy"], ) assert second_returned is not None return_events = [ flag for flag in list(second_returned.risk_flags_json or []) if isinstance(flag, dict) and flag.get("source") == "manual_return" ] assert len(return_events) == 2 assert return_events[0]["return_count"] == 1 assert return_events[0]["stage_return_count"] == 1 assert return_events[0]["return_stage"] == "直属领导审批" assert return_events[0]["reason_codes"] == ["invoice_mismatch", "business_explanation"] assert return_events[0]["risk_points"] == ["票据类型/金额与明细不一致", "业务事由/地点/人员信息不完整"] assert return_events[0]["reason"] == "发票金额与明细金额不一致，请重新核对。" assert return_events[0]["operator_role_codes"] == ["finance"] assert return_events[1]["return_count"] == 2 assert return_events[1]["stage_return_count"] == 1 assert return_events[1]["return_stage"] == "财务审批" assert return_events[1]["risk_points"] == ["超出制度标准或缺少超标说明"] def test_submit_returned_claim_preserves_manual_return_events() -> None: current_user = CurrentUserContext( username="emp-submit-returned@example.com", name="张三", role_codes=[], is_admin=False, ) return_flag = { "source": "manual_return", "return_event_id": "return-event-submit", "message": "第一次退回：业务说明不完整。", "reason": "业务说明不完整。", "return_count": 1, "return_stage": "直属领导审批", "return_stage_key": "direct_manager", "risk_points": ["业务事由/地点/人员信息不完整"], } with build_session() as db: manager = Employee( employee_no="E8200", name="李经理", email="manager-submit-returned@example.com", ) employee = Employee( employee_no="E8201", name="张三", email="emp-submit-returned@example.com", manager=manager, ) claim = build_claim(expense_type="office", location="上海") claim.employee = employee claim.employee_id = employee.id claim.employee_name = "张三" claim.department_name = "市场部" claim.status = "returned" claim.approval_stage = "待提交" claim.risk_flags_json = [return_flag] db.add_all([manager, employee, claim]) db.commit() conversation = AgentConversationService(db).get_or_create_conversation( conversation_id=None, user_id=current_user.username, source="user_message", context_json={ "session_type": "expense", "draft_claim_id": claim.id, }, ) conversation_id = conversation.conversation_id submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user) assert submitted is not None assert submitted.status == "submitted" assert submitted.approval_stage == "直属领导审批" assert any( isinstance(flag, dict) and flag.get("source") == "manual_return" and flag.get("return_event_id") == "return-event-submit" for flag in list(submitted.risk_flags_json or []) ) assert AgentConversationService(db).get_conversation(conversation_id) is None def test_manager_personal_claims_exclude_subordinate_pending_approval_claims() -> None: current_user = CurrentUserContext( username="manager-personal@example.com", name="李经理", role_codes=["manager"], is_admin=False, ) with build_session() as db: manager = Employee( employee_no="E8300", name="李经理", email="manager-personal@example.com", ) employee = Employee( employee_no="E8301", name="张三", email="zhangsan-personal@example.com", manager=manager, ) db.add_all([manager, employee]) db.flush() db.add_all( [ ExpenseClaim( claim_no="EXP-MGR-OWN", employee_id=manager.id, employee_name="李经理", department_name="市场部", project_code="PRJ-MGR", expense_type="office", reason="本人报销", location="上海", amount=Decimal("88.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=None, status="draft", approval_stage="待提交", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-MGR-SUB", employee_id=employee.id, employee_name="张三", department_name="市场部", project_code="PRJ-MGR", expense_type="transport", reason="下属待审批报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 11, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ), ] ) db.commit() service = ExpenseClaimService(db) personal_claims = service.list_claims(current_user) approval_claims = service.list_approval_claims(current_user) assert [claim.claim_no for claim in personal_claims] == ["EXP-MGR-OWN"] assert [claim.claim_no for claim in approval_claims] == ["EXP-MGR-SUB"] def test_list_approval_claims_allows_direct_manager_to_view_pending_claims_for_approval() -> None: current_user = CurrentUserContext( username="manager@example.com", name="李经理", role_codes=["manager"], is_admin=False, ) with build_session() as db: manager = Employee( employee_no="E8000", name="李经理", email="manager@example.com", ) employee = Employee( employee_no="E8001", name="张三", email="zhangsan@example.com", manager=manager, ) outsider_manager = Employee( employee_no="E8002", name="王经理", email="other-manager@example.com", ) outsider = Employee( employee_no="E8003", name="李四", email="lisi@example.com", manager=outsider_manager, ) db.add_all([manager, employee, outsider_manager, outsider]) db.flush() db.add_all( [ ExpenseClaim( claim_no="EXP-MGR-201", employee_id=employee.id, employee_name="张三", department_name="市场部", project_code="PRJ-MGR", expense_type="transport", reason="滴滴报销", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-MGR-202", employee_id=outsider.id, employee_name="李四", department_name="销售部", project_code="PRJ-OTHER", expense_type="meal", reason="客户用餐", location="杭州", amount=Decimal("188.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ), ] ) db.commit() claims = ExpenseClaimService(db).list_approval_claims(current_user) assert len(claims) == 1 assert claims[0].claim_no == "EXP-MGR-201" def test_list_approval_claims_limits_finance_to_finance_stage_claims() -> None: current_user = CurrentUserContext( username="finance-approval-list@example.com", name="财务", role_codes=["finance"], is_admin=False, ) with build_session() as db: db.add_all( [ ExpenseClaim( claim_no="EXP-FIN-LIST-201", employee_name="张三", department_name="市场部", project_code="PRJ-FIN", expense_type="transport", reason="直属领导待审", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC), status="submitted", approval_stage="直属领导审批", risk_flags_json=[], ), ExpenseClaim( claim_no="EXP-FIN-LIST-202", employee_name="李四", department_name="销售部", project_code="PRJ-FIN", expense_type="meal", reason="财务待审", location="杭州", amount=Decimal("188.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC), submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC), status="submitted", approval_stage="财务审批", risk_flags_json=[], ), ] ) db.commit() claims = ExpenseClaimService(db).list_approval_claims(current_user) assert [claim.claim_no for claim in claims] == ["EXP-FIN-LIST-202"]