diff --git a/etl/airflow/tests/test_dags.py b/etl/airflow/tests/test_dags.py index 555849d..1f066f9 100644 --- a/etl/airflow/tests/test_dags.py +++ b/etl/airflow/tests/test_dags.py @@ -1,5 +1,8 @@ -"""Tests d'integrite des DAGs : s'importent sans erreur, structure attendue. Pas d'execution -reelle des taches (ca reclamerait le conteneur avec `uv`/`enervision_ml`), juste la definition.""" +"""Tests d'integrite des DAGs : s'importent sans erreur et ont la structure attendue. + +Les tests ne lancent pas reellement les traitements : ils valident uniquement la definition +des DAGs et leurs commandes. +""" from datetime import timedelta from pathlib import Path @@ -10,12 +13,14 @@ from airflow.models.dagbag import DagBag DAGS_FOLDER = Path(__file__).resolve().parent.parent / "dags" -DAG_IDS = ["ml_train", "ml_score", "alertes"] +DAG_IDS = ["ml_train", "ml_score", "alertes", "historical_import"] + TACHES = [ ("ml_train", "train"), ("ml_score", "score"), ("alertes", "detection"), ("alertes", "recommandations"), + ("historical_import", "import_historical"), ] @@ -37,16 +42,17 @@ def test_ml_train_has_no_schedule(dagbag: DagBag) -> None: def test_ml_score_runs_every_hour(dagbag: DagBag) -> None: - # `@hourly` est un alias Airflow pour ce cron, c'est sous cette forme que `.summary` le rend. assert dagbag.dags["ml_score"].timetable.summary == "0 * * * *" def test_alertes_runs_after_the_hourly_scoring(dagbag: DagBag) -> None: - # Le decalage n'est pas cosmetique : la regle `anomaly` compare une lecture a la `prediction` - # du meme instant, que `ml_score` ecrit a l'heure pile. assert dagbag.dags["alertes"].timetable.summary == "15 * * * *" +def test_historical_import_has_no_schedule(dagbag: DagBag) -> None: + assert dagbag.dags["historical_import"].timetable.summary == "None" + + def test_ml_train_task_calls_the_training_module(dagbag: DagBag) -> None: tache = dagbag.dags["ml_train"].get_task("train") assert "enervision_ml.train" in tache.bash_command @@ -67,15 +73,29 @@ def test_alertes_recommendation_task_calls_the_backend_cli(dagbag: DagBag) -> No assert "app.cli generate-recommendations" in tache.bash_command +def test_historical_import_calls_the_existing_backend_module(dagbag: DagBag) -> None: + tache = dagbag.dags["historical_import"].get_task("import_historical") + assert "app.etl.historical_import" in tache.bash_command + + +def test_historical_import_uses_the_expected_source_files(dagbag: DagBag) -> None: + commande = dagbag.dags["historical_import"].get_task("import_historical").bash_command + + assert "--csv /opt/data/raw/all_sites_combined.csv" in commande + assert "--metadata /opt/data/raw/dataset_metadata.json" in commande + + @pytest.mark.parametrize("task_id", ["detection", "recommandations"]) def test_alertes_tasks_run_in_the_backend_environment(dagbag: DagBag, task_id: str) -> None: - # Le backend a son propre venv dans l'image, distinct de celui de ml/ (ADR 0008). assert "/opt/backend" in dagbag.dags["alertes"].get_task(task_id).bash_command +def test_historical_import_runs_in_the_backend_environment(dagbag: DagBag) -> None: + commande = dagbag.dags["historical_import"].get_task("import_historical").bash_command + assert "/opt/backend" in commande + + def test_alertes_generates_recommendations_after_detecting(dagbag: DagBag) -> None: - # `recommendation.alert_id` est une cle etrangere `NOT NULL` : la generation n'a rien a lire - # tant que la detection n'a pas ecrit. assert dagbag.dags["alertes"].get_task("detection").downstream_task_ids == {"recommandations"} @@ -90,14 +110,11 @@ def test_ml_score_reuses_the_model_path_written_by_ml_train(dagbag: DagBag) -> N @pytest.mark.parametrize("dag_id", DAG_IDS) def test_no_two_runs_of_a_dag_overlap(dagbag: DagBag, dag_id: str) -> None: - # Deux entrainements ecriraient le meme fichier modele, deux scorings inseriraient en meme - # temps dans `prediction`, deux detections analyseraient la meme fenetre. assert dagbag.dags[dag_id].max_active_runs == 1 @pytest.mark.parametrize(("dag_id", "task_id"), TACHES) def test_every_task_has_an_execution_timeout(dagbag: DagBag, dag_id: str, task_id: str) -> None: - # Sans plafond, une connexion pendue immobilise un slot du scheduler indefiniment. assert dagbag.dags[dag_id].get_task(task_id).execution_timeout is not None @@ -108,15 +125,11 @@ def test_ml_score_execution_timeout_stays_below_its_hourly_step(dagbag: DagBag) def duree_au_pire(tache: BaseOperator) -> timedelta: - # `execution_timeout` plafonne une tentative, pas la tache : deux reprises occupent trois - # plafonds et deux delais d'attente. assert tache.execution_timeout is not None return (tache.retries + 1) * tache.execution_timeout + tache.retries * tache.retry_delay def test_alertes_worst_case_stays_below_its_hourly_step(dagbag: DagBag) -> None: - # Les deux taches s'enchainent : c'est leur somme, reprises comprises, qui doit tenir dans le - # pas horaire, sinon `max_active_runs=1` fait attendre l'execution suivante. taches = [ dagbag.dags["alertes"].get_task(task_id) for task_id in ("detection", "recommandations") ] @@ -129,13 +142,15 @@ def test_ml_score_retries_after_a_transient_failure(dagbag: DagBag) -> None: @pytest.mark.parametrize("task_id", ["detection", "recommandations"]) def test_alertes_retries_after_a_transient_failure(dagbag: DagBag, task_id: str) -> None: - # Les deux commandes sont idempotentes en base, une reprise ne duplique rien. assert dagbag.dags["alertes"].get_task(task_id).retries >= 1 +def test_historical_import_retries_after_a_transient_failure(dagbag: DagBag) -> None: + assert dagbag.dags["historical_import"].get_task("import_historical").retries >= 1 + + @pytest.mark.parametrize(("dag_id", "task_id"), TACHES) def test_tasks_never_resync_the_baked_environment( dagbag: DagBag, dag_id: str, task_id: str ) -> None: - # Sans `--no-sync`, `uv run` reconstruit le projet a chaque execution. assert "--no-sync" in dagbag.dags[dag_id].get_task(task_id).bash_command