Initial plan

Definition des jalons
2026-09-15 14:50:42 +00:00 · 2026-09-14 11:30:01 +02:00
204 changed files with 8 additions and 21615 deletions
@@ -1,22 +0,0 @@
 root = true
 [*]
 charset = utf-8
 end_of_line = lf
 insert_final_newline = true
 trim_trailing_whitespace = true
 indent_style = space
 indent_size = 2
 [*.py]
 indent_size = 4
 max_line_length = 100
 [*.{tf,tfvars}]
 indent_size = 2
 [Makefile]
 indent_style = tab
 [*.md]
 trim_trailing_whitespace = false
@@ -1,19 +0,0 @@
 # Variables lues par docker-compose.yml à la racine.
 # Le backend lancé hors conteneur (`make dev`) lit apps/backend/.env, pas ce fichier.
 POSTGRES_USER=enervision
 POSTGRES_PASSWORD=change_me
 POSTGRES_DB=enervision
 # 5432 est souvent déjà pris par une autre base du poste.
 POSTGRES_PORT=5433
 # `basic` renvoie des statistiques d'usage à Timescale.
 TIMESCALEDB_TELEMETRY=off
 APP_ENV=local
 APP_DEBUG=false
 APP_LOG_LEVEL=INFO
 # L'API refuse de démarrer tant que cette valeur reste un exemple ou fait moins de
 # 32 caractères. Générer la vôtre : python -c "import secrets; print(secrets.token_urlsafe(48))"
 APP_SECRET_KEY=change_me
 APP_CORS_ORIGINS=http://localhost:4200
 BACKEND_PORT=8000
@@ -1,58 +0,0 @@
 name: Backend
 # Piège : la version de Python vient de apps/backend/.python-version, et elle doit rester
 # en 3.14. Le code utilise le PEP 758, qu'un interpréteur 3.13 refuse de compiler.
 on:
  push:
    paths:
      - "apps/backend/**"
      - ".github/workflows/backend.yml"
  pull_request:
    paths:
      - "apps/backend/**"
      - ".github/workflows/backend.yml"
 permissions:
  contents: read
 concurrency:
  group: backend-${{ github.ref }}
  cancel-in-progress: true
 jobs:
  verification:
    name: Lint, typage et tests
    runs-on: ubuntu-latest
    defaults:
      run:
        working-directory: apps/backend
    steps:
      - name: Récupère le dépôt
        uses: actions/checkout@v4
      - name: Installe uv
        uses: astral-sh/setup-uv@v5
        with:
          enable-cache: true
          cache-dependency-glob: apps/backend/uv.lock
      - name: Installe l'interpréteur déclaré par .python-version
        run: uv python install
      - name: Synchronise les dépendances sans dévier du verrou
        run: uv sync --all-groups --frozen
      - name: Vérifie le formatage
        run: uv run ruff format --check .
      - name: Analyse statique
        run: uv run ruff check --output-format=github .
      - name: Typage
        run: uv run mypy app
      # Le marqueur `integration` est exclu par défaut, donc aucune base n'est nécessaire ici.
      - name: Tests et couverture
        run: uv run pytest --cov-fail-under=85
@@ -1,65 +0,0 @@
 # Python
 __pycache__/
 *.py[cod]
 .venv/
 venv/
 .pytest_cache/
 .mypy_cache/
 .ruff_cache/
 .coverage
 coverage.xml
 htmlcov/
 test-results/
 dist/
 build/
 *.egg-info/
 # Node / Angular
 node_modules/
 .angular/
 apps/frontend/dist/
 apps/frontend/.angular/
 npm-debug.log*
 yarn-error.log*
 # Terraform
 .terraform/
 # .terraform.lock.hcl est versionne (pas ignore) pour figer les versions de provider entre contributeurs/CI
 *.tfstate
 *.tfstate.*
 *.tfplan
 crash.log
 override.tf
 override.tf.json
 *_override.tf
 *_override.tf.json
 *.tfvars
 !*.tfvars.example
 kubeconfig
 # Airflow
 etl/airflow/logs/
 airflow.db
 airflow-webserver.pid
 standalone_admin_password.txt
 # Environnement et secrets
 .env
 .env.*
 !.env.example
 *.pem
 *.key
 secrets/
 # Donnees locales
 data/
 *.sqlite3
 monitoring/grafana/data/
 monitoring/prometheus/data/
 # IDE et OS
 .idea/
 .vscode/
 *.swp
 .DS_Store
 Thumbs.db
@@ -1,59 +0,0 @@
 BACKEND := apps/backend
 .DEFAULT_GOAL := help
 .PHONY: help install dev lint format typecheck test test-cov test-integration check \
        docker-build db-up db-down db-reset db-logs db-psql migrate bootstrap-admin
 help: ## Liste les cibles disponibles
 	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-16s\033[0m %s\n", $$1, $$2}'
 install: ## Installe les dépendances du backend
 	cd $(BACKEND) && uv sync --all-groups
 dev: ## Lance l'API en rechargement à chaud
 	cd $(BACKEND) && uv run uvicorn app.main:create_app --factory --reload --host 0.0.0.0 --port 8000
 lint: ## Analyse statique du backend
 	cd $(BACKEND) && uv run ruff check .
 format: ## Formate et corrige le backend
 	cd $(BACKEND) && uv run ruff format . && uv run ruff check --fix .
 typecheck: ## Vérifie le typage du backend
 	cd $(BACKEND) && uv run mypy app
 test: ## Exécute les tests backend ne demandant pas de base
 	cd $(BACKEND) && uv run pytest --cov-fail-under=85
 test-cov: ## Rapports de couverture HTML et XML, plus les résultats au format JUnit
 	cd $(BACKEND) && uv run pytest --cov-fail-under=85 --cov-report=html \
 		--cov-report=xml --junitxml=test-results/junit.xml
 test-integration: ## Exécute les tests exigeant une base joignable
 	cd $(BACKEND) && uv run pytest -m integration
 check: lint typecheck test ## Chaîne de vérification complète
 docker-build: ## Construit l'image du backend
 	docker build -t enervision-backend:local $(BACKEND)
 db-up: ## Démarre la base PostgreSQL TimescaleDB
 	docker compose up -d db
 db-down: ## Arrête la base en conservant ses données
 	docker compose stop db
 db-reset: ## Détruit la base et rejoue db/init
 	docker compose down -v && docker compose up -d db
 db-logs: ## Suit les journaux de la base
 	docker compose logs -f db
 db-psql: ## Ouvre une session psql sur la base applicative
 	docker compose exec db psql -U $${POSTGRES_USER:-enervision} -d $${POSTGRES_DB:-enervision}
 migrate: ## Applique les migrations Alembic
 	cd $(BACKEND) && uv run alembic upgrade head
 bootstrap-admin: ## Crée le premier administrateur, mot de passe saisi au clavier
 	cd $(BACKEND) && uv run python -m app.cli create-admin --email $${EMAIL:?EMAIL=... requis}
@@ -1,101 +1,10 @@
 # EnerVision
-Monorepo de la plateforme EnerVision : collecte, stockage, analyse et restitution de
+## Jalons définis
-series temporelles energetiques, deployee sur une machine on-premise.
+J1 - Valider la préparation de l'environnement et du repo
 J2 - Valider le périmètre retenu et les choix technologiques
 J3 - Valider l'architecture et la gestion de la sécurité
 J4 - Valider la robustesse et assurer les livrables
-## Jalons
+## Outil de collaboration utilisé
-
+GitHub
 | Jalon | Intitulé |
 |-------|----------------------------------------------------------|
 | J1    | Valider la préparation de l'environnement et du repo     |
 | J2    | Valider le périmètre retenu et les choix technologiques  |
 | J3    | Valider l'architecture et la gestion de la sécurité      |
 | J4    | Valider la robustesse et assurer les livrables           |
 Ce que la documentation apporte à chacun : [docs/architecture/00-vue-ensemble.md](docs/architecture/00-vue-ensemble.md).
 ## Stack cible
 | Domaine    | Technologie                         | Emplacement         | Etat          |
 |------------|-------------------------------------|---------------------|---------------|
 | Backend    | FastAPI, Python 3.14                | `apps/backend`      | Initialise    |
 | Frontend   | Angular 22, Node 24 LTS             | `apps/frontend`     | Tableau de bord |
 | Base       | PostgreSQL 17 + TimescaleDB         | `db`                | Initialise    |
 | ETL        | Apache Airflow                      | `etl/airflow`       | A initialiser |
 | Infra      | Terraform (k3s single-node)         | `infra/terraform`   | Initialise    |
 | CI/CD      | GitHub Actions                      | `.github/workflows` | Backend en place |
 | Monitoring | Prometheus, Grafana, Alertmanager   | `monitoring`        | A initialiser |
 Le backend, la base et l'infrastructure (Terraform/k3s) sont initialises a ce stade. Le frontend
 sert un tableau de bord sur `/dashboard`, dont les données proviennent de fixtures : les endpoints
 correspondants restent à écrire côté API. Les autres dossiers portent l'arborescence et un README
 de cadrage, leur contenu fait l'objet d'un ticket dedie.
 L'etat detaille de chaque brique et les vues d'architecture sont dans
 [docs/architecture](docs/architecture/README.md).
 ## Arborescence
 ```
 .
 ├── apps/
 │   ├── backend/        API FastAPI
 │   └── frontend/       Application Angular
 ├── db/
 │   ├── init/           Bootstrap PostgreSQL + TimescaleDB
 │   ├── migrations/     Migrations SQL versionnees
 │   └── seeds/          Jeux de donnees de reference
 ├── etl/airflow/
 │   ├── dags/           DAGs d'ingestion et d'agregation
 │   ├── plugins/        Operateurs et hooks maison
 │   ├── include/        Requetes SQL et ressources des DAGs
 │   └── tests/          Tests d'integrite des DAGs
 ├── infra/terraform/
 │   ├── modules/        Modules reutilisables
 │   └── environments/   Racines Terraform, une par environnement
 ├── monitoring/
 │   ├── prometheus/     Collecte et regles d'alerte
 │   ├── grafana/        Provisioning et dashboards
 │   └── alertmanager/   Routage des alertes
 ├── docs/               ADR et vues d'architecture
 └── scripts/            Outillage local
 ```
 ## Demarrage
 Prerequis : uv, Docker. Le poste doit disposer de Python 3.14, que `uv` installe seul.
 ```bash
 cp .env.example .env                               # variables de docker-compose
 cp apps/backend/.env.example apps/backend/.env     # variables du backend hors conteneur
 make db-up     # PostgreSQL + TimescaleDB, publie sur le port 5433
 make install   # dependances du backend
 make migrate   # applique les migrations Alembic
 make dev       # API sur http://localhost:8000, docs sur /docs
 make check     # lint + typage + tests
 ```
 `make help` liste les cibles disponibles.
 Deux fichiers d'environnement, deux usages : `.env` a la racine alimente `docker-compose.yml`,
 `apps/backend/.env` alimente le backend lance sur le poste. Le port 5433 est publie plutot que
 5432, souvent deja pris par une autre base.
 La boucle de developpement est `make db-up` puis `make dev` : seule la base tourne en
 conteneur. Le service `backend` du `docker-compose.yml` sert la stack complete et la recette,
 et n'embarque pas le source, donc toute modification y demande un
 `docker compose up -d --build backend`.
 Verifier que la base repond et que l'extension est chargee :
 ```bash
 curl -s localhost:8000/api/v1/health/ready
 ```
 ## Conventions
 - Branches : `feat/`, `fix/`, `chore/`, `docs/`, `test/` suivi d'un libelle court.
 - Commits : Conventional Commits, portee = dossier de premier niveau concerne.
 - Toute decision structurante donne lieu a un ADR dans `docs/adr`.
 - Toute PR qui change un composant met a jour sa vue dans `docs/architecture`, dans la meme PR.
@@ -1,16 +0,0 @@
 .venv/
 __pycache__/
 *.py[cod]
 .pytest_cache/
 .mypy_cache/
 .ruff_cache/
 .coverage
 coverage.xml
 htmlcov/
 .env
 .env.*
 !.env.example
 tests/
 Dockerfile
 .dockerignore
 README.md
@@ -1,10 +0,0 @@
 APP_ENV=local
 APP_DEBUG=false
 APP_LOG_LEVEL=INFO
 # L'API refuse de démarrer tant que cette valeur reste un exemple ou fait moins de
 # 32 caractères. Générer la vôtre : python -c "import secrets; print(secrets.token_urlsafe(48))"
 APP_SECRET_KEY=change_me
 APP_CORS_ORIGINS=http://localhost:4200
 DATABASE_URL=postgresql+asyncpg://enervision:change_me@localhost:5433/enervision
@@ -1 +0,0 @@
 3.14
@@ -1,42 +0,0 @@
 FROM python:3.14-slim AS builder
 COPY --from=ghcr.io/astral-sh/uv:0.11.26 /uv /uvx /bin/
 ENV UV_COMPILE_BYTECODE=1 \
    UV_LINK_MODE=copy \
    UV_PYTHON_DOWNLOADS=never
 WORKDIR /app
 RUN --mount=type=cache,target=/root/.cache/uv \
    --mount=type=bind,source=uv.lock,target=uv.lock \
    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
    uv sync --locked --no-install-project --no-dev
 COPY . /app
 RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --locked --no-dev
 FROM python:3.14-slim AS runtime
 RUN groupadd --system --gid 1001 app \
    && useradd --system --uid 1001 --gid app --create-home app
 ENV PATH="/app/.venv/bin:${PATH}" \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1
 WORKDIR /app
 COPY --from=builder --chown=app:app /app /app
 USER app
 EXPOSE 8000
 HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/api/v1/health/live')"
 CMD ["uvicorn", "app.main:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000"]
@@ -1,148 +0,0 @@
 # Backend EnerVision
 API FastAPI exposant les series temporelles energetiques.
 | Element     | Choix                                      |
 |-------------|--------------------------------------------|
 | Python      | 3.14                                       |
 | Gestionnaire| uv (`uv.lock` fait foi)                    |
 | Framework   | FastAPI + Uvicorn                          |
 | Persistance | SQLAlchemy 2 async + asyncpg + Alembic     |
 | Lint/format | ruff                                       |
 | Typage      | mypy en mode strict                        |
 | Tests       | pytest + pytest-asyncio + httpx            |
 ## Installation
 ```bash
 cp .env.example .env
 uv sync --all-groups
 ```
 `APP_SECRET_KEY` et `DATABASE_URL` n'ont pas de valeur par defaut : l'application refuse
 de demarrer sans elles.
 `DATABASE_URL` pointe sur `localhost:5433`, le port publie par le service `db` du
 `docker-compose.yml` racine. Demarrer la base depuis la racine avec `make db-up`.
 ## Commandes
 Depuis la racine du monorepo, via le `Makefile` : `make install`, `make dev`, `make lint`,
 `make format`, `make typecheck`, `make test`, `make check`, `make docker-build`.
 Directement depuis ce dossier :
 ```bash
 uv run uvicorn app.main:create_app --factory --reload --port 8000
 uv run ruff check .          # lint
 uv run ruff format .         # format
 uv run mypy app              # typage strict
 uv run pytest                # tests + couverture
 uv run pytest -m integration # tests exigeant une base joignable
 ```
 Les conventions de tests, les gabarits et le detail des marqueurs sont dans
 [`TESTING.md`](TESTING.md).
 `pytest` ecarte par defaut les tests marques `integration`, pour que `make check` reste
 jouable sans Docker. Ces tests visent la base `enervision_test`, creee par
 `db/init/110-test-database.sql` au premier demarrage du conteneur.
 L'application est exposee par une factory (`create_app`) et non par un objet module :
 aucune configuration n'est lue a l'import, ce qui rend les tests et les migrations
 independants de l'environnement.
 ## Structure
 ```
 app/
 ├── api/
 │   ├── deps.py          Dépendances partagées : session, settings, principal, gardes de rôle
 │   ├── errors.py        Gestionnaires 422 et 500
 │   ├── middleware.py    En-têtes de sécurité
 │   ├── security.py      Garde du point /metrics
 │   └── v1/
 │       ├── router.py    Agrégation des routes de la version 1
 │       └── endpoints/   Un module par ressource exposée
 ├── core/
 │   ├── config.py        Settings Pydantic, source unique de configuration
 │   ├── cookies.py       Attributs du cookie de rafraîchissement
 │   ├── hashing.py       Argon2id, poussé dans un fil sous limiteur
 │   ├── logging.py       Journalisation console en local, JSON en production
 │   ├── principal.py     L'identité que voit le code métier
 │   ├── roles.py         Rôles ordonnés
 │   └── security.py      Encodage et décodage des jetons d'accès
 ├── db/
 │   ├── base.py          Base declarative SQLAlchemy
 │   └── session.py       Engine et sessions asynchrones
 ├── models/              Modeles SQLAlchemy
 ├── schemas/             Modeles Pydantic d'entree et de sortie
 ├── repositories/        Acces aux donnees, une classe par agregat
 ├── services/            Regles metier, orchestrent les repositories
 ├── cli.py               Commandes hors HTTP, dont l'amorcage du premier admin
 └── main.py              Factory applicative
 tests/                   Miroir de app/
 alembic/                 Migrations du schema applicatif
 ```
 Le sens de dependance est unique : `endpoints` vers `services` vers `repositories` vers
 `models`. Un endpoint ne touche jamais une session directement.
 ## Routes
 | Route | Rôle | Accès |
 |---|---|---|
 | `/api/v1/health/live` | Sonde de vivacité, aucune dépendance externe | public |
 | `/api/v1/health/ready` | Sonde de disponibilité, vérifie la base et TimescaleDB | public |
 | `/api/v1/auth/login` | Ouvre une session | public |
 | `/api/v1/auth/refresh` | Fait tourner la session | cookie |
 | `/api/v1/auth/logout` | Ferme la session courante | cookie, idempotente |
 | `/api/v1/auth/logout-all` | Ferme toutes les sessions du compte | jeton |
 | `/api/v1/auth/password` | Change son propre mot de passe | jeton |
 | `/api/v1/auth/me` | Décrit le compte connecté | jeton |
 | `/api/v1/users` | Liste et crée des comptes | `admin` |
 | `/api/v1/users/{id}` | Change le rôle ou l'activation | `admin` |
 | `/api/v1/users/{id}/password-reset` | Réinitialise et ferme les sessions | `admin` |
 | `/metrics` | Métriques au format Prometheus | jeton si `APP_METRICS_TOKEN` |
 | `/docs`, `/openapi.json` | Documentation, fermée en `staging` et `prod` | public sinon |
 Le contrat détaillé pour le frontend est dans
 [`docs/architecture/31-contrat-authentification.md`](../../docs/architecture/31-contrat-authentification.md).
 ## Premier administrateur
 Aucun compte n'existe après les migrations. Il s'en crée un en ligne de commande :
 ```bash
 make bootstrap-admin EMAIL=prenom.nom@enervision.fr   # mot de passe saisi au clavier
 # ou, depuis apps/backend :
 uv run python -m app.cli create-admin --email prenom.nom@enervision.fr --generate
 ```
 Le compte est créé avec `must_change_password`, donc la première connexion ne donne accès qu'à
 `/auth/me` et `/auth/password` jusqu'au changement. Le mot de passe ne transite jamais par
 `argv`, visible de tout `ps`, et aucune révision Alembic n'insère de compte : son empreinte
 resterait dans Git pour toujours.
 ## Migrations
 ```bash
 uv run alembic revision --autogenerate -m "libelle"
 uv run alembic upgrade head
 ```
 L'URL de connexion vient de `DATABASE_URL`, pas de `alembic.ini`.
 La premiere revision ne cree aucune table : elle refuse de s'appliquer si l'extension
 TimescaleDB manque, ce qui arrive quand `db/init` n'a pas ete joue. Le DDL propre a
 TimescaleDB qui ne depend pas du schema applicatif vit dans `db/`, pas ici.
 ## Image Docker
 Build multi-stage, dependances resolues par uv depuis `uv.lock`, execution sous un
 utilisateur non root, sonde de sante integree.
 ```bash
 docker build -t enervision-backend:local .
 docker run --rm -p 8000:8000 --env-file .env enervision-backend:local
 ```
@@ -1,175 +0,0 @@
 # Conventions de tests unitaires : Backend
 ## Outil
 pytest, avec pytest-asyncio en mode `auto` : un `async def test_*` est collecte sans
 decorateur. Les appels HTTP passent par httpx sur `ASGITransport`, qui parle a
 l'application en memoire, sans serveur ni port ouvert.
 ## Ou ecrire les tests
 `tests/` est le miroir de `app/` : un test de `app/services/consumption.py` va dans
 `tests/services/test_consumption.py`. Les paquets `core`, `db`, `services` et
 `repositories` existent deja, vides, pour cette raison.
 ## Nommage
 - Fonctions en anglais : `test_<sujet>_<comportement>_when_<condition>`.
 - `ids=` de `parametrize` en francais : `ids=["erreur_sqlalchemy", "erreur_reseau"]`.
 - Pas de docstring : le nom porte l'intention.
 ## Structure attendue (Arrange / Act / Assert)
 Une ligne vide separe les trois temps, sans commentaire pour les annoncer.
 ```python
 async def test_readiness_returns_503_when_the_extension_is_missing(
    fake_session: Callable[..., None], client: AsyncClient
 ) -> None:
    fake_session(result=None)
    response = await client.get("/api/v1/health/ready")
    assert response.status_code == 503
    assert response.json()["detail"] == "Extension TimescaleDB absente"
 ```
 ## Ce qui doit etre teste en priorite
 Le sens de dependance du backend est `endpoints -> services -> repositories -> models`.
 | Couche | Ce qu'on teste |
 |---|---|
 | `services/` | La logique metier, cas nominal et cas d'erreur. C'est la priorite. |
 | `repositories/` | Chaque branche de decision, sous le marqueur `integration`. |
 | `endpoints/` | Le code de statut et la forme de la reponse, pas la logique metier. |
 | `schemas/` | Rien, sauf si le schema porte une validation ecrite a la main. |
 ## Doubles
 On remplace une dependance FastAPI par `app.dependency_overrides`, jamais par
 `unittest.mock`. `tests/factories.py` fournit le necessaire.
 - `fake_session(result=...)` : la session repond `result`.
 - `fake_session(failure=...)` : la session leve l'exception.
 - `make_settings(**overrides)` : fabrique une `Settings`, dont les valeurs priment sur
  l'environnement et sur `.env`. C'est le moyen de tester `create_app` en `prod`.
 ## Gabarit : un endpoint
 ```python
 from collections.abc import Callable
 from httpx import AsyncClient
 async def test_endpoint_returns_the_expected_payload(
    fake_session: Callable[..., None], client: AsyncClient
 ) -> None:
    fake_session(result=42)
    response = await client.get("/api/v1/...")
    assert response.status_code == 200
    assert response.json() == {"valeur": 42}
 ```
 ## Gabarit : un service avec repository factice
 Un service ne connait que son repository : on lui en passe un faux, sans base ni session.
 ```python
 from app.services.consumption import ConsumptionService
 class FakeRepository:
    async def total_for(self, site_id: int) -> float:
        return 12.5
 async def test_service_converts_the_total_to_kilowatt_hours() -> None:
    service = ConsumptionService(FakeRepository())
    total = await service.total_kwh(site_id=1)
    assert total == 12.5
 ```
 ## Gabarit : un repository sur la vraie base
 Un repository parle du SQL : le tester sur un double ne prouve rien. Il porte donc le
 marqueur `integration`, ecarte par defaut.
 ```python
 import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.site import Site
 from app.repositories.site import SiteRepository
@pytest.mark.integration
 async def test_repository_reads_back_what_it_wrote(session: AsyncSession) -> None:
    repository = SiteRepository(session)
    await repository.add(Site(name="Toulouse"))
    assert await repository.by_name("Toulouse") is not None
 ```
 ## Marqueurs
 `integration` designe tout test exigeant une base joignable. `pytest` les ecarte par
 defaut, ce qui garde `make check` jouable sans Docker. Tout autre marqueur doit etre
 declare dans `pyproject.toml` : `--strict-markers` refuse les marqueurs inconnus.
 ## Couverture
 Les branches sont mesurees, pas seulement les lignes. Le seuil de 85 % ne s'applique
 qu'aux cibles qui jouent toute la suite, `make test` et `make test-cov` : un fichier
 joue seul affiche sa couverture sans jamais echouer dessus. Le detail se lit dans
 `htmlcov/index.html` apres `make test-cov`.
 ## Lancer les tests
 ```bash
 make test                          # suite unitaire, sans base
 make test-cov                      # idem, plus les rapports HTML, XML et JUnit
 make db-up && make test-integration # tests exigeant une base, demande Docker
 make check                         # lint + typage + suite unitaire
 uv run pytest tests/api/test_health.py           # un seul fichier
 uv run pytest -k readiness                       # par motif de nom
 ```
 ## Trois fichiers à connaître avant de toucher à l'authentification
 `tests/api/test_route_protection.py` interroge réellement chaque route sans identifiant et
 échoue si l'une d'elles répond autre chose qu'un 401 ou un 403. Il n'inspecte pas l'arbre de
 dépendances : celui-ci n'est accessible que par l'API privée de FastAPI, et surtout une route
 peut porter la bonne dépendance tout en répondant quand même. **Rendre une route publique impose
 donc de modifier la liste `ROUTES_PUBLIQUES` de ce fichier**, ce qui apparaît en clair dans la
 diff d'une pull request.
 `tests/services/test_auth.py` donne au faux hacheur un **compteur d'appels**. C'est ce qui rend
 possibles les deux assertions qui prouvent la conception, et qu'aucune autre forme de test
 n'atteint :
 - adresse inconnue → le compteur vaut 1, donc le haché leurre a bien été vérifié et il n'y a pas
  d'oracle temporel ;
 - limite de débit atteinte → le compteur vaut 0, donc la limite est évaluée avant Argon2.
 `tests/api/test_parcours_authentification.py` joue six parcours complets contre la vraie base,
 sous le marqueur `integration`, sans serveur ni port ouvert. C'est là que se démontrent
 l'atomicité de la rotation, la mort de la famille au rejeu d'un cookie déjà tourné, et la
 révocation immédiate d'un compte désactivé.
 ## Deux pièges d'écriture de test
 **Lire les attributs avant le `rollback`.** Un `session.rollback()` périme les attributs chargés,
 et les relire déclenche une entrée-sortie hors du contexte greenlet, donc un `MissingGreenlet`.
 On capture la valeur dans une variable locale avant d'annuler.
 **`audit_log` ne se nettoie pas.** La table est en ajout seul, garanti par déclencheur : un test
 ne peut pas effacer ce qu'il y écrit, et les lignes d'une exécution précédente sont encore là.
 Chaque test filtre donc sur son propre `target_id` plutôt que de supposer une table vide.
@@ -1,150 +0,0 @@
 # A generic, single database configuration.
 [alembic]
 # path to migration scripts.
 # this is typically a path given in POSIX (e.g. forward slashes)
 # format, relative to the token %(here)s which refers to the location of this
 # ini file
 script_location = %(here)s/alembic
 # template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
 # Uncomment the line below if you want the files to be prepended with date and time
 # see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
 # for all available tokens
 # file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
 # Or organize into date-based subdirectories (requires recursive_version_locations = true)
 # file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s
 # sys.path path, will be prepended to sys.path if present.
 # defaults to the current working directory.  for multiple paths, the path separator
 # is defined by "path_separator" below.
 prepend_sys_path = .
 # timezone to use when rendering the date within the migration file
 # as well as the filename.
 # If specified, requires the tzdata library which can be installed by adding
 # `alembic[tz]` to the pip requirements.
 # string value is passed to ZoneInfo()
 # leave blank for localtime
 # timezone =
 # max length of characters to apply to the "slug" field
 # truncate_slug_length = 40
 # set to 'true' to run the environment during
 # the 'revision' command, regardless of autogenerate
 # revision_environment = false
 # set to 'true' to allow .pyc and .pyo files without
 # a source .py file to be detected as revisions in the
 # versions/ directory
 # sourceless = false
 # version location specification; This defaults
 # to <script_location>/versions.  When using multiple version
 # directories, initial revisions must be specified with --version-path.
 # The path separator used here should be the separator specified by "path_separator"
 # below.
 # version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions
 # path_separator; This indicates what character is used to split lists of file
 # paths, including version_locations and prepend_sys_path within configparser
 # files such as alembic.ini.
 # The default rendered in new alembic.ini files is "os", which uses os.pathsep
 # to provide os-dependent path splitting.
 #
 # Note that in order to support legacy alembic.ini files, this default does NOT
 # take place if path_separator is not present in alembic.ini.  If this
 # option is omitted entirely, fallback logic is as follows:
 #
 # 1. Parsing of the version_locations option falls back to using the legacy
 #    "version_path_separator" key, which if absent then falls back to the legacy
 #    behavior of splitting on spaces and/or commas.
 # 2. Parsing of the prepend_sys_path option falls back to the legacy
 #    behavior of splitting on spaces, commas, or colons.
 #
 # Valid values for path_separator are:
 #
 # path_separator = :
 # path_separator = ;
 # path_separator = space
 # path_separator = newline
 #
 # Use os.pathsep. Default configuration used for new projects.
 path_separator = os
 # set to 'true' to search source files recursively
 # in each "version_locations" directory
 # new in Alembic version 1.10
 # recursive_version_locations = false
 # the output encoding used when revision files
 # are written from script.py.mako
 # output_encoding = utf-8
 # database URL.  This is consumed by the user-maintained env.py script only.
 # other means of configuring database URLs may be customized within the env.py
 # file.
 # L'URL est injectee par alembic/env.py depuis app.core.config.
 sqlalchemy.url =
 [post_write_hooks]
 # post_write_hooks defines scripts or Python functions that are run
 # on newly generated revision scripts.  See the documentation for further
 # detail and examples
 # format using "black" - use the console_scripts runner, against the "black" entrypoint
 # hooks = black
 # black.type = console_scripts
 # black.entrypoint = black
 # black.options = -l 79 REVISION_SCRIPT_FILENAME
 # lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
 # hooks = ruff
 # ruff.type = module
 # ruff.module = ruff
 # ruff.options = check --fix REVISION_SCRIPT_FILENAME
 # Alternatively, use the exec runner to execute a binary found on your PATH
 # hooks = ruff
 # ruff.type = exec
 # ruff.executable = ruff
 # ruff.options = check --fix REVISION_SCRIPT_FILENAME
 # Logging configuration.  This is also consumed by the user-maintained
 # env.py script only.
 [loggers]
 keys = root,sqlalchemy,alembic
 [handlers]
 keys = console
 [formatters]
 keys = generic
 [logger_root]
 level = WARNING
 handlers = console
 qualname =
 [logger_sqlalchemy]
 level = WARNING
 handlers =
 qualname = sqlalchemy.engine
 [logger_alembic]
 level = INFO
 handlers =
 qualname = alembic
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)
 level = NOTSET
 formatter = generic
 [formatter_generic]
 format = %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %H:%M:%S
@@ -1 +0,0 @@
 Generic single-database configuration with an async dbapi.
