sowai-fiscal-svc/tests/test_signed_goldens.py

"""Task 6: fecha o gap da EMENDA 7b (design spec's "os 3 seguros de
portabilidade", item (b)) -- o corpus golden da lib (`sowai_fiscal.
goldens`) só cobre XML PRÉ-assinatura; o caminho de ASSINATURA fica sem
seguro golden até aqui. Para cada caso golden, este teste emite ATRAVÉS DA
API real (`POST /v1/emissoes`, o mesmo caminho de qualquer produto
consumidor) com `cnf`/`dh_emi` PINADOS (via monkeypatch -- os únicos dois
elementos do payload real que a SERVICE gera de forma não-determinística;
`numero` é controlado plantando o `next_number` da série) e o certificado de
TESTE fixo (`tests/fixtures/test_cert.pfx`, NUNCA um certificado real) e
compara o XML assinado resultante, BYTE A BYTE, contra `tests/
goldens_signed/<case>.expected.xml` (gerado uma vez por
`scripts/generate_signed_goldens.py`).

RSA PKCS#1v1.5 (o que `erpbrasil.assinatura` usa) não tem padding
aleatório -- a MESMA chave privada assinando o MESMO XML canonicalizado
produz sempre a MESMA assinatura. Isso é o que torna esta comparação byte a
byte estável entre execuções (provado por `scripts/generate_signed_goldens.
py` sendo rodado duas vezes seguidas e comparado -- ver seu commit)."""
import importlib.resources
import json
import uuid
from datetime import datetime
from pathlib import Path

import pytest
from httpx import ASGITransport, AsyncClient

from fiscal_svc.core.db import get_session
from fiscal_svc.emission import service as emission_service
from fiscal_svc.main import app
from fiscal_svc.tenancy.service import create_product

_GOLDENS_DIR = Path(str(importlib.resources.files("sowai_fiscal") / "goldens"))
_EXPECTED_DIR = Path(__file__).resolve().parent / "goldens_signed"
_CERT_PATH = Path(__file__).resolve().parent / "fixtures" / "test_cert.pfx"
_CERT_PASSWORD = "test-cert-password"

_CASES = sorted(p.name.removesuffix(".input.json") for p in _GOLDENS_DIR.glob("*.input.json"))


@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):
    from cryptography.fernet import Fernet

    from fiscal_svc.certificates import crypto as certificate_lib

    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 _headers(api_key: str) -> dict[str, str]:
    return {"X-Api-Key": api_key}


class _FrozenDatetime:
    """Stand-in for `emission.service`'s module-level `datetime` NAME
    (`from datetime import datetime` -- a plain attribute of the module,
    monkeypatch-replaceable). Exposes only `.now(tz)`, the ONLY method
    `emission.service` calls on it -- returns the SAME fixed instant
    regardless of the requested `tz` (converted via `.astimezone`), so both
    call sites (`datetime.now(_TZ_EMISSAO)` for `dh_emi`/AAMM and `datetime.
    now(timezone.utc)` for the certificate validity check) see a coherent,
    pinned "now"."""

    def __init__(self, fixed: datetime):
        self._fixed = fixed

    def now(self, tz=None):
        return self._fixed.astimezone(tz) if tz is not None else self._fixed


@pytest.mark.parametrize("case", _CASES)
@pytest.mark.asyncio
async def test_signed_golden_matches_byte_for_byte(db_session, monkeypatch, case):
    raw = json.loads((_GOLDENS_DIR / f"{case}.input.json").read_text(encoding="utf-8"))
    expected = (_EXPECTED_DIR / f"{case}.expected.xml").read_text(encoding="utf-8")

    # cNF pinado -- o único elemento aleatório de `emitir_documento` além do
    # relógio.
    monkeypatch.setattr(emission_service, "gerar_cnf", lambda numero: raw["cnf"])
    # dh_emi/AAMM pinados no MESMO instante do golden.
    fixed_now = datetime.fromisoformat(raw["dh_emi"])
    monkeypatch.setattr(emission_service, "datetime", _FrozenDatetime(fixed_now))

    key = f"k-{uuid.uuid4().hex}"
    product = await create_product(db_session, name=f"golden-{case}", api_key=key)

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        pfx_bytes = _CERT_PATH.read_bytes()
        upload_response = await client.post(
            "/v1/certificados",
            params={"tenant_ref": "t1", "branch_ref": "b1"},
            files={"file": ("cert.pfx", pfx_bytes, "application/x-pkcs12")},
            data={"password": _CERT_PASSWORD, "cnpj": raw["emitente"]["cnpj"]},
            headers=_headers(key),
        )
        assert upload_response.status_code == 201, upload_response.text

        # `numero` é controlado plantando o `next_number` da série -- a
        # primeira (e única) alocação desta série devolve exatamente
        # `raw["numero"]`.
        series_response = 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),
        )
        assert series_response.status_code == 201, series_response.text

        payload = {
            "tenant_ref": "t1",
            "branch_ref": "b1",
            "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"),
            # A lib congela `ver_proc="sowai-auto/1b.1"` como default do
            # PRÓPRIO `DadosEmissao` quando o golden JSON não o especifica
            # (todos os 6 casos hoje) -- reproduzido aqui EXPLICITAMENTE
            # (design spec EMENDA F1: este serviço nunca aplica esse default
            # sozinho, o produto chamador sempre declara).
            "ver_proc": raw.get("ver_proc", "sowai-auto/1b.1"),
            # Campos opcionais que `sowai_fiscal.golden_helpers.dados_from_
            # json` só seta quando presentes no JSON golden (caindo no
            # default do próprio `DadosEmissao` quando ausentes) -- mesmo
            # racional aqui: `EmissaoRequest` tem os MESMOS defaults
            # ("Venda"/"1"/"1"/"1"/"1"), então só repassamos quando o golden
            # declara um valor NÃO-default. Faltar isto (bug real corrigido
            # aqui) fazia `caso_devolucao_intra` -- o único dos 6 casos com
            # `nat_op` != "Venda" -- emitir com `<natOp>Venda</natOp>` em vez
            # de `<natOp>Devolucao de venda</natOp>`, divergindo do golden
            # assinado byte a byte.
            "nat_op": raw.get("nat_op", "Venda"),
            "tp_emis": raw.get("tp_emis", "1"),
            "ind_final": raw.get("ind_final", "1"),
            "ind_pres": raw.get("ind_pres", "1"),
            "fin_nfe": raw.get("fin_nfe", "1"),
        }
        emit_response = await client.post(
            "/v1/emissoes", json=payload,
            headers={**_headers(key), "Idempotency-Key": f"golden-{case}-{uuid.uuid4().hex}"},
        )
        assert emit_response.status_code == 201, emit_response.text
        document_id = emit_response.json()["id"]

        xml_response = await client.get(f"/v1/documentos/{document_id}/xml", headers=_headers(key))

    assert emit_response.json()["chave_acesso"] == raw["chave_acesso"], (
        "a chave de acesso montada pelo serviço deveria bater com a do golden -- todos os "
        "componentes (uf_ibge/aamm/cnpj/modelo/serie/numero/tp_emis/cnf) precisam coincidir"
    )
    assert xml_response.text == expected, (
        f"XML assinado do caso {case!r} divergiu byte a byte do golden esperado -- "
        "regenere com scripts/generate_signed_goldens.py SE E SOMENTE SE a mudança for "
        "intencional (ex.: golden novo/atualizado na lib), nunca para 'fazer passar'"
    )