Merge branch 'dev' into feat/openapi-contrat
Complète le contrat OpenAPI de GET /sites et GET /sites/{site_id} (merges depuis dev via #78) :
tag sites décrit, REPONSES_LECTEUR (401 + 403 mot de passe provisoire) posée au niveau du
routeur, 404 et 422 documentés sur la route detail. openapi.json régénéré.
This commit is contained in:
@@ -0,0 +1,141 @@
|
||||
from collections.abc import Callable, Iterator
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from httpx import AsyncClient
|
||||
|
||||
from app.api.deps import get_current_principal, get_site_service
|
||||
from app.core.principal import Principal
|
||||
from app.core.roles import AccountKind, Role
|
||||
from app.models.energy import Site
|
||||
from app.services.site import SiteNotFoundError
|
||||
|
||||
|
||||
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 site(site_id: str = "site-1") -> Site:
|
||||
return Site(
|
||||
site_id=site_id,
|
||||
site_name="Site de test",
|
||||
site_type="industriel",
|
||||
location="Toulouse",
|
||||
capacity_kw=42.0,
|
||||
status="actif",
|
||||
)
|
||||
|
||||
|
||||
class FauxService:
|
||||
def __init__(self, erreur: Exception | None = None) -> None:
|
||||
self._erreur = erreur
|
||||
self.site = site()
|
||||
|
||||
async def list_all(self) -> list[Site]:
|
||||
return [self.site]
|
||||
|
||||
async def get_by_id(self, site_id: str) -> Site:
|
||||
if self._erreur is not None:
|
||||
raise self._erreur
|
||||
return self.site
|
||||
|
||||
|
||||
@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_site_service] = lambda: service
|
||||
return service
|
||||
|
||||
yield installe
|
||||
app.dependency_overrides.pop(get_site_service, None)
|
||||
|
||||
|
||||
async def test_list_sites_returns_the_sites(
|
||||
servi: Callable[..., FauxService], client: AsyncClient
|
||||
) -> None:
|
||||
servi()
|
||||
|
||||
response = await client.get("/api/v1/sites")
|
||||
|
||||
assert response.status_code == 200
|
||||
corps = response.json()
|
||||
assert corps == [
|
||||
{
|
||||
"site_id": "site-1",
|
||||
"site_name": "Site de test",
|
||||
"site_type": "industriel",
|
||||
"location": "Toulouse",
|
||||
"capacity_kw": 42.0,
|
||||
"status": "actif",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
async def test_get_site_returns_the_matching_site(
|
||||
servi: Callable[..., FauxService], client: AsyncClient
|
||||
) -> None:
|
||||
servi()
|
||||
|
||||
response = await client.get("/api/v1/sites/site-1")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["site_id"] == "site-1"
|
||||
|
||||
|
||||
async def test_get_site_returns_404_for_an_unknown_site(
|
||||
servi: Callable[..., FauxService], client: AsyncClient
|
||||
) -> None:
|
||||
servi(SiteNotFoundError("site-inconnu"))
|
||||
|
||||
response = await client.get("/api/v1/sites/site-inconnu")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
async def test_list_sites_reaches_the_repository_through_the_session(
|
||||
lecteur_connecte: None, fake_session: Callable[..., None], client: AsyncClient
|
||||
) -> None:
|
||||
fake_session(result=[site("a"), site("b")])
|
||||
|
||||
response = await client.get("/api/v1/sites")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [s["site_id"] for s in response.json()] == ["a", "b"]
|
||||
|
||||
|
||||
async def test_get_site_reaches_the_repository_through_the_session(
|
||||
lecteur_connecte: None, fake_session: Callable[..., None], client: AsyncClient
|
||||
) -> None:
|
||||
fake_session(result=site("a"))
|
||||
|
||||
response = await client.get("/api/v1/sites/a")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["site_id"] == "a"
|
||||
|
||||
|
||||
async def test_get_site_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/sites/inconnu")
|
||||
|
||||
assert response.status_code == 404
|
||||
@@ -1,3 +1,4 @@
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from app.core.config import Settings
|
||||
@@ -12,6 +13,16 @@ SETTINGS_DE_TEST: dict[str, Any] = {
|
||||
}
|
||||
|
||||
|
||||
class FakeScalars:
|
||||
"""Resultat factice pour `.scalars()` : `.all()` renvoie les lignes fournies."""
|
||||
|
||||
def __init__(self, rows: Sequence[object]) -> None:
|
||||
self._rows = rows
|
||||
|
||||
def all(self) -> Sequence[object]:
|
||||
return self._rows
|
||||
|
||||
|
||||
class FakeSession:
|
||||
"""Session factice : renvoie `result`, ou leve `failure` si elle est fournie."""
|
||||
|
||||
@@ -25,6 +36,9 @@ class FakeSession:
|
||||
async def execute(self, *_: object, **__: object) -> object:
|
||||
return self._repondre()
|
||||
|
||||
async def scalars(self, *_: object, **__: object) -> FakeScalars:
|
||||
return FakeScalars(self._repondre() or [])
|
||||
|
||||
def _repondre(self) -> object:
|
||||
if self._failure is not None:
|
||||
raise self._failure
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.energy import Site
|
||||
from app.repositories.site import SiteRepository
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def identifiant() -> str:
|
||||
return f"site-{uuid.uuid4().hex[:12]}"
|
||||
|
||||
|
||||
async def creer(session: AsyncSession, **overrides: object) -> Site:
|
||||
site = Site(
|
||||
site_id=overrides.get("site_id", identifiant()),
|
||||
site_name=overrides.get("site_name", "Site de test"),
|
||||
site_type=overrides.get("site_type", "industriel"),
|
||||
location=overrides.get("location", "Toulouse"),
|
||||
capacity_kw=overrides.get("capacity_kw", 42.0),
|
||||
status=overrides.get("status", "actif"),
|
||||
)
|
||||
session.add(site)
|
||||
await session.flush()
|
||||
return site
|
||||
|
||||
|
||||
async def test_get_by_id_returns_the_matching_site(session: AsyncSession) -> None:
|
||||
depot = SiteRepository(session)
|
||||
cree = await creer(session)
|
||||
|
||||
trouve = await depot.get_by_id(cree.site_id)
|
||||
nom = trouve.site_name if trouve else None
|
||||
await session.rollback()
|
||||
|
||||
assert nom == "Site de test"
|
||||
|
||||
|
||||
async def test_get_by_id_returns_nothing_for_an_unknown_identifier(
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
trouve = await SiteRepository(session).get_by_id(identifiant())
|
||||
|
||||
assert trouve is None
|
||||
|
||||
|
||||
async def test_list_all_returns_the_sites_sorted_by_identifier(session: AsyncSession) -> None:
|
||||
depot = SiteRepository(session)
|
||||
premier, second = sorted([f"zz-{identifiant()}", f"aa-{identifiant()}"])
|
||||
await creer(session, site_id=second)
|
||||
await creer(session, site_id=premier)
|
||||
|
||||
sites = await depot.list_all()
|
||||
identifiants = [site.site_id for site in sites if site.site_id in (premier, second)]
|
||||
await session.rollback()
|
||||
|
||||
assert identifiants == [premier, second]
|
||||
@@ -0,0 +1,49 @@
|
||||
import pytest
|
||||
|
||||
from app.models.energy import Site
|
||||
from app.services.site import SiteNotFoundError, SiteService
|
||||
|
||||
|
||||
def site(site_id: str = "site-1") -> Site:
|
||||
return Site(
|
||||
site_id=site_id,
|
||||
site_name="Site de test",
|
||||
site_type="industriel",
|
||||
location="Toulouse",
|
||||
capacity_kw=42.0,
|
||||
status="actif",
|
||||
)
|
||||
|
||||
|
||||
class FakeRepository:
|
||||
def __init__(self, sites: list[Site]) -> None:
|
||||
self._sites = sites
|
||||
|
||||
async def list_all(self) -> list[Site]:
|
||||
return self._sites
|
||||
|
||||
async def get_by_id(self, site_id: str) -> Site | None:
|
||||
return next((s for s in self._sites if s.site_id == site_id), None)
|
||||
|
||||
|
||||
async def test_list_all_returns_the_repository_sites() -> None:
|
||||
service = SiteService(sites=FakeRepository([site("a"), site("b")]))
|
||||
|
||||
sites = await service.list_all()
|
||||
|
||||
assert [s.site_id for s in sites] == ["a", "b"]
|
||||
|
||||
|
||||
async def test_get_by_id_returns_the_matching_site() -> None:
|
||||
service = SiteService(sites=FakeRepository([site("a")]))
|
||||
|
||||
trouve = await service.get_by_id("a")
|
||||
|
||||
assert trouve.site_id == "a"
|
||||
|
||||
|
||||
async def test_get_by_id_raises_when_the_site_is_unknown() -> None:
|
||||
service = SiteService(sites=FakeRepository([]))
|
||||
|
||||
with pytest.raises(SiteNotFoundError):
|
||||
await service.get_by_id("inconnu")
|
||||
Reference in New Issue
Block a user