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."