jonatan/sowai-fiscal-svc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
sowai-fiscal-svc/tests/test_contract.py
T

401 lines
16 KiB
Python
Raw Permalink Blame History

"""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 "<NFe" in xml_response.text
list_response = await client.get(
"/v1/documentos", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(product_api_key)
)
assert list_response.status_code == 200, list_response.text
assert any(d["id"] == document_id for d in list_response.json())
not_found_response = await client.get(
f"/v1/documentos/{uuid.uuid4()}", headers=_headers(product_api_key)
)
assert not_found_response.status_code == 404, not_found_response.text
Reference in New Issue View Git Blame Copy Permalink