143 lines
5.1 KiB
Python
143 lines
5.1 KiB
Python
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
pytest.importorskip("email_validator")
|
|
|
|
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_EMPRESAS_ID", "test-companies")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_EMPRESAS_DADOS_ID", "test-companies-data")
|
|
os.environ.setdefault("APPWRITE_COLLECTION_ENDERECOS_ID", "test-addresses")
|
|
|
|
from appwrite.exception import AppwriteException
|
|
|
|
from src.modules.empresas.schemas import EmpresaCreate, EmpresaUpdate
|
|
from src.modules.empresas.service import EmpresasService
|
|
|
|
|
|
class _SyncEmpresasService(EmpresasService):
|
|
async def _run(self, func: Any, *args: Any, **kwargs: Any) -> Any: # pragma: no cover - helper override
|
|
return func(*args, **kwargs)
|
|
|
|
|
|
class _DummyDatabases:
|
|
def __init__(self, *, existing_enderecos: List[str]) -> None:
|
|
self._existing_enderecos = set(existing_enderecos)
|
|
self.created_payloads: List[Dict[str, Any]] = []
|
|
self.updated_payloads: List[Dict[str, Any]] = []
|
|
self.endereco_lookups: List[str] = []
|
|
|
|
def get_document(self, database_id: str, collection_id: str, document_id: str) -> Dict[str, Any]:
|
|
if collection_id != "test-addresses": # pragma: no cover - defensive
|
|
raise AssertionError("Unexpected collection lookup during tests")
|
|
self.endereco_lookups.append(document_id)
|
|
if document_id not in self._existing_enderecos:
|
|
raise AppwriteException("Not found", 404)
|
|
return {"$id": document_id}
|
|
|
|
def create_document(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
document_id: str,
|
|
data: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
self.created_payloads.append(data)
|
|
return {
|
|
"$id": "emp_1",
|
|
"cnpj": data.get("cnpj"),
|
|
"enderecos": data.get("enderecos"),
|
|
"$createdAt": "2024-01-01T00:00:00Z",
|
|
"$updatedAt": "2024-01-01T00:00:00Z",
|
|
}
|
|
|
|
def update_document(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
document_id: str,
|
|
data: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
self.updated_payloads.append(data)
|
|
return {
|
|
"$id": document_id,
|
|
"cnpj": "27865757000102",
|
|
"enderecos": data.get("enderecos"),
|
|
"$createdAt": "2024-01-01T00:00:00Z",
|
|
"$updatedAt": "2024-01-02T00:00:00Z",
|
|
}
|
|
|
|
|
|
def test_create_empresa_validates_and_normalizes_enderecos() -> None:
|
|
databases = _DummyDatabases(existing_enderecos=["addr_1"])
|
|
service = _SyncEmpresasService(databases, "test-db", "test-companies-data", "test-addresses")
|
|
|
|
payload = EmpresaCreate(cnpj="27865757000102", enderecos=["addr_1", "addr_1 "])
|
|
|
|
result = asyncio.run(service.create(payload))
|
|
|
|
assert databases.endereco_lookups == ["addr_1"]
|
|
assert databases.created_payloads[0]["enderecos"] == ["addr_1"]
|
|
assert result.enderecos == ["addr_1"]
|
|
|
|
|
|
def test_create_empresa_accepts_single_endereco_id_alias() -> None:
|
|
databases = _DummyDatabases(existing_enderecos=["addr_1"])
|
|
service = _SyncEmpresasService(databases, "test-db", "test-companies-data", "test-addresses")
|
|
|
|
payload = EmpresaCreate(cnpj="27865757000102", enderecos="addr_1")
|
|
|
|
result = asyncio.run(service.create(payload))
|
|
|
|
assert databases.created_payloads[0]["enderecos"] == ["addr_1"]
|
|
assert result.enderecos == ["addr_1"]
|
|
|
|
|
|
def test_create_empresa_raises_when_endereco_missing() -> None:
|
|
databases = _DummyDatabases(existing_enderecos=[])
|
|
service = _SyncEmpresasService(databases, "test-db", "test-companies-data", "test-addresses")
|
|
|
|
payload = EmpresaCreate(cnpj="27865757000102", enderecos=["missing"])
|
|
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
asyncio.run(service.create(payload))
|
|
|
|
assert excinfo.value.status_code == 404
|
|
assert "Endereco" in excinfo.value.detail
|
|
|
|
|
|
def test_update_empresa_allows_clearing_enderecos() -> None:
|
|
databases = _DummyDatabases(existing_enderecos=["addr_1"])
|
|
service = _SyncEmpresasService(databases, "test-db", "test-companies-data", "test-addresses")
|
|
|
|
payload = EmpresaUpdate(enderecos=[])
|
|
|
|
result = asyncio.run(service.update("emp_1", payload))
|
|
|
|
assert databases.updated_payloads[0]["enderecos"] == []
|
|
assert result.enderecos == []
|
|
|
|
|
|
def test_update_empresa_validates_new_enderecos() -> None:
|
|
databases = _DummyDatabases(existing_enderecos=["addr_1"])
|
|
service = _SyncEmpresasService(databases, "test-db", "test-companies-data", "test-addresses")
|
|
|
|
payload = EmpresaUpdate(enderecos=["addr_1"])
|
|
|
|
result = asyncio.run(service.update("emp_1", payload))
|
|
|
|
assert databases.endereco_lookups == ["addr_1"]
|
|
assert databases.updated_payloads[0]["enderecos"] == ["addr_1"]
|
|
assert result.enderecos == ["addr_1"]
|