@@ -1,90 +0,0 @@
 import asyncio
 from logging.config import fileConfig
 from alembic import context
 from sqlalchemy import pool
 from sqlalchemy.engine import Connection
 from sqlalchemy.ext.asyncio import async_engine_from_config
 import app.models  # noqa: F401
 from app.core.config import get_settings
 from app.db.base import Base
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
 # Interpret the config file for Python logging.
 # This line sets up loggers basically.
 if config.config_file_name is not None:
    fileConfig(config.config_file_name)
 config.set_main_option("sqlalchemy.url", get_settings().database_url.replace("%", "%%"))
 target_metadata = Base.metadata
 # other values from the config, defined by the needs of env.py,
 # can be acquired:
 # my_important_option = config.get_main_option("my_important_option")
 # ... etc.
 def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.
    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.
    Calls to context.execute() here emit the given string to the
    script output.
    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()
 def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()
 async def run_async_migrations() -> None:
    """In this scenario we need to create an Engine
    and associate a connection with the context.
    """
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()
 def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    asyncio.run(run_async_migrations())
 if context.is_offline_mode():
    run_migrations_offline()
 else:
    run_migrations_online()
@@ -1,28 +0,0 @@
 """${message}
 Revision ID: ${up_revision}
 Revises: ${down_revision | comma,n}
 Create Date: ${create_date}
 """
 from typing import Sequence, Union
 from alembic import op
 import sqlalchemy as sa
 ${imports if imports else ""}
 # revision identifiers, used by Alembic.
 revision: str = ${repr(up_revision)}
 down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
 branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
 depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
 def upgrade() -> None:
    """Upgrade schema."""
    ${upgrades if upgrades else "pass"}
 def downgrade() -> None:
    """Downgrade schema."""
    ${downgrades if downgrades else "pass"}
@@ -1,119 +0,0 @@
 """tentatives de connexion et journal d audit
 Revision ID: 517053a3c044
 Revises: b1a7c3d9e240
 Create Date: 2026-09-15 14:31:07.966180
 Deux tables aux vocations opposees. `login_attempt` est le compteur de la limitation
 de debit : son volume est pilote par l'attaquant, donc elle se purge. `audit_log` est
 en ajout seul, garanti par deux declencheurs.
 Le declencheur TRUNCATE n'est pas redondant : TRUNCATE ne passe pas par les
 declencheurs de ligne. Et RAISE EXCEPTION plutot qu'un RETURN NULL, qui annulerait
 l'operation silencieusement.
 """
 from collections.abc import Sequence
 import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 revision: str = "517053a3c044"
 down_revision: str | Sequence[str] | None = "b1a7c3d9e240"
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 FONCTION_AJOUT_SEUL = """
 CREATE FUNCTION audit_log_append_only() RETURNS trigger AS $$
 BEGIN
    RAISE EXCEPTION 'audit_log est en ajout seul : % interdit', TG_OP;
 END
 $$ LANGUAGE plpgsql;
 """
 DECLENCHEUR_LIGNE = """
 CREATE TRIGGER audit_log_no_update_delete
    BEFORE UPDATE OR DELETE ON audit_log
    FOR EACH ROW EXECUTE FUNCTION audit_log_append_only();
 """
 DECLENCHEUR_TRUNCATE = """
 CREATE TRIGGER audit_log_no_truncate
    BEFORE TRUNCATE ON audit_log
    FOR EACH STATEMENT EXECUTE FUNCTION audit_log_append_only();
 """
 def upgrade() -> None:
    op.create_table(
        "login_attempt",
        sa.Column("id", sa.BigInteger(), sa.Identity(always=True), nullable=False),
        sa.Column(
            "occurred_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column("email_tried", sa.String(length=320), nullable=False),
        sa.Column("client_ip", postgresql.INET(), nullable=True),
        sa.Column("outcome", sa.Text(), nullable=False),
        sa.Column("user_id", sa.UUID(), nullable=True),
        sa.CheckConstraint(
            "outcome in ('success', 'bad_credentials', 'throttled', 'inactive')",
            name="ck_login_attempt_outcome",
        ),
        sa.PrimaryKeyConstraint("id", name="pk_login_attempt"),
    )
    op.create_index(
        "ix_login_attempt_email_date", "login_attempt", ["email_tried", "occurred_at"]
    )
    op.create_index("ix_login_attempt_ip_date", "login_attempt", ["client_ip", "occurred_at"])
    op.create_table(
        "audit_log",
        sa.Column("id", sa.BigInteger(), sa.Identity(always=True), nullable=False),
        sa.Column(
            "occurred_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column("actor_id", sa.UUID(), nullable=True),
        sa.Column("actor_email", sa.Text(), nullable=True),
        sa.Column("actor_role", sa.Text(), nullable=True),
        sa.Column("action", sa.Text(), nullable=False),
        sa.Column("target_type", sa.Text(), nullable=True),
        sa.Column("target_id", sa.Text(), nullable=True),
        sa.Column("outcome", sa.Text(), nullable=False),
        sa.Column("client_ip", postgresql.INET(), nullable=True),
        sa.Column("user_agent", sa.Text(), nullable=True),
        sa.Column(
            "detail",
            postgresql.JSONB(astext_type=sa.Text()),
            server_default=sa.text("jsonb_build_object()"),
            nullable=False,
        ),
        sa.CheckConstraint("outcome in ('success', 'failure')", name="ck_audit_log_outcome"),
        sa.PrimaryKeyConstraint("id", name="pk_audit_log"),
    )
    op.create_index("ix_audit_log_date", "audit_log", ["occurred_at"])
    op.create_index("ix_audit_log_action_date", "audit_log", ["action", "occurred_at"])
    op.execute(FONCTION_AJOUT_SEUL)
    op.execute(DECLENCHEUR_LIGNE)
    op.execute(DECLENCHEUR_TRUNCATE)
 def downgrade() -> None:
    op.execute("DROP TRIGGER IF EXISTS audit_log_no_truncate ON audit_log;")
    op.execute("DROP TRIGGER IF EXISTS audit_log_no_update_delete ON audit_log;")
    op.execute("DROP FUNCTION IF EXISTS audit_log_append_only();")
    op.drop_index("ix_audit_log_action_date", table_name="audit_log")
    op.drop_index("ix_audit_log_date", table_name="audit_log")
    op.drop_table("audit_log")
    op.drop_index("ix_login_attempt_ip_date", table_name="login_attempt")
    op.drop_index("ix_login_attempt_email_date", table_name="login_attempt")
    op.drop_table("login_attempt")
@@ -1,37 +0,0 @@
 """socle garde extension timescaledb
 Revision ID: 5353c0e4f094
 Revises:
 Create Date: 2026-09-14 14:17:17.556764
 Premiere revision du schema applicatif. Elle ne cree aucune table : elle etablit
 alembic_version et refuse de s'appliquer sur une base ou l'extension TimescaleDB
 manque, cas qui se produit quand db/init n'a pas ete joue.
 """
 from collections.abc import Sequence
 from alembic import op
 revision: str = "5353c0e4f094"
 down_revision: str | Sequence[str] | None = None
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 GARDE_EXTENSION = """
 DO $$
 BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb') THEN
        RAISE EXCEPTION 'extension timescaledb absente, voir db/init et db/README.md';
    END IF;
 END
 $$;
 """
 def upgrade() -> None:
    op.execute(GARDE_EXTENSION)
 def downgrade() -> None:
    pass
@@ -1,77 +0,0 @@
 """jetons de rafraichissement
 Revision ID: 821f71be74c0
 Revises: 517053a3c044
 Create Date: 2026-09-15 14:42:09.757949
 Le jeton lui-meme n'est jamais stocke : seule son empreinte SHA-256 l'est, dans
 `token_hash`. Un pg_dump qui fuiterait ne livrerait donc aucune session utilisable.
 L'index partiel `ix_refresh_token_vivants` sert la revocation en cascade et la
 recherche des sessions actives, qui ne regardent jamais les lignes deja tournees.
 """
 from collections.abc import Sequence
 import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 revision: str = "821f71be74c0"
 down_revision: str | Sequence[str] | None = "517053a3c044"
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 MOTIFS = "'logout', 'rotation', 'reuse_detected', 'password_change', 'admin'"
 JETONS_VIVANTS = "revoked_at is null and rotated_at is null"
 def upgrade() -> None:
    op.create_table(
        "refresh_token",
        sa.Column(
            "id", sa.UUID(), server_default=sa.text("gen_random_uuid()"), nullable=False
        ),
        sa.Column("family_id", sa.UUID(), nullable=False),
        sa.Column("user_id", sa.UUID(), nullable=False),
        sa.Column("token_hash", sa.LargeBinary(), nullable=False),
        sa.Column(
            "issued_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("rotated_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("revoked_reason", sa.Text(), nullable=True),
        sa.Column("replaced_by", sa.UUID(), nullable=True),
        sa.Column("client_ip", postgresql.INET(), nullable=True),
        sa.Column("user_agent", sa.Text(), nullable=True),
        sa.CheckConstraint(
            f"revoked_reason is null or revoked_reason in ({MOTIFS})",
            name="ck_refresh_token_revoked_reason",
        ),
        sa.ForeignKeyConstraint(
            ["user_id"], ["app_user.id"], name="fk_refresh_token_user", ondelete="CASCADE"
        ),
        sa.PrimaryKeyConstraint("id", name="pk_refresh_token"),
        sa.UniqueConstraint("token_hash", name="uq_refresh_token_hash"),
    )
    op.create_index("ix_refresh_token_family", "refresh_token", ["family_id"])
    op.create_index("ix_refresh_token_user", "refresh_token", ["user_id"])
    op.create_index(
        "ix_refresh_token_vivants",
        "refresh_token",
        ["user_id"],
        postgresql_where=JETONS_VIVANTS,
    )
 def downgrade() -> None:
    op.drop_index(
        "ix_refresh_token_vivants", table_name="refresh_token", postgresql_where=JETONS_VIVANTS
    )
    op.drop_index("ix_refresh_token_user", table_name="refresh_token")
    op.drop_index("ix_refresh_token_family", table_name="refresh_token")
    op.drop_table("refresh_token")
@@ -1,72 +0,0 @@
 """comptes applicatifs
 Revision ID: b1a7c3d9e240
 Revises: 5353c0e4f094
 Create Date: 2026-09-15 14:40:00.000000
 Cree `app_user`, la table des comptes humains et de service. Le nom evite `user`,
 mot reserve de PostgreSQL. `gen_random_uuid()` est au coeur de PG17, aucune
 extension n'est necessaire.
 """
 from collections.abc import Sequence
 import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 revision: str = "b1a7c3d9e240"
 down_revision: str | Sequence[str] | None = "5353c0e4f094"
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 def upgrade() -> None:
    op.create_table(
        "app_user",
        sa.Column(
            "id",
            postgresql.UUID(as_uuid=True),
            server_default=sa.text("gen_random_uuid()"),
            nullable=False,
        ),
        sa.Column("email", sa.String(length=320), nullable=False),
        sa.Column("password_hash", sa.Text(), nullable=False),
        sa.Column("role", sa.Text(), nullable=False),
        sa.Column("kind", sa.Text(), server_default=sa.text("'human'"), nullable=False),
        sa.Column("is_active", sa.Boolean(), server_default=sa.text("true"), nullable=False),
        sa.Column(
            "must_change_password", sa.Boolean(), server_default=sa.text("false"), nullable=False
        ),
        sa.Column(
            "credentials_changed_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column("last_login_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("full_name", sa.Text(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.CheckConstraint("email = lower(email)", name="ck_app_user_email_minuscule"),
        sa.CheckConstraint(
            "role in ('lecteur', 'operateur', 'admin')", name="ck_app_user_role"
        ),
        sa.CheckConstraint("kind in ('human', 'service')", name="ck_app_user_kind"),
        sa.PrimaryKeyConstraint("id", name="pk_app_user"),
        sa.UniqueConstraint("email", name="uq_app_user_email"),
    )
 def downgrade() -> None:
    op.drop_table("app_user")
@@ -1,218 +0,0 @@
 """Création des six tables Data et de l'hypertable reading.
 Revision ID: e6d2026091501
 Revises: 821f71be74c0
 """
 from alembic import op
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 revision = "e6d2026091501"
 down_revision = "821f71be74c0"
 branch_labels = None
 depends_on = None
 def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "dataset",
        sa.Column("dataset_id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("dataset_name", sa.Text(), nullable=False),
        sa.Column("archive_sha256", sa.String(length=64), nullable=False),
        sa.Column("storage_uri", sa.Text(), nullable=False),
        sa.Column("source_timezone", sa.Text(), nullable=True),
        sa.Column(
            "metadata", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False
        ),
        sa.CheckConstraint("dataset_id > 0", name="ck_dataset_positive_id"),
        sa.PrimaryKeyConstraint("dataset_id"),
        sa.UniqueConstraint("archive_sha256", name="uq_dataset_archive_sha256"),
    )
    op.create_table(
        "site",
        sa.Column("site_id", sa.Text(), nullable=False),
        sa.Column("site_name", sa.Text(), nullable=False),
        sa.Column("site_type", sa.Text(), nullable=False),
        sa.Column("location", sa.Text(), nullable=True),
        sa.Column("capacity_kw", sa.Double(), nullable=True),
        sa.Column("status", sa.Text(), nullable=True),
        sa.PrimaryKeyConstraint("site_id"),
    )
    op.create_table(
        "prediction",
        sa.Column("prediction_id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("site_id", sa.Text(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column("target_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("target_metric", sa.Text(), nullable=False),
        sa.Column("period_minutes", sa.Integer(), nullable=True),
        sa.Column("predicted_value", sa.Double(), nullable=True),
        sa.Column("model_reference", sa.Text(), nullable=False),
        sa.Column("status", sa.Text(), nullable=False),
        sa.Column("failure_reason", sa.Text(), nullable=True),
        sa.CheckConstraint(
            "(status = 'available' AND predicted_value IS NOT NULL AND failure_reason IS NULL) OR (status IN ('insufficient_data', 'error') AND predicted_value IS NULL AND failure_reason IS NOT NULL)",
            name="ck_prediction_status",
        ),
        sa.CheckConstraint(
            "target_metric <> 'consumption_kwh' OR period_minutes IS NOT NULL",
            name="ck_prediction_energy_period",
        ),
        sa.CheckConstraint(
            "target_metric IN ('consumption_kwh', 'consumption_kw')", name="ck_prediction_metric"
        ),
        sa.CheckConstraint(
            "period_minutes IS NULL OR period_minutes > 0", name="ck_prediction_period"
        ),
        sa.ForeignKeyConstraint(
            ["site_id"], ["site.site_id"], name="fk_prediction_site", ondelete="RESTRICT"
        ),
        sa.PrimaryKeyConstraint("prediction_id"),
        sa.UniqueConstraint("prediction_id", "site_id", name="uq_prediction_id_site"),
    )
    op.create_index(
        "ix_prediction_site_target", "prediction", ["site_id", "target_at"], unique=False
    )
    op.create_table(
        "reading",
        sa.Column("reading_id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("site_id", sa.Text(), nullable=False),
        sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False),
        sa.Column("source", sa.Text(), nullable=False),
        sa.Column("dataset_id", sa.BigInteger(), nullable=True),
        sa.Column("consumption_kw", sa.Double(), nullable=True),
        sa.Column("consumption_kwh", sa.Double(), nullable=True),
        sa.Column("consumption_euros", sa.Numeric(precision=14, scale=2), nullable=True),
        sa.Column("voltage_v", sa.Double(), nullable=True),
        sa.Column("current_a", sa.Double(), nullable=True),
        sa.Column("power_factor", sa.Double(), nullable=True),
        sa.Column("temperature_celsius", sa.Double(), nullable=True),
        sa.Column("humidity_percent", sa.Double(), nullable=True),
        sa.Column("solar_irradiance_wm2", sa.Double(), nullable=True),
        sa.Column("is_working_hours", sa.Boolean(), nullable=True),
        sa.Column("data_quality", sa.Text(), nullable=True),
        sa.Column("null_reasons", postgresql.ARRAY(sa.Text()), nullable=True),
        sa.Column(
            "imputed_values", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=True
        ),
        sa.Column("imputation_method", sa.Text(), nullable=True),
        sa.Column(
            "ingested_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column(
            "raw_data", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False
        ),
        sa.CheckConstraint(
            "(source = 'csv' AND dataset_id IS NOT NULL) OR (source IN ('api_current', 'api_history') AND dataset_id IS NULL)",
            name="ck_reading_dataset_source",
        ),
        sa.CheckConstraint(
            "data_quality IS NULL OR data_quality IN ('good', 'partial', 'degraded', 'critical')",
            name="ck_reading_quality",
        ),
        sa.CheckConstraint(
            "source IN ('csv', 'api_current', 'api_history')", name="ck_reading_source"
        ),
        sa.CheckConstraint(
            "(imputed_values IS NULL AND imputation_method IS NULL) OR (imputed_values IS NOT NULL AND imputation_method IS NOT NULL)",
            name="ck_reading_imputation",
        ),
        sa.ForeignKeyConstraint(
            ["dataset_id"], ["dataset.dataset_id"], name="fk_reading_dataset", ondelete="RESTRICT"
        ),
        sa.ForeignKeyConstraint(
            ["site_id"], ["site.site_id"], name="fk_reading_site", ondelete="RESTRICT"
        ),
        sa.PrimaryKeyConstraint("reading_id", "timestamp"),
    )
    op.create_index("ix_reading_dataset_id", "reading", ["dataset_id"], unique=False)
    op.create_index(
        "ix_reading_site_timestamp", "reading", ["site_id", "timestamp"], unique=False
    )
    op.create_index(
        "uq_reading_source",
        "reading",
        ["site_id", "timestamp", "source", sa.literal_column("coalesce(dataset_id, 0)")],
        unique=True,
    )
    op.execute(
        "SELECT create_hypertable('reading', by_range('timestamp'), create_default_indexes => FALSE)"
    )
    op.create_table(
        "alert",
        sa.Column("alert_id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("source_alert_id", sa.Text(), nullable=False),
        sa.Column("site_id", sa.Text(), nullable=False),
        sa.Column("source", sa.Text(), nullable=False),
        sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False),
        sa.Column("type", sa.Text(), nullable=False),
        sa.Column("severity", sa.Text(), nullable=False),
        sa.Column("message", sa.Text(), nullable=False),
        sa.Column("value", sa.Double(), nullable=True),
        sa.Column("threshold", sa.Double(), nullable=True),
        sa.Column("metric", sa.Text(), nullable=True),
        sa.Column("prediction_id", sa.BigInteger(), nullable=True),
        sa.Column(
            "raw_data", postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False
        ),
        sa.CheckConstraint(
            "severity IN ('low', 'medium', 'high', 'critical')", name="ck_alert_severity"
        ),
        sa.CheckConstraint("source IN ('api_mock', 'enervision')", name="ck_alert_source"),
        sa.CheckConstraint(
            "type IN ('spike', 'threshold', 'anomaly', 'outage', 'sensor')", name="ck_alert_type"
        ),
        sa.ForeignKeyConstraint(
            ["prediction_id", "site_id"],
            ["prediction.prediction_id", "prediction.site_id"],
            name="fk_alert_prediction_site",
            ondelete="RESTRICT",
        ),
        sa.ForeignKeyConstraint(
            ["site_id"], ["site.site_id"], name="fk_alert_site", ondelete="RESTRICT"
        ),
        sa.PrimaryKeyConstraint("alert_id"),
        sa.UniqueConstraint(
            "source", "site_id", "source_alert_id", name="uq_alert_source_reference"
        ),
    )
    op.create_index("ix_alert_site_timestamp", "alert", ["site_id", "timestamp"], unique=False)
    op.create_table(
        "recommendation",
        sa.Column("recommendation_id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("alert_id", sa.BigInteger(), nullable=False),
        sa.Column("action", sa.Text(), nullable=False),
        sa.Column("explanation", sa.Text(), nullable=False),
        sa.Column("rule_reference", sa.Text(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.ForeignKeyConstraint(
            ["alert_id"], ["alert.alert_id"], name="fk_recommendation_alert", ondelete="RESTRICT"
        ),
        sa.PrimaryKeyConstraint("recommendation_id"),
        sa.UniqueConstraint("alert_id", "rule_reference", name="uq_recommendation_alert_rule"),
    )
    # ### end Alembic commands ###
 def downgrade() -> None:
    op.drop_table("recommendation")
    op.drop_table("alert")
    op.drop_table("reading")
    op.drop_table("prediction")
    op.drop_table("site")
    op.drop_table("dataset")
@@ -1,197 +0,0 @@
 # Piège : `get_current_principal()` relit le compte en base à chaque requête au lieu de faire
 # confiance aux claims. C'est le renoncement assumé à la propriété « sans état » : sur un seul
 # service et une seule base, elle n'achetait rien, et la lecture par clé primaire coûte moins
 # d'un pour cent du budget d'une requête. Ce qu'elle achète, c'est la révocation immédiate.
 # Piège : le `Principal` est construit depuis la ligne, jamais depuis le claim `role`. Un claim
 # périmé ne peut donc pas provoquer d'élévation de privilège.
 from collections.abc import Callable
 from datetime import timedelta
 from functools import lru_cache
 from typing import Annotated
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.config import Settings, get_settings
 from app.core.hashing import Argon2Hasher, build_hasher
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role, has_at_least
 from app.core.security import TokenExpiredError, TokenInvalidError, TokenPolicy
 from app.core.security import decode_access_token as decode_token
 from app.db.session import get_session
 from app.repositories.audit_log import AuditLogRepository
 from app.repositories.login_attempt import LoginAttemptRepository
 from app.repositories.refresh_token import RefreshTokenRepository
 from app.repositories.user import UserRepository
 from app.services.auth import AuthService, LoginPolicy
 from app.services.user import UserService
 SessionDep = Annotated[AsyncSession, Depends(get_session)]
 SettingsDep = Annotated[Settings, Depends(get_settings)]
 CODE_CHANGEMENT_REQUIS = "password_change_required"
 _porteur = HTTPBearer(auto_error=False, scheme_name="Jeton d'accès")
 CredentialsDep = Annotated[HTTPAuthorizationCredentials | None, Depends(_porteur)]
 def _non_authentifie(description: str) -> HTTPException:
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Authentification requise",
        headers={"WWW-Authenticate": f'Bearer error="{description}"'},
    )
 def get_token_policy(settings: SettingsDep) -> TokenPolicy:
    return TokenPolicy(
        secret=settings.secret_key.get_secret_value(),
        issuer=settings.jwt_issuer,
        audience=settings.jwt_audience,
        access_ttl=timedelta(seconds=settings.access_token_ttl_seconds),
    )
 # Construire un `Argon2Hasher` calcule un haché leurre, donc 17 ms : il est mis en cache sur
 # les paramètres plutôt que reconstruit à chaque requête.
@lru_cache
 def _hasher_cache(
    time_cost: int, memory_cost_kib: int, parallelism: int, max_concurrency: int
 ) -> Argon2Hasher:
    return build_hasher(
        time_cost=time_cost,
        memory_cost_kib=memory_cost_kib,
        parallelism=parallelism,
        max_concurrency=max_concurrency,
    )
 def get_hasher(settings: SettingsDep) -> Argon2Hasher:
    return _hasher_cache(
        settings.argon2_time_cost,
        settings.argon2_memory_cost_kib,
        settings.argon2_parallelism,
        settings.argon2_max_concurrency,
    )
 def get_client_ip(request: Request, settings: SettingsDep) -> str | None:
    # Derrière un proxy, `request.client.host` vaut l'IP du proxy : le compteur par IP
    # deviendrait global, donc un déni de service auto-infligé. Le dernier élément est le seul
    # qu'un proxy de confiance ait écrit, les précédents sont fournis par le client.
    if settings.trust_proxy_headers:
        transmis = request.headers.get("x-forwarded-for")
        if transmis:
            return transmis.split(",")[-1].strip()
    return request.client.host if request.client else None
 def get_auth_service(
    session: SessionDep,
    settings: SettingsDep,
    hasher: Annotated[Argon2Hasher, Depends(get_hasher)],
    token_policy: Annotated[TokenPolicy, Depends(get_token_policy)],
 ) -> AuthService:
    return AuthService(
        users=UserRepository(session),
        attempts=LoginAttemptRepository(session),
        refresh_tokens=RefreshTokenRepository(session),
        audit=AuditLogRepository(session),
        hasher=hasher,
        transaction=session,
        token_policy=token_policy,
        login_policy=LoginPolicy(
            window_seconds=settings.login_window_seconds,
            max_failures_per_identifier_and_ip=(settings.login_max_failures_per_identifier_and_ip),
            max_failures_per_ip=settings.login_max_failures_per_ip,
            max_failures_per_identifier=settings.login_max_failures_per_identifier,
        ),
        refresh_ttl=timedelta(seconds=settings.refresh_token_ttl_seconds),
    )
 AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]
 def get_user_service(
    session: SessionDep,
    hasher: Annotated[Argon2Hasher, Depends(get_hasher)],
 ) -> UserService:
    return UserService(
        users=UserRepository(session),
        refresh_tokens=RefreshTokenRepository(session),
        audit=AuditLogRepository(session),
        hasher=hasher,
        transaction=session,
    )
 UserServiceDep = Annotated[UserService, Depends(get_user_service)]
 async def get_current_principal(
    credentials: CredentialsDep,
    session: SessionDep,
    token_policy: Annotated[TokenPolicy, Depends(get_token_policy)],
 ) -> Principal:
    if credentials is None:
        raise _non_authentifie("invalid_request")
    try:
        claims = decode_token(token_policy, credentials.credentials)
    except TokenExpiredError as erreur:
        raise _non_authentifie("expired") from erreur
    except TokenInvalidError as erreur:
        raise _non_authentifie("invalid_token") from erreur
    compte = await UserRepository(session).get_by_id(claims.subject)
    if compte is None or not compte.is_active:
        raise _non_authentifie("invalid_token")
    # Piège : `iat` est une date JWT, donc en secondes entières. Comparer sans tronquer le
    # marqueur rejetterait tout jeton émis dans la même seconde que le changement, c'est-à-dire
    # celui que `/auth/password` vient de rendre pour garder l'appareil courant connecté.
    if int(claims.issued_at.timestamp()) < int(compte.credentials_changed_at.timestamp()):
        raise _non_authentifie("token_stale")
    if claims.role != compte.role:
        raise _non_authentifie("token_stale")
    return Principal(
        id=compte.id,
        email=compte.email,
        role=Role(compte.role),
        kind=AccountKind(compte.kind),
        must_change_password=compte.must_change_password,
    )
 CurrentPrincipalDep = Annotated[Principal, Depends(get_current_principal)]
 def require_role(minimum: Role) -> Callable[[Principal], Principal]:
    def garde(principal: CurrentPrincipalDep) -> Principal:
        if principal.must_change_password:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail=CODE_CHANGEMENT_REQUIS
            )
        if not has_at_least(principal.role, minimum):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Droits insuffisants")
        return principal
    return garde
 LecteurDep = Annotated[Principal, Depends(require_role(Role.LECTEUR))]
 OperateurDep = Annotated[Principal, Depends(require_role(Role.OPERATEUR))]
 AdminDep = Annotated[Principal, Depends(require_role(Role.ADMIN))]
 def require_trusted_origin(request: Request, settings: SettingsDep) -> None:
    # Un navigateur envoie toujours `Origin` sur une requête non sûre. Son absence signale un
    # client hors navigateur, qui ne détient aucun cookie de victime : rien à protéger.
    origine = request.headers.get("origin")
    if origine is None:
        return
    if origine not in settings.allowed_origins:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Origine refusée")
@@ -1,47 +0,0 @@
 # Piège : la réponse 422 par défaut de FastAPI contient la clé `input`, c'est-à-dire la valeur
 # rejetée. Sur `/auth/login`, un corps malformé renverrait donc le mot de passe au client et le
 # déposerait dans les journaux d'erreur. `validation_error_handler()` ne laisse passer que le
 # champ fautif et le type d'erreur.
 import uuid
 from typing import Any
 from fastapi import FastAPI, Request, status
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from app.core.logging import get_logger
 logger = get_logger(__name__)
 async def validation_error_handler(_: Request, exception: RequestValidationError) -> JSONResponse:
    champs: list[dict[str, Any]] = [
        {
            "champ": ".".join(str(element) for element in erreur["loc"]),
            "type": erreur["type"],
        }
        for erreur in exception.errors()
    ]
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, content={"detail": champs}
    )
 async def unhandled_error_handler(request: Request, exception: Exception) -> JSONResponse:
    correlation = uuid.uuid4().hex
    logger.exception(
        "erreur non gérée correlation=%s methode=%s chemin=%s",
        correlation,
        request.method,
        request.url.path,
    )
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Erreur interne", "correlation": correlation},
    )
 def register_error_handlers(application: FastAPI) -> None:
    application.add_exception_handler(RequestValidationError, validation_error_handler)  # type: ignore[arg-type]
    application.add_exception_handler(Exception, unhandled_error_handler)
@@ -1,35 +0,0 @@
 # Pourquoi : `SecurityHeadersMiddleware` ne pose ni HSTS ni CSP, et c'est délibéré.
 # L'application ignore si TLS termine devant elle, donc elle ne peut pas décider d'un HSTS ;
 # et une CSP sur une API JSON ne protège presque rien, celle qui compte protège la page
 # Angular. Les deux appartiennent au terminateur TLS.
 # Contrainte : `/docs` charge Swagger depuis un CDN, une CSP stricte ici casserait la
 # documentation sans rien sécuriser.
 from collections.abc import Awaitable, Callable
 from typing import Final
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
 from starlette.responses import Response
 EN_TETES: Final[dict[str, str]] = {
    "X-Content-Type-Options": "nosniff",
    "X-Frame-Options": "DENY",
    "Referrer-Policy": "no-referrer",
 }
 PREFIXE_AUTHENTIFICATION: Final = "/auth"
 class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        response = await call_next(request)
        for nom, valeur in EN_TETES.items():
            response.headers.setdefault(nom, valeur)
        # Une réponse d'authentification ne doit jamais être conservée par un intermédiaire.
        if PREFIXE_AUTHENTIFICATION in request.url.path:
            response.headers["Cache-Control"] = "no-store"
        return response
@@ -1,23 +0,0 @@
 # Pourquoi : `/metrics` est protégé par un jeton statique et non par un rôle applicatif. Coupler
 # la supervision au modèle d'utilisateurs casserait la collecte à chaque panne
 # d'authentification, c'est-à-dire précisément quand on a besoin des métriques. Le vrai contrôle
 # reste le réseau : Prometheus scrute sur le réseau interne et `/metrics` ne sort pas.
 import secrets
 from fastapi import HTTPException, Request, status
 from app.api.deps import SettingsDep
 def require_metrics_token(request: Request, settings: SettingsDep) -> None:
    attendu = settings.metrics_token
    if attendu is None:
        return
    presente = request.headers.get("authorization", "")
    prefixe = "Bearer "
    if not presente.startswith(prefixe) or not secrets.compare_digest(
        presente[len(prefixe) :], attendu.get_secret_value()
    ):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Jeton requis")
@@ -1,202 +0,0 @@
 # Piège : le jeton de rafraîchissement ne quitte jamais le cookie httpOnly, et le jeton
 # d'accès ne va jamais dans un cookie. C'est ce qui réduit la surface CSRF aux trois routes de
 # ce module : partout ailleurs, le navigateur n'attache rien de lui-même.
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 from app.api.deps import (
    AuthServiceDep,
    CurrentPrincipalDep,
    SettingsDep,
    get_client_ip,
    require_trusted_origin,
 )
 from app.core.cookies import RefreshCookie, cookie_name
 from app.core.logging import get_logger
 from app.schemas.auth import (
    LoginRequest,
    PasswordChangeRequest,
    PrincipalResponse,
    TokenResponse,
 )
 from app.services.auth import (
    AuthenticatedSession,
    InvalidCredentialsError,
    RateLimitedError,
    SessionRejectedError,
 )
 router = APIRouter()
 logger = get_logger(__name__)
 DETAIL_IDENTIFIANTS = "Identifiants invalides"
 DETAIL_SESSION = "Session invalide"
 def repond(
    response: Response, settings: SettingsDep, session: AuthenticatedSession
 ) -> TokenResponse:
    response.headers["Cache-Control"] = "no-store"
    response.set_cookie(**RefreshCookie.build(settings, session.refresh_secret).as_kwargs())
    return TokenResponse(
        access_token=session.access_token,
        expires_in=session.expires_in,
        principal=PrincipalResponse.from_principal(session.principal),
    )
 # Piège : une `HTTPException` construit sa propre réponse, donc tout en-tête posé sur la
 # `Response` injectée est perdu. L'effacement du cookie doit voyager avec l'exception,
 # sans quoi un navigateur garderait un cookie mort après une détection de réutilisation.
 def entete_de_suppression(settings: SettingsDep) -> str:
    temoin = Response()
    temoin.delete_cookie(**RefreshCookie.expired(settings).as_deletion_kwargs())
    return temoin.headers["set-cookie"]
 def lit_le_cookie(request: Request, settings: SettingsDep) -> str:
    secret = request.cookies.get(cookie_name(settings))
    if not secret:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=DETAIL_SESSION)
    return secret
@router.post("/login", response_model=TokenResponse, summary="Ouvre une session")
 async def login(
    payload: LoginRequest,
    request: Request,
    response: Response,
    settings: SettingsDep,
    service: AuthServiceDep,
    client_ip: str | None = Depends(get_client_ip),
 ) -> TokenResponse:
    response.headers["Cache-Control"] = "no-store"
    agent = request.headers.get("user-agent")
    try:
        session = await service.authenticate(
            email=payload.email, password=payload.password, client_ip=client_ip, user_agent=agent
        )
    except RateLimitedError as erreur:
        logger.warning("auth.rate_limited email=%s ip=%s", payload.email, client_ip)
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Trop de tentatives, réessayez plus tard",
            headers={"Retry-After": str(erreur.retry_after)},
        ) from erreur
    except InvalidCredentialsError as erreur:
        logger.warning("auth.login.failure email=%s ip=%s", payload.email, client_ip)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail=DETAIL_IDENTIFIANTS
        ) from erreur
    logger.info("auth.login.success user_id=%s ip=%s", session.principal.id, client_ip)
    return repond(response, settings, session)
@router.post(
    "/refresh",
    response_model=TokenResponse,
    summary="Fait tourner la session",
    dependencies=[Depends(require_trusted_origin)],
 )
 async def refresh(
    request: Request,
    response: Response,
    settings: SettingsDep,
    service: AuthServiceDep,
    client_ip: str | None = Depends(get_client_ip),
 ) -> TokenResponse:
    response.headers["Cache-Control"] = "no-store"
    try:
        session = await service.refresh(
            secret=lit_le_cookie(request, settings),
            client_ip=client_ip,
            user_agent=request.headers.get("user-agent"),
        )
    except SessionRejectedError as erreur:
        logger.warning("auth.refresh.rejected ip=%s", client_ip)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=DETAIL_SESSION,
            headers={
                "Set-Cookie": entete_de_suppression(settings),
                "Cache-Control": "no-store",
            },
        ) from erreur
    return repond(response, settings, session)
@router.post(
    "/logout",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Ferme la session courante",
    dependencies=[Depends(require_trusted_origin)],
 )
 async def logout(
    request: Request, response: Response, settings: SettingsDep, service: AuthServiceDep
 ) -> None:
    response.headers["Cache-Control"] = "no-store"
    secret = request.cookies.get(cookie_name(settings))
    if secret:
        await service.logout(secret=secret)
    response.delete_cookie(**RefreshCookie.expired(settings).as_deletion_kwargs())
@router.post(
    "/logout-all",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Ferme toutes les sessions du compte",
    dependencies=[Depends(require_trusted_origin)],
 )
 async def logout_all(
    principal: CurrentPrincipalDep,
    response: Response,
    settings: SettingsDep,
    service: AuthServiceDep,
 ) -> None:
    response.headers["Cache-Control"] = "no-store"
    revoquees = await service.logout_all(principal)
    logger.info("auth.logout_all user_id=%s sessions=%s", principal.id, revoquees)
    response.delete_cookie(**RefreshCookie.expired(settings).as_deletion_kwargs())
@router.get("/me", response_model=PrincipalResponse, summary="Décrit le compte connecté")
 async def me(principal: CurrentPrincipalDep) -> PrincipalResponse:
    return PrincipalResponse.from_principal(principal)
@router.post(
    "/password",
    response_model=TokenResponse,
    summary="Change son propre mot de passe",
    dependencies=[Depends(require_trusted_origin)],
 )
 async def change_password(
    payload: PasswordChangeRequest,
    principal: CurrentPrincipalDep,
    request: Request,
    response: Response,
    settings: SettingsDep,
    service: AuthServiceDep,
    client_ip: str | None = Depends(get_client_ip),
 ) -> TokenResponse:
    response.headers["Cache-Control"] = "no-store"
    try:
        session = await service.change_password(
            principal=principal,
            current_password=payload.current_password,
            new_password=payload.new_password,
            client_ip=client_ip,
            user_agent=request.headers.get("user-agent"),
        )
    except InvalidCredentialsError as erreur:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail=DETAIL_IDENTIFIANTS
        ) from erreur
    logger.info("auth.password_changed user_id=%s", principal.id)
    return repond(response, settings, session)
