X-Financial/server/tests/test_expense_case_service.py

from __future__ import annotations

from datetime import UTC, date, datetime
from decimal import Decimal

import pytest
from sqlalchemy import create_engine, inspect, select
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.audit_log import AuditLog
from app.models.budget import BudgetAllocation, BudgetReservation, BudgetTransaction
from app.models.employee import Employee
from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
from app.models.organization import OrganizationUnit
from app.services.agent_foundation import AgentFoundationService
from app.services.expense_cases import ExpenseCaseService
from app.services.expense_claim_errors import ExpenseClaimPreReviewBlockedError
from app.services.expense_claim_workflow_constants import (
    APPLICATION_ARCHIVE_STAGE,
    APPLICATION_LINK_STATUS_STAGE,
    DIRECT_MANAGER_APPROVAL_STAGE,
)
from app.services.expense_claims import ExpenseClaimService
from app.services.tenant_registry import TenantRegistryService


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 build_claim(
    *,
    claim_no: str,
    employee: Employee | None = None,
    status: str = "draft",
    approval_stage: str = "待提交",
    expense_type: str = "transport",
) -> ExpenseClaim:
    claim = ExpenseClaim(
        claim_no=claim_no,
        employee_id=employee.id if employee is not None else None,
        employee_name=employee.name if employee is not None else "张三",
        department_name="市场部",
        project_code="PRJ-CASE",
        expense_type=expense_type,
        reason="客户现场差旅",
        location="上海",
        amount=Decimal("88.00"),
        currency="CNY",
        invoice_count=1,
        occurred_at=datetime(2026, 7, 13, 9, 0, tzinfo=UTC),
        submitted_at=(datetime(2026, 7, 13, 10, 0, tzinfo=UTC) if status != "draft" else None),
        status=status,
        approval_stage=approval_stage,
        risk_flags_json=[],
    )
    if status == "draft":
        claim.items = [
            ExpenseClaimItem(
                item_date=date(2026, 7, 13),
                item_type="transport",
                item_reason="客户现场交通",
                item_location="上海",
                item_note="",
                item_amount=Decimal("88.00"),
                invoice_id="invoice-case-1",
            )
        ]
    return claim


def test_event_write_uses_caller_transaction_and_tenant_scope() -> None:
    with build_session() as db:
        claim = build_claim(claim_no="RE-CASE-ROLLBACK")
        db.add(claim)
        db.commit()

        service = ExpenseCaseService(db)
        service.record_claim_event(
            claim,
            event_type="claim_draft_created",
            actor_id="owner@example.com",
            tenant_id="tenant-a",
            correlation_id="run-case-rollback",
        )

        assert db.scalar(select(ExpenseCase)) is not None
        assert db.scalar(select(BusinessEvent)) is not None
        db.rollback()

        assert db.scalar(select(ExpenseCase)) is None
        assert db.scalar(select(ExpenseCaseLink)) is None
        assert db.scalar(select(BusinessEvent)) is None

        service.record_claim_event(
            claim,
            event_type="claim_draft_created",
            actor_id="owner@example.com",
            tenant_id="tenant-a",
        )
        db.commit()

        assert service.get_timeline_for_claim(claim.id, tenant_id="tenant-a") is not None
        assert service.get_timeline_for_claim(claim.id, tenant_id="tenant-b") is None
        with pytest.raises(PermissionError, match="其他租户"):
            service.record_claim_event(
                claim,
                event_type="claim_draft_updated",
                actor_id="other@example.com",
                tenant_id="tenant-b",
            )


