jonatan/sowai-fiscal-svc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
260644c961d6ccef1094017234a135f3121f5c03
sowai-fiscal-svc/tests/emission/test_emissao.py
T
jonatanritterandClaude Opus 4.8 260644c961 fix(emission): idempotency replay ignores soft-deleted documents 
MINOR (F2 review): _get_document_by_idempotency_key lacked the
deleted_at IS NULL filter that get_fiscal_document already has. No F2
trigger today (nothing soft-deletes a FiscalDocument yet), but a future
cancel-by-soft-delete would replay a dead document as if it were still the
valid idempotent response, instead of treating the key as free for a new
emission.

Test: tests/emission/test_emissao.py::
test_idempotency_key_apontando_para_documento_soft_deletado_e_tratada_como_ausente
-- manually inserts a soft-deleted FiscalDocument + its FiscalIdempotencyKey
row (constructing the pre-cancel scenario directly, since nothing else in
F2 produces one yet), asserts the query-level replay returns None, and that
replaying via POST /v1/emissoes with that key emits a brand-new document
(201, new number allocated) rather than resolving to the dead one.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-08 15:35:27 -03:00

640 lines
26 KiB
Python
Raw Blame History

"""Task 5: `POST /v1/emissoes` + `GET /v1/documentos/{id}[/xml]` -- ported
from the auto's `tests/modules/fiscal/test_emissao.py`, with the porte
table's biggest structural change applied: this service receives a
COMPLETE `EmissaoRequest` (no Sale/Branch/Person/Part collection), so the
"caminho feliz" setup here is FAR shorter -- upload a certificate + create a
series + POST a golden's `DadosEmissao` payload, no cadastro at all.
Golden cases (`sowai_fiscal.goldens/*.input.json`) are the payload SOURCE
for the happy path -- proves the service round-trips a real, lib-shaped
`DadosEmissao` end to end. Byte-exact signature determinism against
pre-generated `.expected.xml` fixtures is Task 6's `test_signed_goldens.py`;
this file only proves the document is well-formed/valid and the invariants
(outbox, idempotency, `ver_proc` obrigatório) hold."""
import importlib.resources
import json
import uuid
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from pathlib import Path
import pytest
from cryptography import x509
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.x509.oid import NameOID
from httpx import ASGITransport, AsyncClient
from lxml import etree
from sqlalchemy import select
from fiscal_svc.certificates import crypto as certificate_lib
from fiscal_svc.core.db import get_session
from fiscal_svc.documents.models import FiscalDocument, FiscalIdempotencyKey, FiscalSeries
from fiscal_svc.emission import service as emission_service
from fiscal_svc.emission.schemas import EmissaoRequest
from fiscal_svc.main import app
from fiscal_svc.tenancy.service import create_product
_GOLDENS_DIR = Path(str(importlib.resources.files("sowai_fiscal") / "goldens"))
_CNPJ_EMITENTE = "12345678000190"
@pytest.fixture(autouse=True)
def _override_db(db_session):
async def _get_session_override():
yield db_session
app.dependency_overrides[get_session] = _get_session_override
yield
app.dependency_overrides.clear()
@pytest.fixture(autouse=True)
def _fiscal_cert_encryption_key(monkeypatch):
monkeypatch.setenv("FISCAL_CERT_ENCRYPTION_KEY", Fernet.generate_key().decode("ascii"))
certificate_lib._get_fernet.cache_clear()
yield
certificate_lib._get_fernet.cache_clear()
def _build_test_pfx(cnpj: str = _CNPJ_EMITENTE, password: str = "senha123") -> bytes:
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
subject = issuer = x509.Name(
[
x509.NameAttribute(NameOID.COMMON_NAME, f"AUTOPECAS THIAGO LTDA:{cnpj}"),
x509.NameAttribute(NameOID.SERIAL_NUMBER, cnpj),
]
)
now = datetime.now(timezone.utc)
cert = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now - timedelta(days=1))
.not_valid_after(now + timedelta(days=365))
.sign(key, hashes.SHA256())
)
return pkcs12.serialize_key_and_certificates(
name=b"test", key=key, cert=cert, cas=None,
encryption_algorithm=serialization.BestAvailableEncryption(password.encode("utf-8")),
)
def _headers(api_key: str) -> dict[str, str]:
return {"X-Api-Key": api_key}
async def _product_and_key(db_session, name="auto"):
key = f"k-{uuid.uuid4().hex}"
product = await create_product(db_session, name=name, api_key=key)
return product, key
def _payload_from_golden(
case: str, *, tenant_ref: str = "t1", branch_ref: str = "b1", ver_proc: str = "sowai-auto/1b.1"
) -> tuple[dict, dict]:
raw = json.loads((_GOLDENS_DIR / f"{case}.input.json").read_text(encoding="utf-8"))
payload = {
"tenant_ref": tenant_ref,
"branch_ref": branch_ref,
"document_model": "55",
"serie": raw["serie"],
"emitente": raw["emitente"],
"itens": raw["itens"],
"pagamento": raw["pagamento"],
"ambiente": raw["ambiente"],
"uf_destino_tipo": raw["uf_destino_tipo"],
"destinatario": raw.get("destinatario"),
"ver_proc": ver_proc,
}
return payload, raw
async def _setup_certificate_and_series(
db_session, client, key, *, tenant_ref, branch_ref, serie, next_number, cnpj=_CNPJ_EMITENTE,
):
pfx_bytes = _build_test_pfx(cnpj=cnpj)
upload_response = await client.post(
"/v1/certificados",
params={"tenant_ref": tenant_ref, "branch_ref": branch_ref},
files={"file": ("cert.pfx", pfx_bytes, "application/x-pkcs12")},
data={"password": "senha123", "cnpj": cnpj},
headers=_headers(key),
)
assert upload_response.status_code == 201, upload_response.text
series_response = await client.post(
"/v1/series",
json={
"tenant_ref": tenant_ref, "branch_ref": branch_ref,
"document_model": "55", "serie": serie, "next_number": next_number,
},
headers=_headers(key),
)
assert series_response.status_code == 201, series_response.text
return series_response.json()
def _xsd_schema():
"""Resolvido do pacote `nfelib` INSTALADO -- ausência é FALHA, não
skip, mesma convenção do auto (`tests/modules/fiscal/test_emissao.py::
_xsd_schema`)."""
import nfelib
path = Path(nfelib.__file__).parent / "nfe" / "schemas" / "v4_0" / "nfe_v4.00.xsd"
if not path.exists():
pytest.fail(f"XSD não encontrado em {path} -- validação XSD é prova OBRIGATÓRIA")
return etree.XMLSchema(etree.parse(str(path)))
@pytest.mark.asyncio
async def test_emitir_documento_caminho_feliz_gera_documento_assinado_xsd_valido(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert response.status_code == 201, response.text
body = response.json()
assert body["status"] == "ASSINADO"
assert body["numero"] == raw["numero"]
assert len(body["chave_acesso"]) == 44 and body["chave_acesso"].isdigit()
assert body["tenant_ref"] == "t1"
assert body["branch_ref"] == "b1"
assert "xml_assinado" not in body
result = await db_session.execute(
select(FiscalDocument).where(FiscalDocument.id == uuid.UUID(body["id"]))
)
document = result.scalar_one()
assert "<Signature" in document.xml_assinado
schema = _xsd_schema()
doc = etree.fromstring(document.xml_assinado.encode("utf-8"))
valid = schema.validate(doc)
assert valid, schema.error_log
result = await db_session.execute(
select(FiscalSeries).where(FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1")
)
series = result.scalar_one()
assert series.next_number == raw["numero"] + 1
@pytest.mark.asyncio
async def test_emitir_documento_endpoint_get_and_xml(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_inter", tenant_ref="t2", branch_ref="b2")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t2", branch_ref="b2",
serie=raw["serie"], next_number=raw["numero"],
)
post_response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
document_id = post_response.json()["id"]
get_response = await client.get(f"/v1/documentos/{document_id}", headers=_headers(key))
xml_response = await client.get(f"/v1/documentos/{document_id}/xml", headers=_headers(key))
assert post_response.status_code == 201, post_response.text
assert get_response.status_code == 200, get_response.text
assert "xml_assinado" not in get_response.json()
assert xml_response.status_code == 200
assert xml_response.headers["content-type"].startswith("application/xml")
assert "<NFe" in xml_response.text
@pytest.mark.asyncio
async def test_ver_proc_ausente_e_422(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
del payload["ver_proc"]
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert response.status_code == 422, response.text
@pytest.mark.asyncio
async def test_idempotency_key_repetida_devolve_o_mesmo_documento_com_200(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
idem_key = f"idem-{uuid.uuid4().hex}"
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
first = await client.post(
"/v1/emissoes", json=payload, headers={**_headers(key), "Idempotency-Key": idem_key}
)
second = await client.post(
"/v1/emissoes", json=payload, headers={**_headers(key), "Idempotency-Key": idem_key}
)
assert first.status_code == 201, first.text
assert second.status_code == 200, second.text
assert first.json()["id"] == second.json()["id"]
result = await db_session.execute(
select(FiscalDocument).where(
FiscalDocument.product_id == product.id,
FiscalDocument.tenant_ref == "t1", FiscalDocument.branch_ref == "b1",
)
)
assert len(result.scalars().all()) == 1, "idempotency-key repetida não deveria emitir um segundo documento"
result = await db_session.execute(
select(FiscalSeries).where(FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1")
)
series = result.scalar_one()
assert series.next_number == raw["numero"] + 1, (
"a segunda chamada (mesma idempotency-key) não deveria ter alocado um SEGUNDO número"
)
# --- FIX 4 (F2 review, MINOR): replay de idempotency-key ignora soft-delete
@pytest.mark.asyncio
async def test_idempotency_key_apontando_para_documento_soft_deletado_e_tratada_como_ausente(db_session):
"""Nenhum caminho do F2 soft-deleta um `FiscalDocument` hoje (ver o
docstring de `_get_document_by_idempotency_key`), então esta prova
insere o cenário manualmente -- exatamente o que o brief de fixes
autoriza quando construir o soft-delete "pré-cancelamento" seria
forçado. Replay com a MESMA `Idempotency-Key` de um documento MORTO
precisa se comportar como se a chave nunca tivesse sido usada: uma
emissão NOVA acontece (201, número novo alocado), não um 200 devolvendo
o documento morto."""
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
idem_key = f"idem-dead-{uuid.uuid4().hex}"
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
result = await db_session.execute(
select(FiscalSeries).where(
FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1"
)
)
series = result.scalar_one()
dead_document = FiscalDocument(
product_id=product.id, tenant_ref="t1", branch_ref="b1",
series_id=series.id, document_model="55", serie=raw["serie"],
numero=raw["numero"] - 1 if raw["numero"] > 1 else 999999,
chave_acesso="5" * 44, codigo_numerico="87654321",
status="ASSINADO", ambiente="homologacao", xml_assinado="<NFe/>",
deleted_at=datetime.now(timezone.utc),
)
db_session.add(dead_document)
await db_session.flush()
db_session.add(
FiscalIdempotencyKey(
product_id=product.id, idempotency_key=idem_key, document_id=dead_document.id
)
)
await db_session.commit()
dead_document_id = dead_document.id
# Unidade: a query de replay em si já trata a chave como ausente.
replay = await emission_service._get_document_by_idempotency_key(db_session, product.id, idem_key)
assert replay is None, "replay não pode devolver um FiscalDocument soft-deletado"
response = await client.post(
"/v1/emissoes", json=payload, headers={**_headers(key), "Idempotency-Key": idem_key},
)
assert response.status_code == 201, response.text
assert response.json()["id"] != str(dead_document_id), (
"uma idempotency-key que só apontava para um documento morto deveria emitir um NOVO documento"
)
result = await db_session.execute(
select(FiscalSeries).where(FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1")
)
series = result.scalar_one()
assert series.next_number == raw["numero"] + 1
@pytest.mark.asyncio
async def test_idempotency_key_ausente_e_422(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
response = await client.post("/v1/emissoes", json=payload, headers=_headers(key))
assert response.status_code == 422, response.text
@pytest.mark.asyncio
async def test_certificado_ausente_e_409_fiscal_config_missing(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Only the series, no certificate uploaded.
await client.post(
"/v1/series",
json={
"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55",
"serie": raw["serie"], "next_number": raw["numero"],
},
headers=_headers(key),
)
response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert response.status_code == 409, response.text
assert response.json()["detail"]["code"] == "fiscal_config_missing"
result = await db_session.execute(
select(FiscalSeries).where(FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1")
)
series = result.scalar_one()
assert series.next_number == raw["numero"], "número não pode ter sido queimado sem certificado"
@pytest.mark.asyncio
async def test_serie_ausente_e_409_fiscal_config_missing(db_session):
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
pfx_bytes = _build_test_pfx()
await client.post(
"/v1/certificados",
params={"tenant_ref": "t1", "branch_ref": "b1"},
files={"file": ("cert.pfx", pfx_bytes, "application/x-pkcs12")},
data={"password": "senha123", "cnpj": _CNPJ_EMITENTE},
headers=_headers(key),
)
# No series created.
response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert response.status_code == 409, response.text
assert response.json()["detail"]["code"] == "fiscal_config_missing"
# --- FIX 1 (F2 review): emitente.cnpj vs certificado do branch_ref ----------
@pytest.mark.asyncio
async def test_emitente_cnpj_divergente_do_certificado_e_409_e_nao_queima_numero(db_session):
"""Certificado do branch_ref carrega o CNPJ A; o payload declara
`emitente.cnpj`=B (outro CNPJ válido, 14 dígitos) -- a auto teria
barrado isso estruturalmente (emitente vem de `Branch.cnpj`, o MESMO
vínculo do certificado); este serviço, com `emitente` free-form no
payload, precisa da checagem explícita ou assina B com a chave de A."""
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
cnpj_divergente = "99887766000155"
assert cnpj_divergente != _CNPJ_EMITENTE
payload["emitente"] = {**payload["emitente"], "cnpj": cnpj_divergente}
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"], cnpj=_CNPJ_EMITENTE,
)
mismatch_response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert mismatch_response.status_code == 409, mismatch_response.text
assert mismatch_response.json()["detail"]["code"] == "emitente_certificate_cnpj_mismatch"
# A mensagem NÃO deve vazar o CNPJ real do certificado.
assert _CNPJ_EMITENTE not in mismatch_response.text
result = await db_session.execute(
select(FiscalSeries).where(
FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1"
)
)
series = result.scalar_one()
assert series.next_number == raw["numero"], (
"o número NÃO pode ter sido queimado por um emitente.cnpj divergente do certificado"
)
# O número que seria queimado acima segue disponível -- uma emissão
# com o CNPJ CORRETO recebe exatamente esse número.
matching_payload, _ = _payload_from_golden("caso_padrao_intra")
success_response = await client.post(
"/v1/emissoes", json=matching_payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert success_response.status_code == 201, success_response.text
assert success_response.json()["numero"] == raw["numero"]
# --- FIX 2 (F2 review): certificado é POR TENANT, não só por branch_ref -----
@pytest.mark.asyncio
async def test_dois_tenants_do_mesmo_produto_reusando_branch_ref_tem_certificados_isolados(db_session):
"""Antes do FIX 2, `_get_live_certificate` (emissão) e o pre-check de
upload omitiam `tenant_ref` -- dois tenants do MESMO produto reusando o
MESMO `branch_ref` opaco ("matriz", plausível: refs são strings livres
do produto chamador) colapsavam no MESMO slot. O upload do tenant B
soft-deletava o certificado ainda vivo do tenant A (replace
"legítimo"), e a emissão do tenant A passava a resolver o certificado
de B.
A prova combina FIX 1 (CNPJ do emitente vs certificado) para tornar o
vínculo OBSERVÁVEL: sem o FIX 2, o certificado "vivo" para `branch_ref
="matriz"` seria o de B (CNPJ_B) para AMBOS os tenants -- a emissão do
tenant A com `emitente.cnpj`=CNPJ_A bateria no FIX 1 e devolveria 409
`emitente_certificate_cnpj_mismatch` em vez de 201."""
cnpj_a = "11222333000181"
cnpj_b = "44555666000107"
product, key = await _product_and_key(db_session)
payload_a, raw_a = _payload_from_golden(
"caso_padrao_intra", tenant_ref="tenant-a", branch_ref="matriz"
)
payload_a["emitente"] = {**payload_a["emitente"], "cnpj": cnpj_a}
payload_b, raw_b = _payload_from_golden(
"caso_padrao_inter", tenant_ref="tenant-b", branch_ref="matriz"
)
payload_b["emitente"] = {**payload_b["emitente"], "cnpj": cnpj_b}
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="tenant-a", branch_ref="matriz",
serie=raw_a["serie"], next_number=raw_a["numero"], cnpj=cnpj_a,
)
# Segundo upload, MESMO produto, MESMO branch_ref, tenant DIFERENTE.
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="tenant-b", branch_ref="matriz",
serie=raw_b["serie"], next_number=raw_b["numero"], cnpj=cnpj_b,
)
# O certificado de A segue vivo (GET de A não foi soft-deletado
# pelo upload de B) -- prova direta, sem depender do FIX 1.
get_a = await client.get(
"/v1/certificados",
params={"tenant_ref": "tenant-a", "branch_ref": "matriz"},
headers=_headers(key),
)
assert get_a.status_code == 200, get_a.text
assert get_a.json()["cnpj_certificado"] == cnpj_a
emit_a = await client.post(
"/v1/emissoes", json=payload_a,
headers={**_headers(key), "Idempotency-Key": f"idem-a-{uuid.uuid4().hex}"},
)
emit_b = await client.post(
"/v1/emissoes", json=payload_b,
headers={**_headers(key), "Idempotency-Key": f"idem-b-{uuid.uuid4().hex}"},
)
assert emit_a.status_code == 201, emit_a.text
assert emit_b.status_code == 201, emit_b.text
# --- FIX 3 (F2 review): Σ vPag vs vNF ---------------------------------------
@pytest.mark.asyncio
async def test_pagamento_divergente_do_total_dos_itens_e_409_e_nao_queima_numero(db_session):
"""A SEFAZ rejeita `pagamento.valor != vNF` com "Valor do Pagamento
difere do total" (`sowai_fiscal.xml_builder` module docstring) -- este
serviço promete validar "coerência estrutural (somas)" (spec decisão
#3) antes de queimar um número, não só deixar a rejeição acontecer do
lado de lá depois de já ter um `FiscalDocument` ASSINADO persistido."""
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
valor_correto = Decimal(str(payload["pagamento"]["valor"]))
payload["pagamento"]["valor"] = str(valor_correto + Decimal("10.00"))
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert response.status_code == 409, response.text
assert response.json()["detail"]["code"] == "pagamento_total_diverge"
result = await db_session.execute(
select(FiscalSeries).where(FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1")
)
series = result.scalar_one()
assert series.next_number == raw["numero"], (
"o número NÃO pode ter sido queimado por um pagamento divergente do total dos itens"
)
# --- prova do outbox --------------------------------------------------------
@pytest.mark.asyncio
async def test_outbox_falha_na_assinatura_nao_queima_o_numero_nem_cria_documento(db_session, monkeypatch):
"""Chama `emission.service.emitir_documento` DIRETO (não via HTTP,
mesma escolha do auto's `test_outbox_falha_na_assinatura_...`) -- uma
exceção não mapeada propagando pela pilha ASGI real não é o que este
teste prova; o que importa é o estado do banco DEPOIS do rollback."""
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"],
)
def _boom(*args, **kwargs):
raise RuntimeError("falha simulada na assinatura")
monkeypatch.setattr(emission_service, "sign_nfe_xml", _boom)
# Capturado ANTES da chamada -- `db_session.rollback()` (abaixo) EXPIRA
# todo objeto ORM já carregado nesta sessão (independente de `expire_on_
# commit`, que só rege o comportamento pós-COMMIT); tocar `product.id`
# DEPOIS do rollback, fora do contexto greenlet do SQLAlchemy, estoura
# `MissingGreenlet` -- mesma pegadinha que o auto's próprio teste
# documenta para `series.id`/`sale.id`.
product_id = product.id
request = EmissaoRequest.model_validate(payload)
with pytest.raises(RuntimeError, match="falha simulada"):
await emission_service.emitir_documento(db_session, product, request, f"idem-{uuid.uuid4().hex}")
await db_session.rollback()
result = await db_session.execute(
select(FiscalSeries).where(FiscalSeries.product_id == product_id, FiscalSeries.tenant_ref == "t1")
)
series = result.scalar_one()
assert series.next_number == raw["numero"], (
"o número NÃO pode ter sido queimado -- a falha aconteceu depois da alocação e "
"antes do commit, o rollback deve desfazer as duas coisas"
)
result = await db_session.execute(
select(FiscalDocument).where(
FiscalDocument.product_id == product_id,
FiscalDocument.tenant_ref == "t1", FiscalDocument.branch_ref == "b1",
)
)
assert result.scalars().first() is None, "nenhum FiscalDocument deveria ter sido criado"
Reference in New Issue View Git Blame Copy Permalink