X-Financial/server/tests/test_backfill_legacy_expense_claim_cases_cli.py

from __future__ import annotations

import argparse
import importlib.util
from datetime import UTC, datetime
from pathlib import Path

import pytest


def load_script_module():
    script_path = (
        Path(__file__).resolve().parents[1] / "scripts" / "backfill_legacy_expense_claim_cases.py"
    )
    spec = importlib.util.spec_from_file_location(
        "backfill_legacy_expense_claim_cases_cli",
        script_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


MODULE = load_script_module()


def test_created_before_requires_timezone_and_normalizes_to_utc() -> None:
    assert MODULE.parse_created_before("2026-07-14T08:00:00+08:00") == datetime(
        2026,
        7,
        14,
        tzinfo=UTC,
    )
    with pytest.raises(argparse.ArgumentTypeError, match="时区"):
        MODULE.parse_created_before("2026-07-14T00:00:00")


@pytest.mark.parametrize("value", ["0", "-1", "not-a-number"])
def test_positive_int_rejects_invalid_values(value: str) -> None:
    with pytest.raises(argparse.ArgumentTypeError):
        MODULE.positive_int(value)


def test_parser_defaults_to_dry_run_and_requires_explicit_scope() -> None:
    args = MODULE.build_parser().parse_args(
        [
            "--tenant-id",
            "default",
            "--created-before",
            "2026-07-14T00:00:00Z",
            "--expected-host",
            "migration-probe",
            "--expected-database",
            "migration_probe",
        ]
    )

    assert args.apply is False
    assert args.dry_run is False
    assert args.batch_size == 100
    assert args.tenant_id == "default"


def test_parser_rejects_apply_and_dry_run_together() -> None:
    with pytest.raises(SystemExit):
        MODULE.build_parser().parse_args(
            [
                "--apply",
                "--dry-run",
                "--tenant-id",
                "default",
                "--created-before",
                "2026-07-14T00:00:00Z",
                "--expected-host",
                "migration-probe",
                "--expected-database",
                "migration_probe",
            ]
        )


def test_parser_rejects_blank_tenant() -> None:
    with pytest.raises(SystemExit):
        MODULE.build_parser().parse_args(
            [
                "--tenant-id",
                " ",
                "--created-before",
                "2026-07-14T00:00:00Z",
                "--expected-host",
                "migration-probe",
                "--expected-database",
                "migration_probe",
            ]
        )


def test_apply_requires_exact_target_confirmation_before_connecting(monkeypatch) -> None:
    monkeypatch.setenv(
        "DATABASE_URL",
        "postgresql://finance:secret@migration-probe/migration_probe",
    )
    args = MODULE.build_parser().parse_args(
        [
            "--apply",
            "--tenant-id",
            "default",
            "--created-before",
            "2026-07-14T00:00:00Z",
            "--expected-host",
            "migration-probe",
            "--expected-database",
            "migration_probe",
        ]
    )

    with pytest.raises(MODULE.BackfillCommandError) as exc_info:
        MODULE.run(args)

    assert exc_info.value.code == "confirm_target_required"
    assert exc_info.value.exit_code == MODULE.EXIT_SAFETY


def test_missing_database_url_is_rejected_without_configuration_fallback(monkeypatch) -> None:
    monkeypatch.delenv("DATABASE_URL", raising=False)
    args = MODULE.build_parser().parse_args(
        [
            "--tenant-id",
            "default",
            "--created-before",
            "2026-07-14T00:00:00Z",
            "--expected-host",
            "migration-probe",
            "--expected-database",
            "migration_probe",
        ]
    )

    with pytest.raises(MODULE.MaintenanceDatabaseTargetError) as exc_info:
        MODULE.run(args)

    assert exc_info.value.code == "database_url_required"


def test_partial_failure_payload_discloses_committed_progress() -> None:
    summary = {
        "run_id": "historical-import-run-1",
        "batches": 2,
        "inspected": 4,
        "created": 3,
        "already_linked": 1,
        "last_cursor": {
            "created_at": "2026-07-10T00:00:00Z",
            "claim_id": "claim-4",
        },
    }
    details = MODULE._execution_progress(
        summary,
        current_batch_conflict_ids=["claim-5"],
    )
    error = MODULE.BackfillCommandError(
        "current batch rolled back",
        exit_code=MODULE.EXIT_CONFLICT,
        code="legacy_claim_conflict",
        details=details,
    )

    payload = MODULE._error_payload(error, code=error.code, details=error.details)

    assert payload["details"] == {
        "run_id": "historical-import-run-1",
        "partial_commit": True,
        "committed_batches": 2,
        "committed_claims": 4,
        "created": 3,
        "already_linked": 1,
        "last_committed_cursor": {
            "created_at": "2026-07-10T00:00:00Z",
            "claim_id": "claim-4",
        },
        "current_batch_conflict_ids": ["claim-5"],
    }