109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
import pytest
|
|
|
|
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_USUARIOS_ID", "test-usuarios")
|
|
|
|
from src.modules.usuarios.service import UsuariosService
|
|
|
|
|
|
class _SyncUsuariosService(UsuariosService):
|
|
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, documents: List[Dict[str, Any]]) -> None:
|
|
self._documents = documents
|
|
self.received_queries: List[List[Dict[str, Any]]] = []
|
|
|
|
def list_documents(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
queries: List[Dict[str, Any]] | None = None,
|
|
) -> Dict[str, Any]:
|
|
if queries is not None:
|
|
self.received_queries.append(queries)
|
|
else: # pragma: no cover - defensive
|
|
self.received_queries.append([])
|
|
return {"documents": list(self._documents), "total": len(self._documents)}
|
|
|
|
|
|
def _sample_document(identifier: str, *, ativo: bool) -> Dict[str, Any]:
|
|
return {
|
|
"$id": identifier,
|
|
"identificador": identifier,
|
|
"ativo": ativo,
|
|
"superadmin": False,
|
|
"$createdAt": "2024-01-01T00:00:00Z",
|
|
"$updatedAt": "2024-01-01T00:00:00Z",
|
|
}
|
|
|
|
|
|
def test_list_usuarios_adds_boolean_filter_to_queries() -> None:
|
|
databases = _DummyDatabases([
|
|
_sample_document("user_1", ativo=True),
|
|
_sample_document("user_2", ativo=False),
|
|
])
|
|
service = _SyncUsuariosService(databases, "test-db", "test-usuarios")
|
|
|
|
result = asyncio.run(
|
|
service.list(
|
|
page=1,
|
|
limit=10,
|
|
q=None,
|
|
sort=None,
|
|
order=None,
|
|
empresas_dados_id=None,
|
|
ativo=True,
|
|
)
|
|
)
|
|
|
|
assert result["page"] == 1
|
|
assert any(
|
|
isinstance(query, str)
|
|
and '"attribute":"ativo"' in query
|
|
and '"values":[true]' in query
|
|
for query in databases.received_queries[0]
|
|
)
|
|
|
|
|
|
def test_list_usuarios_fallback_respects_boolean_filter(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
databases = _DummyDatabases([
|
|
_sample_document("alpha", ativo=True),
|
|
_sample_document("bravo", ativo=False),
|
|
])
|
|
service = _SyncUsuariosService(databases, "test-db", "test-usuarios")
|
|
|
|
def _broken_search(attribute: str, value: str) -> Dict[str, Any]: # pragma: no cover - helper for fallback
|
|
raise AttributeError("search not supported")
|
|
|
|
monkeypatch.setattr("src.modules.usuarios.service.Query.search", _broken_search)
|
|
|
|
result = asyncio.run(
|
|
service.list(
|
|
page=1,
|
|
limit=None,
|
|
q="a",
|
|
sort=None,
|
|
order=None,
|
|
empresas_dados_id=None,
|
|
ativo=True,
|
|
)
|
|
)
|
|
|
|
assert result["total"] == 1
|
|
assert result["items"][0]["id"] == "alpha"
|