update failing tests with new requirement regarding getting variable also from env

Author: duprijil

dags/fetch_notices_by_date.py

@@ -26,8 +26,6 @@
 VALIDATE_FETCHED_NOTICES_TASK_ID = "validate_fetched_notices"
 
 DAG_FETCH_DEFAULT_TIMEZONE = "UTC"
-DAG_FETCH_DEFAULT_TIMETABLE = "0 1 * * *"
-SCHEDULE_DAG_FETCH_VAR_NAME = "SCHEDULE_DAG_FETCH"
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      catchup=False,

ted_sws/__init__.py

@@ -35,6 +35,7 @@
 PROJECT_PATH = pathlib.Path(__file__).parent.resolve()
 SPARQL_PREFIXES_PATH = PROJECT_PATH / "resources" / "prefixes" / "prefixes.json"
 
+DAG_FETCH_DEFAULT_TIMETABLE = "0 1 * * *"
 
 class MongoDBConfig:
 
@@ -263,9 +264,16 @@ def S3_PUBLISH_ENABLED(self, config_value: str) -> bool:
         return config_value.lower() in ["1", "true"]
 
 
+class DagSchedulingConfig:
+
+    @env_property(config_resolver_class=AirflowAndEnvConfigResolver, default_value=DAG_FETCH_DEFAULT_TIMETABLE)
+    def SCHEDULE_DAG_FETCH(self, config_value: str) -> str:
+        return config_value
+
+
 class TedConfigResolver(MongoDBConfig, RMLMapperConfig, XMLProcessorConfig, ELKConfig, LoggingConfig,
                         GitHubArtefacts, API, AllegroConfig, TedAPIConfig, SFTPConfig, FusekiConfig,
-                        SPARQLConfig, LimesAlignmentConfig, S3PublishConfig):
+                        SPARQLConfig, LimesAlignmentConfig, S3PublishConfig, DagSchedulingConfig):
     """
         This class resolve the secrets of the ted-sws project.
     """

tests/unit/dags/test_fetch_notices_by_date.py

@@ -1,32 +1,42 @@
+import os
+
 from airflow import DAG
 from airflow.models import DagBag, Variable
 from airflow.timetables.trigger import CronTriggerTimetable
 
-from dags.fetch_notices_by_date import DAG_FETCH_DEFAULT_TIMETABLE, DAG_FETCH_DEFAULT_TIMEZONE, \
-    SCHEDULE_DAG_FETCH_VAR_NAME
+from dags.fetch_notices_by_date import DAG_FETCH_DEFAULT_TIMEZONE
+from ted_sws import DAG_FETCH_DEFAULT_TIMETABLE
 
 
-def test_schedule_dag_fetch_ver_name_is_correct(dag_fetch_schedule_variable_name):
-    assert SCHEDULE_DAG_FETCH_VAR_NAME == dag_fetch_schedule_variable_name
+def test_fetcher_change_timetable_from_airflow_variable_after_reparse(dag_bag: DagBag,
+                                                                      dag_fetch_schedule_variable_name,
+                                                                      fetcher_dag_id: str,
+                                                                      example_dag_cron_table: CronTriggerTimetable,
+                                                                      airflow_timetable_import_error_name: str):
+    fetcher_dag: DAG = dag_bag.get_dag(dag_id=fetcher_dag_id)
+    assert fetcher_dag is not None
+    assert fetcher_dag.schedule_interval != example_dag_cron_table._expression
 
+    Variable.set(key=dag_fetch_schedule_variable_name, value=example_dag_cron_table._expression)
+    dag_bag.collect_dags(only_if_updated=False)
 
-def test_fetcher_has_default_timetable_at_the_beginning(dag_bag: DagBag, fetcher_dag_id: str):
     fetcher_dag: DAG = dag_bag.get_dag(dag_id=fetcher_dag_id)
-    default_dag_timetable = CronTriggerTimetable(cron=DAG_FETCH_DEFAULT_TIMETABLE,
-                                                 timezone=DAG_FETCH_DEFAULT_TIMEZONE)
-
     assert fetcher_dag is not None
-    assert fetcher_dag.schedule_interval == default_dag_timetable._expression
+    assert fetcher_dag.schedule_interval == example_dag_cron_table._expression
+
+    assert all(airflow_timetable_import_error_name not in error for error in dag_bag.import_errors.values())
 
 
-def test_fetcher_gets_correct_timetable_after_reparse(dag_bag: DagBag, fetcher_dag_id: str,
-                                                      example_dag_cron_table: CronTriggerTimetable,
-                                                      airflow_timetable_import_error_name: str):
+def test_fetcher_change_timetable_from_env_variable_after_reparse(dag_bag: DagBag,
+                                                                  dag_fetch_schedule_variable_name: str,
+                                                                  fetcher_dag_id: str,
+                                                                  example_dag_cron_table: CronTriggerTimetable,
+                                                                  airflow_timetable_import_error_name: str):
     fetcher_dag: DAG = dag_bag.get_dag(dag_id=fetcher_dag_id)
     assert fetcher_dag is not None
     assert fetcher_dag.schedule_interval != example_dag_cron_table._expression
 
-    Variable.set(key=SCHEDULE_DAG_FETCH_VAR_NAME, value=example_dag_cron_table._expression)
+    os.environ[dag_fetch_schedule_variable_name] = example_dag_cron_table._expression
     dag_bag.collect_dags(only_if_updated=False)
 
     fetcher_dag: DAG = dag_bag.get_dag(dag_id=fetcher_dag_id)
@@ -36,7 +46,9 @@ def test_fetcher_gets_correct_timetable_after_reparse(dag_bag: DagBag, fetcher_d
     assert all(airflow_timetable_import_error_name not in error for error in dag_bag.import_errors.values())
 
 
-def test_fetcher_gets_incorrect_timetable_after_reparse(dag_bag: DagBag, fetcher_dag_id: str,
+def test_fetcher_gets_incorrect_timetable_after_reparse(dag_bag: DagBag,
+                                                        dag_fetch_schedule_variable_name: str,
+                                                        fetcher_dag_id: str,
                                                         example_wrong_cron_table: str,
                                                         airflow_timetable_import_error_name: str):
     fetcher_dag: DAG = dag_bag.get_dag(dag_id=fetcher_dag_id)
@@ -45,7 +57,7 @@ def test_fetcher_gets_incorrect_timetable_after_reparse(dag_bag: DagBag, fetcher
                                                  timezone=DAG_FETCH_DEFAULT_TIMEZONE)
     assert fetcher_dag.schedule_interval == default_dag_timetable._expression
 
-    Variable.set(key=SCHEDULE_DAG_FETCH_VAR_NAME, value=example_wrong_cron_table)
+    Variable.set(key=dag_fetch_schedule_variable_name, value=example_wrong_cron_table)
     dag_bag.collect_dags(only_if_updated=False)
 
     assert any(airflow_timetable_import_error_name in error for error in dag_bag.import_errors.values())