@@ -1,44 +0,0 @@
 from fastapi import APIRouter, HTTPException, status
 from sqlalchemy import text
 from sqlalchemy.exc import SQLAlchemyError
 from app.api.deps import SessionDep, SettingsDep
 from app.core.logging import get_logger
 from app.schemas.health import LivenessStatus, ReadinessStatus
 logger = get_logger(__name__)
 router = APIRouter(tags=["health"])
 TIMESCALEDB_VERSION = text("SELECT extversion FROM pg_extension WHERE extname = 'timescaledb'")
@router.get("/live", summary="Sonde de vivacite")
 async def liveness(settings: SettingsDep) -> LivenessStatus:
    return LivenessStatus(
        status="ok",
        service=settings.name,
        version=settings.version,
        environment=settings.env,
    )
@router.get("/ready", summary="Sonde de disponibilite")
 async def readiness(session: SessionDep) -> ReadinessStatus:
    try:
        version: str | None = await session.scalar(TIMESCALEDB_VERSION)
    except SQLAlchemyError, OSError:
        logger.exception("Base de données injoignable")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Base de données injoignable",
        ) from None
    if version is None:
        logger.error("Extension TimescaleDB absente de la base")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Extension TimescaleDB absente",
        )
    logger.debug("Extension TimescaleDB en version %s", version)
    return ReadinessStatus(status="ready", database="reachable", timescaledb="loaded")
@@ -1,111 +0,0 @@
 from uuid import UUID
 from fastapi import APIRouter, HTTPException, Response, status
 from app.api.deps import AdminDep, UserServiceDep
 from app.core.logging import get_logger
 from app.schemas.user import (
    TemporaryPasswordResponse,
    UserCreateRequest,
    UserResponse,
    UserUpdateRequest,
 )
 from app.services.user import EmailAlreadyUsedError, LastAdminError, UserNotFoundError
 router = APIRouter()
 logger = get_logger(__name__)
@router.get("", response_model=list[UserResponse], summary="Liste les comptes")
 async def list_users(_: AdminDep, service: UserServiceDep) -> list[UserResponse]:
    comptes = await service.list_all()
    return [UserResponse.model_validate(compte) for compte in comptes]
@router.post(
    "",
    response_model=TemporaryPasswordResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Crée un compte avec un mot de passe provisoire",
 )
 async def create_user(
    payload: UserCreateRequest,
    acteur: AdminDep,
    service: UserServiceDep,
    response: Response,
 ) -> TemporaryPasswordResponse:
    # Le mot de passe provisoire ne doit être conservé par aucun intermédiaire.
    response.headers["Cache-Control"] = "no-store"
    try:
        cree = await service.create(
            actor=acteur,
            email=payload.email,
            role=payload.role,
            full_name=payload.full_name,
        )
    except EmailAlreadyUsedError as erreur:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT, detail="Adresse déjà utilisée"
        ) from erreur
    logger.info("user.created actor=%s target=%s", acteur.id, cree.user.id)
    return TemporaryPasswordResponse(
        user=UserResponse.model_validate(cree.user),
        temporary_password=cree.temporary_password,
    )
@router.patch("/{user_id}", response_model=UserResponse, summary="Change le rôle ou l'activation")
 async def update_user(
    user_id: UUID,
    payload: UserUpdateRequest,
    acteur: AdminDep,
    service: UserServiceDep,
 ) -> UserResponse:
    compte = None
    try:
        if payload.role is not None:
            compte = await service.change_role(actor=acteur, user_id=user_id, role=payload.role)
        if payload.is_active is not None:
            compte = await service.set_active(
                actor=acteur, user_id=user_id, is_active=payload.is_active
            )
    except UserNotFoundError as erreur:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Compte introuvable"
        ) from erreur
    except LastAdminError as erreur:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Dernier administrateur actif, l'opération le laisserait sans successeur",
        ) from erreur
    if compte is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Aucune modification demandée"
        )
    logger.info("user.updated actor=%s target=%s", acteur.id, user_id)
    return UserResponse.model_validate(compte)
@router.post(
    "/{user_id}/password-reset",
    response_model=TemporaryPasswordResponse,
    summary="Réinitialise le mot de passe et ferme les sessions",
 )
 async def reset_password(
    user_id: UUID, acteur: AdminDep, service: UserServiceDep, response: Response
 ) -> TemporaryPasswordResponse:
    response.headers["Cache-Control"] = "no-store"
    try:
        reinitialise = await service.reset_password(actor=acteur, user_id=user_id)
    except UserNotFoundError as erreur:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Compte introuvable"
        ) from erreur
    logger.info("user.password_reset actor=%s target=%s", acteur.id, user_id)
    return TemporaryPasswordResponse(
        user=UserResponse.model_validate(reinitialise.user),
        temporary_password=reinitialise.temporary_password,
    )
@@ -1,8 +0,0 @@
 from fastapi import APIRouter
 from app.api.v1.endpoints import auth, health, users
 api_router = APIRouter()
 api_router.include_router(health.router, prefix="/health", tags=["health"])
 api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
 api_router.include_router(users.router, prefix="/users", tags=["users"])
@@ -1,104 +0,0 @@
 # Pourquoi : `create_admin()` est une commande et non une révision Alembic. Une révision qui
 # insérerait un compte graverait son empreinte dans Git pour toujours, et son mot de passe
 # serait connu de quiconque lit le dépôt. L'ADR 0001 pose par ailleurs qu'Alembic porte le
 # schéma, pas les données.
 # Piège : le mot de passe ne transite jamais par `argv`, visible de tout `ps`, ni par
 # l'historique du shell. Il est saisi par `getpass` ou tiré au sort par la commande.
 import argparse
 import asyncio
 import secrets
 import sys
 from getpass import getpass
 from app.core.config import Settings, get_settings
 from app.core.hashing import build_hasher
 from app.core.roles import Role
 from app.db.session import get_session_factory
 from app.repositories.user import UserRepository
 LONGUEUR_MOT_DE_PASSE_GENERE = 24
 LONGUEUR_MINIMALE = 12
 async def create_admin(
    settings: Settings, *, email: str, password: str, force: bool
 ) -> tuple[bool, str]:
    hacheur = build_hasher(
        time_cost=settings.argon2_time_cost,
        memory_cost_kib=settings.argon2_memory_cost_kib,
        parallelism=settings.argon2_parallelism,
        max_concurrency=settings.argon2_max_concurrency,
    )
    empreinte = await hacheur.hash(password)
    async with get_session_factory()() as session:
        depot = UserRepository(session)
        if not force and await depot.count_active_admins() > 0:
            return False, "Un administrateur actif existe déjà, relancer avec --force pour forcer"
        if await depot.get_by_email(email) is not None:
            return False, f"Le compte {email} existe déjà"
        await depot.create(
            email=email,
            password_hash=empreinte,
            role=Role.ADMIN,
            must_change_password=True,
        )
        await session.commit()
    return (
        True,
        f"Administrateur {email.strip().lower()} créé, mot de passe à changer à la connexion",
    )
 def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="python -m app.cli", description="Outils EnerVision")
    sous_commandes = parser.add_subparsers(dest="commande", required=True)
    admin = sous_commandes.add_parser("create-admin", help="Crée le premier administrateur")
    admin.add_argument("--email", required=True)
    admin.add_argument(
        "--generate", action="store_true", help="Tire un mot de passe au sort et l'affiche une fois"
    )
    admin.add_argument(
        "--force", action="store_true", help="Crée le compte même si un administrateur existe"
    )
    return parser
 def read_password(*, generate: bool) -> str:
    if generate:
        mot_de_passe = secrets.token_urlsafe(LONGUEUR_MOT_DE_PASSE_GENERE)
        print(f"Mot de passe généré, il ne sera plus affiché : {mot_de_passe}")
        return mot_de_passe
    mot_de_passe = getpass("Mot de passe : ")
    if len(mot_de_passe) < LONGUEUR_MINIMALE:
        raise SystemExit(f"Le mot de passe doit faire au moins {LONGUEUR_MINIMALE} caractères")
    if mot_de_passe != getpass("Confirmation : "):
        raise SystemExit("Les deux saisies diffèrent")
    return mot_de_passe
 def main(argv: list[str] | None = None) -> int:
    arguments = build_parser().parse_args(argv)
    mot_de_passe = read_password(generate=arguments.generate)
    succes, message = asyncio.run(
        create_admin(
            get_settings(),
            email=arguments.email,
            password=mot_de_passe,
            force=arguments.force,
        )
    )
    print(message)
    return 0 if succes else 1
 if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())
@@ -1,109 +0,0 @@
 from functools import lru_cache
 from typing import Literal, Self
 from pydantic import Field, SecretStr, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 Environment = Literal["local", "dev", "staging", "prod"]
 SameSite = Literal["lax", "strict", "none"]
 SECRET_KEY_MIN_LENGTH = 32
 SENTINELLES_INTERDITES = frozenset(
    {"change_me", "changeme", "secret", "secret-de-test", "changez-moi", "todo"}
 )
 class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_prefix="APP_",
        env_file_encoding="utf-8",
        extra="ignore",
    )
    name: str = "EnerVision API"
    version: str = "0.1.0"
    env: Environment = "local"
    debug: bool = False
    log_level: str = "INFO"
    api_prefix: str = "/api/v1"
    secret_key: SecretStr
    cors_origins: str = ""
    database_url: str = Field(validation_alias="DATABASE_URL")
    database_pool_size: int = 5
    database_max_overflow: int = 10
    jwt_issuer: str = "enervision-api"
    jwt_audience: str = "enervision-web"
    access_token_ttl_seconds: int = Field(default=900, ge=60, le=3600)
    refresh_token_ttl_seconds: int = Field(default=604800, ge=3600, le=2592000)
    refresh_cookie_name: str = "ev_refresh"
    cookie_path: str = "/api/v1/auth"
    cookie_samesite: SameSite = "strict"
    cookie_secure: bool | None = None
    argon2_time_cost: int = Field(default=2, ge=1, le=10)
    argon2_memory_cost_kib: int = Field(default=19456, ge=8192)
    argon2_parallelism: int = Field(default=1, ge=1, le=4)
    argon2_max_concurrency: int = Field(default=4, ge=1, le=32)
    login_window_seconds: int = Field(default=900, ge=60)
    login_max_failures_per_identifier_and_ip: int = Field(default=5, ge=1)
    login_max_failures_per_ip: int = Field(default=20, ge=1)
    login_max_failures_per_identifier: int = Field(default=50, ge=1)
    trust_proxy_headers: bool = False
    expose_api_docs: bool | None = None
    metrics_token: SecretStr | None = None
    @property
    def allowed_origins(self) -> list[str]:
        return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()]
    @property
    def is_production(self) -> bool:
        return self.env == "prod"
    @property
    def cookies_are_secure(self) -> bool:
        return self.env != "local" if self.cookie_secure is None else self.cookie_secure
    @property
    def api_docs_are_exposed(self) -> bool:
        if self.expose_api_docs is not None:
            return self.expose_api_docs
        return self.env not in ("staging", "prod")
    @model_validator(mode="after")
    def _refuse_les_configurations_dangereuses(self) -> Self:
        secret = self.secret_key.get_secret_value()
        if len(secret) < SECRET_KEY_MIN_LENGTH:
            raise ValueError(
                f"APP_SECRET_KEY doit faire au moins {SECRET_KEY_MIN_LENGTH} caractères"
            )
        if secret.strip().lower() in SENTINELLES_INTERDITES:
            raise ValueError("APP_SECRET_KEY est une valeur d'exemple, il faut en générer une")
        # Piège : `create_app()` passe `debug` à FastAPI, qui renvoie alors la trace complète
        # au client, et à l'engine, qui journalise le SQL et ses paramètres.
        if self.debug and self.env in ("staging", "prod"):
            raise ValueError("APP_DEBUG doit rester faux hors des environnements locaux")
        if "*" in self.cors_origins:
            raise ValueError("APP_CORS_ORIGINS n'accepte pas de joker, les origines sont listées")
        # Sans origines, aucun middleware CORS n'est monté et la vérification d'`Origin` des
        # routes d'authentification n'a plus de référentiel auquel comparer.
        if self.env != "local" and not self.allowed_origins:
            raise ValueError("APP_CORS_ORIGINS doit lister au moins une origine hors local")
        if self.cookie_samesite == "none" and not self.cookies_are_secure:
            raise ValueError("Un cookie SameSite=None est rejeté par les navigateurs sans Secure")
        return self
@lru_cache
 def get_settings() -> Settings:
    return Settings()
@@ -1,61 +0,0 @@
 # Piège : le cookie de suppression doit reprendre exactement le nom et le `Path` du cookie
 # posé, sinon le navigateur en garde une copie et la déconnexion n'est que cosmétique.
 # `RefreshCookie.expired()` existe pour que les deux ne puissent pas diverger.
 from dataclasses import asdict, dataclass
 from typing import Any, Self
 from app.core.config import SameSite, Settings
 SECURE_PREFIX = "__Secure-"
@dataclass(frozen=True, slots=True)
 class RefreshCookie:
    key: str
    value: str
    max_age: int
    path: str
    secure: bool
    httponly: bool
    samesite: SameSite
    @classmethod
    def build(cls, settings: Settings, value: str) -> Self:
        return cls(
            key=cookie_name(settings),
            value=value,
            max_age=settings.refresh_token_ttl_seconds,
            path=settings.cookie_path,
            secure=settings.cookies_are_secure,
            httponly=True,
            samesite=settings.cookie_samesite,
        )
    @classmethod
    def expired(cls, settings: Settings) -> Self:
        return cls(
            key=cookie_name(settings),
            value="",
            max_age=0,
            path=settings.cookie_path,
            secure=settings.cookies_are_secure,
            httponly=True,
            samesite=settings.cookie_samesite,
        )
    def as_kwargs(self) -> dict[str, Any]:
        return asdict(self)
    def as_deletion_kwargs(self) -> dict[str, Any]:
        # `Response.delete_cookie()` n'accepte ni `value` ni `max_age`, mais il exige le même
        # nom, le même chemin et les mêmes attributs, sinon le navigateur garde le cookie.
        arguments = asdict(self)
        del arguments["value"], arguments["max_age"]
        return arguments
 def cookie_name(settings: Settings) -> str:
    if settings.cookies_are_secure:
        return f"{SECURE_PREFIX}{settings.refresh_cookie_name}"
    return settings.refresh_cookie_name
@@ -1,63 +0,0 @@
 # Piège : `PasswordHasher.verify()` bloque 17 ms. Appelé tel quel dans un `async def`, il fige
 # la boucle d'événements et gèle toutes les requêtes en cours, pas seulement la connexion.
 # `Argon2Hasher` le pousse donc dans un fil, sous un `CapacityLimiter` : le pool par défaut
 # d'anyio accepte 40 fils, soit 40 x 19 Mio dans le pire cas sur une machine qui héberge aussi
 # PostgreSQL, Prometheus et Grafana.
 # Piège : `verify_dummy()` doit être appelé quand l'utilisateur est introuvable. Sans lui,
 # l'écart entre 2 ms et 17 ms est un oracle d'existence de compte, mesurable à distance.
 import secrets
 import anyio
 import anyio.to_thread
 from argon2 import PasswordHasher
 from argon2.exceptions import Argon2Error, InvalidHashError, VerificationError
 _ERREURS_DE_VERIFICATION = (VerificationError, InvalidHashError, Argon2Error)
 class Argon2Hasher:
    def __init__(self, hasher: PasswordHasher, *, max_concurrency: int) -> None:
        self._hasher = hasher
        self._limiter = anyio.CapacityLimiter(max_concurrency)
        self._leurre = hasher.hash(secrets.token_urlsafe(32))
    async def hash(self, password: str) -> str:
        return await anyio.to_thread.run_sync(self._hasher.hash, password, limiter=self._limiter)
    async def verify(self, stored: str, password: str) -> bool:
        return await anyio.to_thread.run_sync(self._verify, stored, password, limiter=self._limiter)
    async def verify_dummy(self) -> None:
        await self.verify(self._leurre, "")
    def needs_rehash(self, stored: str) -> bool:
        try:
            return self._hasher.check_needs_rehash(stored)
        except _ERREURS_DE_VERIFICATION:
            return True
    def _verify(self, stored: str, password: str) -> bool:
        try:
            return self._hasher.verify(stored, password)
        except _ERREURS_DE_VERIFICATION:
            return False
 def build_hasher(
    *,
    time_cost: int,
    memory_cost_kib: int,
    parallelism: int,
    max_concurrency: int,
 ) -> Argon2Hasher:
    return Argon2Hasher(
        PasswordHasher(
            time_cost=time_cost,
            memory_cost=memory_cost_kib,
            parallelism=parallelism,
            hash_len=32,
            salt_len=16,
        ),
        max_concurrency=max_concurrency,
    )
@@ -1,92 +0,0 @@
 # Pourquoi : `RedactingFilter` est la troisième ligne de défense, pas la première. La première
 # est de ne jamais passer un secret au logger, la deuxième de ne jamais mettre un jeton dans
 # une URL, que le journal d'accès enregistrerait de toute façon. Le filtre rattrape l'erreur
 # que personne n'a relue, notamment l'écho SQL quand `debug` est actif.
 import logging
 import re
 from logging.config import dictConfig
 from typing import Final
 from app.core.config import Settings
 CAVIARDAGE: Final = "[expurgé]"
 REMPLACEMENTS: Final[tuple[tuple[re.Pattern[str], str], ...]] = (
    (re.compile(r"Bearer\s+[A-Za-z0-9._~+/-]{20,}=*"), f"Bearer {CAVIARDAGE}"),
    (re.compile(r"eyJ[A-Za-z0-9._-]{20,}"), CAVIARDAGE),
    (re.compile(r"\$argon2[a-z0-9]*\$\S+"), CAVIARDAGE),
    (
        re.compile(r'("?(?:password|mot_de_passe|secret|token)"?\s*[:=]\s*")[^"]*(")'),
        rf"\1{CAVIARDAGE}\2",
    ),
    (
        re.compile(r"((?:password|mot_de_passe|secret|token)[A-Za-z_]*=)[^&\s;\"]+"),
        rf"\1{CAVIARDAGE}",
    ),
    (re.compile(r"(ev_refresh=)[^;\s]+"), rf"\1{CAVIARDAGE}"),
 )
 def redact(message: str) -> str:
    for motif, remplacement in REMPLACEMENTS:
        message = motif.sub(remplacement, message)
    return message
 class RedactingFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        expurge = redact(message)
        if expurge != message:
            record.msg = expurge
            record.args = ()
        return True
 def configure_logging(settings: Settings) -> None:
    formatter = "json" if settings.is_production else "console"
    dictConfig(
        {
            "version": 1,
            "disable_existing_loggers": False,
            "filters": {
                "redaction": {"()": "app.core.logging.RedactingFilter"},
            },
            "formatters": {
                "console": {
                    "format": "%(asctime)s %(levelname)-8s %(name)s %(message)s",
                },
                "json": {
                    "()": "pythonjsonlogger.json.JsonFormatter",
                    "format": "%(asctime)s %(levelname)s %(name)s %(message)s",
                },
            },
            "handlers": {
                "default": {
                    "class": "logging.StreamHandler",
                    "formatter": formatter,
                    "filters": ["redaction"],
                    "stream": "ext://sys.stdout",
                },
            },
            "root": {"handlers": ["default"], "level": settings.log_level},
            "loggers": {
                "uvicorn": {
                    "handlers": ["default"],
                    "level": settings.log_level,
                    "propagate": False,
                },
                "uvicorn.access": {
                    "handlers": ["default"],
                    "level": settings.log_level,
                    "propagate": False,
                },
                "sqlalchemy.engine": {"level": "WARNING"},
            },
        }
    )
 def get_logger(name: str) -> logging.Logger:
    return logging.getLogger(name)
@@ -1,18 +0,0 @@
 # Pourquoi : tout le code métier dépend de `Principal` et jamais du modèle ORM ni des claims
 # du jeton. C'est ce qui garde la bascule vers un fournisseur OIDC locale à
 # `get_current_principal()` et à `AuthService.authenticate()`, au lieu de la répandre dans
 # chaque endpoint.
 from dataclasses import dataclass
 from uuid import UUID
 from app.core.roles import AccountKind, Role
@dataclass(frozen=True, slots=True)
 class Principal:
    id: UUID
    email: str
    role: Role
    kind: AccountKind
    must_change_password: bool
@@ -1,26 +0,0 @@
 from enum import StrEnum
 from typing import Final
 class Role(StrEnum):
    # Contrainte : ces valeurs voyagent en base, en JSON et dans les jetons. Elles restent
    # en ASCII, contrairement au libellé « opérateur » affiché à l'utilisateur.
    LECTEUR = "lecteur"
    OPERATEUR = "operateur"
    ADMIN = "admin"
 class AccountKind(StrEnum):
    HUMAIN = "human"
    SERVICE = "service"
 ROLE_RANK: Final[dict[Role, int]] = {
    Role.LECTEUR: 0,
    Role.OPERATEUR: 1,
    Role.ADMIN: 2,
 }
 def has_at_least(actual: Role, required: Role) -> bool:
    return ROLE_RANK[actual] >= ROLE_RANK[required]
@@ -1,117 +0,0 @@
 # Piège : `decode_access_token()` porte trois barrières indépendantes, et retirer l'une
 # d'elles ne casse aucun test évident. L'algorithme est épinglé, sinon un jeton forgé en
 # `alg: none` passerait. L'audience et l'émetteur sont vérifiés, sinon un jeton émis pour
 # un autre service serait accepté. Le claim `typ` est comparé, sinon un jeton de
 # rafraîchissement servirait de jeton d'accès, ce qui transformerait une fenêtre de
 # 15 minutes en fenêtre de 7 jours.
 # Contrainte : ce module ne lit jamais `get_settings()`, qui est mis en cache par
 # `lru_cache` et se contaminerait entre tests. Tout paramètre arrive par `TokenPolicy`.
 import hashlib
 import secrets
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import Final
 from uuid import UUID, uuid4
 import jwt
 ACCESS_TOKEN_TYPE: Final = "access"  # noqa: S105
 REFRESH_SECRET_BYTES: Final = 32
 _ALGORITHME: Final = "HS256"
 _CLAIMS_REQUIS: Final = ["iss", "aud", "sub", "iat", "exp", "jti", "typ", "role", "kind"]
 class TokenInvalidError(Exception):
    pass
 class TokenExpiredError(TokenInvalidError):
    pass
@dataclass(frozen=True, slots=True)
 class TokenPolicy:
    secret: str
    issuer: str
    audience: str
    access_ttl: timedelta
@dataclass(frozen=True, slots=True)
 class AccessClaims:
    subject: UUID
    role: str
    kind: str
    token_id: UUID
    issued_at: datetime
 def encode_access_token(
    policy: TokenPolicy,
    *,
    subject: UUID,
    role: str,
    kind: str,
    now: datetime | None = None,
 ) -> str:
    emis_a = now or datetime.now(UTC)
    return jwt.encode(
        {
            "iss": policy.issuer,
            "aud": policy.audience,
            "sub": str(subject),
            "iat": emis_a,
            "exp": emis_a + policy.access_ttl,
            "jti": str(uuid4()),
            "typ": ACCESS_TOKEN_TYPE,
            "role": role,
            "kind": kind,
        },
        policy.secret,
        algorithm=_ALGORITHME,
    )
 def decode_access_token(policy: TokenPolicy, token: str) -> AccessClaims:
    try:
        charge = jwt.decode(
            token,
            policy.secret,
            algorithms=[_ALGORITHME],
            audience=policy.audience,
            issuer=policy.issuer,
            options={"require": _CLAIMS_REQUIS},
        )
    except jwt.ExpiredSignatureError as erreur:
        raise TokenExpiredError("Jeton expiré") from erreur
    except jwt.InvalidTokenError as erreur:
        raise TokenInvalidError("Jeton invalide") from erreur
    if charge["typ"] != ACCESS_TOKEN_TYPE:
        raise TokenInvalidError("Type de jeton inattendu")
    try:
        sujet = UUID(charge["sub"])
        identifiant = UUID(charge["jti"])
    except (AttributeError, TypeError, ValueError) as erreur:
        raise TokenInvalidError("Identifiants du jeton illisibles") from erreur
    return AccessClaims(
        subject=sujet,
        role=str(charge["role"]),
        kind=str(charge["kind"]),
        token_id=identifiant,
        issued_at=datetime.fromtimestamp(charge["iat"], tz=UTC),
    )
 def generate_refresh_secret() -> str:
    return secrets.token_urlsafe(REFRESH_SECRET_BYTES)
 # SHA-256 nu, pas Argon2id : 256 bits de CSPRNG n'ont ni dictionnaire ni préimage atteignable,
 # et une KDF lente coûterait 17 ms à chaque rafraîchissement pour aucun gain.
 def fingerprint_refresh(secret: str) -> bytes:
    return hashlib.sha256(secret.encode("utf-8")).digest()
@@ -1,5 +0,0 @@
 from sqlalchemy.orm import DeclarativeBase
 class Base(DeclarativeBase):
    """Base déclarative commune à tous les modèles."""
@@ -1,33 +0,0 @@
 from collections.abc import AsyncIterator
 from functools import lru_cache
 from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
 )
 from app.core.config import get_settings
