test(ml,backend): couvre ML vers DB, puis la chaine complete jusqu'a l'API

Le pipeline ML n'avait aucun test touchant PostgreSQL : `ml/README.md` le disait, faute de
base joignable en CI. Le marqueur `integration` de `ml/pyproject.toml` etait declare et porte
par zero test.

- `ml/tests/conftest.py` : deux fixtures d'acces a la base, jamais interchangeables.
  `connexion_ml` annule sa transaction, `parc` valide ses ecritures parce que `run_scoring`
  ouvre sa propre connexion et ne verrait rien d'autre. Garde sur le nom de base, marque uuid
  sur chaque site, nettoyage dans l'ordre des cles etrangeres.
- `test_data_integration.py` : les neuf colonnes du contrat confrontees au schema Alembic
  reel, la borne `since`, l'ordre de tri dont dependent des lags positionnels, et le typage
  des colonnes entierement nulles.
- `test_score_integration.py` : les contraintes de `prediction` vues depuis le code qui
  ecrit, l'empilement volontaire de deux runs, et `run_scoring` de bout en bout sur un
  booster reel.
- `apps/backend/tests/test_chaine_ml_api.py` : lance les vrais binaires `enervision_ml.train`
  et `.score` en sous-processus, comme les DAGs, puis relit par `GET /api/v1/predictions`.
  Marqueur `chaine` distinct : le job `integration` du backend n'a pas l'environnement de ml/.
- `ml.yml` : job `integration`, seul du depot a reunir les deux environnements uv et une base.
  Ses `paths` incluent les migrations du backend, sans quoi le schema deriverait du SQL du
  pipeline sans que rien ne casse.
