"""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"