@lru_cache
 def get_engine() -> AsyncEngine:
    settings = get_settings()
    return create_async_engine(
        settings.database_url,
        echo=settings.debug,
        pool_pre_ping=True,
        pool_size=settings.database_pool_size,
        max_overflow=settings.database_max_overflow,
    )
@lru_cache
 def get_session_factory() -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(get_engine(), class_=AsyncSession, expire_on_commit=False)
 async def get_session() -> AsyncIterator[AsyncSession]:
    async with get_session_factory()() as session:
        yield session
@@ -1,78 +0,0 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from fastapi import Depends, FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from prometheus_fastapi_instrumentator import Instrumentator
 from app.api.errors import register_error_handlers
 from app.api.middleware import SecurityHeadersMiddleware
 from app.api.security import require_metrics_token
 from app.api.v1.router import api_router
 from app.core.config import Settings, get_settings
 from app.core.logging import configure_logging, get_logger
 from app.db.session import get_engine
 logger = get_logger(__name__)
 METHODES_AUTORISEES = ["GET", "POST", "PATCH", "PUT", "DELETE", "OPTIONS"]
 EN_TETES_AUTORISES = ["Authorization", "Content-Type"]
@asynccontextmanager
 async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    settings = get_settings()
    logger.info(
        "Démarrage de %s %s en environnement %s", settings.name, settings.version, settings.env
    )
    yield
    await get_engine().dispose()
 def create_app(settings: Settings | None = None) -> FastAPI:
    resolved = settings or get_settings()
    configure_logging(resolved)
    documentee = resolved.api_docs_are_exposed
    application = FastAPI(
        title=resolved.name,
        version=resolved.version,
        debug=resolved.debug,
        lifespan=lifespan,
        docs_url="/docs" if documentee else None,
        redoc_url="/redoc" if documentee else None,
        openapi_url="/openapi.json" if documentee else None,
    )
    application.add_middleware(SecurityHeadersMiddleware)
    if resolved.allowed_origins:
        # Méthodes et en-têtes listés plutôt que joker : avec `allow_credentials`, la liste
        # d'origines devient l'unique contrôle, autant documenter le contrat exact.
        application.add_middleware(
            CORSMiddleware,
            allow_origins=resolved.allowed_origins,
            allow_credentials=True,
            allow_methods=METHODES_AUTORISEES,
            allow_headers=EN_TETES_AUTORISES,
            expose_headers=["Retry-After"],
            max_age=600,
        )
    register_error_handlers(application)
    Instrumentator().instrument(application).expose(
        application,
        endpoint="/metrics",
        include_in_schema=False,
        dependencies=[Depends(require_metrics_token)],
    )
    application.include_router(api_router, prefix=resolved.api_prefix)
    # Piège : sans cette surcharge, une configuration passée à `create_app()` ne piloterait
    # que la construction, et les dépendances continueraient de lire `get_settings()` depuis
    # l'environnement. Un test « en production » ne testerait alors pas la production.
    if settings is not None:
        application.dependency_overrides[get_settings] = lambda: resolved
    return application
@@ -1,21 +0,0 @@
 # Piège : tout modèle absent de ce module reste invisible de `alembic revision
 # --autogenerate`, qui générerait alors un drop de sa table.
 from app.models.audit_log import AuditLog
 from app.models.energy import Alert, Dataset, Prediction, Reading, Recommendation, Site
 from app.models.login_attempt import LoginAttempt
 from app.models.refresh_token import RefreshToken
 from app.models.user import AppUser
 __all__ = [
    "Alert",
    "AppUser",
    "AuditLog",
    "Dataset",
    "LoginAttempt",
    "Prediction",
    "Reading",
    "Recommendation",
    "RefreshToken",
    "Site",
 ]
@@ -1,64 +0,0 @@
 # Pourquoi : `actor_id` ne porte volontairement aucune clé étrangère. Une contrainte
 # `ON DELETE SET NULL` déclencherait un UPDATE que le déclencheur d'ajout seul refuserait, donc
 # la suppression d'un compte échouerait ; une contrainte `NO ACTION` interdirait toute
 # suppression. `actor_email` et `actor_role` sont dénormalisés pour la même raison : le journal
 # dit ce qui était vrai au moment de l'acte, pas ce qui est vrai aujourd'hui.
 import uuid
 from datetime import datetime
 from enum import StrEnum
 from typing import Any
 from sqlalchemy import BigInteger, CheckConstraint, DateTime, Identity, Index, Text, func
 from sqlalchemy.dialects.postgresql import INET, JSONB
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
 from sqlalchemy.orm import Mapped, mapped_column
 from app.db.base import Base
 class AuditOutcome(StrEnum):
    SUCCES = "success"
    ECHEC = "failure"
 class AuditAction(StrEnum):
    COMPTE_CREE = "user.created"
    COMPTE_ROLE_CHANGE = "user.role_changed"
    COMPTE_DESACTIVE = "user.disabled"
    COMPTE_ACTIVE = "user.enabled"
    COMPTE_MOT_DE_PASSE_REINITIALISE = "user.password_reset_by_admin"
    COMPTE_MOT_DE_PASSE_CHANGE = "user.password_changed"
    REFRESH_REUTILISE = "auth.refresh_reuse_detected"
    SESSIONS_REVOQUEES = "auth.all_sessions_revoked"
    LIMITE_PAR_IDENTIFIANT = "auth.identifier_throttled"
    ADMIN_AMORCE = "bootstrap.admin_created"
 ISSUES_AUTORISEES = ", ".join(f"'{issue.value}'" for issue in AuditOutcome)
 class AuditLog(Base):
    __tablename__ = "audit_log"
    __table_args__ = (
        CheckConstraint(f"outcome in ({ISSUES_AUTORISEES})", name="ck_audit_log_outcome"),
        Index("ix_audit_log_date", "occurred_at"),
        Index("ix_audit_log_action_date", "action", "occurred_at"),
    )
    id: Mapped[int] = mapped_column(BigInteger, Identity(always=True), primary_key=True)
    occurred_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    actor_id: Mapped[uuid.UUID | None] = mapped_column(PG_UUID(as_uuid=True), nullable=True)
    actor_email: Mapped[str | None] = mapped_column(Text, nullable=True)
    actor_role: Mapped[str | None] = mapped_column(Text, nullable=True)
    action: Mapped[str] = mapped_column(Text, nullable=False)
    target_type: Mapped[str | None] = mapped_column(Text, nullable=True)
    target_id: Mapped[str | None] = mapped_column(Text, nullable=True)
    outcome: Mapped[str] = mapped_column(Text, nullable=False)
    client_ip: Mapped[str | None] = mapped_column(INET, nullable=True)
    user_agent: Mapped[str | None] = mapped_column(Text, nullable=True)
    detail: Mapped[dict[str, Any]] = mapped_column(
        JSONB, nullable=False, server_default=func.jsonb_build_object()
    )
@@ -1,210 +0,0 @@
 """Tables du modèle de données EnerVision (CSV, API Mock et résultats ML)."""
 from datetime import datetime
 from decimal import Decimal
 from typing import Any
 from sqlalchemy import (
    BigInteger,
    Boolean,
    CheckConstraint,
    DateTime,
    Double,
    ForeignKey,
    ForeignKeyConstraint,
    Index,
    Integer,
    Numeric,
    String,
    Text,
    UniqueConstraint,
    func,
    text,
 )
 from sqlalchemy.dialects.postgresql import ARRAY, JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 from app.db.base import Base
 class Dataset(Base):
    __tablename__ = "dataset"
    __table_args__ = (
        CheckConstraint("dataset_id > 0", name="ck_dataset_positive_id"),
        UniqueConstraint("archive_sha256", name="uq_dataset_archive_sha256"),
    )
    dataset_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    dataset_name: Mapped[str] = mapped_column(Text)
    archive_sha256: Mapped[str] = mapped_column(String(64))
    storage_uri: Mapped[str] = mapped_column(Text)
    source_timezone: Mapped[str | None] = mapped_column(Text)
    # "metadata" est réservé par SQLAlchemy ; le nom SQL reste inchangé.
    dataset_metadata: Mapped[dict[str, Any]] = mapped_column("metadata", JSONB(none_as_null=True))
 class Site(Base):
    __tablename__ = "site"
    site_id: Mapped[str] = mapped_column(Text, primary_key=True)
    site_name: Mapped[str] = mapped_column(Text)
    site_type: Mapped[str] = mapped_column(Text)
    location: Mapped[str | None] = mapped_column(Text)
    capacity_kw: Mapped[float | None] = mapped_column(Double)
    status: Mapped[str | None] = mapped_column(Text)
 class Reading(Base):
    __tablename__ = "reading"
    __table_args__ = (
        CheckConstraint(
            "source IN ('csv', 'api_current', 'api_history')", name="ck_reading_source"
        ),
        CheckConstraint(
            "(source = 'csv' AND dataset_id IS NOT NULL) OR "
            "(source IN ('api_current', 'api_history') AND dataset_id IS NULL)",
            name="ck_reading_dataset_source",
        ),
        CheckConstraint(
            "data_quality IS NULL OR data_quality IN ('good', 'partial', 'degraded', 'critical')",
            name="ck_reading_quality",
        ),
        CheckConstraint(
            "(imputed_values IS NULL AND imputation_method IS NULL) OR "
            "(imputed_values IS NOT NULL AND imputation_method IS NOT NULL)",
            name="ck_reading_imputation",
        ),
        Index("ix_reading_site_timestamp", "site_id", "timestamp"),
        Index("ix_reading_dataset_id", "dataset_id"),
    )
    reading_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    site_id: Mapped[str] = mapped_column(
        Text, ForeignKey("site.site_id", name="fk_reading_site", ondelete="RESTRICT")
    )
    timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), primary_key=True)
    source: Mapped[str] = mapped_column(Text)
    dataset_id: Mapped[int | None] = mapped_column(
        BigInteger,
        ForeignKey("dataset.dataset_id", name="fk_reading_dataset", ondelete="RESTRICT"),
    )
    consumption_kw: Mapped[float | None] = mapped_column(Double)
    consumption_kwh: Mapped[float | None] = mapped_column(Double)
    consumption_euros: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    voltage_v: Mapped[float | None] = mapped_column(Double)
    current_a: Mapped[float | None] = mapped_column(Double)
    power_factor: Mapped[float | None] = mapped_column(Double)
    temperature_celsius: Mapped[float | None] = mapped_column(Double)
    humidity_percent: Mapped[float | None] = mapped_column(Double)
    solar_irradiance_wm2: Mapped[float | None] = mapped_column(Double)
    is_working_hours: Mapped[bool | None] = mapped_column(Boolean)
    data_quality: Mapped[str | None] = mapped_column(Text)
    null_reasons: Mapped[list[str] | None] = mapped_column(ARRAY(Text))
    imputed_values: Mapped[dict[str, Any] | None] = mapped_column(JSONB(none_as_null=True))
    imputation_method: Mapped[str | None] = mapped_column(Text)
    ingested_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    raw_data: Mapped[dict[str, Any]] = mapped_column(JSONB(none_as_null=True))
 Index(
    "uq_reading_source",
    Reading.site_id,
    Reading.timestamp,
    Reading.source,
    func.coalesce(Reading.dataset_id, text("0")),
    unique=True,
 )
 class Prediction(Base):
    __tablename__ = "prediction"
    __table_args__ = (
        UniqueConstraint("prediction_id", "site_id", name="uq_prediction_id_site"),
        Index("ix_prediction_site_target", "site_id", "target_at"),
        CheckConstraint(
            "target_metric IN ('consumption_kwh', 'consumption_kw')",
            name="ck_prediction_metric",
        ),
        CheckConstraint(
            "period_minutes IS NULL OR period_minutes > 0", name="ck_prediction_period"
        ),
        CheckConstraint(
            "target_metric <> 'consumption_kwh' OR period_minutes IS NOT NULL",
            name="ck_prediction_energy_period",
        ),
        CheckConstraint(
            "(status = 'available' AND predicted_value IS NOT NULL AND failure_reason IS NULL) OR "
            "(status IN ('insufficient_data', 'error') AND predicted_value IS NULL "
            "AND failure_reason IS NOT NULL)",
            name="ck_prediction_status",
        ),
    )
    prediction_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    site_id: Mapped[str] = mapped_column(
        Text, ForeignKey("site.site_id", name="fk_prediction_site", ondelete="RESTRICT")
    )
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    target_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
    target_metric: Mapped[str] = mapped_column(Text)
    period_minutes: Mapped[int | None] = mapped_column(Integer)
    predicted_value: Mapped[float | None] = mapped_column(Double)
    model_reference: Mapped[str] = mapped_column(Text)
    status: Mapped[str] = mapped_column(Text)
    failure_reason: Mapped[str | None] = mapped_column(Text)
 class Alert(Base):
    __tablename__ = "alert"
    __table_args__ = (
        UniqueConstraint("source", "site_id", "source_alert_id", name="uq_alert_source_reference"),
        Index("ix_alert_site_timestamp", "site_id", "timestamp"),
        ForeignKeyConstraint(
            ["prediction_id", "site_id"],
            ["prediction.prediction_id", "prediction.site_id"],
            name="fk_alert_prediction_site",
            ondelete="RESTRICT",
        ),
        CheckConstraint("source IN ('api_mock', 'enervision')", name="ck_alert_source"),
        CheckConstraint(
            "type IN ('spike', 'threshold', 'anomaly', 'outage', 'sensor')", name="ck_alert_type"
        ),
        CheckConstraint(
            "severity IN ('low', 'medium', 'high', 'critical')", name="ck_alert_severity"
        ),
    )
    alert_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    source_alert_id: Mapped[str] = mapped_column(Text)
    site_id: Mapped[str] = mapped_column(
        Text, ForeignKey("site.site_id", name="fk_alert_site", ondelete="RESTRICT")
    )
    source: Mapped[str] = mapped_column(Text)
    timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True))
    type: Mapped[str] = mapped_column(Text)
    severity: Mapped[str] = mapped_column(Text)
    message: Mapped[str] = mapped_column(Text)
    value: Mapped[float | None] = mapped_column(Double)
    threshold: Mapped[float | None] = mapped_column(Double)
    metric: Mapped[str | None] = mapped_column(Text)
    prediction_id: Mapped[int | None] = mapped_column(BigInteger)
    raw_data: Mapped[dict[str, Any]] = mapped_column(JSONB(none_as_null=True))
 class Recommendation(Base):
    __tablename__ = "recommendation"
    __table_args__ = (
        UniqueConstraint("alert_id", "rule_reference", name="uq_recommendation_alert_rule"),
    )
    recommendation_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    alert_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("alert.alert_id", name="fk_recommendation_alert", ondelete="RESTRICT"),
    )
    action: Mapped[str] = mapped_column(Text)
    explanation: Mapped[str] = mapped_column(Text)
    rule_reference: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
@@ -1,44 +0,0 @@
 # Pourquoi : les tentatives vivent ici et non dans `audit_log`, qui est en ajout seul. Leur
 # volume est piloté par l'attaquant : une force brute y écrirait des millions de lignes
 # indestructibles. Cette table-ci se purge, et c'est aussi le compteur de la limitation.
 # Piège : la tentative est enregistrée même quand l'email est inconnu, sinon le 429 dirait
 # qu'un compte existe.
 import uuid
 from datetime import datetime
 from enum import StrEnum
 from sqlalchemy import BigInteger, CheckConstraint, DateTime, Identity, Index, String, Text, func
 from sqlalchemy.dialects.postgresql import INET
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
 from sqlalchemy.orm import Mapped, mapped_column
 from app.db.base import Base
 class LoginOutcome(StrEnum):
    SUCCES = "success"
    IDENTIFIANTS_INVALIDES = "bad_credentials"
    LIMITE = "throttled"
    COMPTE_INDISPONIBLE = "inactive"
 ISSUES_AUTORISEES = ", ".join(f"'{issue.value}'" for issue in LoginOutcome)
 class LoginAttempt(Base):
    __tablename__ = "login_attempt"
    __table_args__ = (
        CheckConstraint(f"outcome in ({ISSUES_AUTORISEES})", name="ck_login_attempt_outcome"),
        Index("ix_login_attempt_email_date", "email_tried", "occurred_at"),
        Index("ix_login_attempt_ip_date", "client_ip", "occurred_at"),
    )
    id: Mapped[int] = mapped_column(BigInteger, Identity(always=True), primary_key=True)
    occurred_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    email_tried: Mapped[str] = mapped_column(String(320), nullable=False)
    client_ip: Mapped[str | None] = mapped_column(INET, nullable=True)
    outcome: Mapped[str] = mapped_column(Text, nullable=False)
    user_id: Mapped[uuid.UUID | None] = mapped_column(PG_UUID(as_uuid=True), nullable=True)
@@ -1,64 +0,0 @@
 # Pourquoi : un jeton de rafraîchissement est une chaîne opaque, jamais un JWT. Il doit être
 # révocable, donc cette ligne existe de toute façon ; le JWT n'ajouterait qu'un second chemin de
 # signature. Surtout, la séparation devient structurelle : un JWT ne figure dans aucune ligne,
 # une chaîne opaque échoue au décodage. Aucune confusion de type n'est possible.
 # Piège : `expires_at` est absolu et hérité du prédécesseur à chaque rotation. S'il glissait,
 # la promesse de sept jours serait fictive et une session active ne finirait jamais.
 import uuid
 from datetime import datetime
 from enum import StrEnum
 from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, LargeBinary, Text, func
 from sqlalchemy.dialects.postgresql import INET
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
 from sqlalchemy.orm import Mapped, mapped_column
 from app.db.base import Base
 class RevocationReason(StrEnum):
    DECONNEXION = "logout"
    ROTATION = "rotation"
    REUTILISATION = "reuse_detected"
    CHANGEMENT_MOT_DE_PASSE = "password_change"
    ADMINISTRATION = "admin"
 MOTIFS_AUTORISES = ", ".join(f"'{motif.value}'" for motif in RevocationReason)
 class RefreshToken(Base):
    __tablename__ = "refresh_token"
    __table_args__ = (
        CheckConstraint(
            f"revoked_reason is null or revoked_reason in ({MOTIFS_AUTORISES})",
            name="ck_refresh_token_revoked_reason",
        ),
        Index("ix_refresh_token_family", "family_id"),
        Index("ix_refresh_token_user", "user_id"),
        Index(
            "ix_refresh_token_vivants",
            "user_id",
            postgresql_where="revoked_at is null and rotated_at is null",
        ),
    )
    id: Mapped[uuid.UUID] = mapped_column(
        PG_UUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()
    )
    family_id: Mapped[uuid.UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
    user_id: Mapped[uuid.UUID] = mapped_column(
        PG_UUID(as_uuid=True), ForeignKey("app_user.id", ondelete="CASCADE"), nullable=False
    )
    token_hash: Mapped[bytes] = mapped_column(LargeBinary, nullable=False, unique=True)
    issued_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    rotated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    revoked_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
    replaced_by: Mapped[uuid.UUID | None] = mapped_column(PG_UUID(as_uuid=True), nullable=True)
    client_ip: Mapped[str | None] = mapped_column(INET, nullable=True)
    user_agent: Mapped[str | None] = mapped_column(Text, nullable=True)
@@ -1,50 +0,0 @@
 # Contrainte : la table s'appelle `app_user` et non `user`, qui est un mot réservé PostgreSQL,
 # raccourci de `CURRENT_USER`. Le nom rappelle aussi qu'il s'agit d'un compte applicatif, par
 # opposition au rôle PostgreSQL qui porte, lui, le cantonnement des accès.
 import uuid
 from datetime import datetime
 from sqlalchemy import Boolean, CheckConstraint, DateTime, String, Text, func, text
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
 from sqlalchemy.orm import Mapped, mapped_column
 from app.core.roles import AccountKind, Role
 from app.db.base import Base
 ROLES_AUTORISES = ", ".join(f"'{role.value}'" for role in Role)
 NATURES_AUTORISEES = ", ".join(f"'{nature.value}'" for nature in AccountKind)
 class AppUser(Base):
    __tablename__ = "app_user"
    __table_args__ = (
        CheckConstraint("email = lower(email)", name="ck_app_user_email_minuscule"),
        CheckConstraint(f"role in ({ROLES_AUTORISES})", name="ck_app_user_role"),
        CheckConstraint(f"kind in ({NATURES_AUTORISEES})", name="ck_app_user_kind"),
    )
    id: Mapped[uuid.UUID] = mapped_column(
        PG_UUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()
    )
    email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(Text, nullable=False)
    role: Mapped[str] = mapped_column(Text, nullable=False)
    kind: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'human'"))
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true"))
    must_change_password: Mapped[bool] = mapped_column(
        Boolean, nullable=False, server_default=text("false")
    )
    # Une seule colonne couvre le changement de mot de passe, le changement de rôle et la
    # désactivation : tout jeton émis avant cet instant est périmé.
    credentials_changed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    full_name: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
    )
@@ -1,62 +0,0 @@
 # Piège : `detail` passe par une liste blanche de clés et jamais par un `dict(**kwargs)`. La
 # table est en ajout seul : une clé inattendue qui porterait un secret ou une donnée
 # personnelle ne pourrait plus en être retirée.
 from collections.abc import Mapping
 from typing import Any
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.principal import Principal
 from app.models.audit_log import AuditAction, AuditLog, AuditOutcome
 CLES_DE_DETAIL_AUTORISEES = frozenset(
    {
        "email",
        "role_avant",
        "role_apres",
        "famille",
        "motif",
        "source",
        "sessions_revoquees",
    }
 )
 def assemble_detail(brut: Mapping[str, Any] | None) -> dict[str, Any]:
    if not brut:
        return {}
    return {cle: valeur for cle, valeur in brut.items() if cle in CLES_DE_DETAIL_AUTORISEES}
 class AuditLogRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session
    async def record(
        self,
        *,
        action: AuditAction,
        outcome: AuditOutcome = AuditOutcome.SUCCES,
        actor: Principal | None = None,
        actor_label: str | None = None,
        target_type: str | None = None,
        target_id: str | None = None,
        client_ip: str | None = None,
        user_agent: str | None = None,
        detail: Mapping[str, Any] | None = None,
    ) -> None:
        self._session.add(
            AuditLog(
                actor_id=actor.id if actor else None,
                actor_email=actor.email if actor else actor_label,
                actor_role=actor.role.value if actor else None,
                action=action.value,
                target_type=target_type,
                target_id=target_id,
                outcome=outcome.value,
                client_ip=client_ip,
                user_agent=user_agent,
                detail=assemble_detail(detail),
            )
        )
@@ -1,67 +0,0 @@
 # Pourquoi : les trois compteurs tiennent en une seule requête, grâce aux clauses FILTER de
 # PostgreSQL. Trois `count(*)` séparés feraient trois allers-retours sur le chemin critique de
 # la connexion, qui est justement celui qu'un attaquant martèle.
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from uuid import UUID
 from sqlalchemy import and_, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.login_attempt import LoginAttempt, LoginOutcome
@dataclass(frozen=True, slots=True)
 class FailureCounts:
    per_identifier_and_ip: int
    per_ip: int
    per_identifier: int
 class LoginAttemptRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session
    async def record(
        self,
        *,
        email: str,
        client_ip: str | None,
        outcome: LoginOutcome,
        user_id: UUID | None = None,
    ) -> None:
        self._session.add(
            LoginAttempt(
                email_tried=email.strip().lower(),
                client_ip=client_ip,
                outcome=outcome.value,
                user_id=user_id,
            )
        )
    async def count_recent_failures(
        self, *, email: str, client_ip: str | None, window_seconds: int
    ) -> FailureCounts:
        identifiant = email.strip().lower()
        meme_email = LoginAttempt.email_tried == identifiant
        meme_ip = LoginAttempt.client_ip == client_ip
        requete = select(
            func.count().filter(and_(meme_email, meme_ip)),
            func.count().filter(meme_ip),
            func.count().filter(meme_email),
        ).where(
            LoginAttempt.outcome != LoginOutcome.SUCCES.value,
            LoginAttempt.occurred_at > datetime.now(UTC) - timedelta(seconds=window_seconds),
            meme_email | meme_ip,
        )
        par_identifiant_et_ip, par_ip, par_identifiant = (
            await self._session.execute(requete)
        ).one()
        return FailureCounts(
            per_identifier_and_ip=par_identifiant_et_ip,
            per_ip=par_ip,
            per_identifier=par_identifiant,
        )
@@ -1,106 +0,0 @@
 # Piège : `claim_for_rotation()` est une seule instruction. Un SELECT puis un UPDATE
 # laisseraient une fenêtre où deux onglets réussissent la même rotation. Zéro ligne retournée
 # signifie donc, sans ambiguïté, que le jeton était déjà tourné, révoqué, expiré ou inconnu, et
 # c'est `inspect()` qui départage ensuite ces cas.
 from dataclasses import dataclass
 from datetime import datetime
 from uuid import UUID
 from sqlalchemy import func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.refresh_token import RefreshToken, RevocationReason
@dataclass(frozen=True, slots=True)
 class ClaimedToken:
    id: UUID
    family_id: UUID
    user_id: UUID
    expires_at: datetime
 class RefreshTokenRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session
    async def create(
        self,
        *,
        user_id: UUID,
        family_id: UUID,
        token_hash: bytes,
        expires_at: datetime,
        client_ip: str | None,
        user_agent: str | None,
    ) -> RefreshToken:
        jeton = RefreshToken(
            user_id=user_id,
            family_id=family_id,
            token_hash=token_hash,
            expires_at=expires_at,
            client_ip=client_ip,
            user_agent=user_agent,
        )
        self._session.add(jeton)
        await self._session.flush()
        return jeton
    async def claim_for_rotation(self, token_hash: bytes) -> ClaimedToken | None:
        requete = (
            update(RefreshToken)
            .where(
                RefreshToken.token_hash == token_hash,
                RefreshToken.rotated_at.is_(None),
                RefreshToken.revoked_at.is_(None),
                RefreshToken.expires_at > func.clock_timestamp(),
            )
            .values(
                rotated_at=func.clock_timestamp(),
                revoked_at=func.clock_timestamp(),
                revoked_reason=RevocationReason.ROTATION.value,
            )
            .returning(
                RefreshToken.id,
                RefreshToken.family_id,
                RefreshToken.user_id,
                RefreshToken.expires_at,
            )
        )
        ligne = (await self._session.execute(requete)).one_or_none()
        if ligne is None:
            return None
        return ClaimedToken(
            id=ligne.id,
            family_id=ligne.family_id,
            user_id=ligne.user_id,
            expires_at=ligne.expires_at,
        )
    async def inspect(self, token_hash: bytes) -> RefreshToken | None:
        requete = select(RefreshToken).where(RefreshToken.token_hash == token_hash)
        return (await self._session.execute(requete)).scalar_one_or_none()
    async def link_replacement(self, ancien_id: UUID, nouveau_id: UUID) -> None:
        await self._session.execute(
            update(RefreshToken).where(RefreshToken.id == ancien_id).values(replaced_by=nouveau_id)
        )
    async def revoke_family(self, family_id: UUID, reason: RevocationReason) -> int:
        resultat = await self._session.execute(
            update(RefreshToken)
            .where(RefreshToken.family_id == family_id, RefreshToken.revoked_at.is_(None))
            .values(revoked_at=func.clock_timestamp(), revoked_reason=reason.value)
            .returning(RefreshToken.id)
        )
        return len(resultat.all())
    async def revoke_all_for_user(self, user_id: UUID, reason: RevocationReason) -> int:
        resultat = await self._session.execute(
            update(RefreshToken)
            .where(RefreshToken.user_id == user_id, RefreshToken.revoked_at.is_(None))
            .values(revoked_at=func.clock_timestamp(), revoked_reason=reason.value)
            .returning(RefreshToken.id)
        )
        return len(resultat.all())
@@ -1,97 +0,0 @@
 # Piège : `set_role()` et `set_active()` avancent `credentials_changed_at`. C'est ce qui rend
 # un changement de rôle ou une désactivation effectifs à la requête suivante au lieu d'attendre
 # l'expiration du jeton d'accès. Une mise à jour qui l'oublierait laisserait 15 minutes de
 # privilèges périmés.
 from collections.abc import Sequence
 from uuid import UUID
 from sqlalchemy import func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.roles import AccountKind, Role
 from app.models.user import AppUser
 class UserRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session
    async def get_by_email(self, email: str) -> AppUser | None:
        requete = select(AppUser).where(AppUser.email == email.strip().lower())
        return (await self._session.execute(requete)).scalar_one_or_none()
    async def get_by_id(self, user_id: UUID) -> AppUser | None:
        return await self._session.get(AppUser, user_id)
    async def list_all(self) -> Sequence[AppUser]:
        requete = select(AppUser).order_by(AppUser.email)
        return (await self._session.execute(requete)).scalars().all()
    async def count_active_admins(self) -> int:
        requete = (
            select(func.count())
            .select_from(AppUser)
            .where(AppUser.role == Role.ADMIN.value, AppUser.is_active.is_(True))
        )
        return (await self._session.execute(requete)).scalar_one()
    async def create(
        self,
        *,
        email: str,
        password_hash: str,
        role: Role,
        kind: AccountKind = AccountKind.HUMAIN,
        full_name: str | None = None,
        must_change_password: bool = False,
    ) -> AppUser:
        compte = AppUser(
            email=email.strip().lower(),
            password_hash=password_hash,
            role=role.value,
            kind=kind.value,
            full_name=full_name,
            must_change_password=must_change_password,
        )
        self._session.add(compte)
        await self._session.flush()
        return compte
    async def update_password(
        self, user_id: UUID, password_hash: str, *, must_change_password: bool
    ) -> None:
        await self._session.execute(
            update(AppUser)
            .where(AppUser.id == user_id)
            .values(
                password_hash=password_hash,
                must_change_password=must_change_password,
                credentials_changed_at=func.clock_timestamp(),
            )
        )
    async def rehash_password(self, user_id: UUID, password_hash: str) -> None:
        # Un simple recalcul avec des paramètres Argon2 plus récents ne périme aucun jeton.
        await self._session.execute(
            update(AppUser).where(AppUser.id == user_id).values(password_hash=password_hash)
        )
    async def touch_last_login(self, user_id: UUID) -> None:
        await self._session.execute(
            update(AppUser).where(AppUser.id == user_id).values(last_login_at=func.now())
        )
    async def set_role(self, user_id: UUID, role: Role) -> None:
        await self._session.execute(
            update(AppUser)
            .where(AppUser.id == user_id)
            .values(role=role.value, credentials_changed_at=func.clock_timestamp())
        )
    async def set_active(self, user_id: UUID, *, is_active: bool) -> None:
        await self._session.execute(
            update(AppUser)
            .where(AppUser.id == user_id)
            .values(is_active=is_active, credentials_changed_at=func.clock_timestamp())
        )
@@ -1,3 +0,0 @@
 from app.schemas.health import LivenessStatus, ReadinessStatus
 __all__ = ["LivenessStatus", "ReadinessStatus"]