def test_pre_review_records_one_idempotent_expense_case_event() -> None:
    current_user = CurrentUserContext(
        username="pre-review-owner@example.com",
        name="张三",
        role_codes=["user"],
        is_admin=False,
        tenant_id="default",
    )
    with build_session() as db:
        manager = Employee(
            employee_no="PRE-REVIEW-MANAGER",
            name="李经理",
            email="pre-review-manager@example.com",
        )
        employee = Employee(
            employee_no="PRE-REVIEW-OWNER",
            name="张三",
            email=current_user.username,
            manager=manager,
        )
        claim = build_claim(claim_no="RE-CASE-PRE-REVIEW", employee=employee)
        claim.risk_flags_json = [
            {
                "source": "manual_risk",
                "severity": "high",
                "label": "票据风险",
                "message": "票据金额与行程不匹配。",
            }
        ]
        db.add_all([manager, employee, claim])
        db.commit()

        service = ExpenseClaimService(db)
        first = service.pre_review_claim(
            claim.id,
            current_user,
            correlation_id="pre-review-request-1",
            idempotency_key="pre-review-request-1",
        )
        repeated = service.pre_review_claim(
            claim.id,
            current_user,
            correlation_id="pre-review-request-1",
            idempotency_key="pre-review-request-1",
        )

        assert first is not None and repeated is not None
        events = list(
            db.scalars(
                select(BusinessEvent).where(
                    BusinessEvent.aggregate_id == claim.id,
                    BusinessEvent.event_type == "claim_pre_review_completed",
                )
            ).all()
        )
        assert len(events) == 1
        event = events[0]
        assert event.correlation_id == "pre-review-request-1"
        assert event.idempotency_key.startswith("pre-review:")
        assert event.payload_json["review_status"] == "failed"
        assert event.payload_json["passed"] is False
        assert event.payload_json["blocking_risk_count"] == 1
        assert event.payload_json["business_stage"] == "reimbursement"
        assert "重大风险" in event.payload_json["message"]
        assert db.scalar(select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == claim.id))