- Makefile : `migrate-test`, qui manquait (`enervision_test` n'a jamais recu de table),
  `ml-test-integration` et `test-chaine`.

.github/workflows/ml.yml

@@ -8,10 +8,28 @@ on:
     paths:
       - "ml/**"
       - ".github/workflows/ml.yml"
+      # Le job `integration` monte son schema avec les migrations du backend et joue le test de
+      # chaine qui vit dans ses tests : sans ces chemins, une migration modifiee ne declencherait
+      # rien et le schema deriverait du SQL du pipeline sans que rien ne casse. Meme raisonnement
+      # que le filtre d'airflow.yml, qui inclut deja des chemins de ml/ et de apps/backend/.
+      - "apps/backend/alembic/**"
+      - "apps/backend/app/models/**"
+      - "apps/backend/tests/test_chaine_ml_api.py"
+      - "apps/backend/pyproject.toml"
+      - "apps/backend/uv.lock"
   pull_request:
     paths:
       - "ml/**"
       - ".github/workflows/ml.yml"
+      # Le job `integration` monte son schema avec les migrations du backend et joue le test de
+      # chaine qui vit dans ses tests : sans ces chemins, une migration modifiee ne declencherait
+      # rien et le schema deriverait du SQL du pipeline sans que rien ne casse. Meme raisonnement
+      # que le filtre d'airflow.yml, qui inclut deja des chemins de ml/ et de apps/backend/.
+      - "apps/backend/alembic/**"
+      - "apps/backend/app/models/**"
+      - "apps/backend/tests/test_chaine_ml_api.py"
+      - "apps/backend/pyproject.toml"
+      - "apps/backend/uv.lock"
 
 permissions:
   contents: read
@@ -53,11 +71,91 @@ jobs:
       - name: Typage
         run: uv run mypy enervision_ml tests
 
-      # Aucun test ne touche PostgreSQL ni MLflow distant : tout tourne sur donnees
+      # Les tests exigeant une base portent le marqueur `integration`, ecarte par defaut et
-      # synthetiques ou un magasin SQLite local jetable (cf. ml/tests/test_train.py).
+      # joue par le job `integration` ci-dessous.
       - name: Tests
         run: uv run pytest
 
+  # Le seul job du depot qui dispose a la fois des deux environnements uv et d'une base. Piege :
+  # le schema de la base ML est celui du backend (apps/backend/alembic, proprietaire du schema).
+  # Le reconstruire ici a la main rendrait ce job vert sur une base qui n'est pas la notre.
+  integration:
+    name: ML - DB et chaîne ML - DB - API
+    runs-on: ubuntu-latest
+
+    services:
+      db:
+        image: timescale/timescaledb-ha:pg17
+        env:
+          POSTGRES_USER: enervision
+          POSTGRES_PASSWORD: change_me
+          POSTGRES_DB: enervision_test
+        ports:
+          - "5433:5432"
+        options: >-
+          --health-cmd "pg_isready -U enervision -d enervision_test"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 12
+          --health-start-period 40s
+
+    env:
+      # Deux variables, deux dialectes : Alembic et l'API parlent asyncpg, le pipeline ML parle
+      # psycopg en synchrone. Cf. docs/ML-START.md, section 1.
+      DATABASE_URL: postgresql+asyncpg://enervision:change_me@localhost:5433/enervision_test
+      ML_DATABASE_URL: postgresql+psycopg://enervision:change_me@localhost:5433/enervision_test
+      APP_SECRET_KEY: secret-de-test-assez-long-pour-le-validateur
+      PGPASSWORD: change_me
+
+    steps:
+      - name: Récupère le dépôt
+        uses: actions/checkout@v7
+
+      - name: Installe uv
+        uses: astral-sh/setup-uv@v7
+        with:
+          enable-cache: true
+          cache-dependency-glob: |
+            ml/uv.lock
+            apps/backend/uv.lock
+
+      - name: Installe l'interpréteur déclaré par .python-version
+        working-directory: ml
+        run: uv python install
+
+      - name: Synchronise le pipeline ML sans dévier du verrou
+        working-directory: ml
+        run: uv sync --all-groups --frozen
+
+      # Le backend est installé ici parce qu'il porte les migrations, seule source du schéma, et
+      # le test de chaîne, qui interroge l'API.
+      - name: Synchronise le backend sans dévier du verrou
+        working-directory: apps/backend
+        run: uv sync --all-groups --frozen
+
+      # db/init/110-test-database.sql n'est pas monté ici, et sans l'extension la première
+      # révision Alembic refuse de s'appliquer.
+      - name: Active TimescaleDB sur la base de test
+        run: psql -h localhost -p 5433 -U enervision -d enervision_test -c "CREATE EXTENSION IF NOT EXISTS timescaledb"
+
+      - name: Applique les migrations du backend, propriétaire du schéma
+        working-directory: apps/backend
+        run: uv run alembic upgrade head
+
+      # `-m` en ligne de commande écrase celui d'addopts. Couverture désactivée : ce job ne joue
+      # qu'une partie de la suite, son taux n'aurait pas de sens (même raison que backend.yml).
+      - name: Tests ML exigeant une base
+        working-directory: ml
+        run: uv run pytest -m integration --no-cov
+
+      # Lance les vrais binaires enervision_ml.train et .score en sous-processus, comme les DAGs
+      # ml_train et ml_score, puis relit le résultat par GET /api/v1/predictions.
+      - name: Chaîne complète ML vers DB vers API
+        working-directory: apps/backend
+        env:
+          ML_PYTHON: ${{ github.workspace }}/ml/.venv/bin/python
+        run: uv run pytest -m chaine --no-cov
+
   sast:
     name: Analyse statique de sécurité
     runs-on: ubuntu-latest

Makefile

@@ -25,6 +25,13 @@ MAILPIT_UI_PORT := $(or $(strip $(call env-val,MAILPIT_UI_PORT)),8025)
 ML_DATABASE_URL ?= postgresql+psycopg://$(PG_USER):$(PG_PASSWORD)@localhost:$(PG_PORT)/$(PG_DB)
 export ML_DATABASE_URL
 
+# Piege : la base des tests d'integration n'est pas la base de developpement. Ces tests ecrivent
+# et suppriment des lignes, et leurs fixtures refusent de demarrer ailleurs que sur
+# `enervision_test` (garde sur le nom, cf. ml/tests/conftest.py).
+PG_TEST_DB ?= enervision_test
+TEST_DATABASE_URL ?= postgresql+asyncpg://$(PG_USER):$(PG_PASSWORD)@localhost:$(PG_PORT)/$(PG_TEST_DB)
+ML_TEST_DATABASE_URL ?= postgresql+psycopg://$(PG_USER):$(PG_PASSWORD)@localhost:$(PG_PORT)/$(PG_TEST_DB)
+
 # Le jeu historique s'arrete au 31/12/2024 : score et detection ancres a l'horloge reelle ne
 # verraient qu'un parc muet depuis des mois. Cf. `--now` de enervision_ml.score.
 DEMO_NOW ?= 2024-12-31T00:00:00Z
@@ -32,9 +39,10 @@ DEMO_NOW ?= 2024-12-31T00:00:00Z
 .DEFAULT_GOAL := help
 .PHONY: help install install-backend install-frontend install-ml install-airflow \
         dev dev-backend dev-frontend \
-        lint format typecheck test test-cov test-integration check \
+        lint format typecheck test test-cov test-integration ml-test-integration \
+        test-chaine check \
         openapi docker-build db-up db-down db-reset db-logs db-psql db-wait db-ensure-airflow \
-        migrate bootstrap-admin services-up demo-data demo-data-force \
+        migrate migrate-test bootstrap-admin services-up demo-data demo-data-force \
         ml-lint ml-typecheck ml-test ml-check ml-train ml-score detect-alerts recommendations \
         airflow-lint airflow-test airflow-check airflow-up airflow-down airflow-logs \
         tls-selfsigned tls-acme tls-renew stack-up stack-down stack-logs
@@ -112,6 +120,16 @@ ml-test: ## Exécute les tests du pipeline ML (donnees synthetiques, sans base n
 
 ml-check: ml-lint ml-typecheck ml-test ## Chaîne de vérification complète du pipeline ML
 
+# La cible surcharge ML_DATABASE_URL, que ce Makefile exporte vers la base de développement : la
+# garde du conftest ferait échouer la cible sans cette surcharge.
+ml-test-integration: ML_DATABASE_URL := $(ML_TEST_DATABASE_URL)
+ml-test-integration: ## Tests ML exigeant une base migrée. Faire `make db-up migrate-test` avant
+	cd $(ML) && uv run pytest -m integration --no-cov
+
+test-chaine: ## Chaîne ML -> DB -> API, vrais binaires. Exige les deux environnements uv
+	cd $(BACKEND) && DATABASE_URL=$(TEST_DATABASE_URL) ML_PYTHON=$(CURDIR)/$(ML)/.venv/bin/python \
+		uv run pytest -m chaine --no-cov
+
 ml-train: ## Entraine le modele LightGBM. CSV=chemin optionnel, sinon lit ML_DATABASE_URL
 	cd $(ML) && uv run python -m enervision_ml.train $(if $(CSV),--csv $(CSV),)
 
@@ -209,6 +227,9 @@ db-ensure-airflow: ## Crée la base de métadonnées Airflow si le volume pgdata
 migrate: ## Applique les migrations Alembic
 	cd $(BACKEND) && uv run alembic upgrade head
 
+migrate-test: ## Applique les migrations sur enervision_test, la base des tests d'intégration
+	cd $(BACKEND) && DATABASE_URL=$(TEST_DATABASE_URL) 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}
 

