from __future__ import annotations import json import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Dict from appwrite.exception import AppwriteException import pytest from fastapi import HTTPException, 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_PEDIDOS_ID", "test-orders") os.environ.setdefault("APPWRITE_COLLECTION_CARRINHOS_ID", "test-carts") 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_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("MERCADO_PAGO_ACCESS_TOKEN", "test-mercado-pago-token") os.environ.setdefault("MERCADO_PAGO_PUBLIC_KEY", "test-mercado-pago-key") os.environ.setdefault("JWT_SECRET", "test-secret") os.environ.setdefault("JWT_ALGORITHM", "HS256") from src.app.main import app from src.core.deps import get_pedidos_service, require_api_key, require_user from src.modules.pedidos.schemas import ( PedidoCreateRequest, PedidoOut, PedidoStatusUpdate, PedidoUpdate, ) from src.modules.pedidos.service import PedidosService class InMemoryPedidosService: 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, usuario_id: str | None, sort: str | None, order: str | None, ) -> Dict[str, object]: limit_value = limit items = list(self._items.values()) items.sort(key=lambda item: item.get("createdAt", ""), reverse=True) if usuario_id: items = [item for item in items if item.get("usuarios") == usuario_id] total = len(items) return { "page": page, "limit": limit_value, "total": total, "items": [PedidoOut(**item).model_dump() for item in items], } async def get(self, pedido_id: str) -> PedidoOut: if pedido_id not in self._items: raise AssertionError("Pedido not found in test store") return PedidoOut(**self._items[pedido_id]) async def create(self, payload: PedidoCreateRequest) -> PedidoOut: data = payload.data.model_dump(by_alias=True, exclude_unset=True, exclude_none=True) self._counter += 1 identifier = f"order_{self._counter}" timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") record = { "id": identifier, "status": data.get("status", "pendente"), "valor_total": data.get("valor-total"), "itens": data.get("itens", []), "quantidade": data.get("quantidade", []), "pagamentos": data.get("pagamentos"), "carrinhos": data.get("carrinhos"), "usuarios": data.get("usuarios"), "faturas": data.get("faturas"), "createdAt": timestamp, "updatedAt": timestamp, } self._items[identifier] = record return PedidoOut(**record) async def update(self, pedido_id: str, payload: PedidoUpdate) -> PedidoOut: if pedido_id not in self._items: raise AssertionError("Pedido not found in test store") stored = self._items[pedido_id].copy() data = payload.model_dump(by_alias=True, exclude_unset=True, exclude_none=True) stored.update({k: v for k, v in data.items() if v is not None}) stored["updatedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") self._items[pedido_id] = stored return PedidoOut(**stored) async def update_status(self, pedido_id: str, payload: PedidoStatusUpdate) -> PedidoOut: if pedido_id not in self._items: raise AssertionError("Pedido not found in test store") stored = self._items[pedido_id].copy() stored["status"] = payload.status stored["updatedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") self._items[pedido_id] = stored return PedidoOut(**stored) async def delete(self, pedido_id: str) -> None: self._items.pop(pedido_id, None) @pytest.fixture() def test_client() -> TestClient: service = InMemoryPedidosService() def override_service() -> InMemoryPedidosService: return service app.dependency_overrides[get_pedidos_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 app.dependency_overrides.clear() client.close() def test_ping_route_is_available(test_client: TestClient) -> None: response = test_client.get("/api/v1/ping") assert response.status_code == 200 def test_create_pedido_returns_201(test_client: TestClient) -> None: payload = { "documentId": "unique()", "data": { "status": "pendente", "valor-total": 99.5, "itens": ["item_1"], "quantidade": [1], "usuarios": "user_1", "carrinhos": "cart_1", "pagamentos": "pay_1", "faturas": "fat_1", }, } response = test_client.post("/api/v1/pedidos", json=payload) assert response.status_code == 201 body = response.json() assert body["status"] == "pendente" assert body["itens"] == ["item_1"] assert body["usuarios"] == "user_1" def test_list_pedidos_returns_created_items(test_client: TestClient) -> None: payload = { "documentId": "unique()", "data": { "status": "pendente", "valor-total": 50.0, "itens": ["item_1"], "quantidade": [1], "usuarios": "user_1", "carrinhos": "cart_1", "pagamentos": "pay_1", "faturas": "fat_1", }, } test_client.post("/api/v1/pedidos", json=payload) response = test_client.get("/api/v1/pedidos") assert response.status_code == 200 data = response.json() assert data["total"] == 1 assert len(data["items"]) == 1 def test_list_pedidos_returns_latest_first(test_client: TestClient) -> None: payload = { "documentId": "unique()", "data": { "status": "pendente", "valor-total": 10.0, "itens": ["item_1"], "quantidade": [1], "usuarios": "user_1", "carrinhos": "cart_1", "pagamentos": "pay_1", "faturas": "fat_1", }, } response_first = test_client.post("/api/v1/pedidos", json=payload) assert response_first.status_code == 201 first_id = response_first.json()["id"] response_second = test_client.post("/api/v1/pedidos", json=payload) assert response_second.status_code == 201 second_id = response_second.json()["id"] list_response = test_client.get("/api/v1/pedidos") assert list_response.status_code == 200 items = list_response.json()["items"] assert [item["id"] for item in items[:2]] == [second_id, first_id] def test_list_pedidos_filters_by_usuario_id(test_client: TestClient) -> None: payload_user1 = { "documentId": "unique()", "data": { "status": "pendente", "valor-total": 50.0, "itens": ["item_1"], "quantidade": [1], "usuarios": "user_1", "carrinhos": "cart_1", "pagamentos": "pay_1", "faturas": "fat_1", }, } payload_user2 = { "documentId": "unique()", "data": { "status": "pendente", "valor-total": 60.0, "itens": ["item_2"], "quantidade": [1], "usuarios": "user_2", "carrinhos": "cart_2", "pagamentos": "pay_2", "faturas": "fat_2", }, } test_client.post("/api/v1/pedidos", json=payload_user1) test_client.post("/api/v1/pedidos", json=payload_user2) response = test_client.get("/api/v1/pedidos", params={"usuario_id": "user_2"}) assert response.status_code == 200 data = response.json() assert data["total"] == 1 assert len(data["items"]) == 1 assert data["items"][0]["usuarios"] == "user_2" def test_update_status_endpoint(test_client: TestClient) -> None: payload = { "documentId": "unique()", "data": { "status": "pendente", "itens": ["item_1"], "quantidade": [1], "usuarios": "user_1", "carrinhos": "cart_1", "pagamentos": "pay_1", "faturas": "fat_1", }, } creation = test_client.post("/api/v1/pedidos", json=payload) identifier = creation.json()["id"] response = test_client.patch( f"/api/v1/pedidos/{identifier}/status", json={"status": "aprovado"}, ) assert response.status_code == 200 assert response.json()["status"] == "aprovado" @pytest.mark.asyncio() async def test_create_pedido_translates_appwrite_401_error() -> None: class FakeDatabases: def create_document(self, *args, **kwargs): raise AppwriteException( "The current user is not authorized to perform the requested action.", 401, ) service = PedidosService( databases=FakeDatabases(), database_id="db", collection_id="col", ) payload = PedidoCreateRequest.model_validate( { "documentId": "unique()", "data": { "status": "pendente", "valor-total": 10.0, "itens": [], "quantidade": [], }, } ) with pytest.raises(HTTPException) as exc_info: await service.create(payload) assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED assert "Falha na autenticação com Appwrite" in exc_info.value.detail @pytest.mark.asyncio() async def test_list_with_usuario_id_preserves_pagination() -> None: documents = [ { "$id": f"order_{index}", "status": "pendente", "valor-total": 9.99, "itens": ["item_1"], "quantidade": [1], "usuarios": "user_1", "pagamentos": None, "carrinhos": None, "faturas": None, "$createdAt": f"2025-01-0{index}T12:00:00.000+00:00", "$updatedAt": f"2025-01-0{index}T12:00:00.000+00:00", } for index in range(1, 6) ] class FakeDatabases: def list_documents(self, _database_id, _collection_id, queries=None): limit = 20 offset = 0 filtered = list(documents) for query in queries or []: query_obj = json.loads(query) if isinstance(query, str) else query method = query_obj.get("method") if method == "limit": limit = int(query_obj["values"][0]) elif method == "offset": offset = int(query_obj["values"][0]) elif method == "equal" and query_obj.get("attribute") == "usuarios": value = query_obj["values"][0] filtered = [doc for doc in filtered if doc.get("usuarios") == value] elif method == "orderDesc": attribute = query_obj.get("attribute") filtered.sort(key=lambda doc: doc.get(attribute), reverse=True) elif method == "orderAsc": attribute = query_obj.get("attribute") filtered.sort(key=lambda doc: doc.get(attribute)) total = len(filtered) sliced = filtered[offset : offset + limit] return {"documents": sliced, "total": total} service = PedidosService( databases=FakeDatabases(), database_id="db", collection_id="col", ) result = await service.list( page=2, limit=2, q=None, usuario_id="user_1", sort=None, order=None, ) assert result["page"] == 2 assert result["limit"] == 2 assert result["total"] == 5 assert [item.id for item in result["items"]] == ["order_3", "order_2"]