X-Financial/server/tests/test_approval_task_backfill.py

from __future__ import annotations

import argparse
import importlib.util
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from pathlib import Path

import pytest
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.db.base import Base
from app.models.approval_task import ApprovalTask, ApprovalTaskEvent
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.schemas.approval_task import ApprovalTaskMutationRead
from app.services.approval_task_backfill import (
    ApprovalTaskBackfillDisposition,
    ApprovalTaskBackfillPreview,
    ApprovalTaskBackfillService,
)
from app.services.approval_task_protocol import ApprovalTaskEventProtocol
from app.services.expense_claim_workflow_constants import (
    DIRECT_MANAGER_APPROVAL_STAGE,
    FINANCE_APPROVAL_STAGE,
)


def _load_cli_module():
    path = Path(__file__).resolve().parents[1] / "scripts" / "backfill_approval_tasks.py"
    spec = importlib.util.spec_from_file_location("backfill_approval_tasks_cli", path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


CLI = _load_cli_module()


def _session() -> Session:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    return sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)()


def _persist_claim(
    db: Session,
    *,
    suffix: str,
    tenant_id: str,
    stage: str = DIRECT_MANAGER_APPROVAL_STAGE,
    status: str = "submitted",
    with_manager: bool = True,
    with_event: bool = True,
) -> tuple[ExpenseClaim, Employee | None, BusinessEvent | None]:
    submitted_at = datetime(2026, 7, 14, 10, 0, tzinfo=UTC)
    manager = (
        Employee(
            id=f"manager-{suffix}",
            tenant_id=tenant_id,
            employee_no=f"M-{suffix}",
            name=f"经理{suffix}",
            email=f"manager-{suffix}@example.com",
        )
        if with_manager
        else None
    )
    employee = Employee(
        id=f"employee-{suffix}",
        tenant_id=tenant_id,
        employee_no=f"E-{suffix}",
        name=f"员工{suffix}",
        email=f"employee-{suffix}@example.com",
        manager=manager,
    )
    claim = ExpenseClaim(
        id=f"claim-{suffix}",
        tenant_id=tenant_id,
        claim_no=f"RE-BACKFILL-{suffix}",
        employee=employee,
        employee_name=employee.name,
        department_name="市场部",
        project_code="PRJ-BACKFILL",
        expense_type="travel",
        reason="客户现场差旅",
        location="上海",
        amount=Decimal("888.00"),
        currency="CNY",
        invoice_count=1,
        occurred_at=submitted_at - timedelta(days=1),
        submitted_at=submitted_at,
        status=status,
        approval_stage=stage,
        risk_flags_json=[],
        created_at=submitted_at - timedelta(hours=2),
        updated_at=submitted_at,
    )
    claim.items = [
        ExpenseClaimItem(
            id=f"item-{suffix}",
            claim_id=claim.id,
            item_date=date(2026, 7, 14),
            item_type="transport",
            item_reason="客户现场交通",
            item_location="上海",
            item_note="",
            item_amount=Decimal("888.00"),
            invoice_id=f"invoice-{suffix}",
            created_at=submitted_at,
            updated_at=submitted_at,
        )
    ]
    expense_case = ExpenseCase(
        id=f"case-{suffix}",
        tenant_id=tenant_id,
        case_no=f"CASE-RE-BACKFILL-{suffix}",
        scene_code="travel",
        title="客户现场差旅",
        owner_employee_id=employee.id,
        current_stage="reviewing",
        status="active",
        created_at=submitted_at - timedelta(hours=2),
        updated_at=submitted_at,
    )
    link = ExpenseCaseLink(
        id=f"link-{suffix}",
        tenant_id=tenant_id,
        expense_case_id=expense_case.id,
        resource_type="expense_claim",
        resource_id=claim.id,
        relation_type="claim",
        created_at=submitted_at,
    )
    event = None
    if with_event:
        event = BusinessEvent(
            id=f"event-{suffix}",
            tenant_id=tenant_id,
            expense_case_id=expense_case.id,
            aggregate_type="expense_claim",
            aggregate_id=claim.id,
            event_type="claim_submitted",
            event_version=1,
            idempotency_key=f"submit-{suffix}",
            correlation_id=f"correlation-{suffix}",
            actor_id=employee.id,
            actor_type="user",
            payload_json={
                "next_status": "submitted",
                "next_approval_stage": stage,
            },
            delivery_status="published",
            occurred_at=submitted_at + timedelta(hours=1),
        )
    db.add_all(
        [
            *([manager] if manager is not None else []),
            employee,
            claim,
            expense_case,
            link,
            *([event] if event is not None else []),
        ]
    )
    db.commit()
    return claim, manager, event