apps/backend/pyproject.toml

@@ -87,8 +87,11 @@ disallow_untyped_defs = false
 testpaths = ["tests"]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
-addopts = "-q --strict-markers -m 'not integration' --cov=app --cov-report=term-missing"
+addopts = "-q --strict-markers -m 'not integration and not chaine' --cov=app --cov-report=term-missing"
-markers = ["integration: requiert une base PostgreSQL joignable, hors `make test`"]
+markers = [
+    "integration: requiert une base PostgreSQL joignable, hors `make test`",
+    "chaine: requiert en plus l'environnement uv de ml/, hors `make test` et hors `-m integration`",
+]
 
 [tool.coverage.run]
 source = ["app"]

apps/backend/tests/test_chaine_ml_api.py

@@ -0,0 +1,223 @@
+"""Piege : ce fichier porte le marqueur `chaine`, pas `integration` - test_the_ml_binaries...()
+
+Il lance les vrais binaires `enervision_ml.train` et `enervision_ml.score` dans l'environnement
+uv de `ml/`, que le job `integration` de `backend.yml` n'installe pas. Un marqueur distinct evite
+que ce job, et `make test`, ne le selectionnent et n'echouent faute de `ml/.venv`.
+"""
+
+import math
+import os
+import subprocess
+from collections.abc import AsyncIterator, Iterator
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from functools import partial
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+import anyio
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+from sqlalchemy import delete, insert, make_url
+
+from app.api.deps import get_current_principal
+from app.core.config import get_settings
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+from app.db.session import get_session_factory
+from app.models.energy import Prediction, Reading, Site
+
+pytestmark = pytest.mark.chaine
+
+RACINE = Path(__file__).resolve().parents[3]
+ML = RACINE / "ml"
+PYTHON_ML = Path(os.environ.get("ML_PYTHON", ML / ".venv" / "bin" / "python"))
+
+HEURES_COMPLETES = 400
+HEURES_INSUFFISANTES = 100
+
+
+def lecteur() -> Principal:
+    return Principal(
+        id=uuid4(),
+        email="lecteur@enervision.fr",
+        role=Role.LECTEUR,
+        kind=AccountKind.HUMAIN,
+        must_change_password=False,
+    )
+
+
+def url_ml() -> str:
+    """Derive la chaine du pipeline de celle du backend plutot que de la recopier : les deux
+    cotes visent ainsi la meme base, dans leur dialecte respectif."""
+    return (
+        make_url(get_settings().database_url)
+        .set(drivername="postgresql+psycopg")
+        .render_as_string(hide_password=False)
+    )
+
+
+def lance_ml(module: str, *arguments: str, journal: Path) -> subprocess.CompletedProcess[str]:
+    if not PYTHON_ML.exists():
+        pytest.fail(
+            f"Environnement ml/ absent ({PYTHON_ML}). Lancer `cd ml && uv sync --all-groups`."
+        )
+
+    return subprocess.run(  # noqa: S603 -- argv en liste, sans shell, binaire resolu dans le depot
+        [str(PYTHON_ML), "-m", module, *arguments],
+        cwd=ML,
+        text=True,
+        capture_output=True,
+        timeout=600,
+        check=False,
+        env={
+            **os.environ,
+            "ML_DATABASE_URL": url_ml(),
+            "MLFLOW_TRACKING_URI": f"sqlite:///{journal}/mlflow.db",
+        },
+    )
+
+
+async def executer(module: str, *arguments: str, journal: Path) -> subprocess.CompletedProcess[str]:
+    resultat = await anyio.to_thread.run_sync(
+        partial(lance_ml, module, *arguments, journal=journal)
+    )
+    assert resultat.returncode == 0, resultat.stderr
+    return resultat
+
+
+@dataclass
+class Parc:
+    sites: list[str] = field(default_factory=list)
+
+
+def lignes_horaires(site_id: str, *, heures: int, fin: datetime) -> list[dict[str, Any]]:
+    return [
+        {
+            "site_id": site_id,
+            "timestamp": fin - timedelta(hours=decalage),
+            "source": "api_history",
+            "consumption_kwh": 50.0 + math.sin(decalage / 12.0) * 10.0,
+            "temperature_celsius": 15.0,
+            "humidity_percent": 50.0,
+            "solar_irradiance_wm2": 0.0,
+            "is_working_hours": True,
+            "raw_data": {},
+        }
+        for decalage in reversed(range(heures))
+    ]
+
+
+@pytest.fixture
+async def parc() -> AsyncIterator[Parc]:
+    """Deux sites dotes d'un historique complet, un troisieme qui n'atteint pas le lag de 168 h.
+
+    Les ecritures sont validees : les binaires ML ouvrent leur propre connexion et ne verraient
+    pas une transaction en cours.
+    """
+    fin = datetime.now(UTC).replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
+    marque = uuid4().hex[:12]
+    complets = [f"TEST-{marque}-A", f"TEST-{marque}-B"]
+    partiel = f"TEST-{marque}-C"
+    parc = Parc(sites=[*complets, partiel])
+
+    async with get_session_factory()() as session:
+        await session.execute(
+            insert(Site),
+            [
+                {
+                    "site_id": site_id,
+                    "site_name": f"Site {site_id}",
+                    "site_type": "office",
+                    "capacity_kw": 100.0,
+                }
+                for site_id in parc.sites
+            ],
+        )
+        for site_id in complets:
+            await session.execute(
+                insert(Reading), lignes_horaires(site_id, heures=HEURES_COMPLETES, fin=fin)
+            )
+        await session.execute(
+            insert(Reading), lignes_horaires(partiel, heures=HEURES_INSUFFISANTES, fin=fin)
+        )
+        await session.commit()
+
+    try:
+        yield parc
+    finally:
+        async with get_session_factory()() as session:
+            await session.execute(delete(Prediction).where(Prediction.site_id.in_(parc.sites)))
+            await session.execute(delete(Reading).where(Reading.site_id.in_(parc.sites)))
+            await session.execute(delete(Site).where(Site.site_id.in_(parc.sites)))
+            await session.commit()
+
+
+@pytest.fixture
+def principal_lecteur(app: FastAPI) -> Iterator[None]:
+    app.dependency_overrides[get_current_principal] = lecteur
+    yield
+    app.dependency_overrides.pop(get_current_principal, None)
+
+
+async def resume_du_site(client: AsyncClient, site_id: str) -> dict[str, Any]:
+    reponse = await client.get("/api/v1/predictions")
+
+    assert reponse.status_code == 200
+    sites = reponse.json()["sites"]
+    return next(site for site in sites if site["site_id"] == site_id)
+
+
+async def entraine_et_score(parc: Parc, tmp_path: Path, *arguments: str) -> Path:
+    modele = tmp_path / "lightgbm-consumption.txt"
+
+    await executer(
+        "enervision_ml.train",
+        "--model-output",
+        str(modele),
+        "--mlflow-tracking-uri",
+        f"sqlite:///{tmp_path}/mlflow.db",
+        journal=tmp_path,
+    )
+    await executer("enervision_ml.score", "--model", str(modele), *arguments, journal=tmp_path)
+
+    return modele
+
+
+async def test_the_ml_binaries_produce_a_prediction_that_the_api_serves(
+    parc: Parc, tmp_path: Path, client: AsyncClient, principal_lecteur: None
+) -> None:
+    await entraine_et_score(parc, tmp_path)
+
+    servi = await resume_du_site(client, parc.sites[0])
+
+    assert servi["prediction"]["status"] == "available"
+    assert servi["prediction"]["predicted_value"] is not None
+    assert servi["prediction"]["target_metric"] == "consumption_kwh"
+
+
+async def test_the_api_exposes_the_failure_reason_of_a_site_without_enough_history(
+    parc: Parc, tmp_path: Path, client: AsyncClient, principal_lecteur: None
+) -> None:
+    await entraine_et_score(parc, tmp_path)
+
+    servi = await resume_du_site(client, parc.sites[-1])
+
+    assert servi["prediction"]["status"] == "insufficient_data"
+    assert servi["prediction"]["predicted_value"] is None
+    assert servi["prediction"]["failure_reason"] is not None
+
+
+async def test_the_api_serves_the_latest_run_when_the_score_cli_runs_twice(
+    parc: Parc, tmp_path: Path, client: AsyncClient, principal_lecteur: None
+) -> None:
+    modele = await entraine_et_score(parc, tmp_path)
+    premier = await resume_du_site(client, parc.sites[0])
+
+    await executer("enervision_ml.score", "--model", str(modele), journal=tmp_path)
+
+    second = await resume_du_site(client, parc.sites[0])
+    assert second["prediction"]["created_at"] >= premier["prediction"]["created_at"]
+    assert second["prediction"]["model_reference"] == premier["prediction"]["model_reference"]

