sowai-fiscal-svc/tests/migrations/test_fiscal_documents_schema.py

"""Migration test (Task 3): `fiscal_series` + `fiscal_certificates` +
`fiscal_documents` -- table/columns exist via `information_schema` over the
`create_all` schema (Teste A), and each round-trips (INSERT cru via the
real ORM) on a database built PURELY by `alembic upgrade head` (Teste B) --
same two-test precedent as `test_products_schema.py` (Task 2) and the
auto's own `tests/migrations/test_fiscal_*_schema.py`. Also proves the
constraints Global Constraints calls out by name: UNIQUE `chave_acesso`,
the partial-unique cert-vivo-per-`(product_id, branch_ref)` index, and the
UNIQUE `(product_id, tenant_ref, branch_ref, document_model, serie)` on
`fiscal_series`."""
from datetime import datetime, timedelta, timezone

import pytest
import pytest_asyncio
from sqlalchemy import select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

# Import de registro: garante que os modelos relevantes estão registrados em
# Base.metadata antes do create_all da fixture test_engine, ao rodar este
# arquivo isolado.
from fiscal_svc.documents import models as _documents_models  # noqa: F401
from fiscal_svc.tenancy import models as _tenancy_models  # noqa: F401
from tests.migrations._helpers import run_alembic, run_psql

_MIGRATION_DB_NAME = "fiscal_svc_test_fiscal_documents_schema"
_MIGRATION_DB_URL = (
    f"postgresql+asyncpg://postgres:postgres@localhost:5432/{_MIGRATION_DB_NAME}"
)


@pytest.mark.asyncio
async def test_fiscal_series_table_and_columns_exist(db_session):
    rows = {
        r[0]
        for r in (
            await db_session.execute(
                text(
                    "select column_name from information_schema.columns "
                    "where table_name='fiscal_series'"
                )
            )
        )
    }
    assert {
        "id", "product_id", "tenant_ref", "branch_ref", "document_model",
        "serie", "next_number", "created_at", "updated_at", "deleted_at",
    } <= rows


@pytest.mark.asyncio
async def test_fiscal_certificates_table_and_columns_exist(db_session):
    rows = {
        r[0]: r[1]
        for r in (
            await db_session.execute(
                text(
                    "select column_name, is_nullable from information_schema.columns "
                    "where table_name='fiscal_certificates'"
                )
            )
        )
    }
    for required in (
        "product_id", "tenant_ref", "branch_ref", "cnpj", "pfx_encrypted",
        "password_encrypted", "subject_cn", "cnpj_certificado",
        "not_valid_before", "not_valid_after", "deleted_at",
    ):
        assert required in rows, f"coluna {required} ausente em fiscal_certificates"
    assert rows["pfx_encrypted"] == "NO"
    assert rows["deleted_at"] == "YES"


@pytest.mark.asyncio
async def test_fiscal_documents_table_and_columns_exist(db_session):
    rows = {
        r[0]: r[1]
        for r in (
            await db_session.execute(
                text(
                    "select column_name, is_nullable from information_schema.columns "
                    "where table_name='fiscal_documents'"
                )
            )
        )
    }
    for required in (
        "product_id", "tenant_ref", "branch_ref", "series_id", "document_model",
        "serie", "numero", "chave_acesso", "codigo_numerico", "status",
        "ambiente", "xml_assinado", "rejeicao_codigo", "rejeicao_motivo",
        "protocolo", "autorizada_em", "deleted_at",
    ):
        assert required in rows, f"coluna {required} ausente em fiscal_documents"
    assert rows["chave_acesso"] == "NO"
    # sale_id/service_order_id are DELIBERATELY absent -- porte table.
    assert "sale_id" not in rows
    assert "service_order_id" not in rows


@pytest_asyncio.fixture
async def migration_database():
    run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
    run_psql("-c", f"CREATE DATABASE {_MIGRATION_DB_NAME};")
    yield
    run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")


async def _make_product(session, name="auto"):
    from fiscal_svc.tenancy.service import create_product

    return await create_product(session, name=name, api_key=f"key-{name}-{id(session)}")


