Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d506e8f7ef |
@@ -8,12 +8,12 @@ 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
|
||||
# Piege : `backend.yml` ne joue jamais `-m chaine`, ce job est le seul. Le test de chaine
|
||||
# traverse tout `apps/backend/app` jusqu'a GET /predictions : restreindre le filtre aux
|
||||
# migrations et aux modeles le laisserait muet sur la PR meme qui le 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/app/**"
|
||||
- "apps/backend/tests/test_chaine_ml_api.py"
|
||||
- "apps/backend/pyproject.toml"
|
||||
- "apps/backend/uv.lock"
|
||||
@@ -21,12 +21,12 @@ 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
|
||||
# Piege : `backend.yml` ne joue jamais `-m chaine`, ce job est le seul. Le test de chaine
|
||||
# traverse tout `apps/backend/app` jusqu'a GET /predictions : restreindre le filtre aux
|
||||
# migrations et aux modeles le laisserait muet sur la PR meme qui le 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/app/**"
|
||||
- "apps/backend/tests/test_chaine_ml_api.py"
|
||||
- "apps/backend/pyproject.toml"
|
||||
- "apps/backend/uv.lock"
|
||||
|
||||
@@ -8,6 +8,10 @@ même heure plusieurs fois, et la moyenne d'erreur pèserait ces sites en double
|
||||
On retient la prédiction du run le plus récent, celle que sert `GET /api/v1/predictions`, avec
|
||||
`prediction_id` en départage : `created_at` vaut l'heure de début de transaction et ne
|
||||
distingue pas deux lignes du même run.
|
||||
|
||||
Côté lectures, le départage est `reading_id` décroissant, la règle même de
|
||||
`GET /api/v1/sites/{site_id}/current`. La dérive se mesure donc contre le réalisé que l'API
|
||||
affiche, et non contre une source élue ici et nulle part ailleurs.
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
@@ -95,7 +95,10 @@ class DriftService:
|
||||
self, *, now: datetime | None = None, site_id: str | None = None
|
||||
) -> list[NouveauRapportDerive]:
|
||||
"""Une ligne par site, plus une ligne globale dont le `site_id` est nul."""
|
||||
fin = (now or datetime.now(UTC)) - self._seuils.grace
|
||||
# Piège : `window_end` est la clé d'idempotence de `uq_drift_report_window`. Sans
|
||||
# troncature à l'heure, deux exécutions ne collident jamais, aux microsecondes près.
|
||||
instant = (now or datetime.now(UTC)).replace(minute=0, second=0, microsecond=0)
|
||||
fin = instant - self._seuils.grace
|
||||
debut = fin - self._seuils.fenetre
|
||||
reference_fin = debut
|
||||
reference_debut = reference_fin - self._seuils.fenetre
|
||||
@@ -132,8 +135,12 @@ class DriftService:
|
||||
anciennes=[p for p in anciennes if p.site_id == site],
|
||||
comptages=[c for c in comptages if c.site_id == site],
|
||||
)
|
||||
# Piège : un site présent dans la référence et absent de la fenêtre récente a cessé
|
||||
# d'être scoré. C'est la panne à crier, pas une ligne à omettre.
|
||||
for site in sorted(
|
||||
{paire.site_id for paire in recentes} | {c.site_id for c in comptages}
|
||||
{paire.site_id for paire in recentes}
|
||||
| {paire.site_id for paire in anciennes}
|
||||
| {c.site_id for c in comptages}
|
||||
)
|
||||
]
|
||||
rapports.append(
|
||||
@@ -191,7 +198,14 @@ class DriftService:
|
||||
)
|
||||
|
||||
plafond = _plafond(reference_mae, ratio=seuils.ratio_derive, plancher=seuils.mae_plancher)
|
||||
if metriques.mae is not None and plafond is not None and metriques.mae > plafond:
|
||||
if plafond is None:
|
||||
return Verdict(
|
||||
STATUT_INDETERMINE,
|
||||
"Fenêtre de référence sans paire vérifiée, et aucun plancher de MAE : "
|
||||
"rien à quoi comparer cette fenêtre.",
|
||||
)
|
||||
|
||||
if metriques.mae is not None and metriques.mae > plafond:
|
||||
return Verdict(
|
||||
STATUT_DERIVE,
|
||||
f"MAE de {metriques.mae:.2f} kWh au-delà de {plafond:.2f} kWh, "
|
||||
|
||||
@@ -25,7 +25,7 @@ from app.api.deps import get_current_principal
|
||||
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 Alert, Prediction, Reading, Recommendation, Site
|
||||
from app.models.energy import Alert, DriftReport, Prediction, Reading, Recommendation, Site
|
||||
from tests.repositories.test_alert import creer_alerte
|
||||
from tests.repositories.test_prediction import creer_prediction
|
||||
from tests.repositories.test_reading import creer_lecture
|
||||
@@ -52,6 +52,7 @@ async def _supprime(session: AsyncSession, sites: list[str]) -> None:
|
||||
alertes = select(Alert.alert_id).where(Alert.site_id.in_(sites))
|
||||
await session.execute(delete(Recommendation).where(Recommendation.alert_id.in_(alertes)))
|
||||
await session.execute(delete(Alert).where(Alert.site_id.in_(sites)))
|
||||
await session.execute(delete(DriftReport).where(DriftReport.site_id.in_(sites)))
|
||||
await session.execute(delete(Prediction).where(Prediction.site_id.in_(sites)))
|
||||
await session.execute(delete(Reading).where(Reading.site_id.in_(sites)))
|
||||
await session.execute(delete(Site).where(Site.site_id.in_(sites)))
|
||||
|
||||
@@ -63,7 +63,8 @@ def servi(app: FastAPI) -> Iterator[list[DriftReport]]:
|
||||
app.dependency_overrides[get_current_principal] = operateur
|
||||
app.dependency_overrides[get_drift_service] = lambda: service
|
||||
yield rapports
|
||||
app.dependency_overrides.clear()
|
||||
app.dependency_overrides.pop(get_current_principal, None)
|
||||
app.dependency_overrides.pop(get_drift_service, None)
|
||||
|
||||
|
||||
async def test_drift_returns_the_latest_report_of_every_site(
|
||||
@@ -86,14 +87,19 @@ async def test_drift_exposes_the_metrics_of_the_stored_report(
|
||||
assert premier["model_references"] == ["lightgbm-aaa"]
|
||||
|
||||
|
||||
async def test_drift_returns_an_empty_list_when_no_report_exists(
|
||||
app: FastAPI, client: AsyncClient
|
||||
) -> None:
|
||||
@pytest.fixture
|
||||
def sans_rapport(app: FastAPI) -> Iterator[None]:
|
||||
app.dependency_overrides[get_current_principal] = operateur
|
||||
app.dependency_overrides[get_drift_service] = lambda: FauxService([])
|
||||
yield
|
||||
app.dependency_overrides.pop(get_current_principal, None)
|
||||
app.dependency_overrides.pop(get_drift_service, None)
|
||||
|
||||
|
||||
async def test_drift_returns_an_empty_list_when_no_report_exists(
|
||||
sans_rapport: None, client: AsyncClient
|
||||
) -> None:
|
||||
reponse = await client.get("/api/v1/monitoring/drift")
|
||||
|
||||
assert reponse.status_code == 200
|
||||
assert reponse.json() == []
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
@@ -142,6 +142,19 @@ async def test_drift_reports_stable_when_the_recent_mae_stays_close_to_the_refer
|
||||
assert global_.reason is None
|
||||
|
||||
|
||||
async def test_drift_reports_indetermine_when_the_reference_window_is_empty() -> None:
|
||||
depot = FauxDepot(
|
||||
recentes=paires(nombre=30, prevu=14.0, reel=10.0),
|
||||
comptages=[ComptageStatut(site_id="SITE001", status="available", nombre=30)],
|
||||
)
|
||||
|
||||
rapports = await service(depot, min_observations=10).evaluate(now=INSTANT)
|
||||
|
||||
global_ = next(rapport for rapport in rapports if rapport.site_id is None)
|
||||
assert global_.status == STATUT_INDETERMINE
|
||||
assert global_.reference_mae is None
|
||||
|
||||
|
||||
async def test_drift_reports_derive_when_the_coverage_ratio_falls_under_the_threshold() -> None:
|
||||
depot = FauxDepot(
|
||||
recentes=paires(nombre=30, prevu=10.0, reel=10.0),
|
||||
@@ -174,6 +187,23 @@ async def test_drift_reports_one_line_per_site_and_one_global_line() -> None:
|
||||
assert next(r for r in rapports if r.site_id is None).n_observations == 24
|
||||
|
||||
|
||||
async def test_drift_still_reports_a_site_that_stopped_being_scored() -> None:
|
||||
depot = FauxDepot(
|
||||
recentes=paires(site_id="SITE001", nombre=30, prevu=10.0, reel=10.0),
|
||||
anciennes=[
|
||||
*paires(site_id="SITE001", nombre=30, prevu=10.0, reel=10.0),
|
||||
*paires(site_id="SITE002", nombre=30, prevu=10.0, reel=10.0),
|
||||
],
|
||||
comptages=[ComptageStatut(site_id="SITE001", status="available", nombre=30)],
|
||||
)
|
||||
|
||||
rapports = await service(depot, min_observations=10).evaluate(now=INSTANT)
|
||||
|
||||
disparu = next(rapport for rapport in rapports if rapport.site_id == "SITE002")
|
||||
assert disparu.status == STATUT_INDETERMINE
|
||||
assert disparu.n_observations == 0
|
||||
|
||||
|
||||
async def test_drift_measures_the_share_of_sites_left_without_enough_history() -> None:
|
||||
depot = FauxDepot(
|
||||
recentes=paires(nombre=30, prevu=10.0, reel=10.0),
|
||||
@@ -201,6 +231,16 @@ async def test_drift_closes_the_window_before_the_grace_delay() -> None:
|
||||
assert reference[1] == recente[0]
|
||||
|
||||
|
||||
async def test_drift_truncates_the_reference_instant_to_the_hour() -> None:
|
||||
premier = FauxDepot()
|
||||
second = FauxDepot()
|
||||
|
||||
await service(premier).evaluate(now=INSTANT + timedelta(minutes=17, microseconds=3))
|
||||
await service(second).evaluate(now=INSTANT + timedelta(minutes=48))
|
||||
|
||||
assert premier.fenetres[0] == second.fenetres[0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("prevu", "attendu"),
|
||||
[(10.0, STATUT_STABLE), (30.0, STATUT_DERIVE)],
|
||||
|
||||
@@ -219,5 +219,5 @@ async def test_the_api_serves_the_latest_run_when_the_score_cli_runs_twice(
|
||||
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"]["created_at"] > premier["prediction"]["created_at"]
|
||||
assert second["prediction"]["model_reference"] == premier["prediction"]["model_reference"]
|
||||
|
||||
@@ -81,14 +81,22 @@ modèle change n'est pas une dérive, c'est une régression de réentraînement.
|
||||
| Écrire le résultat dans `alert` | `ck_alert_source` et `ck_alert_type` bornent les valeurs autorisées, `alert.site_id` est `NOT NULL` et n'accueillerait donc pas la ligne globale, et toute alerte est ensuite relue par le moteur de recommandations, qui devrait apprendre une règle qui ne le concerne pas (ADR 0006). |
|
||||
| Une jauge Prometheus | `monitoring/` ne contient que des `.gitkeep` et aucun collecteur ne lit `/metrics` : une jauge que personne ne scrute n'est pas une preuve. Le calcul est de surcroît un traitement par lot, pas le processus qui sert l'API : la jauge disparaîtrait avec lui. |
|
||||
| Ne rien persister, journaliser seulement | La question posée à un jury est « comment savez-vous que le modèle se dégrade ? ». La réponse est une série dans le temps, pas une ligne de journal perdue avec le conteneur. Sans ligne écrite, l'endpoint n'a rien à lire et le test d'intégration rien à vérifier. |
|
||||
| Une tâche de plus dans le DAG `alertes` | La fenêtre fait 168 h : la recalculer chaque heure écrirait vingt-quatre lignes identiques par jour. Surtout, un échec de dérive ferait rougir `alertes` et laisserait croire que la détection a échoué. |
|
||||
| Une tâche de plus dans le DAG `alertes` | La fenêtre fait 168 h : la recalculer chaque heure écrirait vingt-quatre lignes par jour pour un verdict qui ne bouge pas à cette cadence. Surtout, un échec de dérive ferait rougir `alertes` et laisserait croire que la détection a échoué. |
|
||||
|
||||
## Conséquences
|
||||
|
||||
- Une migration ajoute `drift_report`. Son idempotence passe par un **index unique à
|
||||
`coalesce(site_id, '')`** et non par une `UniqueConstraint` : deux lignes globales ont toutes
|
||||
deux `site_id` à NULL, et NULL n'est égal à rien, pas même à lui-même. Même forme que
|
||||
`uq_reading_source`.
|
||||
`uq_reading_source`. Cet index n'a de sens que parce que `evaluate()` **tronque son instant de
|
||||
référence à l'heure** : avec les microsecondes de `now()`, deux exécutions ne porteraient jamais
|
||||
la même clé et l'index ne dédoublonnerait rien.
|
||||
- **Sans fenêtre de référence, le verdict est `indetermine`, pas `stable`.** Au premier
|
||||
lancement, et après tout trou d'ingestion de plus de 168 h, il n'y a rien à quoi comparer :
|
||||
annoncer `stable` serait affirmer ce que la donnée ne dit pas.
|
||||
- **Un site présent dans la fenêtre de référence et absent de la récente reçoit sa ligne**, à
|
||||
zéro observation. Un site qui cesse d'être scoré est exactement la panne que cette surveillance
|
||||
existe pour dire : le taire en ne produisant aucune ligne serait l'inverse du besoin.
|
||||
- `GET /api/v1/monitoring/drift` est réservé à partir du rôle `operateur` : c'est l'opérateur
|
||||
qui agit sur un pipeline dégradé, pas l'administrateur de comptes. La route est classée dans
|
||||
`tests/api/acces.py`, donc couverte gratuitement par la matrice de rôles rejouée avec de vrais
|
||||
|
||||
@@ -14,9 +14,9 @@ Les quatre couches existent désormais, portées par l'authentification.
|
||||
flowchart TB
|
||||
ep["endpoints<br/>health, auth, users, sites, alerts,<br/>recommendations, stats, readings, sensors,<br/>predictions, monitoring"]
|
||||
sc["schemas<br/>Pydantic"]
|
||||
sv["services<br/>AuthService, UserService,<br/>SiteService, AlertService, RecommendationService,<br/>StatsService, ReadingService, SensorService, PredictionService"]
|
||||
rp["repositories<br/>user, refresh_token,<br/>login_attempt, audit_log,<br/>site, alert, recommendation, reading, prediction"]
|
||||
md["models<br/>10 tables"]
|
||||
sv["services<br/>AuthService, UserService,<br/>SiteService, AlertService, RecommendationService,<br/>StatsService, ReadingService, SensorService,<br/>PredictionService, DriftService"]
|
||||
rp["repositories<br/>user, refresh_token,<br/>login_attempt, audit_log,<br/>site, alert, recommendation, reading,<br/>prediction, drift"]
|
||||
md["models<br/>13 tables"]
|
||||
db[("PostgreSQL")]
|
||||
|
||||
ep --> sc
|
||||
@@ -229,8 +229,9 @@ côté pilote, en `500` plutôt qu'un refus propre.
|
||||
`DriftService.evaluate()` joint `prediction` et `reading` sur `(site_id, target_at = timestamp)`
|
||||
et compare deux fenêtres vives de 168 h, la récente et celle qui la précède. Il rend une ligne par
|
||||
site plus une ligne globale, que `DriftRepository.enregistre()` écrit dans `drift_report` avec
|
||||
`ON CONFLICT DO NOTHING` sur `uq_drift_report_window` : rejouer la commande sur la même fenêtre
|
||||
n'ajoute rien.
|
||||
`ON CONFLICT DO NOTHING` sur `uq_drift_report_window`. L'instant de référence est tronqué à
|
||||
l'heure, ce qui est la condition pour que cet index serve : rejouer la commande dans la même
|
||||
heure n'ajoute rien.
|
||||
|
||||
| Métrique | Ce qu'elle dit |
|
||||
|---|---|
|
||||
@@ -242,7 +243,10 @@ n'ajoute rien.
|
||||
| `model_references` | Les modèles vus dans la fenêtre : une MAE qui saute à l'instant où le modèle change est une régression de réentraînement, pas une dérive |
|
||||
|
||||
Le verdict a trois valeurs, `stable`, `derive` et `indetermine` : sous un nombre minimal
|
||||
d'observations, le service dit qu'il ne sait pas plutôt que de rendre un chiffre trompeur. La
|
||||
d'observations, ou faute de fenêtre de référence à laquelle comparer, le service dit qu'il ne
|
||||
sait pas plutôt que de rendre un chiffre trompeur. Un site qui figure dans la fenêtre de
|
||||
référence mais plus dans la récente reçoit sa ligne à zéro observation : cesser d'être scoré est
|
||||
la panne que cette surveillance existe pour dire. La
|
||||
fenêtre est fermée à droite par un délai de grâce de 2 h, le temps que l'ingestion livre le
|
||||
réalisé de la dernière heure. `python -m app.monitoring.drift` l'exécute, le DAG `derive`
|
||||
l'ordonnance, et `GET /api/v1/monitoring/drift` sert le dernier rapport de chaque site. Les
|
||||
|
||||
@@ -258,7 +258,7 @@ régénère, ce qui invalide les sessions et les connexions chiffrées par Airfl
|
||||
|
||||
### Pourquoi le job d'intégration ML installe aussi le backend
|
||||
|
||||
Le schéma de la base n'a qu'une source, les six révisions Alembic de `apps/backend/alembic` : le
|
||||
Le schéma de la base n'a qu'une source, les sept révisions Alembic de `apps/backend/alembic` : le
|
||||
backend est propriétaire du schéma, `ml/` n'en est que consommateur. Reconstruire ce schéma à la
|
||||
main dans le job ML donnerait un job vert sur une base qui n'est pas la nôtre, exactement l'erreur
|
||||
qu'évite déjà le choix de l'image `timescaledb-ha` plutôt qu'un `postgres` nu. Le job installe
|
||||
@@ -266,12 +266,14 @@ donc les deux environnements uv, applique `alembic upgrade head`, puis joue `-m
|
||||
`ml/` et `-m chaine` côté backend.
|
||||
|
||||
Conséquence sur le déclenchement : les `paths` de `ml.yml` incluent `apps/backend/alembic/**` et
|
||||
`apps/backend/app/models/**`. Sans eux, une migration qui renomme une colonne de `reading` ne
|
||||
`apps/backend/app/**`. Sans eux, une migration qui renomme une colonne de `reading` ne
|
||||
déclencherait pas ce job, le SQL brut du pipeline dériverait du schéma, et **rien ne casserait
|
||||
avant la production**. Le prix est qu'une PR touchant seulement une migration lance aussi le lint
|
||||
et le typage de `ml/` : environ deux minutes de runner, en parallèle. Même arbitrage que le filtre
|
||||
d'`airflow.yml`, qui écoute déjà `ml/**` et `apps/backend/app/**` parce que son image réunit les
|
||||
deux.
|
||||
avant la production**. `app/**` en entier, et non les seuls modèles : ce job est le seul à jouer
|
||||
`-m chaine`, or la chaîne traverse les endpoints, les services et les schémas jusqu'à
|
||||
`GET /predictions`. Un filtre plus étroit laisserait le test muet sur la PR même qui le casse. Le
|
||||
prix est qu'une PR backend lance aussi le lint et le typage de `ml/` : environ deux minutes de
|
||||
runner, en parallèle. Même arbitrage que le filtre d'`airflow.yml`, qui écoute déjà `ml/**` et
|
||||
`apps/backend/app/**` parce que son image réunit les deux.
|
||||
|
||||
Le marqueur `chaine` est distinct d'`integration` pour une raison mécanique : le job `integration`
|
||||
de `backend.yml` n'installe pas `ml/.venv`, et sélectionnerait sinon un test qui lance les
|
||||
@@ -297,7 +299,7 @@ Les tests d'intégration demandent une base **migrée**, et `db/init` ne crée `
|
||||
vide :
|
||||
|
||||
```bash
|
||||
make db-up migrate-test # la base de test reçoit les six révisions Alembic
|
||||
make db-up migrate-test # la base de test reçoit les sept révisions Alembic
|
||||
make test-integration # backend, marqueur `integration`
|
||||
make ml-test-integration # pipeline ML, marqueur `integration`
|
||||
make test-chaine # vrais binaires ML puis relecture par l'API, marqueur `chaine`
|
||||
|
||||
+1
-1
@@ -673,6 +673,6 @@ Le DAG `historical_import` est déclenché manuellement. Il exécute
|
||||
`/opt/data/raw`. L'orchestration de l'import API Mock et la réconciliation globale des deux
|
||||
sources restent couvertes par l'issue #15.
|
||||
|
||||
Airflow permet de planifier les traitements, gérer leur ordre d'exécution, suivre leur état et remonter les erreurs. Il ne remplace pas la logique ETL Python existante : les scripts actuels restent responsables de l'extraction, de la validation, de la transformation et du chargement. `etl/airflow/dags/ml_train.py`, `ml_score.py` et `alertes.py` et `historical_import.py` montrent le patron retenu (des `BashOperator` qui invoquent le script tel quel, dans l'environnement `uv` que l'image embarque pour lui).
|
||||
Airflow permet de planifier les traitements, gérer leur ordre d'exécution, suivre leur état et remonter les erreurs. Il ne remplace pas la logique ETL Python existante : les scripts actuels restent responsables de l'extraction, de la validation, de la transformation et du chargement. `etl/airflow/dags/ml_train.py`, `ml_score.py`, `alertes.py`, `historical_import.py` et `derive.py` montrent le patron retenu (des `BashOperator` qui invoquent le script tel quel, dans l'environnement `uv` que l'image embarque pour lui).
|
||||
|
||||
Le pipeline Data servira ensuite à préparer les données nécessaires au modèle de Machine Learning.
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
"""DAG de surveillance de la dérive du modèle de prévision (issue #45).
|
||||
|
||||
Quotidien, pas horaire : la fenêtre mesurée couvre 168 h, la recalculer chaque heure écrirait
|
||||
vingt-quatre lignes presque identiques par jour et se heurterait à l'index d'idempotence
|
||||
`uq_drift_report_window`. Planifié après les scorings de la nuit, et décalé de `ml_score` (à
|
||||
l'heure pile) comme de `alertes` (à la quinzième minute).
|
||||
vingt-quatre lignes par jour pour un verdict qui ne bouge pas à cette cadence. Planifié après
|
||||
les scorings de la nuit, et décalé de `ml_score` (à l'heure pile) comme de `alertes` (à la
|
||||
quinzième minute).
|
||||
|
||||
Sans `--now` : le service tronque son instant de référence à l'heure, donc deux exécutions de
|
||||
la même heure portent la même clé `uq_drift_report_window` et la seconde n'écrit rien.
|
||||
|
||||
Tâche distincte du DAG `alertes` plutôt qu'ajoutée à lui : un échec de dérive y ferait croire
|
||||
que la détection d'alertes a échoué, et ce DAG porte un budget temporel déjà argumenté face à
|
||||
|
||||
@@ -16,6 +16,12 @@ from tests.conftest import ANCRAGE, insere_lecture, insere_lectures, insere_site
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def du_site(frame: pd.DataFrame, site_id: str) -> pd.DataFrame:
|
||||
"""Piege : les chargeurs ne filtrent pas par site, et `enervision_test` est partagee avec
|
||||
les tests qui valident leurs ecritures. Juger le contenu de toute la fenetre les couplerait."""
|
||||
return frame[frame["site_id"] == site_id].reset_index(drop=True)
|
||||
|
||||
|
||||
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)
|
||||
@@ -49,7 +55,7 @@ def test_load_recent_from_database_excludes_readings_before_the_since_bound(
|
||||
connexion_ml, since=ANCRAGE - timedelta(hours=2), until=ANCRAGE
|
||||
)
|
||||
|
||||
assert list(frame["timestamp"]) == [
|
||||
assert list(du_site(frame, site_id)["timestamp"]) == [
|
||||
ANCRAGE - timedelta(hours=2),
|
||||
ANCRAGE - timedelta(hours=1),
|
||||
ANCRAGE,
|
||||
@@ -66,7 +72,7 @@ def test_load_recent_from_database_includes_a_reading_exactly_at_the_since_bound
|
||||
connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
|
||||
)
|
||||
|
||||
assert len(frame) == 1
|
||||
assert len(du_site(frame, site_id)) == 1
|
||||
|
||||
|
||||
def test_load_recent_from_database_keeps_timestamps_timezone_aware(
|
||||
@@ -93,7 +99,7 @@ def test_load_recent_from_database_orders_readings_by_site_then_timestamp(
|
||||
connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
|
||||
)
|
||||
|
||||
assert list(frame["timestamp"]) == [
|
||||
assert list(du_site(frame, site_id)["timestamp"]) == [
|
||||
ANCRAGE,
|
||||
ANCRAGE + timedelta(hours=1),
|
||||
ANCRAGE + timedelta(hours=2),
|
||||
@@ -122,7 +128,7 @@ def test_load_recent_from_database_types_a_fully_null_capacity_kw_as_float64(
|
||||
)
|
||||
|
||||
assert frame["capacity_kw"].dtype == "float64"
|
||||
assert frame["capacity_kw"].isna().all()
|
||||
assert du_site(frame, site_id)["capacity_kw"].isna().all()
|
||||
|
||||
|
||||
def test_load_recent_from_database_types_a_null_is_working_hours_as_float64(
|
||||
@@ -139,7 +145,7 @@ def test_load_recent_from_database_types_a_null_is_working_hours_as_float64(
|
||||
)
|
||||
|
||||
assert frame["is_working_hours"].dtype == "float64"
|
||||
assert list(frame["is_working_hours"].isna()) == [True, False]
|
||||
assert list(du_site(frame, site_id)["is_working_hours"].isna()) == [True, False]
|
||||
|
||||
|
||||
def test_both_loaders_produce_the_same_columns_in_the_same_order(
|
||||
@@ -180,4 +186,4 @@ def test_load_recent_from_database_excludes_readings_after_the_until_bound(
|
||||
connexion_ml, since=ANCRAGE - timedelta(days=1), until=ANCRAGE
|
||||
)
|
||||
|
||||
assert list(frame["timestamp"]) == [ANCRAGE]
|
||||
assert list(du_site(frame, site_id)["timestamp"]) == [ANCRAGE]
|
||||
|
||||
Reference in New Issue
Block a user