ml/tests/conftest.py

@@ -0,0 +1,332 @@
+"""Piege : deux fixtures d'acces a la base, jamais interchangeables - `connexion_ml` et `parc`.
+
+`connexion_ml` ouvre une transaction annulee a la fin du test : rien ne subsiste, et rien n'est
+visible hors de cette connexion. Elle sert aux fonctions qui recoivent leur connexion en
+argument (`load_from_database`, `load_recent_from_database`, `write_predictions`).
+
+`run_scoring` fabrique en revanche son propre engine depuis `ML_DATABASE_URL` : il ne verrait
+pas des lignes semees dans une transaction non validee, et ses propres ecritures survivraient a
+l'annulation. Les tests qui l'appellent passent donc par `parc`, qui valide ce qu'il ecrit et
+nettoie lui-meme, dans l'ordre impose par les cles etrangeres `RESTRICT`.
+"""
+
+import math
+import os
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+import lightgbm as lgb
+import pandas as pd
+import pytest
+from sqlalchemy import Connection, Engine, Row, bindparam, create_engine, text
+from sqlalchemy.engine import URL, make_url
+
+from enervision_ml.features import TARGET_COLUMN, build_features, feature_columns
+
+BASE_ATTENDUE = "enervision_test"
+
+# Piege : `load_from_database` lit toute la table, et `enervision_test` est partagee entre un run
+# local et la CI. Les tests ancrent donc leurs lectures au-dela de tout jeu de donnees reel
+# (l'historique s'arrete au 31/12/2024) pour que leur borne `since` ne ramene qu'eux.
+ANCRAGE = datetime(2035, 1, 1, tzinfo=UTC)
+
+SITE_TYPE = "office"
+CAPACITY_KW = 100.0
+
+_INSERT_SITE = text(
+    """
+    INSERT INTO site (site_id, site_name, site_type, capacity_kw)
+    VALUES (:site_id, :site_name, :site_type, :capacity_kw)
+    """
+)
+
+# `source = 'api_history'` impose `dataset_id IS NULL` (ck_reading_dataset_source), ce qui evite
+# de creer une ligne `dataset`. `raw_data` est NOT NULL, d'ou le litteral jsonb.
+_INSERT_READING = text(
+    """
+    INSERT INTO reading (
+        site_id, timestamp, source, consumption_kwh, temperature_celsius,
+        humidity_percent, solar_irradiance_wm2, is_working_hours, raw_data
+    ) VALUES (
+        :site_id, :timestamp, :source, :consumption_kwh, :temperature_celsius,
+        :humidity_percent, :solar_irradiance_wm2, :is_working_hours, '{}'::jsonb
+    )
+    """
+)
+
+_SELECT_PREDICTIONS = text(
+    """
+    SELECT target_at, predicted_value, status, failure_reason, model_reference
+    FROM prediction
+    WHERE site_id = :site_id
+    ORDER BY prediction_id
+    """
+)
+
+_INSERT_PREDICTION = text(
+    """
+    INSERT INTO prediction (
+        site_id, target_at, target_metric, period_minutes,
+        predicted_value, model_reference, status, failure_reason
+    ) VALUES (
+        :site_id, :target_at, 'consumption_kwh', 60,
+        :predicted_value, :model_reference, :status, :failure_reason
+    )
+    """
+)
+
+
+# Ordre impose par les cles etrangeres `RESTRICT` : une lecture avant son site, une prediction
+# avant sa lecture.
+_SUPPRESSIONS = tuple(
+    text(requete).bindparams(bindparam("sites", expanding=True))
+    for requete in (
+        "DELETE FROM prediction WHERE site_id IN :sites",
+        "DELETE FROM reading WHERE site_id IN :sites",
+        "DELETE FROM site WHERE site_id IN :sites",
+    )
+)
+
+
+def insere_site(
+    connexion: Connection,
+    *,
+    site_type: str = SITE_TYPE,
+    capacity_kw: float | None = CAPACITY_KW,
+) -> str:
+    site_id = f"TEST-{uuid4().hex[:12]}"
+    connexion.execute(
+        _INSERT_SITE,
+        {
+            "site_id": site_id,
+            "site_name": "Site de test",
+            "site_type": site_type,
+            "capacity_kw": capacity_kw,
+        },
+    )
+    return site_id
+
+
+def insere_lectures(
+    connexion: Connection,
+    site_id: str,
+    *,
+    heures: int,
+    fin: datetime,
+    valeur: float = 50.0,
+    source: str = "api_history",
+    is_working_hours: bool | None = True,
+) -> list[datetime]:
+    """Grille horaire contigue finissant a `fin`, incluse.
+
+    Contigue parce que les lags de `build_features` sont des `shift()` positionnels : un trou
+    dans la grille decalerait le lag de 168 h sans qu'aucune erreur ne se declenche.
+    """
+    instants = [fin - timedelta(hours=decalage) for decalage in reversed(range(heures))]
+    connexion.execute(
+        _INSERT_READING,
+        [
+            {
+                "site_id": site_id,
+                "timestamp": instant,
+                "source": source,
+                "consumption_kwh": valeur + math.sin(rang / 12.0) * 10.0,
+                "temperature_celsius": 15.0,
+                "humidity_percent": 50.0,
+                "solar_irradiance_wm2": 0.0,
+                "is_working_hours": is_working_hours,
+            }
+            for rang, instant in enumerate(instants)
+        ],
+    )
+    return instants
+
+
+def insere_lecture(
+    connexion: Connection,
+    site_id: str,
+    *,
+    instant: datetime,
+    consumption_kwh: float | None = 50.0,
+    source: str = "api_history",
+    is_working_hours: bool | None = True,
+) -> None:
+    """Une lecture isolee, quand le test pilote sa valeur plutot que sa forme."""
+    connexion.execute(
+        _INSERT_READING,
+        {
+            "site_id": site_id,
+            "timestamp": instant,
+            "source": source,
+            "consumption_kwh": consumption_kwh,
+            "temperature_celsius": 15.0,
+            "humidity_percent": 50.0,
+            "solar_irradiance_wm2": 0.0,
+            "is_working_hours": is_working_hours,
+        },
+    )
+
+
+def insere_prediction(
+    connexion: Connection,
+    site_id: str,
+    *,
+    target_at: datetime,
+    predicted_value: float | None = 42.0,
+    model_reference: str = "lightgbm-test000000",
+    status: str = "available",
+    failure_reason: str | None = None,
+) -> None:
+    connexion.execute(
+        _INSERT_PREDICTION,
+        {
+            "site_id": site_id,
+            "target_at": target_at,
+            "predicted_value": predicted_value,
+            "model_reference": model_reference,
+            "status": status,
+            "failure_reason": failure_reason,
+        },
+    )
+
+
+@pytest.fixture(scope="session")
+def url_ml() -> URL:
+    valeur = os.environ.get("ML_DATABASE_URL")
+    if not valeur:
+        pytest.fail("ML_DATABASE_URL absente. Voir `make ml-test-integration`.")
+
+    url = make_url(valeur)
+    if url.database != BASE_ATTENDUE:
+        pytest.fail(
+            f"Ces tests ecrivent et suppriment : ML_DATABASE_URL doit viser {BASE_ATTENDUE}, "
+            f"pas {url.database}."
+        )
+    return url
+
+
+@pytest.fixture(scope="session")
+def moteur_ml(url_ml: URL) -> Iterator[Engine]:
+    moteur = create_engine(url_ml)
+    try:
+        yield moteur
+    finally:
+        moteur.dispose()
+
+
+@pytest.fixture
+def connexion_ml(moteur_ml: Engine) -> Iterator[Connection]:
+    with moteur_ml.connect() as connexion:
+        transaction = connexion.begin()
+        try:
+            yield connexion
+        finally:
+            transaction.rollback()
+
+
+@dataclass
+class Parc:
+    """Semis valide en base, et son nettoyage, pour les tests qui appellent `run_scoring`.
+
+    Chaque `site_id` porte une marque unique : la base de test est partagee entre un run local
+    et la CI.
+    """
+
+    moteur: Engine
+    sites: list[str] = field(default_factory=list)
+
+    def site(self, *, site_type: str = SITE_TYPE, capacity_kw: float | None = CAPACITY_KW) -> str:
+        with self.moteur.begin() as connexion:
+            site_id = insere_site(connexion, site_type=site_type, capacity_kw=capacity_kw)
+        self.sites.append(site_id)
+        return site_id
+
+    def lectures(self, site_id: str, **arguments: Any) -> list[datetime]:
+        with self.moteur.begin() as connexion:
+            return insere_lectures(connexion, site_id, **arguments)
+
+    def lecture(self, site_id: str, **arguments: Any) -> None:
+        with self.moteur.begin() as connexion:
+            insere_lecture(connexion, site_id, **arguments)
+
+    def prediction(self, site_id: str, **arguments: Any) -> None:
+        with self.moteur.begin() as connexion:
+            insere_prediction(connexion, site_id, **arguments)
+
+    def predictions_ecrites(self, site_id: str) -> list[Row[Any]]:
+        with self.moteur.connect() as connexion:
+            return list(connexion.execute(_SELECT_PREDICTIONS, {"site_id": site_id}))
+
+    def nettoie(self) -> None:
+        if not self.sites:
+            return
+
+        with self.moteur.begin() as connexion:
+            for suppression in _SUPPRESSIONS:
+                connexion.execute(suppression, {"sites": self.sites})
+
+
+@pytest.fixture
+def parc(moteur_ml: Engine) -> Iterator[Parc]:
+    semis = Parc(moteur=moteur_ml)
+    try:
+        yield semis
+    finally:
+        semis.nettoie()
+
+
+def trame_synthetique(*, sites: int = 2, heures: int = 400) -> pd.DataFrame:
+    """Lectures horaires deterministes, assez longues pour que le lag de 168 h existe."""
+    depart = datetime(2024, 1, 1, tzinfo=UTC)
+    morceaux = [
+        pd.DataFrame(
+            {
+                "site_id": f"SITE{numero:03d}",
+                "timestamp": [depart + timedelta(hours=rang) for rang in range(heures)],
+                TARGET_COLUMN: [
+                    50.0 + 10.0 * math.sin(rang / 12.0) + numero * 5.0 for rang in range(heures)
+                ],
+                "temperature_celsius": 15.0,
+                "humidity_percent": 50.0,
+                "solar_irradiance_wm2": 0.0,
+                "is_working_hours": True,
+                "site_type": SITE_TYPE,
+                "capacity_kw": CAPACITY_KW,
+            }
+        )
+        for numero in range(sites)
+    ]
+    return pd.concat(morceaux, ignore_index=True)
+
+
+@pytest.fixture(scope="session")
+def modele_jetable(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Booster reel entraine sur une trame synthetique, ecrit dans un repertoire temporaire.
+
+    Ni `ml/models/` (ignore par git, et le polluer serait un effet de bord), ni
+    `enervision_ml.train.train()` (qui journalise dans MLflow sans garde). Le typage `category`
+    de `site_type` reproduit celui de l'entrainement : c'est le `pandas_categorical` enregistre
+    dans le modele que `score()` devra retrouver.
+    """
+    features = build_features(trame_synthetique()).dropna(subset=feature_columns())
+    typee = features.copy()
+    typee["site_type"] = typee["site_type"].astype("category")
+
+    donnees = lgb.Dataset(
+        typee[feature_columns()],
+        label=typee[TARGET_COLUMN],
+        categorical_feature=["site_type"],
+    )
+    booster = lgb.train(
+        {"objective": "regression", "num_leaves": 7, "min_data_in_leaf": 5, "verbosity": -1},
+        donnees,
+        num_boost_round=5,
+    )
+
+    chemin = tmp_path_factory.mktemp("modele") / "lightgbm-consumption.txt"
+    booster.save_model(str(chemin))
+    return chemin

ml/tests/test_data_integration.py

@@ -0,0 +1,154 @@
+from datetime import timedelta
+from pathlib import Path
+
+import pandas as pd
+import pytest
+from sqlalchemy import Connection
+
+from enervision_ml.data import (
+    OUTPUT_COLUMNS,
+    load_from_csv,
+    load_from_database,
+    load_recent_from_database,
+)
+from tests.conftest import ANCRAGE, insere_lecture, insere_lectures, insere_site
+
+pytestmark = pytest.mark.integration
+
+
+def test_load_from_database_returns_the_nine_contract_columns(connexion_ml: Connection) -> None:
+    site_id = insere_site(connexion_ml)
+    insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE)
+
+    frame = load_from_database(connexion_ml)
+
+    assert list(frame.columns) == OUTPUT_COLUMNS
+
+
+def test_load_from_database_joins_the_site_attributes_to_every_reading(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml, site_type="factory", capacity_kw=250.0)
+    insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE)
+
+    frame = load_from_database(connexion_ml)
+
+    mien = frame[frame["site_id"] == site_id]
+    assert len(mien) == 3
+    assert set(mien["site_type"]) == {"factory"}
+    assert set(mien["capacity_kw"]) == {250.0}
+
+
+def test_load_recent_from_database_excludes_readings_before_the_since_bound(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    insere_lectures(connexion_ml, site_id, heures=5, fin=ANCRAGE)
+
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=2))
+
+    assert list(frame["timestamp"]) == [
+        ANCRAGE - timedelta(hours=2),
+        ANCRAGE - timedelta(hours=1),
+        ANCRAGE,
+    ]
+
+
+def test_load_recent_from_database_includes_a_reading_exactly_at_the_since_bound(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    insere_lecture(connexion_ml, site_id, instant=ANCRAGE)
+
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE)
+
+    assert len(frame) == 1
+
+
+def test_load_recent_from_database_keeps_timestamps_timezone_aware(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    insere_lecture(connexion_ml, site_id, instant=ANCRAGE)
+
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE)
+
+    assert frame["timestamp"].dt.tz is not None
+
+
+def test_load_recent_from_database_orders_readings_by_site_then_timestamp(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    for decalage in (2, 0, 1):
+        insere_lecture(connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=decalage))
+
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE)
+
+    assert list(frame["timestamp"]) == [
+        ANCRAGE,
+        ANCRAGE + timedelta(hours=1),
+        ANCRAGE + timedelta(hours=2),
+    ]
+
+
+def test_load_recent_from_database_returns_the_contract_columns_even_without_any_row(
+    connexion_ml: Connection,
+) -> None:
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE + timedelta(days=365))
+
+    assert frame.empty
+    assert list(frame.columns) == OUTPUT_COLUMNS
+
+
+def test_load_recent_from_database_types_a_fully_null_capacity_kw_as_float64(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml, capacity_kw=None)
+    insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE)
+
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=2))
+
+    assert frame["capacity_kw"].dtype == "float64"
+    assert frame["capacity_kw"].isna().all()
+
+
+def test_load_recent_from_database_types_a_null_is_working_hours_as_float64(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    insere_lecture(connexion_ml, site_id, instant=ANCRAGE, is_working_hours=None)
+    insere_lecture(
+        connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=1), is_working_hours=True
+    )
+
+    frame = load_recent_from_database(connexion_ml, since=ANCRAGE)
+
+    assert frame["is_working_hours"].dtype == "float64"
+    assert list(frame["is_working_hours"].isna()) == [True, False]
+
+
+def test_both_loaders_produce_the_same_columns_in_the_same_order(
+    connexion_ml: Connection, tmp_path: Path
+) -> None:
+    site_id = insere_site(connexion_ml)
+    insere_lectures(connexion_ml, site_id, heures=2, fin=ANCRAGE)
+    csv_path = tmp_path / "lectures.csv"
+    pd.DataFrame(
+        {
+            "site_id": [site_id],
+            "timestamp": [ANCRAGE],
+            "consumption_kwh": [50.0],
+            "temperature_celsius": [15.0],
+            "humidity_percent": [50.0],
+            "solar_irradiance_wm2": [0.0],
+            "is_working_hours": [True],
+            "site_type": ["office"],
+        }
+    ).to_csv(csv_path, index=False)
+
+    depuis_la_base = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=1))
+    depuis_le_csv = load_from_csv(csv_path)
+
+    assert list(depuis_la_base.columns) == list(depuis_le_csv.columns)
+    assert depuis_la_base.dtypes.to_dict() == depuis_le_csv.dtypes.to_dict()

ml/tests/test_score_integration.py

@@ -0,0 +1,225 @@
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+from sqlalchemy import Connection, Row, text
+from sqlalchemy.exc import IntegrityError
+
+from enervision_ml.score import (
+    INSUFFICIENT_DATA_REASON,
+    LOOKBACK,
+    MAX_STALENESS,
+    ScoredSite,
+    model_reference,
+    run_scoring,
+    write_predictions,
+)
+from tests.conftest import ANCRAGE, Parc, insere_site
+
+pytestmark = pytest.mark.integration
+
+REFERENCE = "lightgbm-000000000000"
+
+_SELECT = text(
+    """
+    SELECT target_at, target_metric, period_minutes, predicted_value,
+           model_reference, status, failure_reason
+    FROM prediction
+    WHERE site_id = :site_id
+    ORDER BY prediction_id
+    """
+)
+
+
+def lignes(connexion: Connection, site_id: str) -> list[Row[Any]]:
+    return list(connexion.execute(_SELECT, {"site_id": site_id}))
+
+
+def disponible(
+    site_id: str,
+    *,
+    target_at: datetime = ANCRAGE,
+    predicted_value: float | None = 12.5,
+) -> ScoredSite:
+    return ScoredSite(
+        site_id=site_id,
+        target_at=target_at,
+        status="available",
+        predicted_value=predicted_value,
+        failure_reason=None,
+    )
+
+
+def test_write_predictions_inserts_one_row_per_scored_site(connexion_ml: Connection) -> None:
+    premier = insere_site(connexion_ml)
+    second = insere_site(connexion_ml)
+
+    write_predictions(connexion_ml, [disponible(premier), disponible(second)], reference=REFERENCE)
+
+    assert len(lignes(connexion_ml, premier)) == 1
+    assert len(lignes(connexion_ml, second)) == 1
+
+
+def test_write_predictions_stores_the_model_reference_and_the_hourly_period(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+
+    write_predictions(connexion_ml, [disponible(site_id)], reference=REFERENCE)
+
+    ligne = lignes(connexion_ml, site_id)[0]
+    assert ligne.model_reference == REFERENCE
+    assert ligne.target_metric == "consumption_kwh"
+    assert ligne.period_minutes == 60
+
+
+def test_write_predictions_stacks_a_second_run_instead_of_overwriting_the_first(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+
+    write_predictions(
+        connexion_ml, [disponible(site_id, predicted_value=10.0)], reference=REFERENCE
+    )
+    write_predictions(
+        connexion_ml, [disponible(site_id, predicted_value=20.0)], reference=REFERENCE
+    )
+
+    assert [ligne.predicted_value for ligne in lignes(connexion_ml, site_id)] == [10.0, 20.0]
+
+
+def test_write_predictions_writes_nothing_when_no_site_was_scored(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+
+    write_predictions(connexion_ml, [], reference=REFERENCE)
+
+    assert lignes(connexion_ml, site_id) == []
+
+
+def test_write_predictions_rejects_an_available_row_without_a_predicted_value(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+
+    with pytest.raises(IntegrityError, match="ck_prediction_status"):
+        write_predictions(
+            connexion_ml, [disponible(site_id, predicted_value=None)], reference=REFERENCE
+        )
+
+
+def test_write_predictions_rejects_an_insufficient_data_row_carrying_a_value(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    incoherent = ScoredSite(
+        site_id=site_id,
+        target_at=ANCRAGE,
+        status="insufficient_data",
+        predicted_value=12.5,
+        failure_reason=INSUFFICIENT_DATA_REASON,
+    )
+
+    with pytest.raises(IntegrityError, match="ck_prediction_status"):
+        write_predictions(connexion_ml, [incoherent], reference=REFERENCE)
+
+
+def test_write_predictions_rejects_a_prediction_for_an_unknown_site(
+    connexion_ml: Connection,
+) -> None:
+    with pytest.raises(IntegrityError, match="fk_prediction_site"):
+        write_predictions(connexion_ml, [disponible("SITE-INCONNU")], reference=REFERENCE)
+
+
+def test_run_scoring_writes_an_available_prediction_for_a_site_with_a_full_week(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    site_id = parc.site()
+    parc.lectures(site_id, heures=200, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, now=ANCRAGE)
+
+    ligne = parc.predictions_ecrites(site_id)[0]
+    assert ligne.status == "available"
+    assert ligne.predicted_value is not None
+    assert ligne.target_at == ANCRAGE + timedelta(hours=1)
+
+
+def test_run_scoring_writes_insufficient_data_when_the_weekly_lag_is_missing(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    site_id = parc.site()
+    parc.lectures(site_id, heures=100, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, now=ANCRAGE)
+
+    ligne = parc.predictions_ecrites(site_id)[0]
+    assert ligne.status == "insufficient_data"
+    assert ligne.predicted_value is None
+    assert ligne.failure_reason == INSUFFICIENT_DATA_REASON
+
+
+def test_run_scoring_writes_a_staleness_reason_when_the_last_reading_is_too_old(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    site_id = parc.site()
+    parc.lectures(site_id, heures=200, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, now=ANCRAGE + MAX_STALENESS + timedelta(hours=1))
+
+    ligne = parc.predictions_ecrites(site_id)[0]
+    assert ligne.status == "insufficient_data"
+    assert ligne.failure_reason != INSUFFICIENT_DATA_REASON
+
+
+def test_run_scoring_writes_nothing_when_every_reading_is_older_than_the_window(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    site_id = parc.site()
+    parc.lectures(site_id, heures=200, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, now=ANCRAGE + LOOKBACK + timedelta(days=1))
+
+    assert parc.predictions_ecrites(site_id) == []
+
+
+def test_run_scoring_only_writes_the_site_that_was_requested(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    demande = parc.site()
+    ignore = parc.site()
+    parc.lectures(demande, heures=200, fin=ANCRAGE)
+    parc.lectures(ignore, heures=200, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, site_id=demande, now=ANCRAGE)
+
+    assert len(parc.predictions_ecrites(demande)) == 1
+    assert parc.predictions_ecrites(ignore) == []
+
+
+def test_run_scoring_uses_the_model_file_hash_as_model_reference(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    site_id = parc.site()
+    parc.lectures(site_id, heures=200, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, now=ANCRAGE)
+
+    ligne = parc.predictions_ecrites(site_id)[0]
+    assert ligne.model_reference == model_reference(modele_jetable)
+
+
+def test_run_scoring_appends_a_second_row_when_it_runs_twice(
+    parc: Parc, modele_jetable: Path
+) -> None:
+    site_id = parc.site()
+    parc.lectures(site_id, heures=200, fin=ANCRAGE)
+
+    run_scoring(model_path=modele_jetable, now=ANCRAGE)
+    run_scoring(model_path=modele_jetable, now=ANCRAGE)
+
+    ecrites = parc.predictions_ecrites(site_id)
+    assert len(ecrites) == 2
+    assert ecrites[0].target_at == ecrites[1].target_at