@pytest.mark.asyncio
async def test_fiscal_series_and_certificate_and_document_round_trip_on_a_real_migrated_database(
    migration_database,
):
    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
    assert result.returncode == 0, result.stderr

    from fiscal_svc.documents.models import FiscalCertificate, FiscalDocument, FiscalSeries

    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
    session_maker = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with session_maker() as session:
            product = await _make_product(session)

            series = FiscalSeries(
                product_id=product.id,
                tenant_ref="tenant-1",
                branch_ref="branch-1",
                document_model="55",
                serie=1,
                next_number=1014,
            )
            session.add(series)
            await session.flush()

            certificate = FiscalCertificate(
                product_id=product.id,
                tenant_ref="tenant-1",
                branch_ref="branch-1",
                cnpj="14200166000187",
                pfx_encrypted=b"\x00\x01ciphertext-pfx",
                password_encrypted=b"\x00\x02ciphertext-pw",
                subject_cn="EMPRESA TESTE LTDA:14200166000187",
                cnpj_certificado="14200166000187",
                not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
                not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
            )
            session.add(certificate)

            document = FiscalDocument(
                product_id=product.id,
                tenant_ref="tenant-1",
                branch_ref="branch-1",
                series_id=series.id,
                document_model="55",
                serie=1,
                numero=1014,
                chave_acesso="4" * 44,
                codigo_numerico="12345678",
                status="ASSINADO",
                ambiente="homologacao",
                xml_assinado="<NFe/>",
            )
            session.add(document)

            # Would crash here (UndefinedColumn/DataError) before the fix.
            await session.commit()

            series_id, certificate_id, document_id = series.id, certificate.id, document.id

        async with session_maker() as session:
            reloaded_series = await session.get(FiscalSeries, series_id)
            assert reloaded_series.next_number == 1014

            reloaded_cert = await session.get(FiscalCertificate, certificate_id)
            assert reloaded_cert.cnpj_certificado == "14200166000187"
            assert reloaded_cert.pfx_encrypted == b"\x00\x01ciphertext-pfx"

            reloaded_doc = await session.get(FiscalDocument, document_id)
            assert reloaded_doc.status == "ASSINADO"
            assert reloaded_doc.chave_acesso == "4" * 44

            result_q = await session.execute(
                select(FiscalDocument).where(FiscalDocument.chave_acesso == "4" * 44)
            )
            assert result_q.scalar_one().id == document_id
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_chave_acesso_unique_constraint_holds_on_real_migration(migration_database):
    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
    assert result.returncode == 0, result.stderr

    from fiscal_svc.documents.models import FiscalDocument, FiscalSeries

    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
    session_maker = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with session_maker() as session:
            product = await _make_product(session, name="auto-chave")
            series = FiscalSeries(
                product_id=product.id, tenant_ref="t1", branch_ref="b1",
                document_model="55", serie=1, next_number=1,
            )
            session.add(series)
            await session.flush()

            def _doc(numero):
                return FiscalDocument(
                    product_id=product.id, tenant_ref="t1", branch_ref="b1",
                    series_id=series.id, document_model="55", serie=1, numero=numero,
                    chave_acesso="9" * 44, codigo_numerico="12345678",
                    status="ASSINADO", ambiente="homologacao", xml_assinado="<NFe/>",
                )

            session.add(_doc(1))
            await session.commit()

            session.add(_doc(2))
            with pytest.raises(IntegrityError):
                await session.commit()
            await session.rollback()
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_two_live_certificates_for_same_product_branch_violate_unique_index_on_real_migration(
    migration_database,
):
    """Same fix/reasoning as the auto's `a1b2c3d4e5f6` migration: a PARTIAL
    UNIQUE index (here on `(product_id, branch_ref) WHERE deleted_at IS
    NULL`) makes two concurrently-uploaded LIVE certificates for the same
    slot structurally impossible, not just avoided by the service layer."""
    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
    assert result.returncode == 0, result.stderr

    from fiscal_svc.documents.models import FiscalCertificate

    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
    session_maker = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with session_maker() as session:
            product = await _make_product(session, name="auto-cert-corrida")

            def _cert():
                return FiscalCertificate(
                    product_id=product.id,
                    tenant_ref="t1",
                    branch_ref="b1",
                    cnpj="14200166000280",
                    pfx_encrypted=b"\x00pfx",
                    password_encrypted=b"\x00pw",
                    subject_cn="EMPRESA TESTE LTDA:14200166000280",
                    cnpj_certificado="14200166000280",
                    not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
                    not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
                )

            session.add(_cert())
            await session.commit()

            session.add(_cert())
            with pytest.raises(IntegrityError):
                await session.commit()
            await session.rollback()
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_duplicate_fiscal_series_tuple_violates_unique_constraint_on_real_migration(
    migration_database,
):
    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
    assert result.returncode == 0, result.stderr

    from fiscal_svc.documents.models import FiscalSeries

    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
    session_maker = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with session_maker() as session:
            product = await _make_product(session, name="auto-serie-dup")

            def _series():
                return FiscalSeries(
                    product_id=product.id, tenant_ref="t1", branch_ref="b1",
                    document_model="55", serie=1, next_number=1,
                )

            session.add(_series())
            await session.commit()

            session.add(_series())
            with pytest.raises(IntegrityError):
                await session.commit()
            await session.rollback()
    finally:
        await engine.dispose()