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
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_workflow_constants import (
    APPLICATION_ARCHIVE_STAGE,
    APPLICATION_LINK_STATUS_STAGE,
    DIRECT_MANAGER_APPROVAL_STAGE,
)
from app.services.expense_claims import ExpenseClaimService


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_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)

        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(
        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
        event = db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_id == submitted.id))
        assert event is not None
        assert event.event_type == "claim_submitted"
        assert event.delivery_status == "pending"
        assert event.payload_json["previous_status"] == "draft"
        assert 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(
        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(
                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(
        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"