X-Financial/server/tests/test_document_numbering.py

from __future__ import annotations

from datetime import UTC, datetime
from decimal import Decimal

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

from app.db.base import Base
from app.models.financial_record import ExpenseClaim
from app.services.document_numbering import (
    DOCUMENT_NUMBER_EXTRACT_PATTERN,
    DOCUMENT_NUMBER_PATTERN,
    build_document_number,
    generate_unique_expense_claim_no,
    is_application_claim_no,
)


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


def test_build_document_number_uses_kind_prefix_timestamp_and_token() -> None:
    timestamp = datetime(2026, 5, 25, 10, 30, 45, tzinfo=UTC)

    assert (
        build_document_number("application", timestamp=timestamp, token="7K3M9Q2P")
        == "A7K3M9Q2P"
    )
    assert (
        build_document_number("reimbursement", timestamp=timestamp, token="7K3M9Q2P")
        == "R7K3M9Q2P"
    )
    assert (
        build_document_number("audit", timestamp=timestamp, token="7K3M9Q2P")
        == "D7K3M9Q2P"
    )


def test_document_number_pattern_accepts_short_and_legacy_numbers() -> None:
    assert DOCUMENT_NUMBER_PATTERN.fullmatch("A7K3M9Q2P")
    assert DOCUMENT_NUMBER_PATTERN.fullmatch("R7K3M9Q2P")
    assert DOCUMENT_NUMBER_PATTERN.fullmatch("D7K3M9Q2P")
    assert DOCUMENT_NUMBER_PATTERN.fullmatch("AP-20260525103045-ABCDEFGH")
    assert DOCUMENT_NUMBER_PATTERN.fullmatch("RE-20260525103045-HGFEDCBA")


def test_document_number_extract_pattern_finds_short_and_legacy_numbers() -> None:
    query = "查看 A7K3M9Q2P、R7K3M9Q2P 和 AP-20260525103045-ABCDEFGH 的状态"

    assert [match.group(0) for match in DOCUMENT_NUMBER_EXTRACT_PATTERN.finditer(query)] == [
        "A7K3M9Q2P",
        "R7K3M9Q2P",
        "AP-20260525103045-ABCDEFGH",
    ]


def test_build_document_number_rejects_ambiguous_token_chars() -> None:
    timestamp = datetime(2026, 5, 25, 10, 30, 45, tzinfo=UTC)

    with pytest.raises(ValueError):
        build_document_number("application", timestamp=timestamp, token="ABCDEF10")


def test_generate_unique_expense_claim_no_retries_existing_candidate() -> None:
    timestamp = datetime(2026, 5, 25, 10, 30, 45, tzinfo=UTC)
    with build_session() as db:
        db.add(
            ExpenseClaim(
                claim_no="RABCDEFGH",
                employee_name="张三",
                department_name="市场部",
                project_code=None,
                expense_type="transport",
                reason="交通报销",
                location="深圳",
                amount=Decimal("10.00"),
                currency="CNY",
                invoice_count=1,
                occurred_at=timestamp,
                status="draft",
                approval_stage="待提交",
                risk_flags_json=[],
            )
        )
        db.commit()

        tokens = iter(["ABCDEFGH", "HGFEDCBA"])

        assert (
            generate_unique_expense_claim_no(
                db,
                "reimbursement",
                timestamp=timestamp,
                token_factory=lambda: next(tokens),
            )
            == "RHGFEDCBA"
        )


def test_is_application_claim_no_supports_new_and_legacy_prefixes() -> None:
    assert is_application_claim_no("A7K3M9Q2P")
    assert is_application_claim_no("AP-20260525103045-ABCDEFGH")
    assert is_application_claim_no("APP-20260525-ABC123")
    assert not is_application_claim_no("R7K3M9Q2P")
    assert not is_application_claim_no("RE-20260525103045-ABCDEFGH")