jonatanritterandClaude Opus 4.8
72bb089222
IMPORTANT (F2 review): certificate is per-tenant, matching series (already tenant-scoped) and the GET/DELETE anti-oracle boundary. Emission's _get_live_certificate and the upload-replace pre-check (certificates. service, renamed _get_live_certificate_by_branch -> _get_live_certificate_by_tenant_branch) both omitted tenant_ref -- two tenants of one product reusing branch_ref="matriz" collapsed onto the same slot: B's upload soft-deleted A's still-live certificate, and A's emission went on to sign with B's certificate. Migration 8f1a2c9d4b6e replaces the partial-unique index ix_fiscal_certificates_product_branch_live with ix_fiscal_certificates_product_tenant_branch_live on (product_id, tenant_ref, branch_ref) WHERE deleted_at IS NULL, with a working downgrade. Upload's two-layer defense (pre-check + IntegrityError -> CertificateUploadConflictError) still holds against the new index. Tests: - tests/emission/test_emissao.py:: test_dois_tenants_do_mesmo_produto_reusando_branch_ref_tem_certificados_isolados -- two tenants upload for the same product/branch_ref, both stay live; emission for each signs with its OWN certificate (observable via FIX 1's CNPJ check: without FIX 2, tenant A's emission would 409 emitente_certificate_cnpj_mismatch because the "live" cert would actually be B's). - tests/migrations/test_fiscal_documents_schema.py:: test_two_tenants_can_both_hold_a_live_certificate_for_the_same_branch_ref_on_real_migration -- real alembic upgrade head, raw INSERTs proving both tenants' certs land live. - tests/migrations/test_fiscal_documents_schema.py:: test_two_live_certificates_for_same_product_tenant_branch_violate_unique_index_on_real_migration (renamed from ..._product_branch_...) -- same (product, tenant, branch) still rejects a second live certificate on the real migration. - tests/certificates/test_certificates.py:: test_concurrent_uploads_for_same_product_branch_only_one_wins_the_other_gets_409 updated for the renamed/re-scoped precheck function (still same-tenant race, still 1 winner + 1 CertificateUploadConflictError). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
368 lines
14 KiB
Python
368 lines
14 KiB
Python
"""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_tenant_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, tenant_ref, branch_ref) WHERE
|
|
deleted_at IS NULL` -- FIX 2/F2 review, migration `8f1a2c9d4b6e`) makes
|
|
two concurrently-uploaded LIVE certificates for the SAME
|
|
`(product_id, tenant_ref, branch_ref)` slot structurally impossible, not
|
|
just avoided by the service layer. Renamed from `..._product_branch_...`
|
|
(pre-FIX-2 name) -- `tenant_ref` is now PART of the scope this proves."""
|
|
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_two_tenants_can_both_hold_a_live_certificate_for_the_same_branch_ref_on_real_migration(
|
|
migration_database,
|
|
):
|
|
"""FIX 2 (F2 review): the pre-fix index was `(product_id, branch_ref)
|
|
WHERE deleted_at IS NULL` -- ONE live cert per `branch_ref` PER PRODUCT,
|
|
regardless of tenant. Two tenants of the SAME product reusing the
|
|
identical opaque `branch_ref="matriz"` collided on that slot: this
|
|
proves, on a database built PURELY by `alembic upgrade head`, that BOTH
|
|
now insert and stay live simultaneously -- the index scope is
|
|
`(product_id, tenant_ref, branch_ref)`."""
|
|
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-multi-tenant")
|
|
|
|
def _cert(tenant_ref, cnpj):
|
|
return FiscalCertificate(
|
|
product_id=product.id,
|
|
tenant_ref=tenant_ref,
|
|
branch_ref="matriz",
|
|
cnpj=cnpj,
|
|
pfx_encrypted=b"\x00pfx",
|
|
password_encrypted=b"\x00pw",
|
|
subject_cn=f"EMPRESA TESTE LTDA:{cnpj}",
|
|
cnpj_certificado=cnpj,
|
|
not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
|
|
not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
|
|
)
|
|
|
|
cert_a = _cert("tenant-a", "14200166000280")
|
|
cert_b = _cert("tenant-b", "99887766000155")
|
|
session.add(cert_a)
|
|
session.add(cert_b)
|
|
# Would raise IntegrityError on the old (product_id, branch_ref)
|
|
# index before FIX 2 -- the second INSERT collided with the
|
|
# first tenant's still-live row.
|
|
await session.commit()
|
|
|
|
cert_a_id, cert_b_id = cert_a.id, cert_b.id
|
|
|
|
async with session_maker() as session:
|
|
reloaded_a = await session.get(FiscalCertificate, cert_a_id)
|
|
reloaded_b = await session.get(FiscalCertificate, cert_b_id)
|
|
assert reloaded_a.deleted_at is None
|
|
assert reloaded_b.deleted_at is None
|
|
assert reloaded_a.cnpj_certificado == "14200166000280"
|
|
assert reloaded_b.cnpj_certificado == "99887766000155"
|
|
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()
|