feat: read-only internal endpoint for GTI Fiscal to mirror clients
GET /internal/fiscal/clients returns every CLIENTE (active and inactive)
as {external_client_id, code, display_name, active} — CLIENTE.ID is the
identity (CODIGO is editable, same reasoning as FUNCIONARIO_PAINEL.ID in
SSO). Authenticated by a dedicated bearer token (FISCAL_INTERNAL_API_TOKEN,
never reused from SESSION_SECRET/SSO key/admin password), rate-limited per
IP, returns 503 rather than opening anonymously when the token isn't
configured. GTI Gerenciador stays the single source of truth: the Fiscal
Hub never receives MySQL credentials.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
161da90e02
commit
b1f93c2d68
@@ -0,0 +1,81 @@
|
||||
"""GET /internal/fiscal/clients — endpoint read-only server-to-server para o
|
||||
GTI Fiscal espelhar o cadastro de clientes. Usa CODIGO na faixa de teste
|
||||
"999950"-"999959" (ver tests/test_db_clientes.py) e limpa o que criou."""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.db import clientes as db_clientes
|
||||
from app.db.connection import get_connection
|
||||
from app.main import app
|
||||
|
||||
TOKEN = "teste-token-fiscal-interno-suficientemente-longo"
|
||||
|
||||
BASE = {
|
||||
"CNPJ": "11222333000181", "RAZAO": "TESTE INTERNO RAZAO", "IE": "ISENTO",
|
||||
"ENDERECO": "RUA TESTE, 1", "CIDADE": "FEIRA DE SANTANA", "BAIRRO": "CENTRO", "UF": "BA",
|
||||
"CONTATO": "FULANO", "TELEFONE": "75999090000", "CELULAR": "75999090000", "CEP": "44000000",
|
||||
"IMPLANTACAO": "0.00", "MENSALIDADE": "100.00", "JUROS_MORA": "0.00", "JUROS_MULTA": "0.00",
|
||||
"DATA_CONTRATO": "2026-01-01", "EMAIL": "teste@example.com", "OBS": "teste automatizado",
|
||||
"VENCIMENTO_CONTRATO": "2027-01-01", "DIA_VENCIMENTO": "10",
|
||||
}
|
||||
|
||||
|
||||
def _limpar(codigos):
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(f"DELETE FROM CLIENTE WHERE CODIGO IN ({','.join(['%s'] * len(codigos))})", codigos)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clientes_teste():
|
||||
ativo_id = db_clientes.criar_cliente({**BASE, "CODIGO": "999950", "FANTASIA": "Ativo Fiscal Teste", "ATIVO": "1"})
|
||||
inativo_id = db_clientes.criar_cliente({**BASE, "CODIGO": "999951", "FANTASIA": "", "RAZAO": "Inativo Fiscal Teste Razao", "ATIVO": "0"})
|
||||
yield ativo_id, inativo_id
|
||||
_limpar(["999950", "999951"])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch):
|
||||
monkeypatch.setenv("FISCAL_INTERNAL_API_TOKEN", TOKEN)
|
||||
from app import internal_fiscal
|
||||
internal_fiscal._falhas.clear()
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_returns_active_and_inactive_clients_with_minimal_fields(client, clientes_teste):
|
||||
ativo_id, inativo_id = clientes_teste
|
||||
response = client.get("/internal/fiscal/clients", headers={"Authorization": f"Bearer {TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
by_id = {row["external_client_id"]: row for row in response.json()["clients"]}
|
||||
assert by_id[ativo_id] == {"external_client_id": ativo_id, "code": "999950", "display_name": "Ativo Fiscal Teste", "active": True}
|
||||
assert by_id[inativo_id] == {"external_client_id": inativo_id, "code": "999951", "display_name": "Inativo Fiscal Teste Razao", "active": False}
|
||||
# never leaks fields the Fiscal doesn't need
|
||||
assert set(by_id[ativo_id]) == {"external_client_id", "code", "display_name", "active"}
|
||||
|
||||
|
||||
def test_missing_token_rejected(client, clientes_teste):
|
||||
assert client.get("/internal/fiscal/clients").status_code == 401
|
||||
|
||||
|
||||
def test_wrong_token_rejected(client, clientes_teste):
|
||||
response = client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong-token"})
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_endpoint_disabled_when_token_not_configured(monkeypatch):
|
||||
monkeypatch.delenv("FISCAL_INTERNAL_API_TOKEN", raising=False)
|
||||
from app import internal_fiscal
|
||||
internal_fiscal._falhas.clear()
|
||||
client = TestClient(app)
|
||||
response = client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer anything"})
|
||||
assert response.status_code == 503
|
||||
|
||||
|
||||
def test_rate_limited_after_repeated_invalid_attempts(client):
|
||||
from app import internal_fiscal
|
||||
for _ in range(internal_fiscal.MAX_TENTATIVAS):
|
||||
assert client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong"}).status_code == 401
|
||||
assert client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong"}).status_code == 429
|
||||
Reference in New Issue
Block a user