213 lines
8.6 KiB
Python
213 lines
8.6 KiB
Python
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
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_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("JWT_SECRET", "test-secret")
|
|
os.environ.setdefault("JWT_ALGORITHM", "HS256")
|
|
os.environ.setdefault("SECURITY_API_KEY", "")
|
|
|
|
from src.app.main import app
|
|
from src.core.deps import get_empresas_socios_service, require_api_key, require_user
|
|
from src.modules.empresas_socios.schemas import (
|
|
EmpresasSociosCreate,
|
|
EmpresasSociosOut,
|
|
EmpresasSociosUpdate,
|
|
)
|
|
|
|
|
|
class InMemoryEmpresasSociosService:
|
|
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,
|
|
empresas_dados_id: str | None,
|
|
) -> Dict[str, object]:
|
|
items = list(self._items.values())
|
|
if empresas_dados_id:
|
|
items = [item for item in items if item.get("empresas_dados_id") == empresas_dados_id]
|
|
total = len(items)
|
|
return {
|
|
"items": [EmpresasSociosOut(**item).model_dump() for item in items],
|
|
"total": total,
|
|
}
|
|
|
|
async def get(self, socio_id: str) -> EmpresasSociosOut:
|
|
if socio_id not in self._items:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Sócio não encontrado.")
|
|
return EmpresasSociosOut(**self._items[socio_id])
|
|
|
|
async def create(self, payload: EmpresasSociosCreate) -> EmpresasSociosOut:
|
|
data = payload.model_dump(exclude_unset=True)
|
|
self._counter += 1
|
|
identifier = f"soc_{self._counter}"
|
|
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
record = {
|
|
"id": identifier,
|
|
"empresas_dados_id": data.get("empresas_dados_id"),
|
|
"nome": data.get("nome", ""),
|
|
"qualificacao": data.get("qualificacao", ""),
|
|
"data_entrada": data.get("data_entrada"),
|
|
"faixa_etaria": data.get("faixa_etaria"),
|
|
"identificador": data.get("identificador"),
|
|
"created_at": timestamp,
|
|
"updated_at": timestamp,
|
|
}
|
|
self._items[identifier] = record
|
|
return EmpresasSociosOut(**record)
|
|
|
|
async def update(self, socio_id: str, payload: EmpresasSociosUpdate) -> EmpresasSociosOut:
|
|
if socio_id not in self._items:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Sócio não encontrado.")
|
|
stored = self._items[socio_id].copy()
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
for key, value in updates.items():
|
|
field = key if key != "empresas_dados_id" else "empresas_dados_id"
|
|
stored[field] = value
|
|
stored["updated_at"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
self._items[socio_id] = stored
|
|
return EmpresasSociosOut(**stored)
|
|
|
|
async def delete(self, socio_id: str) -> None:
|
|
self._items.pop(socio_id, None)
|
|
|
|
|
|
@pytest.fixture()
|
|
def test_client() -> TestClient:
|
|
service = InMemoryEmpresasSociosService()
|
|
|
|
def override_service() -> InMemoryEmpresasSociosService:
|
|
return service
|
|
|
|
app.dependency_overrides[get_empresas_socios_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_create_empresas_socio_returns_created_item(test_client: TestClient) -> None:
|
|
payload = {
|
|
"empresas_dados_id": "emp-dados_1",
|
|
"nome": "Maria de Souza Lima",
|
|
"qualificacao": "10-Diretor",
|
|
"data_entrada": "2014-03-12",
|
|
"faixa_etaria": "31-40",
|
|
"identificador": "12345678901",
|
|
}
|
|
response = test_client.post("/appwrite/empresas-socios", json=payload)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
data = response.json()
|
|
assert data["nome"] == "Maria de Souza Lima"
|
|
assert data["empresas_dados_id"] == "emp-dados_1"
|
|
|
|
|
|
def test_list_empresas_socios_filters_by_empresa(test_client: TestClient) -> None:
|
|
payload = {
|
|
"empresas_dados_id": "emp-dados_1",
|
|
"nome": "João Pereira",
|
|
"qualificacao": "16-Sócio",
|
|
}
|
|
test_client.post("/appwrite/empresas-socios", json=payload)
|
|
test_client.post(
|
|
"/appwrite/empresas-socios",
|
|
json={"empresas_dados_id": "emp-dados_2", "nome": "Ana Lima", "qualificacao": "49-Sócio"},
|
|
)
|
|
|
|
response = test_client.get("/appwrite/empresas-socios", params={"empresas_dados_id": "emp-dados_1"})
|
|
assert response.status_code == status.HTTP_200_OK
|
|
body = response.json()
|
|
assert body["total"] == 1
|
|
assert body["items"][0]["nome"] == "João Pereira"
|
|
|
|
|
|
def test_get_empresas_socio_returns_item(test_client: TestClient) -> None:
|
|
creation = test_client.post(
|
|
"/appwrite/empresas-socios",
|
|
json={"empresas_dados_id": "emp-dados_1", "nome": "João", "qualificacao": "49-Sócio"},
|
|
)
|
|
identifier = creation.json()["id"]
|
|
|
|
response = test_client.get(f"/appwrite/empresas-socios/{identifier}")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["id"] == identifier
|
|
|
|
|
|
def test_update_empresas_socio_updates_fields(test_client: TestClient) -> None:
|
|
creation = test_client.post(
|
|
"/appwrite/empresas-socios",
|
|
json={"empresas_dados_id": "emp-dados_1", "nome": "João", "qualificacao": "49-Sócio"},
|
|
)
|
|
identifier = creation.json()["id"]
|
|
|
|
response = test_client.patch(
|
|
f"/appwrite/empresas-socios/{identifier}",
|
|
json={"nome": "João Silva", "faixa_etaria": "41-50"},
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
body = response.json()
|
|
assert body["nome"] == "João Silva"
|
|
assert body["faixa_etaria"] == "41-50"
|
|
|
|
|
|
def test_delete_empresas_socio_removes_item(test_client: TestClient) -> None:
|
|
creation = test_client.post(
|
|
"/appwrite/empresas-socios",
|
|
json={"empresas_dados_id": "emp-dados_1", "nome": "João", "qualificacao": "49-Sócio"},
|
|
)
|
|
identifier = creation.json()["id"]
|
|
|
|
delete_response = test_client.delete(f"/appwrite/empresas-socios/{identifier}")
|
|
assert delete_response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
list_response = test_client.get("/appwrite/empresas-socios")
|
|
assert list_response.json()["total"] == 0
|
|
|
|
|
|
def test_get_empresas_socio_not_found_returns_404(test_client: TestClient) -> None:
|
|
response = test_client.get("/appwrite/empresas-socios/unknown")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert response.json()["detail"] == "Sócio não encontrado."
|