jonatan/sowai-fiscal-svc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
sowai-fiscal-svc/tests/emission/test_emissao.py
T
jonatanritterandClaude Opus 4.8 1625e8a257 test(emission): scope the FIX 4 test to the filter's real contract 
The FIX 4 test asserted a full new 201 emission after the only document
under an idempotency key was soft-deleted. That over-reached: the surviving
FiscalIdempotencyKey (never soft-deleted, by design) still collides on the
UNIQUE(product_id, idempotency_key), and the camada-2 winner lookup — now
filtered by deleted_at IS NULL — returns None and re-raises. Re-emission
under a burned key is a future CANCEL feature (no soft-delete path exists in
F2), out of scope for this MINOR. The fix's real contract is narrow: a replay
must not RETURN a soft-deleted document. Test now proves exactly that on the
query (seeded on a separate session so it doesn't pollute the app's shared
db_session — the source of the MissingGreenlet the end-to-end POST hit).
Detection proven: filter removed -> FAILED; restored -> passed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-08 16:14:40 -03:00

645 lines
27 KiB
Python
Raw Permalink 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 sqlalchemy.ext.asyncio import async_sessionmaker
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_get_document_by_idempotency_key_ignora_documento_soft_deletado(
db_session, test_engine
):
"""FIX 4 (F2 review, MINOR): `_get_document_by_idempotency_key` filtra
`deleted_at IS NULL`. O CONTRATO deste fix é estreito -- um REPLAY nunca
DEVOLVE um documento morto -- e é exatamente isso que se prova aqui, na
query, não num POST.
Escopo (deliberado): nenhum caminho da F2 soft-deleta um `FiscalDocument`
(a `FiscalIdempotencyKey` que aponta pra ele, aliás, sobrevive PARA
SEMPRE -- sem soft-delete mixin, por design), então o cenário é semeado à
mão. O fix NÃO promete "re-emitir sob a mesma chave": a chave de
idempotência sobrevivente ainda colide no UNIQUE `(product_id,
idempotency_key)`, e o que fazer nessa colisão é decisão da futura
feature de CANCELAMENTO (que ainda não existe) -- fora do escopo desta
MINOR. Testar a query é o que casa com o que o fix entrega; um teste
end-to-end de nova emissão exigiria semântica ainda não construída."""
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"],
)
# dead-doc + sua idempotency-key: sessão PRÓPRIA (não polui o db_session
# que o app reusa via override), commit -> visível por READ COMMITTED.
# Chave ÚNICA por invocação (o banco do pod persiste entre execuções e a
# chave é global-unique -- hardcode colidiria no 2º run).
chave_morta = f"{uuid.uuid4().int:044d}"[:44]
seed_maker = async_sessionmaker(test_engine, expire_on_commit=False)
async with seed_maker() as seed:
series = (
await seed.execute(
select(FiscalSeries).where(
FiscalSeries.product_id == product.id,
FiscalSeries.tenant_ref == "t1",
)
)
).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"], chave_acesso=chave_morta,
codigo_numerico="87654321", status="ASSINADO",
ambiente="homologacao", xml_assinado="<NFe/>",
deleted_at=datetime.now(timezone.utc),
)
seed.add(dead_document)
await seed.flush()
seed.add(
FiscalIdempotencyKey(
product_id=product.id, idempotency_key=idem_key, document_id=dead_document.id
)
)
await seed.commit()
# SEM o `deleted_at IS NULL` do fix esta query devolveria o dead-doc.
replay = await emission_service._get_document_by_idempotency_key(
db_session, product.id, idem_key
)
assert replay is None, (
"sem o filtro `deleted_at IS NULL` a query devolveria o documento morto"
)
@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