264 lines
9.1 KiB
Python
264 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
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))
|
|
|
|
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_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("JWT_SECRET", "test-secret")
|
|
os.environ.setdefault("JWT_ALGORITHM", "HS256")
|
|
|
|
from src.app.main import app
|
|
from src.core.deps import get_carrinhos_service, require_api_key, require_user
|
|
from src.modules.carrinhos.schemas import (
|
|
CarrinhoCreateRequest,
|
|
CarrinhoOut,
|
|
CarrinhoUpdateRequest,
|
|
)
|
|
from src.modules.carrinhos.service import CarrinhosService
|
|
|
|
|
|
class InMemoryCarrinhosService:
|
|
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": [CarrinhoOut(**item).model_dump() for item in items],
|
|
}
|
|
|
|
async def get(self, carrinho_id: str) -> CarrinhoOut:
|
|
if carrinho_id not in self._items:
|
|
raise AssertionError("Carrinho not found in test store")
|
|
return CarrinhoOut(**self._items[carrinho_id])
|
|
|
|
async def create(
|
|
self, payload: CarrinhoCreateRequest
|
|
) -> CarrinhoOut:
|
|
data = payload.data.model_dump(exclude_unset=True)
|
|
self._counter += 1
|
|
requested_id = payload.documentId
|
|
identifier = (
|
|
requested_id.strip()
|
|
if isinstance(requested_id, str) and requested_id.strip()
|
|
else f"cart_{self._counter}"
|
|
)
|
|
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
record = {
|
|
"id": identifier,
|
|
"itens": data.get("itens", []),
|
|
"codigoInterno": data.get("codigoInterno"),
|
|
"labNome": data.get("labNome", []),
|
|
"quantidade": data.get("quantidade", []),
|
|
"valorTotal": data.get("valorTotal", []),
|
|
"usuarios": data.get("usuarios"),
|
|
"createdAt": timestamp,
|
|
"updatedAt": timestamp,
|
|
}
|
|
self._items[identifier] = record
|
|
return CarrinhoOut(**record)
|
|
|
|
async def update(self, carrinho_id: str, payload: CarrinhoUpdateRequest) -> CarrinhoOut:
|
|
if carrinho_id not in self._items:
|
|
raise AssertionError("Carrinho not found in test store")
|
|
stored = self._items[carrinho_id].copy()
|
|
data = payload.data.model_dump(exclude_unset=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[carrinho_id] = stored
|
|
return CarrinhoOut(**stored)
|
|
|
|
async def delete(self, carrinho_id: str) -> None:
|
|
self._items.pop(carrinho_id, None)
|
|
|
|
|
|
@pytest.fixture()
|
|
def test_client() -> TestClient:
|
|
service = InMemoryCarrinhosService()
|
|
|
|
def override_service() -> InMemoryCarrinhosService:
|
|
return service
|
|
|
|
app.dependency_overrides[get_carrinhos_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_carrinho_returns_201(test_client: TestClient) -> None:
|
|
payload = {
|
|
"documentId": "unique()",
|
|
"data": {
|
|
"itens": ["item_1"],
|
|
"codigoInterno": "CART-001",
|
|
"labNome": ["Lab A"],
|
|
"quantidade": [1],
|
|
"valorTotal": [19.9],
|
|
"usuarios": "user_1",
|
|
},
|
|
}
|
|
response = test_client.post("/api/v1/carrinhos", json=payload)
|
|
assert response.status_code == 201
|
|
body = response.json()
|
|
assert body["codigoInterno"] == "CART-001"
|
|
assert body["itens"] == ["item_1"]
|
|
|
|
|
|
def test_list_carrinhos_returns_created_items(test_client: TestClient) -> None:
|
|
payload = {
|
|
"data": {
|
|
"itens": ["item_1"],
|
|
"codigoInterno": "CART-001",
|
|
"labNome": ["Lab A"],
|
|
"quantidade": [1],
|
|
"valorTotal": [19.9],
|
|
"usuarios": "user_1",
|
|
}
|
|
}
|
|
test_client.post("/api/v1/carrinhos", json=payload)
|
|
|
|
response = test_client.get("/api/v1/carrinhos")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 1
|
|
assert len(data["items"]) == 1
|
|
|
|
|
|
def test_update_carrinho_accepts_document_structure(test_client: TestClient) -> None:
|
|
create_payload = {
|
|
"documentId": "cart_patch",
|
|
"data": {
|
|
"itens": ["item_1", "item_2"],
|
|
"codigoInterno": "CART-123",
|
|
"labNome": ["Lab A", "Lab B"],
|
|
"quantidade": [1, 2],
|
|
"valorTotal": [10.5, 20.0],
|
|
"usuarios": "user_1",
|
|
},
|
|
}
|
|
response = test_client.post("/api/v1/carrinhos", json=create_payload)
|
|
assert response.status_code == 201
|
|
|
|
patch_payload = {
|
|
"data": {
|
|
"quantidade": [2, 3],
|
|
"valorTotal": [21.0, 30.0],
|
|
}
|
|
}
|
|
response = test_client.patch(
|
|
"/api/v1/carrinhos/cart_patch", json=patch_payload
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == "cart_patch"
|
|
assert body["quantidade"] == [2, 3]
|
|
assert body["valorTotal"] == [21.0, 30.0]
|
|
|
|
|
|
def test_map_document_handles_relation_objects() -> None:
|
|
document = {
|
|
"$id": "cart_rel",
|
|
"itens": ["item_1"],
|
|
"codigoInterno": "CART-REL",
|
|
"labNome": ["Lab"],
|
|
"quantidade": [1],
|
|
"valorTotal": [10.0],
|
|
"usuarios": {"$id": "user_rel", "name": "User"},
|
|
"$createdAt": "2025-09-29T12:00:00.000Z",
|
|
"$updatedAt": "2025-09-29T12:05:00.000Z",
|
|
}
|
|
|
|
result = CarrinhosService._map_document(document)
|
|
|
|
assert result.usuarios == "user_rel"
|
|
|
|
|
|
def test_map_document_handles_relation_lists() -> None:
|
|
document = {
|
|
"$id": "cart_rel",
|
|
"itens": ["item_1"],
|
|
"codigoInterno": "CART-REL",
|
|
"labNome": ["Lab"],
|
|
"quantidade": [1],
|
|
"valorTotal": [10.0],
|
|
"usuarios": [
|
|
{"$id": "user_rel", "name": "User"},
|
|
{"$id": "user_secondary"},
|
|
],
|
|
"$createdAt": "2025-09-29T12:00:00.000Z",
|
|
"$updatedAt": "2025-09-29T12:05:00.000Z",
|
|
}
|
|
|
|
result = CarrinhosService._map_document(document)
|
|
|
|
assert result.usuarios == "user_rel"
|
|
|
|
|
|
def test_sanitize_payload_accepts_relation_objects() -> None:
|
|
service = CarrinhosService(databases=object(), database_id="db", collection_id="col")
|
|
|
|
sanitized = service._sanitize_payload(
|
|
{
|
|
"codigoInterno": " CART-REL ",
|
|
"usuarios": {"$id": "user_rel"},
|
|
}
|
|
)
|
|
|
|
assert sanitized["codigoInterno"] == "CART-REL"
|
|
assert sanitized["usuarios"] == "user_rel"
|