190 lines
7.1 KiB
Python
190 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
import pytest
|
|
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))
|
|
|
|
from src.modules.pagamentos.service import PagamentosService
|
|
|
|
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_CATEGORIAS_ID", "test-categorias")
|
|
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-cat-rel")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_SUBCATEGORIAS_ID", "test-subcat")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_CARRINHOS_ID", "test-carts")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_PEDIDOS_ID", "test-orders")
|
|
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("JWT_SECRET", "test-secret")
|
|
os.environ.setdefault("JWT_ALGORITHM", "HS256")
|
|
|
|
|
|
class InMemoryPagamentosService:
|
|
def __init__(self) -> None:
|
|
self._items: Dict[str, Dict[str, object]] = {}
|
|
self._counter = 0
|
|
|
|
async def list(
|
|
self,
|
|
page: int,
|
|
limit: int | None,
|
|
q: str | None,
|
|
sort: str | None,
|
|
order: str | None,
|
|
) -> Dict[str, object]:
|
|
limit_value = limit or 20
|
|
items = list(self._items.values())
|
|
total = len(items)
|
|
return {
|
|
"page": page,
|
|
"limit": limit_value,
|
|
"total": total,
|
|
"items": items,
|
|
}
|
|
|
|
async def get(self, pagamento_id: str):
|
|
if pagamento_id not in self._items:
|
|
raise AssertionError("Pagamento not found in test store")
|
|
return self._items[pagamento_id]
|
|
|
|
async def create(self, payload) -> Dict[str, object]: # pragma: no cover - not used
|
|
self._counter += 1
|
|
identifier = f"pay_{self._counter}"
|
|
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
record = {
|
|
"id": identifier,
|
|
"status": payload.status,
|
|
"metodo": payload.metodo,
|
|
"valor": payload.valor,
|
|
"pedidos": payload.pedidos,
|
|
"createdAt": timestamp,
|
|
"updatedAt": timestamp,
|
|
}
|
|
self._items[identifier] = record
|
|
return record
|
|
|
|
async def update(self, pagamento_id: str, payload) -> Dict[str, object]: # pragma: no cover - not used
|
|
if pagamento_id not in self._items:
|
|
raise AssertionError("Pagamento not found in test store")
|
|
stored = self._items[pagamento_id].copy()
|
|
for key, value in payload.model_dump(exclude_unset=True).items():
|
|
if value is not None:
|
|
stored[key] = value
|
|
stored["updatedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
self._items[pagamento_id] = stored
|
|
return stored
|
|
|
|
async def delete(self, pagamento_id: str) -> None: # pragma: no cover - not used
|
|
self._items.pop(pagamento_id, None)
|
|
|
|
|
|
@pytest.fixture()
|
|
def pagamentos_client() -> tuple[TestClient, str, InMemoryPagamentosService]:
|
|
from src.core import config as config_module
|
|
from src.core.deps import get_pagamentos_service, require_api_key, require_user
|
|
|
|
original_api_key = os.environ.get("SECURITY_API_KEY", "")
|
|
api_key = "test-pay-key"
|
|
os.environ["SECURITY_API_KEY"] = api_key
|
|
|
|
config_module.get_settings.cache_clear() # type: ignore[attr-defined]
|
|
|
|
main_module = importlib.import_module("src.main")
|
|
importlib.reload(main_module)
|
|
app_module = importlib.import_module("src.app.main")
|
|
importlib.reload(app_module)
|
|
app = app_module.app
|
|
|
|
service = InMemoryPagamentosService()
|
|
|
|
def override_service() -> InMemoryPagamentosService:
|
|
return service
|
|
|
|
app.dependency_overrides[get_pagamentos_service] = override_service
|
|
app.dependency_overrides[require_api_key] = lambda: None
|
|
app.dependency_overrides[require_user] = lambda: {"id": "user_1"}
|
|
|
|
client = TestClient(app)
|
|
yield client, api_key, service
|
|
|
|
client.close()
|
|
app.dependency_overrides.clear()
|
|
os.environ["SECURITY_API_KEY"] = original_api_key
|
|
config_module.get_settings.cache_clear() # type: ignore[attr-defined]
|
|
importlib.reload(main_module)
|
|
importlib.reload(app_module)
|
|
|
|
|
|
def test_list_pagamentos_requires_api_key(
|
|
pagamentos_client: tuple[TestClient, str, InMemoryPagamentosService]
|
|
) -> None:
|
|
client, api_key, _ = pagamentos_client
|
|
|
|
response = client.get("/api/v1/pagamentos")
|
|
assert response.status_code == 401
|
|
|
|
response_with_key = client.get("/api/v1/pagamentos", headers={"x-api-key": api_key})
|
|
assert response_with_key.status_code == 200
|
|
|
|
|
|
def test_get_pagamento_route_available(
|
|
pagamentos_client: tuple[TestClient, str, InMemoryPagamentosService]
|
|
) -> None:
|
|
client, api_key, service = pagamentos_client
|
|
|
|
service._items["pay_test"] = {
|
|
"id": "pay_test",
|
|
"status": "pendente",
|
|
"metodo": "pix",
|
|
"valor": 10.0,
|
|
"pedidos": "ped_test",
|
|
"createdAt": "2025-10-01T12:00:00.000Z",
|
|
"updatedAt": "2025-10-01T12:00:00.000Z",
|
|
}
|
|
|
|
response = client.get("/api/v1/pagamentos/pay_test", headers={"x-api-key": api_key})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == "pay_test"
|
|
assert body["status"] == "pendente"
|
|
|
|
|
|
def test_pagamentos_service_map_document_handles_relation_dict() -> None:
|
|
document = {
|
|
"$id": "pay_123",
|
|
"status": "pendente",
|
|
"metodo": "pix",
|
|
"valor": 29.9,
|
|
"pedidos": {"$id": "ped_456", "$collectionId": "pedidos"},
|
|
"$createdAt": "2024-01-01T00:00:00.000Z",
|
|
"$updatedAt": "2024-01-01T00:00:00.000Z",
|
|
}
|
|
|
|
mapped = PagamentosService._map_document(document)
|
|
|
|
assert mapped.pedidos == "ped_456"
|