@@ -1,44 +0,0 @@
 # Contrainte : le mot de passe est borné à 128 caractères. Sans plafond, une chaîne de dix
 # mégaoctets ferait travailler Argon2 gratuitement, à la charge du serveur.
 from typing import Literal, Self
 from uuid import UUID
 from pydantic import BaseModel, ConfigDict, EmailStr, Field
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 PASSWORD_MIN_LENGTH = 12
 PASSWORD_MAX_LENGTH = 128
 class LoginRequest(BaseModel):
    email: EmailStr
    password: str = Field(min_length=1, max_length=PASSWORD_MAX_LENGTH)
 class PasswordChangeRequest(BaseModel):
    current_password: str = Field(min_length=1, max_length=PASSWORD_MAX_LENGTH)
    new_password: str = Field(min_length=PASSWORD_MIN_LENGTH, max_length=PASSWORD_MAX_LENGTH)
 class PrincipalResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    email: str
    role: Role
    kind: AccountKind
    must_change_password: bool
    @classmethod
    def from_principal(cls, principal: Principal) -> Self:
        return cls.model_validate(principal)
 class TokenResponse(BaseModel):
    access_token: str
    token_type: Literal["bearer"] = "bearer"  # noqa: S105
    expires_in: int
    principal: PrincipalResponse
@@ -1,19 +0,0 @@
 from typing import Literal
 from pydantic import BaseModel
 class LivenessStatus(BaseModel):
    status: Literal["ok"]
    service: str
    version: str
    environment: str
 # Contrainte : la sonde ne publie pas la version de TimescaleDB. Une version exacte de
 # composant, servie sans authentification, est de la reconnaissance gratuite pour qui
 # cherche une CVE. Elle part dans le journal, où elle sert au diagnostic.
 class ReadinessStatus(BaseModel):
    status: Literal["ready"]
    database: Literal["reachable"]
    timescaledb: Literal["loaded"]
@@ -1,41 +0,0 @@
 # Contrainte : les schémas de lecture et d'écriture sont séparés. Un modèle unique laisserait
 # passer `role` ou `is_active` depuis un corps de requête, et renverrait `password_hash` en
 # réponse. C'est l'attribution de masse, API3 du top 10 API.
 from datetime import datetime
 from uuid import UUID
 from pydantic import BaseModel, ConfigDict, EmailStr, Field
 from app.core.roles import AccountKind, Role
 class UserCreateRequest(BaseModel):
    email: EmailStr
    role: Role
    full_name: str | None = Field(default=None, max_length=200)
 class UserUpdateRequest(BaseModel):
    role: Role | None = None
    is_active: bool | None = None
 class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    email: str
    role: Role
    kind: AccountKind
    is_active: bool
    must_change_password: bool
    full_name: str | None
    last_login_at: datetime | None
    created_at: datetime
 class TemporaryPasswordResponse(BaseModel):
    # Affiché une seule fois : l'empreinte seule est conservée côté serveur.
    user: UserResponse
    temporary_password: str
@@ -1,322 +0,0 @@
 # Piège : les compteurs de limitation sont lus AVANT le hachage Argon2. Dans l'autre ordre,
 # chaque requête rejetée coûterait quand même 17 ms de processeur et 19 Mio de mémoire, et la
 # protection deviendrait l'amplificateur de déni de service qu'elle est censée empêcher.
 # Piège : quand l'email est inconnu, `verify_dummy()` consomme le même temps qu'une
 # vérification réelle. Sans lui, l'écart de temps de réponse est un oracle d'existence.
 # Piège : la tentative échouée est validée en base AVANT que l'erreur ne soit levée.
 # `get_session()` ne valide pas de lui-même, donc la preuve disparaîtrait avec la transaction.
 # Piège : dans `refresh()`, un jeton expiré ne révoque PAS la famille, un jeton déjà tourné si.
 # La rotation ne protège de rien par elle-même : elle rend la réutilisation détectable, et
 # c'est la détection qui termine le vol.
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import NoReturn, Protocol
 from uuid import UUID, uuid4
 from app.core.hashing import Argon2Hasher
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 from app.core.security import (
    TokenPolicy,
    encode_access_token,
    fingerprint_refresh,
    generate_refresh_secret,
 )
 from app.models.audit_log import AuditAction, AuditOutcome
 from app.models.login_attempt import LoginOutcome
 from app.models.refresh_token import RevocationReason
 from app.repositories.audit_log import AuditLogRepository
 from app.repositories.login_attempt import LoginAttemptRepository
 from app.repositories.refresh_token import RefreshTokenRepository
 from app.repositories.user import UserRepository
 class Transaction(Protocol):
    async def commit(self) -> None: ...
 class AuthError(Exception):
    pass
 class InvalidCredentialsError(AuthError):
    pass
 class SessionRejectedError(AuthError):
    pass
 class RateLimitedError(AuthError):
    def __init__(self, retry_after: int) -> None:
        super().__init__("Trop de tentatives")
        self.retry_after = retry_after
@dataclass(frozen=True, slots=True)
 class LoginPolicy:
    window_seconds: int
    max_failures_per_identifier_and_ip: int
    max_failures_per_ip: int
    max_failures_per_identifier: int
@dataclass(frozen=True, slots=True)
 class AuthenticatedSession:
    principal: Principal
    access_token: str
    expires_in: int
    refresh_secret: str
 class AuthService:
    def __init__(
        self,
        *,
        users: UserRepository,
        attempts: LoginAttemptRepository,
        refresh_tokens: RefreshTokenRepository,
        audit: AuditLogRepository,
        hasher: Argon2Hasher,
        transaction: Transaction,
        token_policy: TokenPolicy,
        login_policy: LoginPolicy,
        refresh_ttl: timedelta,
    ) -> None:
        self._users = users
        self._attempts = attempts
        self._refresh = refresh_tokens
        self._audit = audit
        self._hasher = hasher
        self._transaction = transaction
        self._token_policy = token_policy
        self._login_policy = login_policy
        self._refresh_ttl = refresh_ttl
    async def authenticate(
        self, *, email: str, password: str, client_ip: str | None, user_agent: str | None
    ) -> AuthenticatedSession:
        await self._refuse_si_limite(email=email, client_ip=client_ip, user_agent=user_agent)
        compte = await self._users.get_by_email(email)
        if compte is None:
            await self._hasher.verify_dummy()
            await self._echoue(email, client_ip, LoginOutcome.IDENTIFIANTS_INVALIDES)
        if not await self._hasher.verify(compte.password_hash, password):
            await self._echoue(
                email, client_ip, LoginOutcome.IDENTIFIANTS_INVALIDES, user_id=compte.id
            )
        if not compte.is_active or compte.kind != AccountKind.HUMAIN.value:
            await self._echoue(
                email, client_ip, LoginOutcome.COMPTE_INDISPONIBLE, user_id=compte.id
            )
        if self._hasher.needs_rehash(compte.password_hash):
            await self._users.rehash_password(compte.id, await self._hasher.hash(password))
        await self._users.touch_last_login(compte.id)
        await self._attempts.record(
            email=email, client_ip=client_ip, outcome=LoginOutcome.SUCCES, user_id=compte.id
        )
        secret = await self._ouvre_une_famille(
            user_id=compte.id, client_ip=client_ip, user_agent=user_agent
        )
        await self._transaction.commit()
        return self._session(self._en_principal(compte), secret)
    async def refresh(
        self, *, secret: str, client_ip: str | None, user_agent: str | None
    ) -> AuthenticatedSession:
        empreinte = fingerprint_refresh(secret)
        revendique = await self._refresh.claim_for_rotation(empreinte)
        if revendique is None:
            await self._traite_rotation_refusee(empreinte, client_ip, user_agent)
        compte = await self._users.get_by_id(revendique.user_id)
        if compte is None or not compte.is_active:
            await self._refresh.revoke_family(revendique.family_id, RevocationReason.ADMINISTRATION)
            await self._transaction.commit()
            raise SessionRejectedError("Session révoquée")
        nouveau_secret = generate_refresh_secret()
        nouveau = await self._refresh.create(
            user_id=revendique.user_id,
            family_id=revendique.family_id,
            token_hash=fingerprint_refresh(nouveau_secret),
            expires_at=revendique.expires_at,
            client_ip=client_ip,
            user_agent=user_agent,
        )
        await self._refresh.link_replacement(revendique.id, nouveau.id)
        await self._transaction.commit()
        return self._session(self._en_principal(compte), nouveau_secret)
    async def logout(self, *, secret: str) -> None:
        ligne = await self._refresh.inspect(fingerprint_refresh(secret))
        if ligne is not None:
            await self._refresh.revoke_family(ligne.family_id, RevocationReason.DECONNEXION)
        await self._transaction.commit()
    async def change_password(
        self,
        *,
        principal: Principal,
        current_password: str,
        new_password: str,
        client_ip: str | None,
        user_agent: str | None,
    ) -> AuthenticatedSession:
        compte = await self._users.get_by_id(principal.id)
        if compte is None or not await self._hasher.verify(compte.password_hash, current_password):
            raise InvalidCredentialsError("Identifiants invalides")
        await self._users.update_password(
            principal.id, await self._hasher.hash(new_password), must_change_password=False
        )
        # Toutes les sessions tombent, puis on en rouvre une : l'appareil courant reste
        # connecté et tous les autres sont déconnectés.
        revoquees = await self._refresh.revoke_all_for_user(
            principal.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE
        )
        secret = await self._ouvre_une_famille(
            user_id=principal.id, client_ip=client_ip, user_agent=user_agent
        )
        await self._audit.record(
            action=AuditAction.COMPTE_MOT_DE_PASSE_CHANGE,
            actor=principal,
            target_type="app_user",
            target_id=str(principal.id),
            client_ip=client_ip,
            user_agent=user_agent,
            detail={"sessions_revoquees": revoquees},
        )
        await self._transaction.commit()
        rafraichi = await self._users.get_by_id(principal.id)
        return self._session(self._en_principal(rafraichi or compte), secret)
    async def logout_all(self, principal: Principal) -> int:
        revoquees = await self._refresh.revoke_all_for_user(
            principal.id, RevocationReason.DECONNEXION
        )
        await self._audit.record(
            action=AuditAction.SESSIONS_REVOQUEES,
            actor=principal,
            detail={"sessions_revoquees": revoquees},
        )
        await self._transaction.commit()
        return revoquees
    def _session(self, principal: Principal, refresh_secret: str) -> AuthenticatedSession:
        jeton = encode_access_token(
            self._token_policy,
            subject=principal.id,
            role=principal.role.value,
            kind=principal.kind.value,
        )
        return AuthenticatedSession(
            principal=principal,
            access_token=jeton,
            expires_in=int(self._token_policy.access_ttl.total_seconds()),
            refresh_secret=refresh_secret,
        )
    def _en_principal(self, compte: object) -> Principal:
        return Principal(
            id=compte.id,  # type: ignore[attr-defined]
            email=compte.email,  # type: ignore[attr-defined]
            role=Role(compte.role),  # type: ignore[attr-defined]
            kind=AccountKind(compte.kind),  # type: ignore[attr-defined]
            must_change_password=compte.must_change_password,  # type: ignore[attr-defined]
        )
    async def _ouvre_une_famille(
        self, *, user_id: UUID, client_ip: str | None, user_agent: str | None
    ) -> str:
        secret = generate_refresh_secret()
        await self._refresh.create(
            user_id=user_id,
            family_id=uuid4(),
            token_hash=fingerprint_refresh(secret),
            expires_at=datetime.now(UTC) + self._refresh_ttl,
            client_ip=client_ip,
            user_agent=user_agent,
        )
        return secret
    async def _traite_rotation_refusee(
        self, empreinte: bytes, client_ip: str | None, user_agent: str | None
    ) -> NoReturn:
        ligne = await self._refresh.inspect(empreinte)
        if ligne is None:
            raise SessionRejectedError("Session inconnue")
        if ligne.expires_at <= datetime.now(UTC):
            raise SessionRejectedError("Session expirée")
        # Présenter un jeton déjà tourné est une preuve de compromission, pas un accident : toute
        # la famille tombe, y compris la session encore vivante du voleur ou de la victime.
        revoquees = await self._refresh.revoke_family(
            ligne.family_id, RevocationReason.REUTILISATION
        )
        await self._audit.record(
            action=AuditAction.REFRESH_REUTILISE,
            outcome=AuditOutcome.ECHEC,
            target_type="refresh_token",
            target_id=str(ligne.family_id),
            client_ip=client_ip,
            user_agent=user_agent,
            detail={"famille": str(ligne.family_id), "sessions_revoquees": revoquees},
        )
        await self._transaction.commit()
        raise SessionRejectedError("Session révoquée")
    async def _refuse_si_limite(
        self, *, email: str, client_ip: str | None, user_agent: str | None
    ) -> None:
        politique = self._login_policy
        compteurs = await self._attempts.count_recent_failures(
            email=email, client_ip=client_ip, window_seconds=politique.window_seconds
        )
        depasse = (
            compteurs.per_identifier_and_ip >= politique.max_failures_per_identifier_and_ip
            or compteurs.per_ip >= politique.max_failures_per_ip
            or compteurs.per_identifier >= politique.max_failures_per_identifier
        )
        if not depasse:
            return
        await self._attempts.record(email=email, client_ip=client_ip, outcome=LoginOutcome.LIMITE)
        # Un blocage déclenché par l'identifiant seul signe une attaque distribuée : lui seul
        # mérite une trace durable, les échecs ordinaires restent dans `login_attempt`.
        if compteurs.per_identifier >= politique.max_failures_per_identifier:
            await self._audit.record(
                action=AuditAction.LIMITE_PAR_IDENTIFIANT,
                outcome=AuditOutcome.ECHEC,
                actor_label=email.strip().lower(),
                client_ip=client_ip,
                user_agent=user_agent,
                detail={"motif": "seuil par identifiant depasse"},
            )
        await self._transaction.commit()
        raise RateLimitedError(politique.window_seconds)
    async def _echoue(
        self,
        email: str,
        client_ip: str | None,
        outcome: LoginOutcome,
        *,
        user_id: UUID | None = None,
    ) -> NoReturn:
        await self._attempts.record(
            email=email, client_ip=client_ip, outcome=outcome, user_id=user_id
        )
        await self._transaction.commit()
        raise InvalidCredentialsError("Identifiants invalides")
@@ -1,164 +0,0 @@
 # Piège : `change_role()` et `set_active()` refusent de toucher au dernier administrateur actif.
 # Sans cette garde, un administrateur peut se rétrograder ou se désactiver lui-même, et plus
 # personne ne peut administrer la plateforme sans repasser par `psql`.
 import secrets
 from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Protocol
 from uuid import UUID
 from app.core.hashing import Argon2Hasher
 from app.core.principal import Principal
 from app.core.roles import Role
 from app.models.audit_log import AuditAction
 from app.models.refresh_token import RevocationReason
 from app.models.user import AppUser
 from app.repositories.audit_log import AuditLogRepository
 from app.repositories.refresh_token import RefreshTokenRepository
 from app.repositories.user import UserRepository
 LONGUEUR_MOT_DE_PASSE_TEMPORAIRE = 18
 class Transaction(Protocol):
    async def commit(self) -> None: ...
 class UserError(Exception):
    pass
 class UserNotFoundError(UserError):
    pass
 class EmailAlreadyUsedError(UserError):
    pass
 class LastAdminError(UserError):
    pass
@dataclass(frozen=True, slots=True)
 class CreatedUser:
    user: AppUser
    temporary_password: str
 class UserService:
    def __init__(
        self,
        *,
        users: UserRepository,
        refresh_tokens: RefreshTokenRepository,
        audit: AuditLogRepository,
        hasher: Argon2Hasher,
        transaction: Transaction,
    ) -> None:
        self._users = users
        self._refresh = refresh_tokens
        self._audit = audit
        self._hasher = hasher
        self._transaction = transaction
    async def list_all(self) -> Sequence[AppUser]:
        return await self._users.list_all()
    async def create(
        self, *, actor: Principal, email: str, role: Role, full_name: str | None
    ) -> CreatedUser:
        if await self._users.get_by_email(email) is not None:
            raise EmailAlreadyUsedError(email)
        provisoire = secrets.token_urlsafe(LONGUEUR_MOT_DE_PASSE_TEMPORAIRE)
        compte = await self._users.create(
            email=email,
            password_hash=await self._hasher.hash(provisoire),
            role=role,
            full_name=full_name,
            must_change_password=True,
        )
        await self._audit.record(
            action=AuditAction.COMPTE_CREE,
            actor=actor,
            target_type="app_user",
            target_id=str(compte.id),
            detail={"email": compte.email, "role_apres": role.value},
        )
        await self._transaction.commit()
        return CreatedUser(user=compte, temporary_password=provisoire)
    async def change_role(self, *, actor: Principal, user_id: UUID, role: Role) -> AppUser:
        compte = await self._exige(user_id)
        if compte.role == role.value:
            return compte
        await self._refuse_si_dernier_admin(compte, futur_role=role, futur_actif=compte.is_active)
        avant = compte.role
        await self._users.set_role(user_id, role)
        await self._refresh.revoke_all_for_user(user_id, RevocationReason.ADMINISTRATION)
        await self._audit.record(
            action=AuditAction.COMPTE_ROLE_CHANGE,
            actor=actor,
            target_type="app_user",
            target_id=str(user_id),
            detail={"role_avant": avant, "role_apres": role.value},
        )
        await self._transaction.commit()
        return await self._exige(user_id)
    async def set_active(self, *, actor: Principal, user_id: UUID, is_active: bool) -> AppUser:
        compte = await self._exige(user_id)
        if compte.is_active == is_active:
            return compte
        await self._refuse_si_dernier_admin(
            compte, futur_role=Role(compte.role), futur_actif=is_active
        )
        await self._users.set_active(user_id, is_active=is_active)
        if not is_active:
            await self._refresh.revoke_all_for_user(user_id, RevocationReason.ADMINISTRATION)
        await self._audit.record(
            action=AuditAction.COMPTE_ACTIVE if is_active else AuditAction.COMPTE_DESACTIVE,
            actor=actor,
            target_type="app_user",
            target_id=str(user_id),
        )
        await self._transaction.commit()
        return await self._exige(user_id)
    async def reset_password(self, *, actor: Principal, user_id: UUID) -> CreatedUser:
        compte = await self._exige(user_id)
        provisoire = secrets.token_urlsafe(LONGUEUR_MOT_DE_PASSE_TEMPORAIRE)
        await self._users.update_password(
            user_id, await self._hasher.hash(provisoire), must_change_password=True
        )
        await self._refresh.revoke_all_for_user(user_id, RevocationReason.CHANGEMENT_MOT_DE_PASSE)
        await self._audit.record(
            action=AuditAction.COMPTE_MOT_DE_PASSE_REINITIALISE,
            actor=actor,
            target_type="app_user",
            target_id=str(user_id),
            detail={"email": compte.email},
        )
        await self._transaction.commit()
        return CreatedUser(user=await self._exige(user_id), temporary_password=provisoire)
    async def _exige(self, user_id: UUID) -> AppUser:
        compte = await self._users.get_by_id(user_id)
        if compte is None:
            raise UserNotFoundError(str(user_id))
        return compte
    async def _refuse_si_dernier_admin(
        self, compte: AppUser, *, futur_role: Role, futur_actif: bool
    ) -> None:
        etait_admin = compte.role == Role.ADMIN.value and compte.is_active
        reste_admin = futur_role is Role.ADMIN and futur_actif
        if not etait_admin or reste_admin:
            return
        if await self._users.count_active_admins() <= 1:
            raise LastAdminError(str(compte.id))
@@ -1,96 +0,0 @@
 [project]
 name = "enervision-backend"
 version = "0.1.0"
 description = "API EnerVision : exposition des series temporelles energetiques"
 requires-python = ">=3.14,<3.15"
 dependencies = [
    "fastapi>=0.141.1",
    "uvicorn[standard]>=0.53.0",
    "pydantic[email]>=2.13.5",
    "pydantic-settings>=2.15.0",
    "sqlalchemy[asyncio]>=2.0.52",
    "asyncpg>=0.31.0",
    "alembic>=1.20.0",
    "prometheus-fastapi-instrumentator>=8.1.0",
    "python-json-logger>=4.2.0",
    "pyjwt>=2.10",
    "argon2-cffi>=23.1",
    "anyio>=4.0",
 ]
 [dependency-groups]
 dev = [
    "ruff>=0.16.7",
    "mypy>=2.3.1",
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
    "pytest-cov>=7.1.0",
    "httpx>=0.28.1",
 ]
 [build-system]
 requires = ["hatchling>=1.32.0"]
 build-backend = "hatchling.build"
 [tool.hatch.build.targets.wheel]
 packages = ["app"]
 [tool.ruff]
 line-length = 100
 target-version = "py314"
 src = ["app", "tests"]
 extend-exclude = ["alembic/versions"]
 [tool.ruff.lint]
 select = [
    "E", "W",
    "F",
    "I",
    "N",
    "UP",
    "B",
    "C4",
    "SIM",
    "TID",
    "RUF",
    "ASYNC",
    "S",
    "PT",
 ]
 ignore = ["B008"]
 [tool.ruff.lint.per-file-ignores]
 # S105 à S107 signalent les secrets en dur, qui sont justement la matière des tests d'auth.
 "tests/**/*.py" = ["S101", "S105", "S106", "S107"]
 [tool.ruff.lint.isort]
 known-first-party = ["app"]
 [tool.ruff.format]
 quote-style = "double"
 [tool.mypy]
 python_version = "3.14"
 strict = true
 warn_unreachable = true
 plugins = ["pydantic.mypy"]
 exclude = ["^alembic/"]
 [[tool.mypy.overrides]]
 module = ["tests.*"]
 disallow_untyped_defs = false
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 addopts = "-q --strict-markers -m 'not integration' --cov=app --cov-report=term-missing"
 markers = ["integration: requiert une base PostgreSQL joignable, hors `make test`"]
 [tool.coverage.run]
 source = ["app"]
 branch = true
 omit = ["alembic/*"]
 [tool.coverage.report]
 show_missing = true
@@ -1,208 +0,0 @@
 from collections.abc import Iterator
 from uuid import uuid4
 import pytest
 from fastapi import FastAPI
 from httpx import AsyncClient
 from app.api.deps import get_auth_service
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 from app.services.auth import (
    AuthenticatedSession,
    InvalidCredentialsError,
    RateLimitedError,
    SessionRejectedError,
 )
 IDENTIFIANTS = {"email": "operateur@enervision.fr", "password": "un-mot-de-passe-valide"}
 PRINCIPAL = Principal(
    id=uuid4(),
    email="operateur@enervision.fr",
    role=Role.OPERATEUR,
    kind=AccountKind.HUMAIN,
    must_change_password=False,
 )
 class FauxService:
    def __init__(self, erreur: Exception | None = None) -> None:
        self._erreur = erreur
    async def refresh(self, **_: object) -> AuthenticatedSession:
        return await self.authenticate()
    async def logout(self, **_: object) -> None:
        return None
    async def authenticate(self, **_: object) -> AuthenticatedSession:
        if self._erreur is not None:
            raise self._erreur
        return AuthenticatedSession(
            principal=PRINCIPAL,
            access_token="un.jeton.factice",
            expires_in=900,
            refresh_secret="un-secret-opaque",
        )
