"""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="",
)
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="",
)
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()