feat(airflow): orchestre l'import de la Mock API

etl/airflow/tests/test_dags.py

@@ -10,13 +10,20 @@ from airflow.sdk import BaseOperator
 
 DAGS_FOLDER = Path(__file__).resolve().parent.parent / "dags"
 
-DAG_IDS = ["ml_train", "ml_score", "alertes", "historical_import"]
+DAG_IDS = [
+    "ml_train",
+    "ml_score",
+    "alertes",
+    "historical_import",
+    "mock_api_import",
+]
 TACHES = [
     ("ml_train", "train"),
     ("ml_score", "score"),
     ("alertes", "detection"),
     ("alertes", "recommandations"),
     ("historical_import", "import_historical"),
+    ("mock_api_import", "import_mock_api"),
 ]
 
 
@@ -52,6 +59,10 @@ def test_historical_import_has_no_schedule(dagbag: DagBag) -> None:
     assert dagbag.dags["historical_import"].schedule is None
 
 
+def test_mock_api_import_runs_every_hour(dagbag: DagBag) -> None:
+    assert dagbag.dags["mock_api_import"].timetable.expression == "0 * * * *"
+
+
 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
@@ -84,6 +95,20 @@ def test_historical_import_uses_the_expected_source_files(dagbag: DagBag) -> Non
     assert "--metadata /opt/data/raw/dataset_metadata.json" in commande
 
 
+def test_mock_api_import_calls_the_existing_backend_module(dagbag: DagBag) -> None:
+    commande = dagbag.dags["mock_api_import"].get_task("import_mock_api").bash_command
+
+    assert "app.etl.mock_api_import" in commande
+
+
+def test_mock_api_import_uses_the_airflow_data_interval(dagbag: DagBag) -> None:
+    commande = dagbag.dags["mock_api_import"].get_task("import_mock_api").bash_command
+
+    assert '--start-time "{{ data_interval_start.isoformat() }}"' in commande
+    assert '--end-time "{{ data_interval_end.isoformat() }}"' in commande
+    assert "--limit 60" 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).
@@ -95,6 +120,12 @@ def test_historical_import_runs_in_the_backend_environment(dagbag: DagBag) -> No
     assert "/opt/backend" in commande
 
 
+def test_mock_api_import_runs_in_the_backend_environment(dagbag: DagBag) -> None:
+    commande = dagbag.dags["mock_api_import"].get_task("import_mock_api").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.
@@ -136,6 +167,14 @@ def duree_au_pire(tache: BaseOperator) -> timedelta:
     return (tache.retries + 1) * tache.execution_timeout + tache.retries * tache.retry_delay
 
 
+def test_mock_api_import_worst_case_stays_below_its_hourly_step(
+    dagbag: DagBag,
+) -> None:
+    tache = dagbag.dags["mock_api_import"].get_task("import_mock_api")
+
+    assert duree_au_pire(tache) < timedelta(hours=1)
+
+
 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.
@@ -159,6 +198,10 @@ def test_historical_import_retries_after_a_transient_failure(dagbag: DagBag) ->
     assert dagbag.dags["historical_import"].get_task("import_historical").retries >= 1
 
 
+def test_mock_api_import_retries_after_a_transient_failure(dagbag: DagBag) -> None:
+    assert dagbag.dags["mock_api_import"].get_task("import_mock_api").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