"""Task 6: suíte de CONTRATO HTTP -- design spec's "os 3 seguros de portabilidade", item (a): "suíte de contrato HTTP language-agnostic (bate na API, não no interior)". A REGRA DESTE ARQUIVO: zero import de `fiscal_svc.*` além de `fiscal_svc.main.app` (para montar o `ASGITransport` -- inevitável em Python; um cliente HTTP real bateria numa URL, não precisaria nem disso). Nenhum import de model/service/schema interno -- é a suíte que uma implementação GO deste MESMO contrato (`docs/openapi-v1.json`) teria que passar batendo na porta 8140 de fora, sem NUNCA ver este código-fonte. Provisionamento de fixtures (criar um Product/API key) é feito via `scripts/create_product.py`'s MESMA função (`tenancy.service.create_ product`) em qualquer OUTRO arquivo de teste deste repo -- mas aqui, para manter a regra "zero import do interior", cada teste usa o endpoint HTTP disponível (upload de certificado, criação de série) e trata a criação do Product como um pré-requisito de infraestrutura resolvido pela fixture `product_api_key` (que É a única concessão: sem UM jeito de provisionar um Product por HTTP -- design spec decisão #5, "provisionar um Product é uma ação de operador, não self-service" -- não há como popular um cenário sem tocar a camada de app UMA vez por teste; a asserção em si nunca o faz).""" import uuid from datetime import datetime, timedelta, timezone import pytest import pytest_asyncio 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 fiscal_svc.core.db import get_session from fiscal_svc.main import app _CNPJ = "14200166000187" @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")) yield @pytest_asyncio.fixture async def product_api_key(db_session): """A ÚNICA concessão à regra "zero import do interior" -- provisionar um Product não tem endpoint HTTP (decisão de design deliberada, ver o docstring do módulo), então esta fixture chama a função de serviço diretamente. Nenhum teste abaixo importa nada além disto.""" from fiscal_svc.tenancy.service import create_product key = f"k-{uuid.uuid4().hex}" await create_product(db_session, name=f"contract-{uuid.uuid4().hex[:8]}", api_key=key) return key def _headers(api_key: str) -> dict[str, str]: return {"X-Api-Key": api_key} def _build_test_pfx(cnpj: str = _CNPJ, 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"EMPRESA TESTE 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")), ) # --- health ------------------------------------------------------------ @pytest.mark.asyncio async def test_health_returns_200_ok_shape(): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/v1/health") assert response.status_code == 200 assert response.json() == {"status": "ok"} # --- auth (require_product) -------------------------------------------- @pytest.mark.asyncio @pytest.mark.parametrize( "method,path,kwargs", [ ("get", "/v1/certificados", {"params": {"tenant_ref": "t", "branch_ref": "b"}}), ("get", "/v1/series", {"params": {"tenant_ref": "t", "branch_ref": "b"}}), ("get", "/v1/documentos", {}), ("post", "/v1/series", {"json": {"tenant_ref": "t", "branch_ref": "b", "document_model": "55", "serie": 1, "next_number": 1}}), ], ) async def test_every_v1_route_requires_api_key(method, path, kwargs): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await getattr(client, method)(path, **kwargs) assert response.status_code == 401, response.text @pytest.mark.asyncio async def test_wrong_api_key_is_401(): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get( "/v1/series", params={"tenant_ref": "t", "branch_ref": "b"}, headers=_headers("not-a-real-key"), ) assert response.status_code == 401, response.text # --- series -------------------------------------------------------------- @pytest.mark.asyncio async def test_series_create_list_patch_shapes(product_api_key): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: create_response = await client.post( "/v1/series", json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, headers=_headers(product_api_key), ) assert create_response.status_code == 201, create_response.text body = create_response.json() for field in ("id", "product_id", "tenant_ref", "branch_ref", "document_model", "serie", "next_number"): assert field in body, f"campo {field!r} ausente no shape de FiscalSeriesRead" series_id = body["id"] duplicate_response = await client.post( "/v1/series", json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 2}, headers=_headers(product_api_key), ) assert duplicate_response.status_code == 409, duplicate_response.text assert duplicate_response.json()["detail"]["code"] == "duplicate_fiscal_series" list_response = await client.get( "/v1/series", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(product_api_key) ) assert list_response.status_code == 200, list_response.text assert isinstance(list_response.json(), list) assert len(list_response.json()) == 1 patch_response = await client.patch( f"/v1/series/{series_id}", json={"next_number": 5}, headers=_headers(product_api_key) ) assert patch_response.status_code == 200, patch_response.text assert patch_response.json()["next_number"] == 5 not_found_response = await client.patch( f"/v1/series/{uuid.uuid4()}", json={"next_number": 5}, headers=_headers(product_api_key) ) assert not_found_response.status_code == 404, not_found_response.text # --- certificados ---------------------------------------------------------- @pytest.mark.asyncio async def test_certificados_upload_get_delete_shapes(product_api_key): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: upload_response = await client.post( "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, files={"file": ("cert.pfx", _build_test_pfx(), "application/x-pkcs12")}, data={"password": "senha123", "cnpj": _CNPJ}, headers=_headers(product_api_key), ) assert upload_response.status_code == 201, upload_response.text body = upload_response.json() for field in ( "id", "product_id", "tenant_ref", "branch_ref", "cnpj", "subject_cn", "cnpj_certificado", "not_valid_before", "not_valid_after", "created_at", ): assert field in body, f"campo {field!r} ausente no shape de FiscalCertificateRead" assert "pfx_encrypted" not in body assert "password_encrypted" not in body get_response = await client.get( "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(product_api_key) ) assert get_response.status_code == 200, get_response.text delete_response = await client.delete( "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(product_api_key) ) assert delete_response.status_code == 204, delete_response.text get_after_delete_response = await client.get( "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(product_api_key) ) assert get_after_delete_response.status_code == 404, get_after_delete_response.text @pytest.mark.asyncio async def test_certificado_invalido_e_422(product_api_key): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, files={"file": ("cert.pfx", b"nao e um pfx", "application/x-pkcs12")}, data={"password": "qualquer", "cnpj": _CNPJ}, headers=_headers(product_api_key), ) assert response.status_code == 422, response.text # --- emissão --------------------------------------------------------------- def _minimal_emissao_payload(*, tenant_ref="t1", branch_ref="b1", serie=1) -> dict: return { "tenant_ref": tenant_ref, "branch_ref": branch_ref, "document_model": "55", "serie": serie, "emitente": { "cnpj": _CNPJ, "razao_social": "EMPRESA TESTE LTDA", "nome_fantasia": None, "ie": "1234567890", "crt": "1", "address_street": "Rua Teste", "address_number": "100", "address_complement": None, "address_district": "Centro", "address_city": "Curitiba", "address_state": "PR", "address_zip": "80000000", "address_city_ibge_code": "4106902", }, "itens": [ { "codigo": "PC001", "descricao": "Peca teste", "ncm": "84212300", "cfop": "5102", "unidade_comercial": "UN", "unidade_tributavel": "UN", "quantidade": "1", "valor_unitario": "10.00", "fiscal_result": { "cfop": "5102", "cst": None, "csosn": "102", "origem": "0", "consumidor_final": True, "indicador_ie": "9", "tributos": [], }, } ], "pagamento": {"tpag": "01", "valor": "10.00", "indpag": "0"}, "ambiente": "homologacao", "uf_destino_tipo": "interna", "destinatario": None, "ver_proc": "contract-test/1.0", } @pytest.mark.asyncio async def test_emissao_requires_idempotency_key_header(product_api_key): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/v1/emissoes", json=_minimal_emissao_payload(), headers=_headers(product_api_key) ) assert response.status_code == 422, response.text @pytest.mark.asyncio async def test_emissao_requires_ver_proc(product_api_key): payload = _minimal_emissao_payload() del payload["ver_proc"] transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/v1/emissoes", json=payload, headers={**_headers(product_api_key), "Idempotency-Key": f"k-{uuid.uuid4().hex}"}, ) assert response.status_code == 422, response.text @pytest.mark.asyncio async def test_emissao_without_certificate_or_series_is_409_fiscal_config_missing(product_api_key): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/v1/emissoes", json=_minimal_emissao_payload(), headers={**_headers(product_api_key), "Idempotency-Key": f"k-{uuid.uuid4().hex}"}, ) assert response.status_code == 409, response.text body = response.json() assert body["detail"]["code"] == "fiscal_config_missing" assert "message" in body["detail"] @pytest.mark.asyncio async def test_full_emissao_happy_path_shapes_and_idempotency(product_api_key): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: upload_response = await client.post( "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, files={"file": ("cert.pfx", _build_test_pfx(), "application/x-pkcs12")}, data={"password": "senha123", "cnpj": _CNPJ}, headers=_headers(product_api_key), ) assert upload_response.status_code == 201, upload_response.text series_response = await client.post( "/v1/series", json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, headers=_headers(product_api_key), ) assert series_response.status_code == 201, series_response.text idem_key = f"k-{uuid.uuid4().hex}" first_response = await client.post( "/v1/emissoes", json=_minimal_emissao_payload(), headers={**_headers(product_api_key), "Idempotency-Key": idem_key}, ) assert first_response.status_code == 201, first_response.text body = first_response.json() for field in ( "id", "product_id", "tenant_ref", "branch_ref", "series_id", "document_model", "serie", "numero", "chave_acesso", "codigo_numerico", "status", "ambiente", "rejeicao_codigo", "rejeicao_motivo", "protocolo", "autorizada_em", "created_at", ): assert field in body, f"campo {field!r} ausente no shape de FiscalDocumentRead" assert "xml_assinado" not in body assert body["status"] == "ASSINADO" document_id = body["id"] second_response = await client.post( "/v1/emissoes", json=_minimal_emissao_payload(), headers={**_headers(product_api_key), "Idempotency-Key": idem_key}, ) assert second_response.status_code == 200, second_response.text assert second_response.json()["id"] == document_id get_response = await client.get(f"/v1/documentos/{document_id}", headers=_headers(product_api_key)) assert get_response.status_code == 200, get_response.text assert "xml_assinado" not in get_response.json() xml_response = await client.get(f"/v1/documentos/{document_id}/xml", headers=_headers(product_api_key)) assert xml_response.status_code == 200 assert xml_response.headers["content-type"].startswith("application/xml") assert "