117 lines
4.8 KiB
Python
117 lines
4.8 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
import pytest
|
|
from fastapi import status
|
|
from fastapi.testclient import TestClient
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
os.environ.setdefault("APPWRITE_ENDPOINT", "http://localhost/v1")
|
|
os.environ.setdefault("APPWRITE_PROJECT_ID", "test-project")
|
|
os.environ.setdefault("APPWRITE_API_KEY", "test-key")
|
|
os.environ.setdefault("APPWRITE_DATABASE_ID", "test-db")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_CARRINHOS_ID", "test-carts")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_PEDIDOS_ID", "test-orders")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_PRODUTOS_CATALOGO_ID", "test-produtos")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_LABORATORIOS_ID", "test-labs")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_CATEGORIAS_ID", "test-categorias")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_SUBCATEGORIAS_ID", "test-subcat")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_PAGAMENTOS_ID", "test-payments")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_PRODUTOS_ESTOQUE_ID", "test-produtos-estoque")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_PRODUTOS_VENDA_ID", "test-produtos-venda")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_USUARIOS_DATA_ID", "test-usuarios-data")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_USUARIOS_PERFIL_ID", "test-usuarios-perfil")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_USUARIO_EMPRESA_PERFIS_ID", "test-usuario-empresa-perfis")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_FATURAS_ID", "test-faturas")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_USUARIOS_ID", "test-users")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_EMPRESAS_ID", "test-companies")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_EMPRESAS_DADOS_ID", "test-companies-data")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_EMPRESAS_SOCIOS_ID", "test-empresas-socios")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_ENDERECOS_ID", "test-addresses")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_ENTREGAS_ID", "test-entregas")
|
|
os.environ.setdefault("SECURITY_API_KEY", "")
|
|
os.environ.setdefault("JWT_SECRET", "test-secret")
|
|
os.environ.setdefault("JWT_ALGORITHM", "HS256")
|
|
|
|
from src.app.main import app
|
|
from src.routers.public import cnpj_router
|
|
|
|
|
|
class MockResponse:
|
|
def __init__(self, status_code: int, json_data: Any) -> None:
|
|
self.status_code = status_code
|
|
self._json_data = json_data
|
|
|
|
def json(self) -> Any:
|
|
if isinstance(self._json_data, Exception):
|
|
raise self._json_data
|
|
return self._json_data
|
|
|
|
|
|
def _mock_async_client(response: MockResponse):
|
|
class _Client:
|
|
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
|
self._response = response
|
|
|
|
async def __aenter__(self) -> "_Client":
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb) -> None: # pragma: no cover - interface
|
|
return None
|
|
|
|
async def get(self, *args: Any, **kwargs: Any) -> MockResponse:
|
|
return self._response
|
|
|
|
return _Client
|
|
|
|
|
|
@pytest.fixture()
|
|
def client() -> TestClient:
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
|
|
def test_consultar_cnpj_success(monkeypatch: pytest.MonkeyPatch, client: TestClient) -> None:
|
|
payload: Dict[str, Any] = {
|
|
"status": "OK",
|
|
"cnpj": "27865757000102",
|
|
"nome": "SAVE IN MED COMERCIO DE MEDICAMENTOS LTDA",
|
|
}
|
|
response = MockResponse(status.HTTP_200_OK, payload)
|
|
monkeypatch.setattr(cnpj_router.httpx, "AsyncClient", _mock_async_client(response))
|
|
|
|
result = client.get("/public/cnpj/27.865.757/0001-02")
|
|
|
|
assert result.status_code == status.HTTP_200_OK
|
|
assert result.json() == payload
|
|
|
|
|
|
def test_consultar_cnpj_invalid_param(client: TestClient) -> None:
|
|
result = client.get("/public/cnpj/1234")
|
|
assert result.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
assert result.json()["detail"] == "CNPJ deve conter 14 dígitos numéricos."
|
|
|
|
|
|
def test_consultar_cnpj_not_found(monkeypatch: pytest.MonkeyPatch, client: TestClient) -> None:
|
|
response = MockResponse(status.HTTP_404_NOT_FOUND, {"message": "CNPJ não encontrado"})
|
|
monkeypatch.setattr(cnpj_router.httpx, "AsyncClient", _mock_async_client(response))
|
|
|
|
result = client.get("/public/cnpj/27865757000102")
|
|
|
|
assert result.status_code == status.HTTP_404_NOT_FOUND
|
|
assert result.json()["detail"] == "CNPJ não encontrado"
|
|
|
|
|
|
def test_consultar_cnpj_rate_limit(monkeypatch: pytest.MonkeyPatch, client: TestClient) -> None:
|
|
response = MockResponse(status.HTTP_429_TOO_MANY_REQUESTS, {"message": "Too Many Requests"})
|
|
monkeypatch.setattr(cnpj_router.httpx, "AsyncClient", _mock_async_client(response))
|
|
|
|
result = client.get("/public/cnpj/27865757000102")
|
|
|
|
assert result.status_code == status.HTTP_429_TOO_MANY_REQUESTS
|
|
assert result.json()["detail"].startswith("Serviço externo indisponível")
|