Fusionne dev dans feat/stats-summary
Resout les conflits de deps.py, router.py, repositories/site.py et test_site.py entre l'ajout de stats et le merge de sites/openapi-contrat sur dev. Generalise ROUTES_A_ROLE dans test_openapi.py et documente la checklist d'ajout d'une route metier, absentes de dev au moment du fork.
This commit is contained in:
Johan LEROY
@@ -0,0 +1,120 @@
|
||||
# Pourquoi : `openapi.json` est versionné, donc une route qui change son contrat public le montre
|
||||
# dans la diff d'une pull request. `test_the_committed_contract_matches_the_generated_one` est ce
|
||||
# qui empêche le fichier de dériver du code sans que personne ne le voie.
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app import cli
|
||||
|
||||
METHODES = {"get", "post", "patch", "put", "delete"}
|
||||
|
||||
# `/auth/logout` lit le cookie mais ne le réclame pas : sans session elle répond 204, et un 401
|
||||
# documenté y serait faux.
|
||||
SANS_REFUS = {("POST", "/api/v1/auth/logout")}
|
||||
|
||||
ORIGINE_VERIFIEE = {
|
||||
("POST", "/api/v1/auth/refresh"),
|
||||
("POST", "/api/v1/auth/logout"),
|
||||
("POST", "/api/v1/auth/logout-all"),
|
||||
("POST", "/api/v1/auth/password"),
|
||||
}
|
||||
|
||||
# Toute route derrière `require_role` (LecteurDep, OperateurDep, AdminDep) peut rendre 403 pour
|
||||
# `password_change_required`, pas seulement les routes `admin`.
|
||||
ROUTES_A_ROLE = {
|
||||
("GET", "/api/v1/users"),
|
||||
("POST", "/api/v1/users"),
|
||||
("PATCH", "/api/v1/users/{id}"),
|
||||
("POST", "/api/v1/users/{id}/password-reset"),
|
||||
("GET", "/api/v1/sites"),
|
||||
("GET", "/api/v1/sites/{site_id}"),
|
||||
("GET", "/api/v1/stats/summary"),
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def schema() -> dict[str, Any]:
|
||||
return cli.schema_du_contrat()
|
||||
|
||||
|
||||
def operations(schema: dict[str, Any]) -> list[tuple[str, str, dict[str, Any]]]:
|
||||
return [
|
||||
(methode.upper(), chemin, operation)
|
||||
for chemin, operations_du_chemin in schema["paths"].items()
|
||||
for methode, operation in operations_du_chemin.items()
|
||||
if methode in METHODES
|
||||
]
|
||||
|
||||
|
||||
def test_the_committed_contract_matches_the_generated_one(schema: dict[str, Any]) -> None:
|
||||
publie = json.loads(cli.CHEMIN_CONTRAT.read_text(encoding="utf-8"))
|
||||
|
||||
assert publie == schema, "lancer `make openapi` et versionner le fichier obtenu"
|
||||
|
||||
|
||||
def test_every_route_demanding_an_identity_says_how_it_refuses(schema: dict[str, Any]) -> None:
|
||||
muettes = [
|
||||
(methode, chemin)
|
||||
for methode, chemin, operation in operations(schema)
|
||||
if operation.get("security")
|
||||
and (methode, chemin) not in SANS_REFUS
|
||||
and "401" not in operation["responses"]
|
||||
]
|
||||
|
||||
assert muettes == []
|
||||
|
||||
|
||||
def test_every_role_guarded_route_documents_the_role_refusal(schema: dict[str, Any]) -> None:
|
||||
sans_403 = [
|
||||
(methode, chemin)
|
||||
for methode, chemin, operation in operations(schema)
|
||||
if (methode, chemin) in ROUTES_A_ROLE and "403" not in operation["responses"]
|
||||
]
|
||||
|
||||
assert sans_403 == []
|
||||
|
||||
|
||||
def test_every_origin_checked_route_documents_the_csrf_refusal(schema: dict[str, Any]) -> None:
|
||||
sans_403 = [
|
||||
(methode, chemin)
|
||||
for methode, chemin, operation in operations(schema)
|
||||
if (methode, chemin) in ORIGINE_VERIFIEE and "403" not in operation["responses"]
|
||||
]
|
||||
|
||||
assert sans_403 == []
|
||||
|
||||
|
||||
def test_the_validation_model_matches_what_the_handler_returns(schema: dict[str, Any]) -> None:
|
||||
modeles = {
|
||||
operation["responses"]["422"]["content"]["application/json"]["schema"]["$ref"]
|
||||
for _, _, operation in operations(schema)
|
||||
if "422" in operation["responses"]
|
||||
}
|
||||
|
||||
assert modeles == {"#/components/schemas/ValidationErrorResponse"}
|
||||
assert "HTTPValidationError" not in schema["components"]["schemas"]
|
||||
|
||||
|
||||
def test_the_rate_limit_documents_the_delay_header(schema: dict[str, Any]) -> None:
|
||||
trop_de_tentatives = schema["paths"]["/api/v1/auth/login"]["post"]["responses"]["429"]
|
||||
|
||||
assert "Retry-After" in trop_de_tentatives["headers"]
|
||||
|
||||
|
||||
def test_the_refresh_cookie_appears_in_the_security_schemes(schema: dict[str, Any]) -> None:
|
||||
schemes = schema["components"]["securitySchemes"]
|
||||
|
||||
assert schemes["Cookie de rafraîchissement"]["in"] == "cookie"
|
||||
assert schemes["Cookie de rafraîchissement"]["name"] == "ev_refresh"
|
||||
|
||||
|
||||
def test_each_tag_used_by_a_route_is_described(schema: dict[str, Any]) -> None:
|
||||
decrits = {tag["name"] for tag in schema["tags"]}
|
||||
|
||||
for methode, chemin, operation in operations(schema):
|
||||
poses = operation.get("tags", [])
|
||||
assert len(poses) == len(set(poses)), f"tag en double sur {methode} {chemin}"
|
||||
assert set(poses) <= decrits, f"tag non décrit sur {methode} {chemin}"
|
||||
@@ -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
|
||||
|
||||
@@ -10,19 +10,47 @@ pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def identifiant() -> str:
|
||||
return f"SITE-{uuid.uuid4().hex[:8]}"
|
||||
return f"site-{uuid.uuid4().hex[:12]}"
|
||||
|
||||
|
||||
async def test_list_all_returns_every_site_sorted_by_id(session: AsyncSession) -> None:
|
||||
premier, second = sorted([identifiant(), identifiant()])
|
||||
session.add_all(
|
||||
[
|
||||
Site(site_id=second, site_name="B", site_type="bureau", capacity_kw=100),
|
||||
Site(site_id=premier, site_name="A", site_type="bureau", capacity_kw=50),
|
||||
]
|
||||
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)]
|
||||
|
||||
@@ -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")
|
||||
@@ -1,3 +1,6 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app import cli
|
||||
@@ -55,3 +58,52 @@ def test_read_password_refuses_two_different_entries(monkeypatch: pytest.MonkeyP
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli.read_password(generate=False)
|
||||
|
||||
|
||||
def test_build_parser_reads_the_export_openapi_arguments() -> None:
|
||||
arguments = cli.build_parser().parse_args(
|
||||
["export-openapi", "--output", "ailleurs/contrat.json"]
|
||||
)
|
||||
|
||||
assert arguments.commande == "export-openapi"
|
||||
assert arguments.output == "ailleurs/contrat.json"
|
||||
|
||||
|
||||
def test_build_parser_defaults_the_export_to_the_versioned_contract() -> None:
|
||||
arguments = cli.build_parser().parse_args(["export-openapi"])
|
||||
|
||||
assert arguments.output == str(cli.CHEMIN_CONTRAT)
|
||||
|
||||
|
||||
def test_settings_of_the_contract_ignore_the_local_environment(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setenv("APP_API_PREFIX", "/api/v9")
|
||||
monkeypatch.setenv("APP_NAME", "API du poste de Johan")
|
||||
|
||||
settings = cli.settings_du_contrat()
|
||||
|
||||
assert settings.api_prefix == "/api/v1"
|
||||
assert settings.name == "EnerVision API"
|
||||
|
||||
|
||||
def test_export_openapi_writes_a_readable_schema_where_asked(tmp_path: Path) -> None:
|
||||
destination = tmp_path / "contrat.json"
|
||||
|
||||
cli.export_openapi(destination)
|
||||
|
||||
assert json.loads(destination.read_text(encoding="utf-8"))["openapi"].startswith("3.")
|
||||
|
||||
|
||||
# Piège : `main()` réclamait un mot de passe avant de lire la commande. Sans le branchement,
|
||||
# l'export resterait bloqué sur `getpass` et aucune CI ne pourrait le rejouer.
|
||||
def test_main_exports_the_contract_without_asking_for_a_password(
|
||||
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
destination = tmp_path / "contrat.json"
|
||||
|
||||
code = cli.main(["export-openapi", "--output", str(destination)])
|
||||
|
||||
assert code == 0
|
||||
assert destination.exists()
|
||||
assert str(destination) in capsys.readouterr().out
|
||||
|
||||
Reference in New Issue
Block a user