71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncGenerator
|
|
from contextlib import asynccontextmanager
|
|
|
|
import pytest
|
|
|
|
from src.services import stats_service as stats_service_module
|
|
from src.services.stats_service import StatsService
|
|
|
|
|
|
class _NoopRedis:
|
|
async def get(self, key: str) -> None:
|
|
return None
|
|
|
|
async def set(self, key: str, value: str, ex: int) -> None:
|
|
return None
|
|
|
|
|
|
class _EmptyResult:
|
|
def all(self) -> list:
|
|
return []
|
|
|
|
|
|
class _FakeDb:
|
|
async def execute(self, statement) -> _EmptyResult:
|
|
return _EmptyResult()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_service_uses_default_project_for_db_context(monkeypatch) -> None:
|
|
seen_project_ids: list[str | None] = []
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str | None = None) -> AsyncGenerator[_FakeDb, None]:
|
|
seen_project_ids.append(project_id)
|
|
yield _FakeDb()
|
|
|
|
monkeypatch.setattr(stats_service_module, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(stats_service_module, "get_redis", lambda: _NoopRedis())
|
|
|
|
result = await StatsService().get_resolution_stats(days=7)
|
|
|
|
assert result["sample_size"] == 0
|
|
assert seen_project_ids == ["awoooi"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_service_allows_explicit_project_for_db_context(monkeypatch) -> None:
|
|
seen_project_ids: list[str | None] = []
|
|
seen_cache_keys: list[str] = []
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str | None = None) -> AsyncGenerator[_FakeDb, None]:
|
|
seen_project_ids.append(project_id)
|
|
yield _FakeDb()
|
|
|
|
class TrackingRedis(_NoopRedis):
|
|
async def get(self, key: str) -> None:
|
|
seen_cache_keys.append(key)
|
|
return None
|
|
|
|
monkeypatch.setattr(stats_service_module, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(stats_service_module, "get_redis", lambda: TrackingRedis())
|
|
|
|
result = await StatsService().get_ai_performance(days=7, project_id="vibework")
|
|
|
|
assert result["total_proposals"] == 0
|
|
assert seen_project_ids == ["vibework"]
|
|
assert seen_cache_keys == ["stats:vibework:ai_performance:7"]
|