def test_pre_review_event_failure_rolls_back_claim_and_case(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    current_user = CurrentUserContext(
        username="pre-review-rollback@example.com",
        name="张三",
        role_codes=["user"],
        is_admin=False,
        tenant_id="default",
    )
    with build_session() as db:
        manager = Employee(
            employee_no="PRE-REVIEW-ROLLBACK-MANAGER",
            name="李经理",
            email="pre-review-rollback-manager@example.com",
        )
        employee = Employee(
            employee_no="PRE-REVIEW-ROLLBACK",
            name="张三",
            email=current_user.username,
            manager=manager,
        )
        claim = build_claim(claim_no="RE-CASE-PRE-REVIEW-ROLLBACK", employee=employee)
        original_flags = [
            {
                "source": "manual_risk",
                "severity": "high",
                "label": "原始风险",
                "message": "必须保留。",
            }
        ]
        claim.risk_flags_json = original_flags
        db.add_all([manager, employee, claim])
        db.commit()
        claim_id = claim.id

        def fail_event(*_args, **_kwargs):
            raise RuntimeError("simulated pre-review event failure")

        monkeypatch.setattr(ExpenseCaseService, "record_claim_event", fail_event)
        with pytest.raises(RuntimeError, match="simulated pre-review event failure"):
            ExpenseClaimService(db).pre_review_claim(
                claim_id,
                current_user,
                correlation_id="pre-review-rollback",
            )

        persisted_claim = db.get(ExpenseClaim, claim_id)
        assert persisted_claim is not None
        assert persisted_claim.risk_flags_json == original_flags
        assert db.scalar(select(ExpenseCase)) is None
        assert db.scalar(select(ExpenseCaseLink)) is None
        assert db.scalar(select(BusinessEvent)) is None


def test_submit_blocks_fixable_pre_review_before_budget_and_submission_event(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    current_user = CurrentUserContext(
        username="pre-review-block-owner@example.com",
        name="张三",
        role_codes=["user"],
        is_admin=False,
        tenant_id="default",
    )
    with build_session() as db:
        manager = Employee(
            employee_no="PRE-REVIEW-BLOCK-MANAGER",
            name="李经理",
            email="pre-review-block-manager@example.com",
        )
        employee = Employee(
            employee_no="PRE-REVIEW-BLOCK-OWNER",
            name="张三",
            email=current_user.username,
            manager=manager,
        )
        claim = build_claim(claim_no="RE-CASE-PRE-REVIEW-BLOCK", employee=employee)
        claim.risk_flags_json = [
            {
                "source": "manual_risk",
                "severity": "high",
                "actionability": "fixable_by_submitter",
                "business_stage": "reimbursement",
                "label": "票据与明细不一致",
                "message": "请更正费用明细或重新上传正确票据。",
                "item_ids": [claim.items[0].id],
            }
        ]
        db.add_all([manager, employee, claim])
        db.commit()

        service = ExpenseClaimService(db)
        reviewed = service.pre_review_claim(claim.id, current_user)
        assert reviewed is not None
        pre_review_flag = next(
            flag
            for flag in reviewed.risk_flags_json
            if isinstance(flag, dict) and flag.get("source") == "ai_pre_review"
        )
        assert pre_review_flag["decision"] == "needs_fix"

        def fail_budget(*_args, **_kwargs):
            raise AssertionError("预审阻断后不应占用预算")

        monkeypatch.setattr(service, "_reserve_budget_for_submission", fail_budget)
        with pytest.raises(ExpenseClaimPreReviewBlockedError) as error_info:
            service.submit_claim(
                claim.id,
                current_user,
                pre_review_id=pre_review_flag["review_id"],
                pre_review_input_fingerprint=pre_review_flag["input_fingerprint"],
            )

        assert error_info.value.review["decision"] == "needs_fix"
        persisted_claim = db.get(ExpenseClaim, claim.id)
        assert persisted_claim is not None
        assert persisted_claim.status == "draft"
        assert persisted_claim.submitted_at is None
        pre_review_events = list(
            db.scalars(
                select(BusinessEvent).where(
                    BusinessEvent.aggregate_id == claim.id,
                    BusinessEvent.event_type == "claim_pre_review_completed",
                )
            ).all()
        )
        assert len(pre_review_events) == 1
        assert (
            db.scalar(
                select(BusinessEvent).where(
                    BusinessEvent.aggregate_id == claim.id,
                    BusinessEvent.event_type == "claim_submitted",
                )
            )
            is None
        )
        assert db.scalar(select(BudgetReservation)) is None
        assert db.scalar(select(BudgetTransaction)) is None


def test_submit_rechecks_dynamic_risk_context_and_rejects_stale_ready_review(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    current_user = CurrentUserContext(
        username="dynamic-review-owner@example.com",
        name="张三",
        role_codes=["user"],
        is_admin=False,
        tenant_id="default",
    )
    with build_session() as db:
        manager = Employee(
            employee_no="DYNAMIC-REVIEW-MANAGER",
            name="李经理",
            email="dynamic-review-manager@example.com",
        )
        employee = Employee(
            employee_no="DYNAMIC-REVIEW-OWNER",
            name="张三",
            email=current_user.username,
            manager=manager,
        )
        claim = build_claim(
            claim_no="RE-CASE-DYNAMIC-REVIEW",
            employee=employee,
        )
        db.add_all([manager, employee, claim])
        db.commit()

        service = ExpenseClaimService(db)
        review_calls = 0

        def dynamic_review(_claim):
            nonlocal review_calls
            review_calls += 1
            risk_flags = []
            if review_calls > 1:
                risk_flags = [
                    {
                        "source": "submission_review",
                        "severity": "high",
                        "actionability": "fixable_by_submitter",
                        "business_stage": "reimbursement",
                        "label": "重复发票",
                        "message": "预审后发现同一发票已被其他单据使用。",
                    }
                ]
            return {
                "risk_flags": risk_flags,
                "rule_set_fingerprint": "rules-v1",
            }

        monkeypatch.setattr(service, "_run_ai_submission_review", dynamic_review)
        reviewed = service.pre_review_claim(claim.id, current_user)
        assert reviewed is not None
        ready_flag = next(
            flag
            for flag in reviewed.risk_flags_json
            if isinstance(flag, dict) and flag.get("source") == "ai_pre_review"
        )
        assert ready_flag["decision"] == "ready"

        with pytest.raises(ExpenseClaimPreReviewBlockedError) as error_info:
            service.submit_claim(
                claim.id,
                current_user,
                pre_review_id=ready_flag["review_id"],
                pre_review_input_fingerprint=ready_flag["input_fingerprint"],
            )

        assert error_info.value.code == "PRE_REVIEW_NEEDS_FIX"
        assert error_info.value.review["review_id"] != ready_flag["review_id"]
        assert error_info.value.review["decision"] == "needs_fix"
        assert db.get(ExpenseClaim, claim.id).status == "draft"
        assert (
            db.scalar(
                select(BusinessEvent).where(
                    BusinessEvent.aggregate_id == claim.id,
                    BusinessEvent.event_type == "claim_submitted",
                )
            )
            is None
        )
        pre_review_events = list(
            db.scalars(
                select(BusinessEvent).where(
                    BusinessEvent.aggregate_id == claim.id,
                    BusinessEvent.event_type == "claim_pre_review_completed",
                )
            ).all()
        )
        assert len(pre_review_events) == 2


def test_legacy_bootstrap_excludes_migration_owned_tables(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    with session_factory() as db:
        service = AgentFoundationService(db)
        monkeypatch.setattr(service, "_ensure_agent_asset_schema", lambda: None)
        monkeypatch.setattr(service, "_ensure_financial_record_schema", lambda: None)
        monkeypatch.setattr(service, "_seed_agent_assets", lambda: None)
        monkeypatch.setattr(service, "_sync_demo_financial_records", lambda: None)
        monkeypatch.setattr(service, "_seed_runs_and_logs", lambda: None)
        monkeypatch.setattr(TenantRegistryService, "ensure_builtin", lambda self: None)

        service._prepare_foundation()

    table_names = set(inspect(engine).get_table_names())
    assert "employees" in table_names
    assert {
        "auth_sessions",
        "expense_cases",
        "expense_case_links",
        "business_events",
    }.isdisjoint(table_names)


def test_event_write_is_idempotent_for_same_business_operation() -> None:
    with build_session() as db:
        claim = build_claim(claim_no="RE-CASE-IDEMPOTENT")
        db.add(claim)
        db.commit()

        service = ExpenseCaseService(db)
        _case, first_event = service.record_claim_event(
            claim,
            event_type="claim_draft_updated",
            actor_id="owner@example.com",
            tenant_id="tenant-a",
            correlation_id="run-idempotent",
            idempotency_key="save-operation-1",
        )
        _case, repeated_event = service.record_claim_event(
            claim,
            event_type="claim_draft_updated",
            actor_id="owner@example.com",
            tenant_id="tenant-a",
            correlation_id="run-idempotent",
            idempotency_key="save-operation-1",
        )
        db.commit()

        assert repeated_event.id == first_event.id
        assert len(list(db.scalars(select(BusinessEvent)).all())) == 1


def test_submit_claim_creates_case_link_and_structured_event() -> None:
    current_user = CurrentUserContext(
        tenant_id="default",
        username="employee-case@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="CASE-MANAGER",
            name="李经理",
            email="manager-case@example.com",
        )
        employee = Employee(
            employee_no="CASE-EMPLOYEE",
            name="张三",
            email=current_user.username,
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        claim = build_claim(claim_no="RE-CASE-SUBMIT", employee=employee)
        claim.risk_flags_json = [
            {
                "source": "ai_pre_review",
                "status": "passed",
                "passed": True,
                "severity": "info",
                "blocking_risk_count": 0,
            }
        ]
        db.add(claim)
        db.commit()

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        link = db.scalar(select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == submitted.id))
        assert link is not None
        pre_review_event = db.scalar(
            select(BusinessEvent).where(
                BusinessEvent.aggregate_id == submitted.id,
                BusinessEvent.event_type == "claim_pre_review_completed",
            )
        )
        submitted_event = db.scalar(
            select(BusinessEvent).where(
                BusinessEvent.aggregate_id == submitted.id,
                BusinessEvent.event_type == "claim_submitted",
            )
        )
        assert pre_review_event is not None and submitted_event is not None
        assert submitted_event.delivery_status == "pending"
        assert submitted_event.correlation_id == pre_review_event.correlation_id
        assert submitted_event.causation_id == pre_review_event.id
        assert submitted_event.payload_json["previous_status"] == "draft"
        assert submitted_event.payload_json["next_status"] == "submitted"


def test_payment_event_failure_rolls_back_payment_archive_and_nested_audit(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    current_user = CurrentUserContext(
        tenant_id="default",
        username="finance-case@example.com",
        name="财务付款",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        application_claim = build_claim(
            claim_no="AP-CASE-ARCHIVE",
            status="approved",
            approval_stage="关联单据状态",
            expense_type="travel_application",
        )
        reimbursement_claim = build_claim(
            claim_no="RE-CASE-PAY",
            status="pending_payment",
            approval_stage="待付款",
            expense_type="travel",
        )
        reimbursement_claim.risk_flags_json = [
            {
                "source": "application_handoff",
                "application_claim_id": application_claim.id,
                "application_claim_no": application_claim.claim_no,
            }
        ]
        db.add_all([application_claim, reimbursement_claim])
        db.commit()

        service = ExpenseClaimService(db)

        def fail_event(*args, **kwargs):
            raise RuntimeError("outbox unavailable")

        monkeypatch.setattr(service._expense_cases, "record_claim_event", fail_event)
        with pytest.raises(RuntimeError, match="outbox unavailable"):
            service.mark_claim_paid(reimbursement_claim.id, current_user)
        db.rollback()

        db.refresh(reimbursement_claim)
        db.refresh(application_claim)
        assert reimbursement_claim.status == "pending_payment"
        assert reimbursement_claim.approval_stage == "待付款"
        assert application_claim.status == "approved"
        assert application_claim.approval_stage != APPLICATION_ARCHIVE_STAGE
        assert db.scalar(select(AuditLog)) is None


def test_application_approval_links_generated_reimbursement_to_same_case() -> None:
    with build_session() as db:
        department = OrganizationUnit(
            unit_code="CASE-TRAVEL",
            name="差旅试点部",
            unit_type="department",
        )
        manager = Employee(
            employee_no="CASE-APP-MANAGER",
            name="差旅经理",
            email="travel-manager@example.com",
            organization_unit=department,
        )
        employee = Employee(
            employee_no="CASE-APP-EMPLOYEE",
            name="差旅员工",
            email="travel-employee@example.com",
            manager=manager,
            organization_unit=department,
        )
        db.add_all([department, manager, employee])
        db.flush()
        db.add(
            BudgetAllocation(
                budget_no="BUD-CASE-TRAVEL",
                fiscal_year=2026,
                period_type="year",
                period_key="2026",
                department_id=department.id,
                department_name=department.name,
                cost_center=None,
                project_code=None,
                subject_code="travel",
                subject_name="差旅费",
                original_amount=Decimal("50000.00"),
                adjusted_amount=Decimal("0.00"),
                status="active",
                warning_threshold=Decimal("80.00"),
                control_action="block",
            )
        )
        application_claim = build_claim(
            claim_no="AP-CASE-GENERATE",
            employee=employee,
            status="submitted",
            approval_stage=DIRECT_MANAGER_APPROVAL_STAGE,
            expense_type="travel_application",
        )
        application_claim.amount = Decimal("500.00")
        db.add(application_claim)
        db.commit()

        approved = ExpenseClaimService(db).approve_claim(
            application_claim.id,
            CurrentUserContext(
                tenant_id="default",
                username=manager.email,
                name=manager.name,
                role_codes=["manager"],
                is_admin=False,
            ),
            opinion="业务必要，同意申请",
        )

        assert approved is not None
        assert approved.status == "approved"
        assert approved.approval_stage == APPLICATION_LINK_STATUS_STAGE
        expense_case = db.scalar(select(ExpenseCase))
        assert expense_case is not None
        links = list(
            db.scalars(
                select(ExpenseCaseLink)
                .where(ExpenseCaseLink.expense_case_id == expense_case.id)
                .order_by(ExpenseCaseLink.created_at)
            ).all()
        )
        assert {link.relation_type for link in links} == {
            "application",
            "generated_reimbursement",
        }
        events = list(
            db.scalars(
                select(BusinessEvent)
                .where(BusinessEvent.expense_case_id == expense_case.id)
                .order_by(BusinessEvent.occurred_at)
            ).all()
        )
        assert [event.event_type for event in events] == [
            "application_approved",
            "reimbursement_draft_generated",
        ]
        assert len({event.correlation_id for event in events}) == 1
        assert events[1].causation_id == events[0].id


def test_payment_records_application_archive_event_in_same_case() -> None:
    current_user = CurrentUserContext(
        tenant_id="default",
        username="finance-archive@example.com",
        name="财务付款",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        application_claim = build_claim(
            claim_no="AP-CASE-ARCHIVE-EVENT",
            status="approved",
            approval_stage="关联单据状态",
            expense_type="travel_application",
        )
        reimbursement_claim = build_claim(
            claim_no="RE-CASE-PAY-EVENT",
            status="pending_payment",
            approval_stage="待付款",
            expense_type="travel",
        )
        reimbursement_claim.risk_flags_json = [
            {
                "source": "application_handoff",
                "application_claim_no": application_claim.claim_no,
            }
        ]
        db.add_all([application_claim, reimbursement_claim])
        db.commit()

        paid = ExpenseClaimService(db).mark_claim_paid(reimbursement_claim.id, current_user)

        assert paid is not None
        expense_case = db.scalar(select(ExpenseCase))
        assert expense_case is not None
        links = list(
            db.scalars(
                select(ExpenseCaseLink).where(ExpenseCaseLink.expense_case_id == expense_case.id)
            ).all()
        )
        assert {link.resource_id for link in links} == {
            application_claim.id,
            reimbursement_claim.id,
        }
        events = list(
            db.scalars(
                select(BusinessEvent)
                .where(BusinessEvent.expense_case_id == expense_case.id)
                .order_by(BusinessEvent.occurred_at)
            ).all()
        )
        assert [event.event_type for event in events] == [
            "payment_completed",
            "application_archived",
        ]
        assert events[1].causation_id == events[0].id
        assert expense_case.current_stage == "accounting"