@pytest.fixture
 def fake_auth_service(app: FastAPI) -> Iterator[list[Exception | None]]:
    programme: list[Exception | None] = [None]
    app.dependency_overrides[get_auth_service] = lambda: FauxService(programme[0])
    yield programme
    app.dependency_overrides.pop(get_auth_service, None)
 async def test_login_returns_the_token_and_the_principal_when_credentials_match(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
    assert response.status_code == 200
    corps = response.json()
    assert corps["access_token"] == "un.jeton.factice"
    assert corps["token_type"] == "bearer"
    assert corps["principal"]["role"] == "operateur"
 async def test_login_forbids_intermediaries_from_caching_the_response(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
    assert response.headers["cache-control"] == "no-store"
 async def test_login_never_reveals_which_half_of_the_credentials_was_wrong(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    fake_auth_service[0] = InvalidCredentialsError("Identifiants invalides")
    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
    assert response.status_code == 401
    assert response.json() == {"detail": "Identifiants invalides"}
 async def test_login_returns_429_with_a_retry_after_when_the_rate_limit_is_reached(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    fake_auth_service[0] = RateLimitedError(900)
    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
    assert response.status_code == 429
    assert response.headers["retry-after"] == "900"
@pytest.mark.parametrize(
    "corps",
    [
        {"email": "pas-une-adresse", "password": "un-mot-de-passe-valide"},
        {"email": "operateur@enervision.fr"},
        {"email": "operateur@enervision.fr", "password": "x" * 129},
    ],
    ids=["adresse_invalide", "mot_de_passe_absent", "mot_de_passe_trop_long"],
 )
 async def test_login_rejects_a_malformed_body_without_echoing_the_password(
    fake_auth_service: list[Exception | None], client: AsyncClient, corps: dict[str, str]
 ) -> None:
    response = await client.post("/api/v1/auth/login", json=corps)
    assert response.status_code == 422
    assert "un-mot-de-passe-valide" not in response.text
    assert "x" * 129 not in response.text
 async def test_login_posts_an_http_only_refresh_cookie_scoped_to_the_auth_routes(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
    depose = response.headers["set-cookie"]
    assert depose.startswith("ev_refresh=un-secret-opaque")
    assert "HttpOnly" in depose
    assert "SameSite=strict" in depose
    assert "Path=/api/v1/auth" in depose
 async def test_login_keeps_the_refresh_secret_out_of_the_response_body(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
    assert "un-secret-opaque" not in response.text
 async def test_refresh_returns_401_when_no_cookie_is_presented(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/refresh")
    assert response.status_code == 401
 async def test_refresh_rotates_the_cookie_when_the_session_is_still_valid(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    client.cookies.set("ev_refresh", "un-secret-opaque")
    response = await client.post("/api/v1/auth/refresh")
    assert response.status_code == 200
    assert "ev_refresh=" in response.headers["set-cookie"]
 async def test_refresh_clears_the_cookie_when_the_session_is_rejected(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    fake_auth_service[0] = SessionRejectedError("Session révoquée")
    client.cookies.set("ev_refresh", "un-secret-rejoue")
    response = await client.post("/api/v1/auth/refresh")
    assert response.status_code == 401
    assert 'ev_refresh=""' in response.headers["set-cookie"]
    assert "Path=/api/v1/auth" in response.headers["set-cookie"]
 async def test_logout_answers_204_and_clears_the_cookie(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    client.cookies.set("ev_refresh", "un-secret-opaque")
    response = await client.post("/api/v1/auth/logout")
    assert response.status_code == 204
    assert 'ev_refresh=""' in response.headers["set-cookie"]
 async def test_logout_stays_idempotent_without_a_cookie(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/logout")
    assert response.status_code == 204
@pytest.mark.parametrize(
    "chemin",
    ["/api/v1/auth/refresh", "/api/v1/auth/logout"],
    ids=["rotation", "deconnexion"],
 )
 async def test_a_cookie_bearing_route_refuses_a_foreign_origin(
    fake_auth_service: list[Exception | None], client: AsyncClient, chemin: str
 ) -> None:
    response = await client.post(chemin, headers={"Origin": "https://malveillant.example"})
    assert response.status_code == 403
 async def test_a_cookie_bearing_route_accepts_a_request_without_origin(
    fake_auth_service: list[Exception | None], client: AsyncClient
 ) -> None:
    response = await client.post("/api/v1/auth/logout")
    assert response.status_code != 403
@@ -1,113 +0,0 @@
 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 AdminDep, get_current_principal, require_role
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 CHEMIN_ADMIN = "/api/v1/essai-admin"
 def principal(role: Role = Role.LECTEUR, *, must_change_password: bool = False) -> Principal:
    return Principal(
        id=uuid4(),
        email=f"{role.value}@enervision.fr",
        role=role,
        kind=AccountKind.HUMAIN,
        must_change_password=must_change_password,
    )
@pytest.fixture
 def route_admin(app: FastAPI) -> None:
    @app.get(CHEMIN_ADMIN)
    async def _reserve_aux_admins(acteur: AdminDep) -> dict[str, str]:
        return {"email": acteur.email}
@pytest.fixture
 def connecte(app: FastAPI) -> Iterator[Callable[[Principal], None]]:
    def installe(acteur: Principal) -> None:
        app.dependency_overrides[get_current_principal] = lambda: acteur
    yield installe
    app.dependency_overrides.pop(get_current_principal, None)
 async def test_me_returns_401_when_no_credentials_are_sent(client: AsyncClient) -> None:
    response = await client.get("/api/v1/auth/me")
    assert response.status_code == 401
    assert "Bearer" in response.headers["www-authenticate"]
 async def test_me_returns_401_when_the_token_is_not_readable(client: AsyncClient) -> None:
    response = await client.get(
        "/api/v1/auth/me", headers={"Authorization": "Bearer nimporte.quoi.ici"}
    )
    assert response.status_code == 401
    assert 'error="invalid_token"' in response.headers["www-authenticate"]
 async def test_me_describes_the_connected_account(
    connecte: Callable[[Principal], None], client: AsyncClient
 ) -> None:
    acteur = principal(Role.OPERATEUR)
    connecte(acteur)
    response = await client.get("/api/v1/auth/me")
    assert response.status_code == 200
    assert response.json()["email"] == acteur.email
@pytest.mark.parametrize(
    ("role", "attendu"),
    [(Role.LECTEUR, 403), (Role.OPERATEUR, 403), (Role.ADMIN, 200)],
    ids=["lecteur_refuse", "operateur_refuse", "admin_accepte"],
 )
 async def test_an_admin_route_only_answers_to_an_admin(
    route_admin: None,
    connecte: Callable[[Principal], None],
    client: AsyncClient,
    role: Role,
    attendu: int,
 ) -> None:
    connecte(principal(role))
    response = await client.get(CHEMIN_ADMIN)
    assert response.status_code == attendu
 async def test_a_pending_password_change_blocks_every_business_route(
    route_admin: None, connecte: Callable[[Principal], None], client: AsyncClient
 ) -> None:
    connecte(principal(Role.ADMIN, must_change_password=True))
    response = await client.get(CHEMIN_ADMIN)
    assert response.status_code == 403
    assert response.json()["detail"] == "password_change_required"
 async def test_a_pending_password_change_still_allows_reading_ones_own_account(
    connecte: Callable[[Principal], None], client: AsyncClient
 ) -> None:
    connecte(principal(Role.LECTEUR, must_change_password=True))
    response = await client.get("/api/v1/auth/me")
    assert response.status_code == 200
    assert response.json()["must_change_password"] is True
 def test_require_role_builds_one_guard_per_minimum_level() -> None:
    garde = require_role(Role.OPERATEUR)
    assert callable(garde)
@@ -1,111 +0,0 @@
 import pytest
 from httpx import ASGITransport, AsyncClient
 from httpx import Response as HttpResponse
 from app.main import create_app
 from tests.factories import make_settings
 ORIGINE = "https://enervision.fr"
 async def interroge(
    settings_overrides: dict[str, object], chemin: str, **kwargs: object
 ) -> HttpResponse:
    application = create_app(make_settings(**settings_overrides))
    transport = ASGITransport(app=application)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        return await client.get(chemin, **kwargs)  # type: ignore[arg-type]
@pytest.mark.parametrize(
    ("entete", "valeur"),
    [
        ("x-content-type-options", "nosniff"),
        ("x-frame-options", "DENY"),
        ("referrer-policy", "no-referrer"),
    ],
    ids=["nosniff", "anti_iframe", "referrer"],
 )
 async def test_every_response_carries_the_security_headers(
    client: AsyncClient, entete: str, valeur: str
 ) -> None:
    response = await client.get("/api/v1/health/live")
    assert response.headers[entete] == valeur
 async def test_the_application_never_sets_hsts_itself(client: AsyncClient) -> None:
    response = await client.get("/api/v1/health/live")
    assert "strict-transport-security" not in response.headers
@pytest.mark.parametrize(
    "env",
    ["staging", "prod"],
    ids=["preproduction", "production"],
 )
 async def test_the_documentation_disappears_outside_development(env: str) -> None:
    surcharges = {"env": env, "cors_origins": ORIGINE}
    for chemin in ("/docs", "/openapi.json"):
        assert (await interroge(surcharges, chemin)).status_code == 404
@pytest.mark.parametrize("env", ["local", "dev"], ids=["local", "developpement"])
 async def test_the_documentation_stays_available_while_developing(env: str) -> None:
    surcharges = {"env": env, "cors_origins": ORIGINE}
    assert (await interroge(surcharges, "/openapi.json")).status_code == 200
 async def test_an_explicit_override_can_reopen_the_documentation() -> None:
    surcharges = {"env": "prod", "cors_origins": ORIGINE, "expose_api_docs": True}
    assert (await interroge(surcharges, "/openapi.json")).status_code == 200
 async def test_metrics_stay_open_when_no_token_is_configured(client: AsyncClient) -> None:
    response = await client.get("/metrics")
    assert response.status_code == 200
 async def test_metrics_demand_the_token_once_one_is_configured() -> None:
    surcharges = {"metrics_token": "un-jeton-de-supervision-assez-long"}
    assert (await interroge(surcharges, "/metrics")).status_code == 401
 async def test_metrics_answer_to_the_right_token() -> None:
    surcharges = {"metrics_token": "un-jeton-de-supervision-assez-long"}
    entetes = {"Authorization": "Bearer un-jeton-de-supervision-assez-long"}
    response = await interroge(surcharges, "/metrics", headers=entetes)
    assert response.status_code == 200
 async def test_metrics_refuse_a_token_that_is_almost_right() -> None:
    surcharges = {"metrics_token": "un-jeton-de-supervision-assez-long"}
    entetes = {"Authorization": "Bearer un-jeton-de-supervision-assez-lon"}
    response = await interroge(surcharges, "/metrics", headers=entetes)
    assert response.status_code == 401
 async def test_an_unhandled_error_returns_a_correlation_id_and_no_traceback() -> None:
    application = create_app(make_settings())
    @application.get("/api/v1/essai-panne")
    async def _casse() -> None:
        raise RuntimeError("secret interne de la pile")
    transport = ASGITransport(app=application, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/api/v1/essai-panne")
    assert response.status_code == 500
    assert "secret interne de la pile" not in response.text
    assert response.json()["correlation"]
@@ -1,79 +0,0 @@
 from collections.abc import Callable
 import pytest
 from httpx import AsyncClient
 from sqlalchemy.exc import OperationalError
 async def test_liveness_exposes_service_metadata(client: AsyncClient) -> None:
    response = await client.get("/api/v1/health/live")
    assert response.status_code == 200
    assert response.json() == {
        "status": "ok",
        "service": "EnerVision API",
        "version": "0.1.0",
        "environment": "local",
    }
 async def test_readiness_confirms_the_extension_without_leaking_its_version(
    fake_session: Callable[..., None], client: AsyncClient
 ) -> None:
    fake_session(result="2.22.1")
    response = await client.get("/api/v1/health/ready")
    assert response.status_code == 200
    assert response.json() == {
        "status": "ready",
        "database": "reachable",
        "timescaledb": "loaded",
    }
    assert "2.22.1" not in response.text
 async def test_readiness_returns_503_when_the_extension_is_missing(
    fake_session: Callable[..., None], client: AsyncClient
 ) -> None:
    fake_session(result=None)
    response = await client.get("/api/v1/health/ready")
    assert response.status_code == 503
    assert response.json()["detail"] == "Extension TimescaleDB absente"
@pytest.mark.parametrize(
    "failure",
    [
        OperationalError("SELECT 1", {}, Exception("connexion refusee")),
        ConnectionRefusedError(111, "Connection refused"),
    ],
    ids=["erreur_sqlalchemy", "erreur_reseau_asyncpg"],
 )
 async def test_readiness_returns_503_when_database_is_unreachable(
    fake_session: Callable[..., None], client: AsyncClient, failure: Exception
 ) -> None:
    fake_session(failure=failure)
    response = await client.get("/api/v1/health/ready")
    assert response.status_code == 503
    assert response.json()["detail"] == "Base de données injoignable"
@pytest.mark.parametrize("path", ["/openapi.json", "/metrics"])
 async def test_technical_endpoints_are_served(client: AsyncClient, path: str) -> None:
    assert (await client.get(path)).status_code == 200
@pytest.mark.integration
 async def test_readiness_reaches_the_real_database(client: AsyncClient) -> None:
    response = await client.get("/api/v1/health/ready")
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["status"] == "ready"
    assert body["database"] == "reachable"
    assert body["timescaledb"] == "loaded"
@@ -1,162 +0,0 @@
 # Parcours complet contre la vraie base, sans serveur ni port ouvert. C'est ce fichier qui
 # prouve que le câblage tient : la connexion, la rotation, la détection de réutilisation et la
 # révocation immédiate passent par les vrais dépôts, les vraies transactions et les vrais
 # déclencheurs PostgreSQL.
 import uuid
 from collections.abc import AsyncIterator
 import pytest
 from fastapi import FastAPI
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy import text
 from app.core.hashing import build_hasher
 from app.core.roles import Role
 from app.db.session import get_session_factory
 from app.repositories.user import UserRepository
 pytestmark = pytest.mark.integration
 MOT_DE_PASSE = "un-mot-de-passe-de-recette"
@pytest.fixture
 async def compte_operateur() -> AsyncIterator[str]:
    email = f"parcours-{uuid.uuid4().hex[:12]}@enervision.fr"
    hacheur = build_hasher(time_cost=1, memory_cost_kib=8192, parallelism=1, max_concurrency=2)
    empreinte = await hacheur.hash(MOT_DE_PASSE)
    async with get_session_factory()() as session:
        await UserRepository(session).create(
            email=email, password_hash=empreinte, role=Role.OPERATEUR
        )
        await session.commit()
    yield email
    async with get_session_factory()() as session:
        await session.execute(text("delete from app_user where email = :e"), {"e": email})
        await session.commit()
@pytest.fixture
 async def navigateur(app: FastAPI) -> AsyncIterator[AsyncClient]:
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield client
 async def connecte(navigateur: AsyncClient, email: str) -> dict[str, str]:
    reponse = await navigateur.post(
        "/api/v1/auth/login", json={"email": email, "password": MOT_DE_PASSE}
    )
    assert reponse.status_code == 200, reponse.text
    return {"Authorization": f"Bearer {reponse.json()['access_token']}"}
 async def test_a_full_session_runs_from_login_to_logout(
    compte_operateur: str, navigateur: AsyncClient
 ) -> None:
    entetes = await connecte(navigateur, compte_operateur)
    identite = await navigateur.get("/api/v1/auth/me", headers=entetes)
    rotation = await navigateur.post("/api/v1/auth/refresh")
    deconnexion = await navigateur.post("/api/v1/auth/logout")
    assert identite.status_code == 200
    assert identite.json()["role"] == "operateur"
    assert rotation.status_code == 200
    assert deconnexion.status_code == 204
 async def test_replaying_a_rotated_cookie_kills_the_whole_family(
    compte_operateur: str, navigateur: AsyncClient
 ) -> None:
    await connecte(navigateur, compte_operateur)
    vole = navigateur.cookies["ev_refresh"]
    premiere_rotation = await navigateur.post("/api/v1/auth/refresh")
    vivant = navigateur.cookies["ev_refresh"]
    navigateur.cookies.set("ev_refresh", vole)
    rejeu = await navigateur.post("/api/v1/auth/refresh")
    navigateur.cookies.set("ev_refresh", vivant)
    apres = await navigateur.post("/api/v1/auth/refresh")
    assert premiere_rotation.status_code == 200
    assert rejeu.status_code == 401
    assert apres.status_code == 401, "la session vivante doit tomber avec sa famille"
 async def test_the_reuse_leaves_a_trace_in_the_append_only_audit_log(
    compte_operateur: str, navigateur: AsyncClient
 ) -> None:
    await connecte(navigateur, compte_operateur)
    vole = navigateur.cookies["ev_refresh"]
    await navigateur.post("/api/v1/auth/refresh")
    navigateur.cookies.set("ev_refresh", vole)
    await navigateur.post("/api/v1/auth/refresh")
    async with get_session_factory()() as session:
        traces = await session.scalar(
            text("select count(*) from audit_log where action = 'auth.refresh_reuse_detected'")
        )
    assert traces is not None
    assert traces >= 1
 async def test_disabling_an_account_invalidates_its_access_token_at_once(
    compte_operateur: str, navigateur: AsyncClient
 ) -> None:
    entetes = await connecte(navigateur, compte_operateur)
    avant = await navigateur.get("/api/v1/auth/me", headers=entetes)
    async with get_session_factory()() as session:
        depot = UserRepository(session)
        compte = await depot.get_by_email(compte_operateur)
        assert compte is not None
        await depot.set_active(compte.id, is_active=False)
        await session.commit()
    apres = await navigateur.get("/api/v1/auth/me", headers=entetes)
    assert avant.status_code == 200
    assert apres.status_code == 401, "la révocation doit être immédiate, pas dans 15 minutes"
 async def test_changing_a_role_invalidates_the_token_that_still_carries_the_old_one(
    compte_operateur: str, navigateur: AsyncClient
 ) -> None:
    entetes = await connecte(navigateur, compte_operateur)
    async with get_session_factory()() as session:
        depot = UserRepository(session)
        compte = await depot.get_by_email(compte_operateur)
        assert compte is not None
        await depot.set_role(compte.id, Role.LECTEUR)
        await session.commit()
    apres = await navigateur.get("/api/v1/auth/me", headers=entetes)
    assert apres.status_code == 401
    assert "token_stale" in apres.headers["www-authenticate"]
 async def test_a_failed_login_is_recorded_even_for_an_unknown_address(
    navigateur: AsyncClient,
 ) -> None:
    inconnu = f"inconnu-{uuid.uuid4().hex[:12]}@enervision.fr"
    reponse = await navigateur.post(
        "/api/v1/auth/login", json={"email": inconnu, "password": "peu-importe-ici"}
    )
    async with get_session_factory()() as session:
        tentatives = await session.scalar(
            text("select count(*) from login_attempt where email_tried = :e"), {"e": inconnu}
        )
    assert reponse.status_code == 401
    assert reponse.json() == {"detail": "Identifiants invalides"}
    assert tentatives == 1, "sans cette ligne, le 429 deviendrait un oracle d'existence"
@@ -1,76 +0,0 @@
 # Ce test est le garde-fou de l'autorisation : rendre une route publique oblige à modifier
 # `ROUTES_PUBLIQUES` ci-dessous, ce qui apparaît en clair dans la diff d'une pull request et
 # demande une justification au relecteur.
 # Pourquoi : il interroge réellement chaque route sans jeton au lieu d'inspecter l'arbre de
 # dépendances. L'arbre n'est accessible que par l'API privée de FastAPI, et surtout une route
 # peut porter la bonne dépendance tout en répondant quand même.
 from typing import Any
 import pytest
 from fastapi import FastAPI
 from httpx import AsyncClient
 ROUTES_PUBLIQUES = frozenset(
    {
        ("GET", "/api/v1/health/live"),
        ("GET", "/api/v1/health/ready"),
        ("POST", "/api/v1/auth/login"),
        # Sans cookie, la déconnexion ne fait rien et répond 204 : elle est idempotente.
        ("POST", "/api/v1/auth/logout"),
        ("GET", "/metrics"),
    }
 )
 VALEURS_DE_SUBSTITUTION = "00000000-0000-0000-0000-000000000000"
 STATUTS_DE_REFUS = {401, 403}
 def routes_declarees(app: FastAPI) -> list[tuple[str, str]]:
    schema: dict[str, Any] = app.openapi()
    return [
        (methode.upper(), chemin)
        for chemin, operations in schema["paths"].items()
        for methode in operations
        if methode.upper() in {"GET", "POST", "PATCH", "PUT", "DELETE"}
    ]
 def routes_protegees(app: FastAPI) -> list[tuple[str, str]]:
    return [route for route in routes_declarees(app) if route not in ROUTES_PUBLIQUES]
 def test_the_public_allow_list_has_no_stale_entry(app: FastAPI) -> None:
    declarees = set(routes_declarees(app)) | {("GET", "/metrics")}
    inconnues = ROUTES_PUBLIQUES - declarees
    assert inconnues == set()
 async def test_every_route_rejects_an_anonymous_caller_unless_explicitly_public(
    app: FastAPI, client: AsyncClient
 ) -> None:
    ouvertes: list[tuple[str, str, int]] = []
    for methode, chemin in routes_protegees(app):
        concret = chemin.replace("{user_id}", VALEURS_DE_SUBSTITUTION)
        response = await client.request(methode, concret, json={})
        if response.status_code not in STATUTS_DE_REFUS:
            ouvertes.append((methode, chemin, response.status_code))
    assert ouvertes == []
 async def test_the_declared_routes_are_actually_reachable(app: FastAPI) -> None:
    assert ("POST", "/api/v1/auth/login") in routes_declarees(app)
    assert ("GET", "/api/v1/auth/me") in routes_declarees(app)
@pytest.mark.parametrize(
    "chemin",
    ["/api/v1/health/live", "/api/v1/health/ready"],
    ids=["sonde_de_vie", "sonde_de_disponibilite"],
 )
 def test_the_health_probes_stay_public(app: FastAPI, chemin: str) -> None:
    assert ("GET", chemin) in ROUTES_PUBLIQUES
@@ -1,208 +0,0 @@
 from collections.abc import Callable, Iterator
 from datetime import UTC, datetime
 from uuid import UUID, uuid4
 import pytest
 from fastapi import FastAPI
 from httpx import AsyncClient
 from app.api.deps import get_current_principal, get_user_service
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 from app.services.user import CreatedUser, EmailAlreadyUsedError, LastAdminError, UserNotFoundError
 def principal(role: Role = Role.ADMIN) -> Principal:
    return Principal(
        id=uuid4(),
        email=f"{role.value}@enervision.fr",
        role=role,
        kind=AccountKind.HUMAIN,
        must_change_password=False,
    )
 class FauxCompte:
    def __init__(self, role: Role = Role.LECTEUR) -> None:
        self.id = uuid4()
        self.email = "cible@enervision.fr"
        self.role = role.value
        self.kind = "human"
        self.is_active = True
        self.must_change_password = True
        self.full_name = None
        self.last_login_at: datetime | None = None
        self.created_at = datetime.now(UTC)
 class FauxService:
    def __init__(self, erreur: Exception | None = None) -> None:
        self._erreur = erreur
        self.compte = FauxCompte()
    def _leve(self) -> None:
        if self._erreur is not None:
            raise self._erreur
    async def list_all(self) -> list[FauxCompte]:
        return [self.compte]
    async def create(self, **_: object) -> CreatedUser:
        self._leve()
        return CreatedUser(user=self.compte, temporary_password="mot-de-passe-provisoire")  # type: ignore[arg-type]
    async def change_role(self, **_: object) -> FauxCompte:
        self._leve()
        return self.compte
    async def set_active(self, **_: object) -> FauxCompte:
        self._leve()
        return self.compte
    async def reset_password(self, **_: object) -> CreatedUser:
        self._leve()
        return CreatedUser(user=self.compte, temporary_password="mot-de-passe-provisoire")  # type: ignore[arg-type]
@pytest.fixture
 def administre(app: FastAPI) -> Iterator[Callable[[Exception | None], FauxService]]:
    services: list[FauxService] = []
    def installe(erreur: Exception | None = None) -> FauxService:
        service = FauxService(erreur)
        services.append(service)
        app.dependency_overrides[get_user_service] = lambda: service
        app.dependency_overrides[get_current_principal] = lambda: principal()
        return service
    yield installe
    app.dependency_overrides.pop(get_user_service, None)
    app.dependency_overrides.pop(get_current_principal, None)
@pytest.fixture
 def lecteur_connecte(app: FastAPI) -> Iterator[None]:
    app.dependency_overrides[get_current_principal] = lambda: principal(Role.LECTEUR)
    yield
    app.dependency_overrides.pop(get_current_principal, None)
 async def test_list_users_returns_the_accounts_without_their_digest(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre()
    response = await client.get("/api/v1/users")
    assert response.status_code == 200
    corps = response.json()
    assert "password_hash" not in corps[0]
    assert corps[0]["email"] == "cible@enervision.fr"
 async def test_create_user_returns_the_temporary_password_once(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre()
    response = await client.post(
        "/api/v1/users", json={"email": "nouveau@enervision.fr", "role": "operateur"}
    )
    assert response.status_code == 201
    assert response.json()["temporary_password"] == "mot-de-passe-provisoire"
    assert response.headers["cache-control"] == "no-store"
 async def test_create_user_refuses_an_address_already_taken(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre(EmailAlreadyUsedError("cible@enervision.fr"))
    response = await client.post(
        "/api/v1/users", json={"email": "cible@enervision.fr", "role": "lecteur"}
    )
    assert response.status_code == 409
 async def test_create_user_never_accepts_a_caller_chosen_digest(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre()
    response = await client.post(
        "/api/v1/users",
        json={
            "email": "nouveau@enervision.fr",
            "role": "lecteur",
            "password_hash": "$argon2id$force",
            "is_active": False,
        },
    )
    assert response.status_code == 201
 async def test_update_user_refuses_to_strand_the_last_administrator(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre(LastAdminError("x"))
    response = await client.patch(f"/api/v1/users/{uuid4()}", json={"is_active": False})
    assert response.status_code == 409
 async def test_update_user_returns_404_for_an_unknown_account(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre(UserNotFoundError("x"))
    response = await client.patch(f"/api/v1/users/{uuid4()}", json={"role": "admin"})
    assert response.status_code == 404
 async def test_update_user_refuses_an_empty_body(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre()
    response = await client.patch(f"/api/v1/users/{uuid4()}", json={})
    assert response.status_code == 400
 async def test_reset_password_returns_a_new_temporary_password(
    administre: Callable[..., FauxService], client: AsyncClient
 ) -> None:
    administre()
    response = await client.post(f"/api/v1/users/{uuid4()}/password-reset")
    assert response.status_code == 200
    assert response.json()["temporary_password"] == "mot-de-passe-provisoire"
    assert response.headers["cache-control"] == "no-store"
@pytest.mark.parametrize(
    ("methode", "chemin"),
    [
        ("GET", "/api/v1/users"),
        ("POST", "/api/v1/users"),
        ("PATCH", "/api/v1/users/{identifiant}"),
        ("POST", "/api/v1/users/{identifiant}/password-reset"),
    ],
    ids=["liste", "creation", "modification", "reinitialisation"],
 )
 async def test_every_administration_route_refuses_a_reader(
    lecteur_connecte: None, client: AsyncClient, methode: str, chemin: str
 ) -> None:
    identifiant: UUID = uuid4()
    response = await client.request(
        methode, chemin.format(identifiant=identifiant), json={"role": "admin"}
    )
    assert response.status_code == 403
@@ -1,74 +0,0 @@
 import os
 from collections.abc import AsyncIterator, Callable, Iterator
 import pytest
 from fastapi import FastAPI
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.config import get_settings
 from app.db.session import get_engine, get_session, get_session_factory
 from app.main import create_app
 from tests.factories import FakeSession
 # Piège : les variables d'environnement priment sur apps/backend/.env. Celles qu'on ne
 # pose pas ici, c'est le .env du poste qui les décide, et les assertions avec.
@pytest.fixture(autouse=True, scope="session")
 def environment() -> Iterator[None]:
    os.environ.update(
        {
            "APP_ENV": "local",
            "APP_DEBUG": "false",
            "APP_LOG_LEVEL": "WARNING",
            "APP_CORS_ORIGINS": "",
            "APP_SECRET_KEY": "secret-de-test-assez-long-pour-le-validateur",
        }
    )
    os.environ.setdefault(
        "DATABASE_URL", "postgresql+asyncpg://enervision:change_me@localhost:5433/enervision_test"
    )
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()
 # Piège : get_engine est lru_cache et pytest-asyncio ouvre une boucle par test. Sans ce
 # recyclage, le 2e test touchant vraiment la base hériterait d'une boucle morte.
@pytest.fixture(autouse=True)
 async def engine_per_test() -> AsyncIterator[None]:
    yield
    if get_engine.cache_info().currsize:
        await get_engine().dispose()
    get_engine.cache_clear()
    get_session_factory.cache_clear()
@pytest.fixture
 def app() -> FastAPI:
    return create_app()
@pytest.fixture
 async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as async_client:
        yield async_client
@pytest.fixture
 def fake_session(app: FastAPI) -> Callable[..., None]:
    def install(result: object = None, failure: Exception | None = None) -> None:
        async def override() -> AsyncIterator[FakeSession]:
            yield FakeSession(result=result, failure=failure)
        app.dependency_overrides[get_session] = override
    return install
 # Contrainte : ouvre une vraie connexion, donc réservée aux tests `integration`.
@pytest.fixture
 async def session() -> AsyncIterator[AsyncSession]:
    async with get_session_factory()() as async_session:
        yield async_session
@@ -1,78 +0,0 @@
 import pytest
 from pydantic import ValidationError
 from tests.factories import make_settings
 SECRET_VALIDE = "un-secret-de-test-de-plus-de-trente-deux-caracteres"
@pytest.mark.parametrize(
    "surcharges",
    [
        {"secret_key": "trop-court"},
        {"secret_key": "change_me"},
        {"env": "prod", "debug": True, "cors_origins": "https://enervision.fr"},
        {"cors_origins": "*"},
        {"env": "prod", "cors_origins": ""},
        {"cookie_samesite": "none", "cookie_secure": False},
    ],
    ids=[
        "secret_trop_court",
        "secret_sentinelle",
        "debug_en_production",
        "joker_dans_les_origines",
        "origines_vides_hors_local",
        "samesite_none_sans_secure",
    ],
 )
 def test_settings_refuses_to_build_when_the_configuration_is_unsafe(
    surcharges: dict[str, object],
 ) -> None:
    with pytest.raises(ValidationError):
        make_settings(**surcharges)
 def test_settings_accepts_debug_in_local_environment() -> None:
    settings = make_settings(env="local", debug=True)
    assert settings.debug is True
@pytest.mark.parametrize(
    ("env", "attendu"),
    [("local", False), ("dev", True), ("staging", True), ("prod", True)],
    ids=["local", "dev", "staging", "production"],
 )
 def test_cookies_are_secure_follows_the_environment(env: str, attendu: bool) -> None:
    settings = make_settings(env=env, cors_origins="https://enervision.fr")
    assert settings.cookies_are_secure is attendu
 def test_cookies_are_secure_honours_an_explicit_override() -> None:
    settings = make_settings(env="prod", cors_origins="https://enervision.fr", cookie_secure=False)
    assert settings.cookies_are_secure is False
@pytest.mark.parametrize(
    ("env", "attendu"),
    [("local", True), ("dev", True), ("staging", False), ("prod", False)],
    ids=["local", "dev", "staging", "production"],
 )
 def test_api_docs_are_exposed_closes_staging_and_production(env: str, attendu: bool) -> None:
    settings = make_settings(env=env, cors_origins="https://enervision.fr")
    assert settings.api_docs_are_exposed is attendu
 def test_api_docs_are_exposed_honours_an_explicit_override() -> None:
    settings = make_settings(env="prod", cors_origins="https://enervision.fr", expose_api_docs=True)
    assert settings.api_docs_are_exposed is True
 def test_allowed_origins_splits_and_trims_the_list() -> None:
    settings = make_settings(cors_origins=" http://localhost:4200 , https://enervision.fr ")
    assert settings.allowed_origins == ["http://localhost:4200", "https://enervision.fr"]
@@ -1,58 +0,0 @@
 from app.core.cookies import RefreshCookie, cookie_name
 from tests.factories import make_settings
 def test_build_marks_the_cookie_http_only_and_scopes_it_to_the_auth_routes() -> None:
    settings = make_settings(env="local")
    cookie = RefreshCookie.build(settings, "un-secret-opaque")
    assert cookie.httponly is True
    assert cookie.samesite == "strict"
    assert cookie.path == "/api/v1/auth"
    assert cookie.max_age == settings.refresh_token_ttl_seconds
 def test_build_prefixes_and_secures_the_cookie_outside_local() -> None:
    settings = make_settings(env="prod", cors_origins="https://enervision.fr")
    cookie = RefreshCookie.build(settings, "un-secret-opaque")
    assert cookie.secure is True
    assert cookie.key.startswith("__Secure-")
 def test_build_leaves_the_cookie_unprefixed_in_local() -> None:
    settings = make_settings(env="local")
    cookie = RefreshCookie.build(settings, "un-secret-opaque")
    assert cookie.key == "ev_refresh"
 def test_expired_reuses_the_exact_name_and_path_of_the_posted_cookie() -> None:
    settings = make_settings(env="prod", cors_origins="https://enervision.fr")
    pose = RefreshCookie.build(settings, "un-secret-opaque")
    suppression = RefreshCookie.expired(settings)
    assert suppression.key == pose.key
    assert suppression.path == pose.path
    assert suppression.secure == pose.secure
    assert suppression.samesite == pose.samesite
    assert suppression.max_age == 0
    assert suppression.value == ""
 def test_as_kwargs_matches_the_starlette_set_cookie_signature() -> None:
    settings = make_settings(env="local")
    arguments = RefreshCookie.build(settings, "un-secret-opaque").as_kwargs()
    assert set(arguments) == {"key", "value", "max_age", "path", "secure", "httponly", "samesite"}
 def test_cookie_name_follows_the_configured_name() -> None:
    settings = make_settings(env="local", refresh_cookie_name="autre_nom")
    assert cookie_name(settings) == "autre_nom"
@@ -1,61 +0,0 @@
 from app.core.hashing import Argon2Hasher, build_hasher
 MOT_DE_PASSE = "un-mot-de-passe-de-test-assez-long"
 def fabrique(time_cost: int = 1, max_concurrency: int = 2) -> Argon2Hasher:
    return build_hasher(
        time_cost=time_cost,
        memory_cost_kib=8192,
        parallelism=1,
        max_concurrency=max_concurrency,
    )
 async def test_hash_produces_a_distinct_digest_for_the_same_password() -> None:
    hacheur = fabrique()
    premier = await hacheur.hash(MOT_DE_PASSE)
    second = await hacheur.hash(MOT_DE_PASSE)
    assert premier != second
    assert premier.startswith("$argon2id$")
 async def test_verify_accepts_the_right_password_and_rejects_the_others() -> None:
    hacheur = fabrique()
    empreinte = await hacheur.hash(MOT_DE_PASSE)
    assert await hacheur.verify(empreinte, MOT_DE_PASSE) is True
    assert await hacheur.verify(empreinte, "un-autre-mot-de-passe") is False
 async def test_verify_returns_false_when_the_stored_digest_is_malformed() -> None:
    hacheur = fabrique()
    accorde = await hacheur.verify("pas-une-empreinte-argon2", MOT_DE_PASSE)
    assert accorde is False
 async def test_needs_rehash_is_true_when_the_parameters_changed() -> None:
    ancien = fabrique(time_cost=1)
    recent = fabrique(time_cost=3)
    empreinte = await ancien.hash(MOT_DE_PASSE)
    assert ancien.needs_rehash(empreinte) is False
    assert recent.needs_rehash(empreinte) is True
 def test_needs_rehash_is_true_when_the_stored_digest_is_malformed() -> None:
    hacheur = fabrique()
    assert hacheur.needs_rehash("pas-une-empreinte-argon2") is True
 async def test_verify_dummy_completes_without_revealing_anything() -> None:
    hacheur = fabrique()
    await hacheur.verify_dummy()
@@ -1,72 +0,0 @@
 import logging
 import pytest
 from app.core.logging import CAVIARDAGE, RedactingFilter, redact
@pytest.mark.parametrize(
    "message",
    [
        "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.charge-utile-assez-longue.signature",
        "jeton brut eyJhbGciOiJIUzI1NiJ9abcdefghijklmnopqrstuvwxyz",
        "INSERT ... ('$argon2id$v=19$m=19456,t=2,p=1$sel-en-clair$empreinte-en-clair')",
        '{"password": "le-mot-de-passe-du-client"}',
        "current_password=le-mot-de-passe",
        "Cookie: ev_refresh=abcdefghijklmnopqrstuvwxyz0123456789",
    ],
    ids=[
        "en_tete_bearer",
        "jeton_jwt_nu",
        "empreinte_argon2",
        "mot_de_passe_json",
        "mot_de_passe_en_paire",
        "cookie_de_rafraichissement",
    ],
 )
 def test_redact_removes_every_known_secret_shape(message: str) -> None:
    expurge = redact(message)
    assert CAVIARDAGE in expurge
    for suspect in ("le-mot-de-passe", "empreinte-en-clair", "abcdefghijklmnopqrstuvwxyz"):
        assert suspect not in expurge
 def test_redact_leaves_an_innocent_message_untouched() -> None:
    message = "auth.login.success user_id=3f2a ip=203.0.113.10"
    assert redact(message) == message
 def test_the_filter_rewrites_the_record_before_it_reaches_the_handler() -> None:
    enregistrement = logging.LogRecord(
        name="app",
        level=logging.INFO,
        pathname=__file__,
        lineno=1,
        msg='requete {"password": "%s"}',
        args=("secret-du-client",),
        exc_info=None,
    )
    conserve = RedactingFilter().filter(enregistrement)
    assert conserve is True
    assert "secret-du-client" not in enregistrement.getMessage()
 def test_the_filter_keeps_a_record_that_holds_no_secret() -> None:
    enregistrement = logging.LogRecord(
        name="app",
        level=logging.INFO,
        pathname=__file__,
        lineno=1,
        msg="requete %s",
        args=("/api/v1/health/live",),
        exc_info=None,
    )
    conserve = RedactingFilter().filter(enregistrement)
    assert conserve is True
    assert enregistrement.getMessage() == "requete /api/v1/health/live"
@@ -1,40 +0,0 @@
 import pytest
 from app.core.roles import Role, has_at_least
@pytest.mark.parametrize(
    ("actual", "required", "expected"),
    [
        (Role.LECTEUR, Role.LECTEUR, True),
        (Role.LECTEUR, Role.OPERATEUR, False),
        (Role.LECTEUR, Role.ADMIN, False),
        (Role.OPERATEUR, Role.LECTEUR, True),
        (Role.OPERATEUR, Role.OPERATEUR, True),
        (Role.OPERATEUR, Role.ADMIN, False),
        (Role.ADMIN, Role.LECTEUR, True),
        (Role.ADMIN, Role.OPERATEUR, True),
        (Role.ADMIN, Role.ADMIN, True),
    ],
    ids=[
        "lecteur_sur_lecteur",
        "lecteur_sur_operateur",
        "lecteur_sur_admin",
        "operateur_sur_lecteur",
        "operateur_sur_operateur",
        "operateur_sur_admin",
        "admin_sur_lecteur",
        "admin_sur_operateur",
        "admin_sur_admin",
    ],
 )
 def test_has_at_least_orders_the_three_roles(actual: Role, required: Role, expected: bool) -> None:
    accorde = has_at_least(actual, required)
    assert accorde is expected
 def test_role_values_stay_ascii_for_the_wire_format() -> None:
    valeurs = [role.value for role in Role]
    assert all(valeur.isascii() for valeur in valeurs)
@@ -1,194 +0,0 @@
 import base64
 import json
 from datetime import UTC, datetime, timedelta
 from uuid import uuid4
 import jwt
 import pytest
 from app.core.security import (
    AccessClaims,
    TokenExpiredError,
    TokenInvalidError,
    TokenPolicy,
    decode_access_token,
    encode_access_token,
    fingerprint_refresh,
    generate_refresh_secret,
 )
 POLITIQUE = TokenPolicy(
    secret="un-secret-de-test-de-plus-de-trente-deux-caracteres",
    issuer="enervision-api",
    audience="enervision-web",
    access_ttl=timedelta(minutes=15),
 )
 def emets(**surcharges: object) -> str:
    charge = {
        "iss": POLITIQUE.issuer,
        "aud": POLITIQUE.audience,
        "sub": str(uuid4()),
        "iat": datetime.now(UTC),
        "exp": datetime.now(UTC) + timedelta(minutes=15),
        "jti": str(uuid4()),
        "typ": "access",
        "role": "lecteur",
        "kind": "human",
    }
    charge.update(surcharges)
    return jwt.encode(charge, POLITIQUE.secret, algorithm="HS256")
 def test_decode_access_token_returns_the_claims_when_the_token_is_valid() -> None:
    sujet = uuid4()
    jeton = encode_access_token(POLITIQUE, subject=sujet, role="operateur", kind="human")
    claims = decode_access_token(POLITIQUE, jeton)
    assert isinstance(claims, AccessClaims)
    assert claims.subject == sujet
    assert claims.role == "operateur"
    assert claims.kind == "human"
 def test_decode_access_token_raises_expired_when_the_lifetime_has_passed() -> None:
    passe = datetime.now(UTC) - timedelta(hours=2)
    jeton = encode_access_token(POLITIQUE, subject=uuid4(), role="lecteur", kind="human", now=passe)
    with pytest.raises(TokenExpiredError):
        decode_access_token(POLITIQUE, jeton)
 def test_decode_access_token_raises_invalid_when_the_signature_was_forged() -> None:
    autre = TokenPolicy(
        secret="un-autre-secret-tout-aussi-long-que-le-premier",
        issuer=POLITIQUE.issuer,
        audience=POLITIQUE.audience,
        access_ttl=POLITIQUE.access_ttl,
    )
    jeton = encode_access_token(autre, subject=uuid4(), role="lecteur", kind="human")
    with pytest.raises(TokenInvalidError):
        decode_access_token(POLITIQUE, jeton)
@pytest.mark.parametrize(
    "surcharges",
    [
        {"aud": "un-autre-public"},
        {"iss": "un-autre-emetteur"},
        {"typ": "refresh"},
    ],
    ids=["audience_invalide", "emetteur_invalide", "jeton_de_rafraichissement"],
 )
 def test_decode_access_token_raises_invalid_when_a_claim_is_wrong(
    surcharges: dict[str, object],
 ) -> None:
    jeton = emets(**surcharges)
    with pytest.raises(TokenInvalidError):
        decode_access_token(POLITIQUE, jeton)
@pytest.mark.parametrize(
    "claim",
    ["jti", "typ", "role", "kind"],
    ids=["identifiant", "type", "role", "nature_du_compte"],
 )
 def test_decode_access_token_raises_invalid_when_a_required_claim_is_missing(claim: str) -> None:
    charge = {
        "iss": POLITIQUE.issuer,
        "aud": POLITIQUE.audience,
        "sub": str(uuid4()),
        "iat": datetime.now(UTC),
        "exp": datetime.now(UTC) + timedelta(minutes=15),
        "jti": str(uuid4()),
        "typ": "access",
        "role": "lecteur",
        "kind": "human",
    }
    del charge[claim]
    jeton = jwt.encode(charge, POLITIQUE.secret, algorithm="HS256")
    with pytest.raises(TokenInvalidError):
        decode_access_token(POLITIQUE, jeton)
 def test_decode_access_token_rejects_a_token_forged_with_the_none_algorithm() -> None:
    def encode(donnees: dict[str, object]) -> str:
        brut = json.dumps(donnees, separators=(",", ":")).encode()
        return base64.urlsafe_b64encode(brut).rstrip(b"=").decode()
    entete = encode({"alg": "none", "typ": "JWT"})
    charge = encode(
        {
            "iss": POLITIQUE.issuer,
            "aud": POLITIQUE.audience,
            "sub": str(uuid4()),
            "iat": int(datetime.now(UTC).timestamp()),
            "exp": int((datetime.now(UTC) + timedelta(minutes=15)).timestamp()),
            "jti": str(uuid4()),
            "typ": "access",
            "role": "admin",
            "kind": "human",
        }
    )
    with pytest.raises(TokenInvalidError):
        decode_access_token(POLITIQUE, f"{entete}.{charge}.")
 def test_decode_access_token_rejects_a_token_signed_with_another_algorithm() -> None:
    charge = {
        "iss": POLITIQUE.issuer,
        "aud": POLITIQUE.audience,
        "sub": str(uuid4()),
        "iat": datetime.now(UTC),
        "exp": datetime.now(UTC) + timedelta(minutes=15),
        "jti": str(uuid4()),
        "typ": "access",
        "role": "admin",
        "kind": "human",
    }
    jeton = jwt.encode(charge, POLITIQUE.secret * 2, algorithm="HS512")
    with pytest.raises(TokenInvalidError):
        decode_access_token(POLITIQUE, jeton)
@pytest.mark.parametrize(
    "surcharges",
    [{"sub": "pas-un-uuid"}, {"jti": "pas-un-uuid"}],
    ids=["sujet_illisible", "identifiant_illisible"],
 )
 def test_decode_access_token_raises_invalid_when_an_identifier_is_not_a_uuid(
    surcharges: dict[str, object],
 ) -> None:
    jeton = emets(**surcharges)
    with pytest.raises(TokenInvalidError):
        decode_access_token(POLITIQUE, jeton)
 def test_generate_refresh_secret_returns_distinct_url_safe_values() -> None:
    secrets_generes = {generate_refresh_secret() for _ in range(100)}
    assert len(secrets_generes) == 100
    assert all(len(valeur) >= 43 for valeur in secrets_generes)
 def test_fingerprint_refresh_is_stable_and_distinguishes_two_secrets() -> None:
    premier = generate_refresh_secret()
    second = generate_refresh_secret()
    empreinte = fingerprint_refresh(premier)
    assert len(empreinte) == 32
    assert empreinte == fingerprint_refresh(premier)
    assert empreinte != fingerprint_refresh(second)
@@ -1,261 +0,0 @@
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime
 from uuid import uuid4
 import pytest
 from sqlalchemy import insert, select, text
 from sqlalchemy.engine import make_url
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncConnection, create_async_engine
 from app.core.config import get_settings
 from app.models.energy import Alert, Dataset, Prediction, Reading, Recommendation, Site
 pytestmark = pytest.mark.integration
 MOMENT = datetime(2024, 1, 1, tzinfo=UTC)
@pytest.fixture
 async def data_connection() -> AsyncIterator[AsyncConnection]:
    url = make_url(get_settings().database_url)
    if url.database != "enervision_test":
        pytest.fail("Ces tests exigent DATABASE_URL vers enervision_test.")
    engine = create_async_engine(url)
    try:
        async with engine.connect() as connection:
            transaction = await connection.begin()
            try:
                yield connection
            finally:
                await transaction.rollback()
    finally:
        await engine.dispose()
@pytest.fixture
 async def data_site(data_connection: AsyncConnection) -> str:
    site_id = f"TEST-{uuid4()}"
    await data_connection.execute(
        insert(Site).values(site_id=site_id, site_name="Site de test", site_type="office")
    )
    return site_id
 async def test_reading_is_a_time_hypertable_when_migrated(
    data_connection: AsyncConnection,
 ) -> None:
    query = text(
        "SELECT column_name FROM timescaledb_information.dimensions "
        "WHERE hypertable_schema = 'public' AND hypertable_name = 'reading'"
    )
    result = await data_connection.execute(query)
    assert result.scalars().all() == ["timestamp"]
 async def test_reading_preserves_null_and_zero_when_inserted(
    data_connection: AsyncConnection, data_site: str
 ) -> None:
    statement = insert(Reading).values(
        site_id=data_site,
        timestamp=MOMENT,
        source="api_current",
        consumption_kw=None,
        consumption_kwh=0,
        data_quality="partial",
        null_reasons=["sensor_failure"],
        raw_data={"consumption_kw": None},
        imputed_values=None,
        imputation_method=None,
    )
    await data_connection.execute(statement)
    result = (
        await data_connection.execute(
            select(
                Reading.consumption_kw,
                Reading.consumption_kwh,
                Reading.raw_data,
                Reading.imputed_values,
            ).where(Reading.site_id == data_site)
        )
    ).one()
    assert tuple(result) == (None, 0, {"consumption_kw": None}, None)
@pytest.mark.parametrize("source", ["csv", "api_current", "api_history"])
 async def test_duplicate_reading_is_rejected_when_key_matches(
    data_connection: AsyncConnection, data_site: str, source: str
 ) -> None:
    dataset_id = None
    if source == "csv":
        dataset_id = (
            await data_connection.execute(
                insert(Dataset.__table__)
                .values(
                    dataset_name="Archive de test",
                    archive_sha256=uuid4().hex + uuid4().hex,
                    storage_uri="test://archive",
                    metadata={},
                )
                .returning(Dataset.dataset_id)
            )
        ).scalar_one()
    statement = insert(Reading).values(
        site_id=data_site,
        timestamp=MOMENT,
        source=source,
        dataset_id=dataset_id,
        raw_data={},
    )
    await data_connection.execute(statement)
    with pytest.raises(IntegrityError):
        async with data_connection.begin_nested():
            await data_connection.execute(statement)
@pytest.mark.parametrize(
    "changes",
    [
        {"source": "csv"},
        {"source": "unknown"},
        {"site_id": "UNKNOWN-SITE"},
        {"data_quality": "unknown"},
        {"imputed_values": {"consumption_kw": 12}},
        {"imputation_method": "mean-v1"},
    ],
    ids=[
        "csv_sans_dataset",
        "source_inconnue",
        "site_absent",
        "qualite_inconnue",
        "imputation_sans_methode",
        "methode_sans_imputation",
    ],
 )
 async def test_invalid_reading_is_rejected_when_constraints_fail(
    data_connection: AsyncConnection, data_site: str, changes: dict[str, object]
 ) -> None:
    values: dict[str, object] = {
        "site_id": data_site,
        "timestamp": MOMENT,
        "source": "api_current",
        "raw_data": {},
    }
    values.update(changes)
    with pytest.raises(IntegrityError):
        async with data_connection.begin_nested():
            await data_connection.execute(insert(Reading).values(**values))
 async def test_prediction_requires_period_when_energy_is_predicted(
    data_connection: AsyncConnection, data_site: str
 ) -> None:
    statement = insert(Prediction).values(
        site_id=data_site,
        target_at=MOMENT,
        target_metric="consumption_kwh",
        predicted_value=12,
        status="available",
        model_reference="test-model/1",
    )
    with pytest.raises(IntegrityError):
        async with data_connection.begin_nested():
            await data_connection.execute(statement)
 async def test_unavailable_prediction_preserves_null_when_inserted(
    data_connection: AsyncConnection, data_site: str
 ) -> None:
    statement = (
        insert(Prediction)
        .values(
            site_id=data_site,
            target_at=MOMENT,
            target_metric="consumption_kw",
            status="insufficient_data",
            failure_reason="Historique trop court",
            model_reference="test-model/1",
        )
        .returning(Prediction.predicted_value)
    )
    value = (await data_connection.execute(statement)).scalar_one()
    assert value is None
 async def test_alert_rejects_prediction_when_site_differs(
    data_connection: AsyncConnection, data_site: str
 ) -> None:
    other_site = f"TEST-{uuid4()}"
    await data_connection.execute(
        insert(Site).values(site_id=other_site, site_name="Autre site", site_type="office")
    )
    prediction_id = (
        await data_connection.execute(
            insert(Prediction)
            .values(
                site_id=data_site,
                target_at=MOMENT,
                target_metric="consumption_kw",
                predicted_value=12,
                status="available",
                model_reference="test-model/1",
            )
            .returning(Prediction.prediction_id)
        )
    ).scalar_one()
    with pytest.raises(IntegrityError):
        async with data_connection.begin_nested():
            await data_connection.execute(
                insert(Alert).values(
                    source_alert_id=str(uuid4()),
                    site_id=other_site,
                    source="enervision",
                    timestamp=MOMENT,
                    type="spike",
                    severity="high",
                    message="Test",
                    prediction_id=prediction_id,
                    raw_data={},
                )
            )
 async def test_recommendation_is_unique_when_alert_and_rule_match(
    data_connection: AsyncConnection, data_site: str
 ) -> None:
    alert_id = (
        await data_connection.execute(
            insert(Alert)
            .values(
                source_alert_id=str(uuid4()),
                site_id=data_site,
                source="api_mock",
                timestamp=MOMENT,
                type="spike",
                severity="high",
                message="Test",
                raw_data={},
            )
            .returning(Alert.alert_id)
        )
    ).scalar_one()
    statement = insert(Recommendation).values(
        alert_id=alert_id,
        action="Vérifier la consommation",
        explanation="Pic détecté",
        rule_reference="spike-v1",
    )
    await data_connection.execute(statement)
    with pytest.raises(IntegrityError):
        async with data_connection.begin_nested():
            await data_connection.execute(statement)
@@ -1,37 +0,0 @@
 from typing import Any
 from app.core.config import Settings
 SETTINGS_DE_TEST: dict[str, Any] = {
    "env": "local",
    "debug": False,
    "log_level": "WARNING",
    "cors_origins": "",
    "secret_key": "secret-de-test-assez-long-pour-le-validateur",
    "database_url": "postgresql+asyncpg://enervision:change_me@localhost:5433/enervision_test",
 }
 class FakeSession:
    """Session factice : renvoie `result`, ou leve `failure` si elle est fournie."""
    def __init__(self, result: object = None, failure: Exception | None = None) -> None:
        self._result = result
        self._failure = failure
    async def scalar(self, *_: object, **__: object) -> object:
        return self._repondre()
    async def execute(self, *_: object, **__: object) -> object:
        return self._repondre()
    def _repondre(self) -> object:
        if self._failure is not None:
            raise self._failure
        return self._result
 # Piège : les arguments nommés priment sur l'environnement et sur .env, contrairement
 # aux variables posées par la fixture `environment`, qui restent surchargeables.
 def make_settings(**overrides: Any) -> Settings:
    return Settings(**{**SETTINGS_DE_TEST, **overrides})
@@ -1,142 +0,0 @@
 # Les trois refus ci-dessous sont la preuve que l'ajout seul est une propriété de la base et
 # non une convention de code Python. Ce sont eux qu'il faut montrer, pas la classe du dépôt.
 import uuid
 import pytest
 from sqlalchemy import text
 from sqlalchemy.exc import DBAPIError
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 from app.models.audit_log import AuditAction, AuditOutcome
 from app.repositories.audit_log import (
    CLES_DE_DETAIL_AUTORISEES,
    AuditLogRepository,
    assemble_detail,
 )
 pytestmark = pytest.mark.integration
 ACTEUR = Principal(
    id=uuid.uuid4(),
    email="admin@enervision.fr",
    role=Role.ADMIN,
    kind=AccountKind.HUMAIN,
    must_change_password=False,
 )
 async def une_ligne(session: AsyncSession) -> None:
    await AuditLogRepository(session).record(
        action=AuditAction.COMPTE_CREE, actor=ACTEUR, target_type="app_user", target_id="x"
    )
    await session.flush()
@pytest.mark.parametrize(
    "instruction",
    [
        "update audit_log set action = 'falsifie'",
        "delete from audit_log",
        "truncate audit_log",
    ],
    ids=["modification", "suppression", "vidage"],
 )
 async def test_the_database_refuses_to_mutate_the_audit_log(
    session: AsyncSession, instruction: str
 ) -> None:
    await une_ligne(session)
    with pytest.raises(DBAPIError, match="ajout seul"):
        await session.execute(text(instruction))
    await session.rollback()
 async def test_record_keeps_a_snapshot_of_the_actor(session: AsyncSession) -> None:
    depot = AuditLogRepository(session)
    cible = uuid.uuid4().hex
    await depot.record(action=AuditAction.COMPTE_DESACTIVE, actor=ACTEUR, target_id=cible)
    await session.flush()
    ligne = (
        await session.execute(
            text(
                "select actor_id, actor_email, actor_role, outcome from audit_log "
                "where target_id = :c"
            ),
            {"c": cible},
        )
    ).one()
    await session.rollback()
    assert ligne.actor_id == ACTEUR.id
    assert ligne.actor_email == ACTEUR.email
    assert ligne.actor_role == Role.ADMIN.value
    assert ligne.outcome == AuditOutcome.SUCCES.value
 async def test_record_accepts_a_label_when_there_is_no_authenticated_actor(
    session: AsyncSession,
 ) -> None:
    depot = AuditLogRepository(session)
    cible = uuid.uuid4().hex
    await depot.record(action=AuditAction.ADMIN_AMORCE, actor_label="cli", target_id=cible)
    await session.flush()
    ligne = (
        await session.execute(
            text("select actor_id, actor_email from audit_log where target_id = :c"),
            {"c": cible},
        )
    ).one()
    await session.rollback()
    assert ligne.actor_id is None
    assert ligne.actor_email == "cli"
 async def test_record_drops_the_detail_keys_outside_the_allow_list(
    session: AsyncSession,
 ) -> None:
    depot = AuditLogRepository(session)
    cible = uuid.uuid4().hex
    await depot.record(
        action=AuditAction.COMPTE_ROLE_CHANGE,
        actor=ACTEUR,
        target_id=cible,
        detail={"role_avant": "lecteur", "mot_de_passe": "ne-doit-pas-passer"},
    )
    await session.flush()
    detail = (
        await session.execute(
            text("select detail from audit_log where target_id = :c"), {"c": cible}
        )
    ).scalar_one()
    await session.rollback()
    assert detail == {"role_avant": "lecteur"}
@pytest.mark.parametrize(
    ("brut", "attendu"),
    [
        (None, {}),
        ({}, {}),
        ({"motif": "reutilisation"}, {"motif": "reutilisation"}),
        ({"password": "x"}, {}),
    ],
    ids=["absent", "vide", "cle_autorisee", "cle_refusee"],
 )
 def test_assemble_detail_only_keeps_the_allowed_keys(
    brut: dict[str, str] | None, attendu: dict[str, str]
 ) -> None:
    assert assemble_detail(brut) == attendu
 def test_the_allow_list_never_mentions_a_secret() -> None:
    suspects = {"password", "mot_de_passe", "token", "jeton", "secret", "hash"}
    assert CLES_DE_DETAIL_AUTORISEES & suspects == set()
@@ -1,118 +0,0 @@
 import uuid
 import pytest
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.login_attempt import LoginOutcome
 from app.repositories.login_attempt import LoginAttemptRepository
 pytestmark = pytest.mark.integration
 IP = "203.0.113.10"
 AUTRE_IP = "198.51.100.7"
 def adresse() -> str:
    return f"tentative-{uuid.uuid4().hex[:12]}@enervision.fr"
 async def echoue(
    depot: LoginAttemptRepository, email: str, ip: str | None, combien: int = 1
 ) -> None:
    for _ in range(combien):
        await depot.record(email=email, client_ip=ip, outcome=LoginOutcome.IDENTIFIANTS_INVALIDES)
 async def test_count_recent_failures_separates_the_three_counters(
    session: AsyncSession,
 ) -> None:
    depot = LoginAttemptRepository(session)
    cible, voisin = adresse(), adresse()
    await echoue(depot, cible, IP, combien=3)
    await echoue(depot, cible, AUTRE_IP, combien=2)
    await echoue(depot, voisin, IP, combien=4)
    await session.flush()
    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
    await session.rollback()
    assert compteurs.per_identifier_and_ip == 3
    assert compteurs.per_identifier == 5
    assert compteurs.per_ip == 7
 async def test_count_recent_failures_ignores_successful_attempts(
    session: AsyncSession,
 ) -> None:
    depot = LoginAttemptRepository(session)
    cible = adresse()
    await echoue(depot, cible, IP, combien=2)
    await depot.record(email=cible, client_ip=IP, outcome=LoginOutcome.SUCCES)
    await session.flush()
    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
    await session.rollback()
    assert compteurs.per_identifier_and_ip == 2
 async def test_count_recent_failures_forgets_what_falls_outside_the_window(
    session: AsyncSession,
 ) -> None:
    depot = LoginAttemptRepository(session)
    cible = adresse()
    await echoue(depot, cible, IP, combien=2)
    await session.flush()
    await session.execute(
        text(
            "update login_attempt set occurred_at = now() - interval '2 hours' "
            "where email_tried = :e"
        ),
        {"e": cible},
    )
    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
    await session.rollback()
    assert compteurs.per_identifier_and_ip == 0
 async def test_count_recent_failures_still_counts_when_the_address_is_unknown(
    session: AsyncSession,
 ) -> None:
    depot = LoginAttemptRepository(session)
    inconnu = adresse()
    await echoue(depot, inconnu, IP, combien=5)
    await session.flush()
    compteurs = await depot.count_recent_failures(email=inconnu, client_ip=IP, window_seconds=900)
    await session.rollback()
    assert compteurs.per_identifier_and_ip == 5
 async def test_record_normalises_the_address_before_counting(session: AsyncSession) -> None:
    depot = LoginAttemptRepository(session)
    cible = adresse()
    await echoue(depot, cible.upper(), IP, combien=2)
    await session.flush()
    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
    await session.rollback()
    assert compteurs.per_identifier_and_ip == 2
 async def test_count_recent_failures_tolerates_a_missing_client_address(
    session: AsyncSession,
 ) -> None:
    depot = LoginAttemptRepository(session)
    cible = adresse()
    await echoue(depot, cible, None, combien=2)
    await session.flush()
    compteurs = await depot.count_recent_failures(email=cible, client_ip=None, window_seconds=900)
    await session.rollback()
    assert compteurs.per_identifier == 2
@@ -1,190 +0,0 @@
 # Le premier test de ce fichier est le seul endroit où l'atomicité de la rotation se démontre :
 # sur un double, deux appels concurrents réussiraient tous les deux.
 import uuid
 from datetime import UTC, datetime, timedelta
 import pytest
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.roles import Role
 from app.core.security import fingerprint_refresh, generate_refresh_secret
 from app.models.refresh_token import RevocationReason
 from app.repositories.refresh_token import RefreshTokenRepository
 from app.repositories.user import UserRepository
 pytestmark = pytest.mark.integration
 DUREE = timedelta(days=7)
 async def un_compte(session: AsyncSession) -> uuid.UUID:
    compte = await UserRepository(session).create(
        email=f"jeton-{uuid.uuid4().hex[:12]}@enervision.fr",
        password_hash="$argon2id$x",
        role=Role.LECTEUR,
    )
    return compte.id
 async def un_jeton(
    depot: RefreshTokenRepository,
    user_id: uuid.UUID,
    *,
    family_id: uuid.UUID | None = None,
    duree: timedelta = DUREE,
 ) -> tuple[str, uuid.UUID]:
    secret = generate_refresh_secret()
    jeton = await depot.create(
        user_id=user_id,
        family_id=family_id or uuid.uuid4(),
        token_hash=fingerprint_refresh(secret),
        expires_at=datetime.now(UTC) + duree,
        client_ip="203.0.113.10",
        user_agent="pytest",
    )
    return secret, jeton.family_id
 async def test_claim_for_rotation_only_succeeds_once(session: AsyncSession) -> None:
    depot = RefreshTokenRepository(session)
    secret, _ = await un_jeton(depot, await un_compte(session))
    premier = await depot.claim_for_rotation(fingerprint_refresh(secret))
    second = await depot.claim_for_rotation(fingerprint_refresh(secret))
    await session.rollback()
    assert premier is not None
    assert second is None
 async def test_claim_for_rotation_refuses_an_expired_token(session: AsyncSession) -> None:
    depot = RefreshTokenRepository(session)
    secret, _ = await un_jeton(depot, await un_compte(session), duree=-timedelta(minutes=1))
    revendique = await depot.claim_for_rotation(fingerprint_refresh(secret))
    await session.rollback()
    assert revendique is None
 async def test_claim_for_rotation_returns_nothing_for_an_unknown_fingerprint(
    session: AsyncSession,
 ) -> None:
    revendique = await RefreshTokenRepository(session).claim_for_rotation(
        fingerprint_refresh(generate_refresh_secret())
    )
    assert revendique is None
 async def test_inspect_finds_a_token_that_rotation_already_refused(
    session: AsyncSession,
 ) -> None:
    depot = RefreshTokenRepository(session)
    secret, _ = await un_jeton(depot, await un_compte(session))
    await depot.claim_for_rotation(fingerprint_refresh(secret))
    ligne = await depot.inspect(fingerprint_refresh(secret))
    rotation, motif = (ligne.rotated_at, ligne.revoked_reason) if ligne else (None, None)
    await session.rollback()
    assert rotation is not None
    assert motif == RevocationReason.ROTATION.value
 async def test_revoke_family_touches_every_living_token_of_that_family_only(
    session: AsyncSession,
 ) -> None:
    depot = RefreshTokenRepository(session)
    compte = await un_compte(session)
    famille = uuid.uuid4()
    await un_jeton(depot, compte, family_id=famille)
    await un_jeton(depot, compte, family_id=famille)
    autre_secret, _ = await un_jeton(depot, compte)
    revoquees = await depot.revoke_family(famille, RevocationReason.REUTILISATION)
    intacte = await depot.claim_for_rotation(fingerprint_refresh(autre_secret))
    await session.rollback()
    assert revoquees == 2
    assert intacte is not None
 async def test_revoke_family_is_idempotent(session: AsyncSession) -> None:
    depot = RefreshTokenRepository(session)
    compte = await un_compte(session)
    famille = uuid.uuid4()
    await un_jeton(depot, compte, family_id=famille)
    premier = await depot.revoke_family(famille, RevocationReason.DECONNEXION)
    second = await depot.revoke_family(famille, RevocationReason.DECONNEXION)
    await session.rollback()
    assert premier == 1
    assert second == 0
 async def test_revoke_all_for_user_closes_every_family_at_once(session: AsyncSession) -> None:
    depot = RefreshTokenRepository(session)
    compte = await un_compte(session)
    await un_jeton(depot, compte)
    await un_jeton(depot, compte)
    await un_jeton(depot, compte)
    revoquees = await depot.revoke_all_for_user(compte, RevocationReason.CHANGEMENT_MOT_DE_PASSE)
    await session.rollback()
    assert revoquees == 3
 async def test_link_replacement_records_the_successor(session: AsyncSession) -> None:
    depot = RefreshTokenRepository(session)
    compte = await un_compte(session)
    ancien_secret, famille = await un_jeton(depot, compte)
    revendique = await depot.claim_for_rotation(fingerprint_refresh(ancien_secret))
    assert revendique is not None
    nouveau_secret = generate_refresh_secret()
    nouveau = await depot.create(
        user_id=compte,
        family_id=famille,
        token_hash=fingerprint_refresh(nouveau_secret),
        expires_at=revendique.expires_at,
        client_ip=None,
        user_agent=None,
    )
    await depot.link_replacement(revendique.id, nouveau.id)
    ligne = await depot.inspect(fingerprint_refresh(ancien_secret))
    successeur = ligne.replaced_by if ligne else None
    await session.rollback()
    assert successeur == nouveau.id
 async def test_the_database_refuses_two_tokens_sharing_a_fingerprint(
    session: AsyncSession,
 ) -> None:
    depot = RefreshTokenRepository(session)
    compte = await un_compte(session)
    secret = generate_refresh_secret()
    await depot.create(
        user_id=compte,
        family_id=uuid.uuid4(),
        token_hash=fingerprint_refresh(secret),
        expires_at=datetime.now(UTC) + DUREE,
        client_ip=None,
        user_agent=None,
    )
    with pytest.raises(IntegrityError):
        await depot.create(
            user_id=compte,
            family_id=uuid.uuid4(),
            token_hash=fingerprint_refresh(secret),
            expires_at=datetime.now(UTC) + DUREE,
            client_ip=None,
            user_agent=None,
        )
    await session.rollback()
@@ -1,196 +0,0 @@
 import uuid
 import pytest
 from sqlalchemy import text
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.roles import AccountKind, Role
 from app.repositories.user import UserRepository
 pytestmark = pytest.mark.integration
 def adresse() -> str:
    return f"compte-{uuid.uuid4().hex[:12]}@enervision.fr"
 async def test_create_normalises_the_email_to_lower_case(session: AsyncSession) -> None:
    depot = UserRepository(session)
    saisie = adresse().upper()
    compte = await depot.create(email=saisie, password_hash="$argon2id$x", role=Role.LECTEUR)
    enregistre = compte.email
    await session.rollback()
    assert enregistre == saisie.lower()
 async def test_the_database_refuses_an_email_written_in_upper_case(
    session: AsyncSession,
 ) -> None:
    saisie = adresse().upper()
    with pytest.raises(IntegrityError):
        await session.execute(
            text(
                "insert into app_user (email, password_hash, role) "
                "values (:e, '$argon2id$x', 'lecteur')"
            ),
            {"e": saisie},
        )
    await session.rollback()
 async def test_the_database_refuses_two_accounts_sharing_an_email(
    session: AsyncSession,
 ) -> None:
    depot = UserRepository(session)
    saisie = adresse()
    await depot.create(email=saisie, password_hash="$argon2id$x", role=Role.LECTEUR)
    with pytest.raises(IntegrityError):
        await depot.create(email=saisie, password_hash="$argon2id$y", role=Role.ADMIN)
    await session.rollback()
 async def test_get_by_email_is_case_insensitive(session: AsyncSession) -> None:
    depot = UserRepository(session)
    saisie = adresse()
    await depot.create(email=saisie, password_hash="$argon2id$x", role=Role.OPERATEUR)
    trouve = await depot.get_by_email(saisie.upper())
    role = trouve.role if trouve else None
    await session.rollback()
    assert role == Role.OPERATEUR.value
 async def test_get_by_email_returns_nothing_for_an_unknown_address(
    session: AsyncSession,
 ) -> None:
    trouve = await UserRepository(session).get_by_email(adresse())
    assert trouve is None
 async def test_set_role_moves_the_credentials_marker_forward(session: AsyncSession) -> None:
    depot = UserRepository(session)
    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
    avant = compte.credentials_changed_at
    await depot.set_role(compte.id, Role.ADMIN)
    await session.refresh(compte)
    apres, role = compte.credentials_changed_at, compte.role
    await session.rollback()
    assert role == Role.ADMIN.value
    assert apres > avant
 async def test_set_active_moves_the_credentials_marker_forward(session: AsyncSession) -> None:
    depot = UserRepository(session)
    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
    avant = compte.credentials_changed_at
    await depot.set_active(compte.id, is_active=False)
    await session.refresh(compte)
    apres, actif = compte.credentials_changed_at, compte.is_active
    await session.rollback()
    assert actif is False
    assert apres > avant
 async def test_rehash_password_leaves_the_credentials_marker_untouched(
    session: AsyncSession,
 ) -> None:
    depot = UserRepository(session)
    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
    avant = compte.credentials_changed_at
    await depot.rehash_password(compte.id, "$argon2id$plus-recent")
    await session.refresh(compte)
    apres, empreinte = compte.credentials_changed_at, compte.password_hash
    await session.rollback()
    assert empreinte == "$argon2id$plus-recent"
    assert apres == avant
 async def test_update_password_moves_the_credentials_marker_forward(
    session: AsyncSession,
 ) -> None:
    depot = UserRepository(session)
    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
    avant = compte.credentials_changed_at
    await depot.update_password(compte.id, "$argon2id$neuf", must_change_password=False)
    await session.refresh(compte)
    apres = compte.credentials_changed_at
    await session.rollback()
    assert apres > avant
 async def test_touch_last_login_records_the_connection_date(session: AsyncSession) -> None:
    depot = UserRepository(session)
    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
    await depot.touch_last_login(compte.id)
    await session.refresh(compte)
    date = compte.last_login_at
    await session.rollback()
    assert date is not None
 async def test_count_active_admins_only_counts_enabled_administrators(
    session: AsyncSession,
 ) -> None:
    depot = UserRepository(session)
    depart = await depot.count_active_admins()
    await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.ADMIN)
    desactive = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.ADMIN)
    await depot.set_active(desactive.id, is_active=False)
    total = await depot.count_active_admins()
    await session.rollback()
    assert total == depart + 1
 async def test_create_accepts_a_service_account(session: AsyncSession) -> None:
    depot = UserRepository(session)
    compte = await depot.create(
        email=adresse(),
        password_hash="$argon2id$x",
        role=Role.OPERATEUR,
        kind=AccountKind.SERVICE,
    )
    nature = compte.kind
    await session.rollback()
    assert nature == AccountKind.SERVICE.value
 async def test_list_all_returns_the_accounts_sorted_by_email(session: AsyncSession) -> None:
    depot = UserRepository(session)
    await depot.create(email=f"zz-{adresse()}", password_hash="$argon2id$x", role=Role.LECTEUR)
    await depot.create(email=f"aa-{adresse()}", password_hash="$argon2id$x", role=Role.LECTEUR)
    comptes = await depot.list_all()
    emails = [compte.email for compte in comptes]
    await session.rollback()
    assert emails == sorted(emails)
 async def test_get_by_id_returns_nothing_for_an_unknown_identifier(
    session: AsyncSession,
 ) -> None:
    trouve = await UserRepository(session).get_by_id(uuid.uuid4())
    assert trouve is None
@@ -1,495 +0,0 @@
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import UUID, uuid4
 import pytest
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 from app.core.security import (
    TokenPolicy,
    decode_access_token,
    fingerprint_refresh,
 )
 from app.models.login_attempt import LoginOutcome
 from app.models.refresh_token import RevocationReason
 from app.repositories.login_attempt import FailureCounts
 from app.repositories.refresh_token import ClaimedToken
 from app.services.auth import (
    AuthService,
    InvalidCredentialsError,
    LoginPolicy,
    RateLimitedError,
    SessionRejectedError,
 )
 POLITIQUE_JETON = TokenPolicy(
    secret="un-secret-de-test-de-plus-de-trente-deux-caracteres",
    issuer="enervision-api",
    audience="enervision-web",
    access_ttl=timedelta(minutes=15),
 )
 POLITIQUE_CONNEXION = LoginPolicy(
    window_seconds=900,
    max_failures_per_identifier_and_ip=5,
    max_failures_per_ip=20,
    max_failures_per_identifier=50,
 )
@dataclass
 class FauxCompte:
    id: UUID = field(default_factory=uuid4)
    email: str = "operateur@enervision.fr"
    password_hash: str = "$argon2id$factice"
    role: str = "operateur"
    kind: str = "human"
    is_active: bool = True
    must_change_password: bool = False
    credentials_changed_at: datetime = field(default_factory=lambda: datetime.now(UTC))
 class FauxDepotComptes:
    def __init__(self, compte: FauxCompte | None) -> None:
        self.compte = compte
        self.rehachages = 0
        self.connexions_datees = 0
        self.mots_de_passe_changes = 0
    async def get_by_email(self, email: str) -> FauxCompte | None:
        return self.compte
    async def get_by_id(self, user_id: UUID) -> FauxCompte | None:
        return self.compte
    async def rehash_password(self, user_id: UUID, password_hash: str) -> None:
        self.rehachages += 1
    async def update_password(self, user_id: UUID, password_hash: str, **_: object) -> None:
        self.mots_de_passe_changes += 1
    async def touch_last_login(self, user_id: UUID) -> None:
        self.connexions_datees += 1
 class FauxDepotTentatives:
    def __init__(self, compteurs: FailureCounts | None = None) -> None:
        self.compteurs = compteurs or FailureCounts(0, 0, 0)
        self.enregistrees: list[str] = []
    async def count_recent_failures(self, **_: object) -> FailureCounts:
        return self.compteurs
    async def record(self, *, outcome: object, **_: object) -> None:
        self.enregistrees.append(str(outcome))
 class FauxDepotAudit:
    def __init__(self) -> None:
        self.lignes: list[tuple[str, Mapping[str, Any] | None]] = []
    async def record(self, *, action: object, detail: Any = None, **_: object) -> None:
        self.lignes.append((str(action), detail))
@dataclass
 class FauxJeton:
    id: UUID = field(default_factory=uuid4)
    family_id: UUID = field(default_factory=uuid4)
    user_id: UUID = field(default_factory=uuid4)
    expires_at: datetime = field(default_factory=lambda: datetime.now(UTC) + timedelta(days=7))
    rotated_at: datetime | None = None
    revoked_at: datetime | None = None
 class FauxDepotJetons:
    def __init__(
        self, revendique: ClaimedToken | None = None, connu: FauxJeton | None = None
    ) -> None:
        self.revendique = revendique
        self.connu = connu
        self.crees: list[UUID] = []
        self.familles_revoquees: list[tuple[UUID, str]] = []
        self.revocations_par_compte: list[tuple[UUID, str]] = []
        self.liaisons: list[tuple[UUID, UUID]] = []
    async def create(self, *, user_id: UUID, family_id: UUID, **_: object) -> FauxJeton:
        jeton = FauxJeton(user_id=user_id, family_id=family_id)
        self.crees.append(jeton.id)
        return jeton
    async def claim_for_rotation(self, token_hash: bytes) -> ClaimedToken | None:
        return self.revendique
    async def inspect(self, token_hash: bytes) -> FauxJeton | None:
        return self.connu
    async def link_replacement(self, ancien_id: UUID, nouveau_id: UUID) -> None:
        self.liaisons.append((ancien_id, nouveau_id))
    async def revoke_family(self, family_id: UUID, reason: RevocationReason) -> int:
        self.familles_revoquees.append((family_id, reason.value))
        return 2
    async def revoke_all_for_user(self, user_id: UUID, reason: RevocationReason) -> int:
        self.revocations_par_compte.append((user_id, reason.value))
        return 3
 class FauxHacheur:
    def __init__(self, *, accepte: bool = True, rehachage_requis: bool = False) -> None:
        self.verifications = 0
        self.hachages = 0
        self._accepte = accepte
        self._rehachage_requis = rehachage_requis
    async def hash(self, password: str) -> str:
        self.hachages += 1
        return "$argon2id$nouvelle"
    async def verify(self, stored: str, password: str) -> bool:
        self.verifications += 1
        return self._accepte
    async def verify_dummy(self) -> None:
        self.verifications += 1
    def needs_rehash(self, stored: str) -> bool:
        return self._rehachage_requis
 class FausseTransaction:
    def __init__(self) -> None:
        self.validations = 0
    async def commit(self) -> None:
        self.validations += 1
@dataclass
 class Attirail:
    service: AuthService
    comptes: FauxDepotComptes
    tentatives: FauxDepotTentatives
    jetons: FauxDepotJetons
    audit: FauxDepotAudit
    hacheur: FauxHacheur
 def fabrique_service(
    *,
    compte: FauxCompte | None = None,
    compteurs: FailureCounts | None = None,
    hacheur: FauxHacheur | None = None,
    jetons: FauxDepotJetons | None = None,
 ) -> Attirail:
    comptes = FauxDepotComptes(compte)
    tentatives = FauxDepotTentatives(compteurs)
    depot_jetons = jetons or FauxDepotJetons()
    audit = FauxDepotAudit()
    hacheur = hacheur or FauxHacheur()
    service = AuthService(
        users=comptes,  # type: ignore[arg-type]
        attempts=tentatives,  # type: ignore[arg-type]
        refresh_tokens=depot_jetons,  # type: ignore[arg-type]
        audit=audit,  # type: ignore[arg-type]
        hasher=hacheur,  # type: ignore[arg-type]
        transaction=FausseTransaction(),
        token_policy=POLITIQUE_JETON,
        login_policy=POLITIQUE_CONNEXION,
        refresh_ttl=timedelta(days=7),
    )
    return Attirail(service, comptes, tentatives, depot_jetons, audit, hacheur)
 async def connecte(service: AuthService, mot_de_passe: str = "un-mot-de-passe-valide") -> object:
    return await service.authenticate(
        email="operateur@enervision.fr",
        password=mot_de_passe,
        client_ip="203.0.113.10",
        user_agent="pytest",
    )
 async def rafraichit(service: AuthService, secret: str = "un-secret-opaque") -> object:
    return await service.refresh(secret=secret, client_ip="203.0.113.10", user_agent="pytest")
 async def test_authenticate_returns_a_readable_access_token_when_credentials_match() -> None:
    compte = FauxCompte()
    attirail = fabrique_service(compte=compte)
    session = await connecte(attirail.service)
    claims = decode_access_token(POLITIQUE_JETON, session.access_token)  # type: ignore[attr-defined]
    assert claims.subject == compte.id
    assert claims.role == "operateur"
    assert attirail.tentatives.enregistrees == [LoginOutcome.SUCCES.value]
    assert attirail.comptes.connexions_datees == 1
 async def test_authenticate_opens_one_refresh_family_per_login() -> None:
    attirail = fabrique_service(compte=FauxCompte())
    session = await connecte(attirail.service)
    assert len(attirail.jetons.crees) == 1
    assert session.refresh_secret  # type: ignore[attr-defined]
 async def test_authenticate_verifies_a_decoy_digest_when_the_email_is_unknown() -> None:
    attirail = fabrique_service(compte=None)
    with pytest.raises(InvalidCredentialsError):
        await connecte(attirail.service)
    assert attirail.hacheur.verifications == 1
    assert attirail.tentatives.enregistrees == [LoginOutcome.IDENTIFIANTS_INVALIDES.value]
 async def test_authenticate_skips_hashing_entirely_when_the_rate_limit_is_reached() -> None:
    compteurs = FailureCounts(per_identifier_and_ip=5, per_ip=5, per_identifier=5)
    attirail = fabrique_service(compte=FauxCompte(), compteurs=compteurs)
    with pytest.raises(RateLimitedError):
        await connecte(attirail.service)
    assert attirail.hacheur.verifications == 0
    assert attirail.hacheur.hachages == 0
    assert attirail.tentatives.enregistrees == [LoginOutcome.LIMITE.value]
    assert attirail.audit.lignes == []
 async def test_authenticate_audits_when_the_identifier_threshold_alone_is_reached() -> None:
    compteurs = FailureCounts(per_identifier_and_ip=0, per_ip=0, per_identifier=50)
    attirail = fabrique_service(compte=FauxCompte(), compteurs=compteurs)
    with pytest.raises(RateLimitedError):
        await connecte(attirail.service)
    assert len(attirail.audit.lignes) == 1
    assert "identifier_throttled" in attirail.audit.lignes[0][0]
 async def test_authenticate_rejects_a_wrong_password_with_the_generic_error() -> None:
    attirail = fabrique_service(compte=FauxCompte(), hacheur=FauxHacheur(accepte=False))
    with pytest.raises(InvalidCredentialsError):
        await connecte(attirail.service)
    assert attirail.tentatives.enregistrees == [LoginOutcome.IDENTIFIANTS_INVALIDES.value]
@pytest.mark.parametrize(
    "compte",
    [FauxCompte(is_active=False), FauxCompte(kind="service")],
    ids=["compte_desactive", "compte_de_service"],
 )
 async def test_authenticate_rejects_unavailable_accounts_after_checking_the_password(
    compte: FauxCompte,
 ) -> None:
    attirail = fabrique_service(compte=compte)
    with pytest.raises(InvalidCredentialsError):
        await connecte(attirail.service)
    assert attirail.hacheur.verifications == 1
    assert attirail.tentatives.enregistrees == [LoginOutcome.COMPTE_INDISPONIBLE.value]
 async def test_authenticate_rehashes_the_password_when_the_parameters_changed() -> None:
    attirail = fabrique_service(compte=FauxCompte(), hacheur=FauxHacheur(rehachage_requis=True))
    await connecte(attirail.service)
    assert attirail.comptes.rehachages == 1
 async def test_authenticate_leaves_the_digest_alone_when_the_parameters_match() -> None:
    attirail = fabrique_service(compte=FauxCompte())
    await connecte(attirail.service)
    assert attirail.comptes.rehachages == 0
 async def test_refresh_rotates_the_token_and_keeps_the_family() -> None:
    compte = FauxCompte()
    revendique = ClaimedToken(
        id=uuid4(),
        family_id=uuid4(),
        user_id=compte.id,
        expires_at=datetime.now(UTC) + timedelta(days=5),
    )
    attirail = fabrique_service(compte=compte, jetons=FauxDepotJetons(revendique=revendique))
    session = await rafraichit(attirail.service)
    assert session.refresh_secret  # type: ignore[attr-defined]
    assert len(attirail.jetons.crees) == 1
    assert attirail.jetons.liaisons == [(revendique.id, attirail.jetons.crees[0])]
    assert attirail.jetons.familles_revoquees == []
 async def test_refresh_inherits_the_absolute_expiry_of_its_predecessor() -> None:
    compte = FauxCompte()
    echeance = datetime.now(UTC) + timedelta(days=2)
    revendique = ClaimedToken(id=uuid4(), family_id=uuid4(), user_id=compte.id, expires_at=echeance)
    attirail = fabrique_service(compte=compte, jetons=FauxDepotJetons(revendique=revendique))
    await rafraichit(attirail.service)
    assert revendique.expires_at == echeance
 async def test_refresh_rejects_an_unknown_secret_without_touching_any_family() -> None:
    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons())
    with pytest.raises(SessionRejectedError):
        await rafraichit(attirail.service)
    assert attirail.jetons.familles_revoquees == []
    assert attirail.audit.lignes == []
 async def test_refresh_rejects_an_expired_token_without_revoking_its_family() -> None:
    perime = FauxJeton(expires_at=datetime.now(UTC) - timedelta(minutes=1))
    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons(connu=perime))
    with pytest.raises(SessionRejectedError):
        await rafraichit(attirail.service)
    assert attirail.jetons.familles_revoquees == []
    assert attirail.audit.lignes == []
 async def test_refresh_revokes_the_whole_family_when_a_rotated_token_comes_back() -> None:
    rejoue = FauxJeton(rotated_at=datetime.now(UTC), revoked_at=datetime.now(UTC))
    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons(connu=rejoue))
    with pytest.raises(SessionRejectedError):
        await rafraichit(attirail.service)
    assert attirail.jetons.familles_revoquees == [
        (rejoue.family_id, RevocationReason.REUTILISATION.value)
    ]
    assert "refresh_reuse_detected" in attirail.audit.lignes[0][0]
 async def test_refresh_revokes_the_family_when_the_account_was_disabled_meanwhile() -> None:
    compte = FauxCompte(is_active=False)
    revendique = ClaimedToken(
        id=uuid4(),
        family_id=uuid4(),
        user_id=compte.id,
        expires_at=datetime.now(UTC) + timedelta(days=5),
    )
    attirail = fabrique_service(compte=compte, jetons=FauxDepotJetons(revendique=revendique))
    with pytest.raises(SessionRejectedError):
        await rafraichit(attirail.service)
    assert attirail.jetons.familles_revoquees == [
        (revendique.family_id, RevocationReason.ADMINISTRATION.value)
    ]
 async def test_logout_revokes_only_the_presented_family() -> None:
    connu = FauxJeton()
    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons(connu=connu))
    await attirail.service.logout(secret="un-secret-opaque")
    assert attirail.jetons.familles_revoquees == [
        (connu.family_id, RevocationReason.DECONNEXION.value)
    ]
    assert attirail.jetons.revocations_par_compte == []
 async def test_logout_stays_silent_when_the_cookie_points_at_nothing() -> None:
    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons())
    await attirail.service.logout(secret="un-secret-inconnu")
    assert attirail.jetons.familles_revoquees == []
 async def test_logout_all_revokes_every_session_and_leaves_an_audit_trail() -> None:
    compte = FauxCompte()
    attirail = fabrique_service(compte=compte)
    acteur = Principal(
        id=compte.id,
        email=compte.email,
        role=Role.OPERATEUR,
        kind=AccountKind.HUMAIN,
        must_change_password=False,
    )
    revoquees = await attirail.service.logout_all(acteur)
    assert revoquees == 3
    assert attirail.jetons.revocations_par_compte == [
        (compte.id, RevocationReason.DECONNEXION.value)
    ]
    assert "all_sessions_revoked" in attirail.audit.lignes[0][0]
 def test_fingerprint_is_what_the_service_stores_not_the_secret_itself() -> None:
    secret = "un-secret-opaque"
    empreinte = fingerprint_refresh(secret)
    assert secret.encode() not in empreinte
 async def test_change_password_revokes_every_session_then_reopens_the_current_one() -> None:
    compte = FauxCompte()
    attirail = fabrique_service(compte=compte)
    acteur = Principal(
        id=compte.id,
        email=compte.email,
        role=Role.OPERATEUR,
        kind=AccountKind.HUMAIN,
        must_change_password=True,
    )
    session = await attirail.service.change_password(
        principal=acteur,
        current_password="l-ancien-mot-de-passe",
        new_password="le-nouveau-mot-de-passe",
        client_ip="203.0.113.10",
        user_agent="pytest",
    )
    assert attirail.jetons.revocations_par_compte == [
        (compte.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE.value)
    ]
    assert len(attirail.jetons.crees) == 1, "l'appareil courant doit repartir avec une session"
    assert session.refresh_secret
    assert "password_changed" in attirail.audit.lignes[0][0]
 async def test_change_password_refuses_a_wrong_current_password() -> None:
    compte = FauxCompte()
    attirail = fabrique_service(compte=compte, hacheur=FauxHacheur(accepte=False))
    acteur = Principal(
        id=compte.id,
        email=compte.email,
        role=Role.OPERATEUR,
        kind=AccountKind.HUMAIN,
        must_change_password=False,
    )
    with pytest.raises(InvalidCredentialsError):
        await attirail.service.change_password(
            principal=acteur,
            current_password="mauvais",
            new_password="le-nouveau-mot-de-passe",
            client_ip=None,
            user_agent=None,
        )
    assert attirail.jetons.revocations_par_compte == []
    assert attirail.jetons.crees == []
@@ -1,239 +0,0 @@
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
 from uuid import UUID, uuid4
 import pytest
 from app.core.principal import Principal
 from app.core.roles import AccountKind, Role
 from app.models.refresh_token import RevocationReason
 from app.services.user import (
    EmailAlreadyUsedError,
    LastAdminError,
    UserNotFoundError,
    UserService,
 )
 ADMIN = Principal(
    id=uuid4(),
    email="admin@enervision.fr",
    role=Role.ADMIN,
    kind=AccountKind.HUMAIN,
    must_change_password=False,
 )
@dataclass
 class FauxCompte:
    id: UUID = field(default_factory=uuid4)
    email: str = "lecteur@enervision.fr"
    password_hash: str = "$argon2id$factice"
    role: str = "lecteur"
    kind: str = "human"
    is_active: bool = True
    must_change_password: bool = False
    full_name: str | None = None
    last_login_at: datetime | None = None
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
 class FauxDepotComptes:
    def __init__(
        self, compte: FauxCompte | None = None, *, admins_actifs: int = 2, existe: bool = False
    ) -> None:
        self.compte = compte
        self.admins_actifs = admins_actifs
        self.existe = existe
        self.crees: list[str] = []
        self.roles_poses: list[tuple[UUID, str]] = []
        self.activations: list[tuple[UUID, bool]] = []
        self.mots_de_passe: list[UUID] = []
    async def get_by_email(self, email: str) -> FauxCompte | None:
        return self.compte if self.existe else None
    async def get_by_id(self, user_id: UUID) -> FauxCompte | None:
        return self.compte
    async def count_active_admins(self) -> int:
        return self.admins_actifs
    async def create(self, *, email: str, **_: object) -> FauxCompte:
        self.crees.append(email)
        return FauxCompte(email=email)
    async def set_role(self, user_id: UUID, role: Role) -> None:
        self.roles_poses.append((user_id, role.value))
    async def set_active(self, user_id: UUID, *, is_active: bool) -> None:
        self.activations.append((user_id, is_active))
    async def update_password(self, user_id: UUID, password_hash: str, **_: object) -> None:
        self.mots_de_passe.append(user_id)
 class FauxDepotJetons:
    def __init__(self) -> None:
        self.revocations: list[tuple[UUID, str]] = []
    async def revoke_all_for_user(self, user_id: UUID, reason: RevocationReason) -> int:
        self.revocations.append((user_id, reason.value))
        return 2
 class FauxDepotAudit:
    def __init__(self) -> None:
        self.lignes: list[tuple[str, Any]] = []
    async def record(self, *, action: object, detail: Any = None, **_: object) -> None:
        self.lignes.append((str(action), detail))
 class FauxHacheur:
    async def hash(self, password: str) -> str:
        return "$argon2id$nouvelle"
 class FausseTransaction:
    async def commit(self) -> None:
        return None
@dataclass
 class Attirail:
    service: UserService
    comptes: FauxDepotComptes
    jetons: FauxDepotJetons
    audit: FauxDepotAudit
 def fabrique(
    compte: FauxCompte | None = None, *, admins_actifs: int = 2, existe: bool = False
 ) -> Attirail:
    comptes = FauxDepotComptes(compte, admins_actifs=admins_actifs, existe=existe)
    jetons = FauxDepotJetons()
    audit = FauxDepotAudit()
    service = UserService(
        users=comptes,  # type: ignore[arg-type]
        refresh_tokens=jetons,  # type: ignore[arg-type]
        audit=audit,  # type: ignore[arg-type]
        hasher=FauxHacheur(),  # type: ignore[arg-type]
        transaction=FausseTransaction(),
    )
    return Attirail(service, comptes, jetons, audit)
 async def test_create_returns_a_temporary_password_shown_once() -> None:
    attirail = fabrique()
    cree = await attirail.service.create(
        actor=ADMIN, email="nouveau@enervision.fr", role=Role.LECTEUR, full_name=None
    )
    assert len(cree.temporary_password) >= 18
    assert attirail.comptes.crees == ["nouveau@enervision.fr"]
    assert "user.created" in attirail.audit.lignes[0][0]
 async def test_create_refuses_an_address_already_taken() -> None:
    attirail = fabrique(FauxCompte(), existe=True)
    with pytest.raises(EmailAlreadyUsedError):
        await attirail.service.create(
            actor=ADMIN, email="lecteur@enervision.fr", role=Role.LECTEUR, full_name=None
        )
 async def test_change_role_revokes_every_session_of_the_target() -> None:
    cible = FauxCompte()
    attirail = fabrique(cible)
    await attirail.service.change_role(actor=ADMIN, user_id=cible.id, role=Role.OPERATEUR)
    assert attirail.comptes.roles_poses == [(cible.id, "operateur")]
    assert attirail.jetons.revocations == [(cible.id, RevocationReason.ADMINISTRATION.value)]
 async def test_change_role_does_nothing_when_the_role_is_already_the_right_one() -> None:
    cible = FauxCompte(role="operateur")
    attirail = fabrique(cible)
    await attirail.service.change_role(actor=ADMIN, user_id=cible.id, role=Role.OPERATEUR)
    assert attirail.comptes.roles_poses == []
    assert attirail.jetons.revocations == []
 async def test_change_role_refuses_to_demote_the_last_active_administrator() -> None:
    dernier = FauxCompte(role="admin")
    attirail = fabrique(dernier, admins_actifs=1)
    with pytest.raises(LastAdminError):
        await attirail.service.change_role(actor=ADMIN, user_id=dernier.id, role=Role.LECTEUR)
 async def test_change_role_accepts_a_demotion_when_another_administrator_remains() -> None:
    admin = FauxCompte(role="admin")
    attirail = fabrique(admin, admins_actifs=2)
    await attirail.service.change_role(actor=ADMIN, user_id=admin.id, role=Role.LECTEUR)
    assert attirail.comptes.roles_poses == [(admin.id, "lecteur")]
 async def test_set_active_refuses_to_disable_the_last_active_administrator() -> None:
    dernier = FauxCompte(role="admin")
    attirail = fabrique(dernier, admins_actifs=1)
    with pytest.raises(LastAdminError):
        await attirail.service.set_active(actor=ADMIN, user_id=dernier.id, is_active=False)
 async def test_set_active_revokes_the_sessions_when_disabling() -> None:
    cible = FauxCompte()
    attirail = fabrique(cible)
    await attirail.service.set_active(actor=ADMIN, user_id=cible.id, is_active=False)
    assert attirail.comptes.activations == [(cible.id, False)]
    assert attirail.jetons.revocations == [(cible.id, RevocationReason.ADMINISTRATION.value)]
 async def test_set_active_leaves_the_sessions_alone_when_enabling() -> None:
    cible = FauxCompte(is_active=False)
    attirail = fabrique(cible)
    await attirail.service.set_active(actor=ADMIN, user_id=cible.id, is_active=True)
    assert attirail.jetons.revocations == []
 async def test_reset_password_closes_every_session_and_forces_a_change() -> None:
    cible = FauxCompte()
    attirail = fabrique(cible)
    reinitialise = await attirail.service.reset_password(actor=ADMIN, user_id=cible.id)
    assert len(reinitialise.temporary_password) >= 18
    assert attirail.comptes.mots_de_passe == [cible.id]
    assert attirail.jetons.revocations == [
        (cible.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE.value)
    ]
@pytest.mark.parametrize(
    "action",
    ["change_role", "set_active", "reset_password"],
    ids=["changement_de_role", "activation", "reinitialisation"],
 )
 async def test_every_operation_refuses_an_unknown_account(action: str) -> None:
    attirail = fabrique(None)
    arguments: dict[str, Any] = {"actor": ADMIN, "user_id": uuid4()}
    if action == "change_role":
        arguments["role"] = Role.ADMIN
    if action == "set_active":
        arguments["is_active"] = False
    with pytest.raises(UserNotFoundError):
        await getattr(attirail.service, action)(**arguments)
@@ -1,57 +0,0 @@
 import pytest
 from app import cli
 def test_build_parser_reads_the_create_admin_arguments() -> None:
    arguments = cli.build_parser().parse_args(
        ["create-admin", "--email", "admin@enervision.fr", "--generate", "--force"]
    )
    assert arguments.commande == "create-admin"
    assert arguments.email == "admin@enervision.fr"
    assert arguments.generate is True
    assert arguments.force is True
 def test_build_parser_requires_a_subcommand() -> None:
    with pytest.raises(SystemExit):
        cli.build_parser().parse_args([])
 def test_build_parser_requires_an_email() -> None:
    with pytest.raises(SystemExit):
        cli.build_parser().parse_args(["create-admin"])
 def test_read_password_generates_a_long_secret_when_asked(
    capsys: pytest.CaptureFixture[str],
 ) -> None:
    mot_de_passe = cli.read_password(generate=True)
    assert len(mot_de_passe) >= cli.LONGUEUR_MOT_DE_PASSE_GENERE
    assert mot_de_passe in capsys.readouterr().out
 def test_read_password_accepts_two_matching_entries(monkeypatch: pytest.MonkeyPatch) -> None:
    saisies = iter(["un-mot-de-passe-valide", "un-mot-de-passe-valide"])
    monkeypatch.setattr(cli, "getpass", lambda _: next(saisies))
    assert cli.read_password(generate=False) == "un-mot-de-passe-valide"
 def test_read_password_refuses_a_password_below_the_minimum_length(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(cli, "getpass", lambda _: "court")
    with pytest.raises(SystemExit):
        cli.read_password(generate=False)
 def test_read_password_refuses_two_different_entries(monkeypatch: pytest.MonkeyPatch) -> None:
    saisies = iter(["un-mot-de-passe-valide", "un-autre-mot-de-passe"])
    monkeypatch.setattr(cli, "getpass", lambda _: next(saisies))
    with pytest.raises(SystemExit):
        cli.read_password(generate=False)
@@ -1,17 +0,0 @@
 # Editor configuration, see https://editorconfig.org
 root = true
 [*]
 charset = utf-8
 indent_style = space
 indent_size = 2
 insert_final_newline = true
 trim_trailing_whitespace = true
 [*.ts]
 quote_type = single
 ij_typescript_use_double_quotes = false
 [*.md]
 max_line_length = off
 trim_trailing_whitespace = false
@@ -1,45 +0,0 @@
 # See https://docs.github.com/get-started/getting-started-with-git/ignoring-files for more about ignoring files.
 # Compiled output
 /dist
 /tmp
 /out-tsc
 /bazel-out
 # Node
 /node_modules
 npm-debug.log
 yarn-error.log
 # IDEs and editors
 .idea/
 .project
 .classpath
 .c9/
 *.launch
 .settings/
 *.sublime-workspace
 # Visual Studio Code
 .vscode/*
 !.vscode/settings.json
 !.vscode/tasks.json
 !.vscode/launch.json
 !.vscode/extensions.json
 !.vscode/mcp.json
 .history/*
 # Miscellaneous
 /.angular/cache
 .sass-cache/
 /connect.lock
 /coverage
 /test-results
 /libpeerconnection.log
 testem.log
 /typings
 __screenshots__/
 # System files
 .DS_Store
 Thumbs.db
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
copilot-swe-agent[bot]andGitHub	9ae7e924a7	Initial plan	2026-09-15 14:50:42 +00:00
PhyriosandGitHub	c3fd9327ea	Definition des jalons	2026-09-14 11:30:01 +02:00
		`@@ -1 +0,0 @@`
			`Generic single-database configuration with an async dbapi.`
		`@@ -1,3 +0,0 @@`
			`from app.schemas.health import LivenessStatus, ReadinessStatus`

			`__all__ = ["LivenessStatus", "ReadinessStatus"]`