def test_preview_is_default_read_only_tenant_scoped_and_prefers_business_event() -> None:
    with _session() as db:
        claim, manager, event = _persist_claim(db, suffix="A", tenant_id="tenant-a")
        _persist_claim(db, suffix="B", tenant_id="tenant-b")
        assert manager is not None and event is not None
        service = ApprovalTaskBackfillService(db, tenant_id="tenant-a")

        result = service.run()

        assert isinstance(result, ApprovalTaskBackfillPreview)
        assert result.inspected == 1
        assert result.eligible == 1
        assert result.existing == 0
        assert result.skipped == 0
        assert result.items[0].claim_id == claim.id
        assert result.items[0].entered_at_source == "workflow_event"
        assert result.items[0].entered_at == event.occurred_at
        assert result.items[0].assignee_key == manager.id
        assert db.scalar(select(func.count()).select_from(ApprovalTask)) == 0
        assert db.scalar(select(func.count()).select_from(ApprovalTaskEvent)) == 0


def test_apply_creates_audited_projected_root_task_and_is_idempotent() -> None:
    applied_at = datetime(2026, 7, 16, 12, 0, tzinfo=UTC)
    with _session() as db:
        claim, manager, source_event = _persist_claim(
            db,
            suffix="APPLY",
            tenant_id="tenant-a",
        )
        assert manager is not None and source_event is not None
        service = ApprovalTaskBackfillService(db, tenant_id="tenant-a")

        first = service.apply_batch(
            run_id="approval-task-backfill-run-1",
            backfilled_at=applied_at,
        )
        repeated = service.apply_batch(
            run_id="approval-task-backfill-run-1",
            backfilled_at=applied_at,
        )

        task = db.scalar(select(ApprovalTask).where(ApprovalTask.claim_id == claim.id))
        event = db.scalar(select(ApprovalTaskEvent).where(ApprovalTaskEvent.task_id == task.id))
        assert first.created == 1
        assert repeated.created == 0
        assert repeated.existing == 1
        assert task.task_kind == "root"
        assert task.status == "pending"
        assert task.tenant_id == "tenant-a"
        assert task.expense_case_id == "case-APPLY"
        assert task.node_key == "direct_manager"
        assert task.assignee_employee_id == manager.id
        assert task.entered_at_source == "workflow_event"
        assert task.due_at > task.entered_at
        assert task.projection_updated_at.replace(tzinfo=UTC) == applied_at
        assert task.risk_level == "low"
        assert task.evidence_completeness == Decimal("1.0000")
        assert task.batch_eligible is False
        assert "sla_overdue" in task.batch_block_reasons_json
        assert event.event_type == "node_entered_backfilled"
        assert event.expected_task_version == 0
        assert event.result_task_version == 1
        assert event.business_event_id == source_event.id
        assert event.payload_json["history_reconstructed"] is False
        assert event.payload_json["current_node_reconstructed"] is True
        assert event.response_json["task"]["id"] == task.id
        snapshot = ApprovalTaskMutationRead.model_validate(event.response_json)
        replay = ApprovalTaskEventProtocol(db).find_replay(
            tenant_id="tenant-a",
            actor_id="approval-task-backfill",
            request_id=event.request_id,
            fingerprint=event.payload_fingerprint,
        )
        assert snapshot.event.id == event.id
        assert replay is not None and replay.replayed is True
        assert replay.task.id == task.id
        assert db.scalar(select(func.count()).select_from(ApprovalTask)) == 1
        assert db.scalar(select(func.count()).select_from(ApprovalTaskEvent)) == 1


def test_unresolved_direct_manager_is_fail_closed_but_finance_role_is_safe_fallback() -> None:
    with _session() as db:
        unresolved, _manager, _event = _persist_claim(
            db,
            suffix="NO-MANAGER",
            tenant_id="tenant-a",
            with_manager=False,
        )
        inactive, inactive_manager, _event = _persist_claim(
            db,
            suffix="INACTIVE-MANAGER",
            tenant_id="tenant-a",
        )
        assert inactive_manager is not None
        inactive_manager.employment_status = "离职"
        db.commit()
        finance, _manager, _event = _persist_claim(
            db,
            suffix="FINANCE",
            tenant_id="tenant-a",
            stage=FINANCE_APPROVAL_STAGE,
            with_manager=False,
        )
        service = ApprovalTaskBackfillService(db, tenant_id="tenant-a")

        preview = service.preview()

        by_claim = {item.claim_id: item for item in preview.items}
        assert by_claim[unresolved.id].disposition is (
            ApprovalTaskBackfillDisposition.ASSIGNEE_UNRESOLVED
        )
        assert by_claim[inactive.id].disposition is (
            ApprovalTaskBackfillDisposition.ASSIGNEE_UNRESOLVED
        )
        assert by_claim[finance.id].disposition is ApprovalTaskBackfillDisposition.ELIGIBLE
        assert by_claim[finance.id].assignee_key == "finance"
        assert preview.eligible == 1
        assert preview.skipped == 2


