feat(backend): ajoute GET /recommendations et GET /recommendations/{recommendation_id}
This commit is contained in:
Dorian PESCE
@@ -0,0 +1,144 @@
|
||||
from collections.abc import Callable, Iterator
|
||||
from datetime import UTC, datetime
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from httpx import AsyncClient
|
||||
|
||||
from app.api.deps import get_current_principal, get_recommendation_service
|
||||
from app.core.principal import Principal
|
||||
from app.core.roles import AccountKind, Role
|
||||
from app.models.energy import Recommendation
|
||||
from app.services.recommendation import RecommendationNotFoundError
|
||||
|
||||
MOMENT = datetime(2024, 1, 1, tzinfo=UTC)
|
||||
|
||||
|
||||
def principal(role: Role = Role.LECTEUR) -> Principal:
|
||||
return Principal(
|
||||
id=uuid4(),
|
||||
email=f"{role.value}@enervision.fr",
|
||||
role=role,
|
||||
kind=AccountKind.HUMAIN,
|
||||
must_change_password=False,
|
||||
)
|
||||
|
||||
|
||||
def recommendation(recommendation_id: int = 1) -> Recommendation:
|
||||
return Recommendation(
|
||||
recommendation_id=recommendation_id,
|
||||
alert_id=1,
|
||||
action="Vérifier la consommation",
|
||||
explanation="Pic détecté",
|
||||
rule_reference="spike-v1",
|
||||
created_at=MOMENT,
|
||||
)
|
||||
|
||||
|
||||
class FauxService:
|
||||
def __init__(self, erreur: Exception | None = None) -> None:
|
||||
self._erreur = erreur
|
||||
self.recommendation = recommendation()
|
||||
|
||||
async def list_all(self) -> list[Recommendation]:
|
||||
return [self.recommendation]
|
||||
|
||||
async def get_by_id(self, recommendation_id: int) -> Recommendation:
|
||||
if self._erreur is not None:
|
||||
raise self._erreur
|
||||
return self.recommendation
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lecteur_connecte(app: FastAPI) -> Iterator[None]:
|
||||
app.dependency_overrides[get_current_principal] = lambda: principal()
|
||||
yield
|
||||
app.dependency_overrides.pop(get_current_principal, None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def servi(
|
||||
app: FastAPI, lecteur_connecte: None
|
||||
) -> Iterator[Callable[[Exception | None], FauxService]]:
|
||||
def installe(erreur: Exception | None = None) -> FauxService:
|
||||
service = FauxService(erreur)
|
||||
app.dependency_overrides[get_recommendation_service] = lambda: service
|
||||
return service
|
||||
|
||||
yield installe
|
||||
app.dependency_overrides.pop(get_recommendation_service, None)
|
||||
|
||||
|
||||
async def test_list_recommendations_returns_the_recommendations(
|
||||
servi: Callable[..., FauxService], client: AsyncClient
|
||||
) -> None:
|
||||
servi()
|
||||
|
||||
response = await client.get("/api/v1/recommendations")
|
||||
|
||||
assert response.status_code == 200
|
||||
corps = response.json()
|
||||
assert corps == [
|
||||
{
|
||||
"recommendation_id": 1,
|
||||
"alert_id": 1,
|
||||
"action": "Vérifier la consommation",
|
||||
"explanation": "Pic détecté",
|
||||
"rule_reference": "spike-v1",
|
||||
"created_at": "2024-01-01T00:00:00Z",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
async def test_get_recommendation_returns_the_matching_recommendation(
|
||||
servi: Callable[..., FauxService], client: AsyncClient
|
||||
) -> None:
|
||||
servi()
|
||||
|
||||
response = await client.get("/api/v1/recommendations/1")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["recommendation_id"] == 1
|
||||
|
||||
|
||||
async def test_get_recommendation_returns_404_for_an_unknown_recommendation(
|
||||
servi: Callable[..., FauxService], client: AsyncClient
|
||||
) -> None:
|
||||
servi(RecommendationNotFoundError(404))
|
||||
|
||||
response = await client.get("/api/v1/recommendations/404")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
async def test_list_recommendations_reaches_the_repository_through_the_session(
|
||||
lecteur_connecte: None, fake_session: Callable[..., None], client: AsyncClient
|
||||
) -> None:
|
||||
fake_session(result=[recommendation(1), recommendation(2)])
|
||||
|
||||
response = await client.get("/api/v1/recommendations")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [r["recommendation_id"] for r in response.json()] == [1, 2]
|
||||
|
||||
|
||||
async def test_get_recommendation_reaches_the_repository_through_the_session(
|
||||
lecteur_connecte: None, fake_session: Callable[..., None], client: AsyncClient
|
||||
) -> None:
|
||||
fake_session(result=recommendation(1))
|
||||
|
||||
response = await client.get("/api/v1/recommendations/1")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["recommendation_id"] == 1
|
||||
|
||||
|
||||
async def test_get_recommendation_returns_404_when_the_session_finds_nothing(
|
||||
lecteur_connecte: None, fake_session: Callable[..., None], client: AsyncClient
|
||||
) -> None:
|
||||
fake_session(result=None)
|
||||
|
||||
response = await client.get("/api/v1/recommendations/404")
|
||||
|
||||
assert response.status_code == 404
|
||||
@@ -0,0 +1,85 @@
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.energy import Alert, Recommendation, Site
|
||||
from app.repositories.recommendation import RecommendationRepository
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
MOMENT = datetime(2024, 1, 1, tzinfo=UTC)
|
||||
|
||||
|
||||
async def creer_site(session: AsyncSession) -> str:
|
||||
site_id = f"TEST-{uuid.uuid4()}"
|
||||
session.add(Site(site_id=site_id, site_name="Site de test", site_type="office"))
|
||||
await session.flush()
|
||||
return site_id
|
||||
|
||||
|
||||
async def creer_alerte(session: AsyncSession) -> int:
|
||||
site_id = await creer_site(session)
|
||||
alerte = Alert(
|
||||
source_alert_id=str(uuid.uuid4()),
|
||||
site_id=site_id,
|
||||
source="api_mock",
|
||||
timestamp=MOMENT,
|
||||
type="spike",
|
||||
severity="high",
|
||||
message="Test",
|
||||
raw_data={},
|
||||
)
|
||||
session.add(alerte)
|
||||
await session.flush()
|
||||
return alerte.alert_id
|
||||
|
||||
|
||||
async def creer(session: AsyncSession, **overrides: object) -> Recommendation:
|
||||
recommendation = Recommendation(
|
||||
alert_id=overrides.get("alert_id") or await creer_alerte(session),
|
||||
action=overrides.get("action", "Vérifier la consommation"),
|
||||
explanation=overrides.get("explanation", "Pic détecté"),
|
||||
rule_reference=overrides.get("rule_reference", f"spike-{uuid.uuid4().hex[:8]}"),
|
||||
)
|
||||
session.add(recommendation)
|
||||
await session.flush()
|
||||
return recommendation
|
||||
|
||||
|
||||
async def test_get_by_id_returns_the_matching_recommendation(session: AsyncSession) -> None:
|
||||
depot = RecommendationRepository(session)
|
||||
cree = await creer(session)
|
||||
|
||||
trouve = await depot.get_by_id(cree.recommendation_id)
|
||||
action = trouve.action if trouve else None
|
||||
await session.rollback()
|
||||
|
||||
assert action == "Vérifier la consommation"
|
||||
|
||||
|
||||
async def test_get_by_id_returns_nothing_for_an_unknown_identifier(
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
trouve = await RecommendationRepository(session).get_by_id(0)
|
||||
|
||||
assert trouve is None
|
||||
|
||||
|
||||
async def test_list_all_returns_the_recommendations_sorted_by_identifier(
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
depot = RecommendationRepository(session)
|
||||
premiere = await creer(session)
|
||||
seconde = await creer(session)
|
||||
|
||||
recommendations = await depot.list_all()
|
||||
identifiants = [
|
||||
r.recommendation_id
|
||||
for r in recommendations
|
||||
if r.recommendation_id in (premiere.recommendation_id, seconde.recommendation_id)
|
||||
]
|
||||
await session.rollback()
|
||||
|
||||
assert identifiants == sorted(identifiants)
|
||||
@@ -0,0 +1,55 @@
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models.energy import Recommendation
|
||||
from app.services.recommendation import RecommendationNotFoundError, RecommendationService
|
||||
|
||||
|
||||
def recommendation(recommendation_id: int = 1) -> Recommendation:
|
||||
return Recommendation(
|
||||
recommendation_id=recommendation_id,
|
||||
alert_id=1,
|
||||
action="Vérifier la consommation",
|
||||
explanation="Pic détecté",
|
||||
rule_reference="spike-v1",
|
||||
created_at=datetime(2024, 1, 1, tzinfo=UTC),
|
||||
)
|
||||
|
||||
|
||||
class FakeRepository:
|
||||
def __init__(self, recommendations: list[Recommendation]) -> None:
|
||||
self._recommendations = recommendations
|
||||
|
||||
async def list_all(self) -> list[Recommendation]:
|
||||
return self._recommendations
|
||||
|
||||
async def get_by_id(self, recommendation_id: int) -> Recommendation | None:
|
||||
return next(
|
||||
(r for r in self._recommendations if r.recommendation_id == recommendation_id), None
|
||||
)
|
||||
|
||||
|
||||
async def test_list_all_returns_the_repository_recommendations() -> None:
|
||||
service = RecommendationService(
|
||||
recommendations=FakeRepository([recommendation(1), recommendation(2)])
|
||||
)
|
||||
|
||||
recommendations = await service.list_all()
|
||||
|
||||
assert [r.recommendation_id for r in recommendations] == [1, 2]
|
||||
|
||||
|
||||
async def test_get_by_id_returns_the_matching_recommendation() -> None:
|
||||
service = RecommendationService(recommendations=FakeRepository([recommendation(1)]))
|
||||
|
||||
trouve = await service.get_by_id(1)
|
||||
|
||||
assert trouve.recommendation_id == 1
|
||||
|
||||
|
||||
async def test_get_by_id_raises_when_the_recommendation_is_unknown() -> None:
|
||||
service = RecommendationService(recommendations=FakeRepository([]))
|
||||
|
||||
with pytest.raises(RecommendationNotFoundError):
|
||||
await service.get_by_id(404)
|
||||
Reference in New Issue
Block a user