"""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/.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"), } 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'" )