def test_legacy_in_approval_state_is_canonicalized_without_mutating_claim() -> None:
    with _session() as db:
        claim, manager, _event = _persist_claim(
            db,
            suffix="LEGACY",
            tenant_id="tenant-a",
            status="in_approval",
            stage="",
            with_event=False,
        )
        assert manager is not None
        service = ApprovalTaskBackfillService(db, tenant_id="tenant-a")

        preview = service.preview()
        applied = service.apply_batch(
            run_id="approval-task-backfill-legacy",
            backfilled_at=datetime(2026, 7, 16, 12, 0, tzinfo=UTC),
        )
        task = db.scalar(select(ApprovalTask).where(ApprovalTask.claim_id == claim.id))

        assert preview.eligible == 1
        assert preview.items[0].entered_at_source == "submitted_at"
        assert applied.created == 1
        assert task.claim_status_snapshot == "submitted"
        assert task.claim_stage_snapshot == DIRECT_MANAGER_APPROVAL_STAGE
        assert task.assignee_employee_id == manager.id
        assert claim.status == "in_approval"
        assert claim.approval_stage == ""


def test_entered_at_fallback_order_is_explicit_and_deterministic() -> None:
    submitted_at = datetime(2026, 7, 14, 10, 0, tzinfo=UTC)
    claim = ExpenseClaim(
        id="claim-time-fallback",
        tenant_id="tenant-a",
        claim_no="RE-TIME-FALLBACK",
        employee_name="张三",
        department_name="市场部",
        expense_type="travel",
        reason="差旅",
        location="上海",
        amount=Decimal("100"),
        occurred_at=submitted_at - timedelta(days=1),
        submitted_at=submitted_at,
        status="submitted",
        approval_stage=DIRECT_MANAGER_APPROVAL_STAGE,
        created_at=submitted_at - timedelta(hours=2),
        updated_at=submitted_at - timedelta(hours=1),
    )
    with _session() as db:
        service = ApprovalTaskBackfillService(db, tenant_id="tenant-a")
        entered_at, source, _event = service.derive_entered_at(
            claim,
            events=[],
            approval_stage=DIRECT_MANAGER_APPROVAL_STAGE,
        )
        assert (entered_at, source) == (submitted_at, "submitted_at")

        claim.submitted_at = None
        entered_at, source, _event = service.derive_entered_at(
            claim,
            events=[],
            approval_stage=DIRECT_MANAGER_APPROVAL_STAGE,
        )
        assert (entered_at, source) == (claim.updated_at, "updated_at")

        claim.updated_at = None
        entered_at, source, _event = service.derive_entered_at(
            claim,
            events=[],
            approval_stage=DIRECT_MANAGER_APPROVAL_STAGE,
        )
        assert (entered_at, source) == (claim.created_at, "backfill")


def test_backfill_cli_defaults_to_dry_run_and_requires_exact_apply_confirmation(
    monkeypatch,
) -> None:
    args = CLI.build_parser().parse_args(
        [
            "--tenant-id",
            "tenant-a",
            "--created-before",
            "2026-07-16T00:00:00+08:00",
            "--expected-host",
            "migration-probe",
            "--expected-database",
            "migration_probe",
        ]
    )
    assert args.apply is False
    assert args.created_before == datetime(2026, 7, 15, 16, 0, tzinfo=UTC)

    apply_args = CLI.build_parser().parse_args(
        [
            "--apply",
            "--tenant-id",
            "tenant-a",
            "--created-before",
            "2026-07-16T00:00:00Z",
            "--expected-host",
            "migration-probe",
            "--expected-database",
            "migration_probe",
        ]
    )
    monkeypatch.delenv("DATABASE_URL", raising=False)
    with pytest.raises(CLI.BackfillCommandError) as exc_info:
        CLI.run(apply_args)
    assert exc_info.value.code == "confirm_target_required"


@pytest.mark.parametrize("value", ["", "2026-07-16T00:00:00", "not-a-time"])
def test_backfill_cli_rejects_timestamp_without_explicit_timezone(value: str) -> None:
    with pytest.raises(argparse.ArgumentTypeError):
        CLI.parse_timestamp(value)