Update __init__.py

Author: CaptainOfHacks

tests/e2e/dags/__init__.py

@@ -1,7 +1,7 @@
 from datetime import timezone
 
-from airflow.models import DagRun, TaskInstance
-from airflow.utils.types import DagRunType
+# from airflow.models import DagRun, TaskInstance
+# from airflow.utils.types import DagRunType
 
 INDEX_NOTICE_XML_CONTENT_TASK_ID = "index_notice_xml_content"
 NORMALISE_NOTICE_METADATA_TASK_ID = "normalise_notice_metadata"
@@ -46,35 +46,35 @@
 FULL_BRANCH_TASK_IDS = [START_PROCESSING_NOTICE_TASK_ID,
                         NORMALISE_NOTICE_METADATA_TASK_ID] + TRANSFORM_BRANCH_TASK_IDS + PACKAGE_BRANCH_TASK_IDS + PUBLISH_BRANCH_TASK_IDS
 
-
-def run_task(dag, task, conf: dict, xcom_push_data: dict = None, ignore_first_depends_on_past=True) -> TaskInstance:
-    start_date = dag.default_args["start_date"]
-    end_date = dag.default_args["start_date"]
-    start_date = start_date or task.start_date
-    end_date = end_date or task.end_date or timezone.utcnow()
-
-    info = list(task.dag.iter_dagrun_infos_between(start_date, end_date, align=False))[0]
-    ignore_depends_on_past = info.logical_date == start_date and ignore_first_depends_on_past
-    dr = DagRun(
-        dag_id=task.dag_id,
-        run_id=DagRun.generate_run_id(DagRunType.MANUAL, info.logical_date),
-        run_type=DagRunType.MANUAL,
-        execution_date=info.logical_date,
-        data_interval=info.data_interval,
-        conf=conf
-    )
-    ti = TaskInstance(task, run_id=None)
-    ti.dag_run = dr
-
-    if xcom_push_data is not None:
-        for key, value in xcom_push_data.items():
-            ti.xcom_push(key=str(key), value=value)
-
-    ti.run(
-        mark_success=False,
-        ignore_task_deps=True,
-        ignore_depends_on_past=ignore_depends_on_past,
-        ignore_ti_state=False,
-        test_mode=True,
-    )
-    return ti
+#
+# def run_task(dag, task, conf: dict, xcom_push_data: dict = None, ignore_first_depends_on_past=True) -> TaskInstance:
+#     start_date = dag.default_args["start_date"]
+#     end_date = dag.default_args["start_date"]
+#     start_date = start_date or task.start_date
+#     end_date = end_date or task.end_date or timezone.utcnow()
+#
+#     info = list(task.dag.iter_dagrun_infos_between(start_date, end_date, align=False))[0]
+#     ignore_depends_on_past = info.logical_date == start_date and ignore_first_depends_on_past
+#     dr = DagRun(
+#         dag_id=task.dag_id,
+#         run_id=DagRun.generate_run_id(DagRunType.MANUAL, info.logical_date),
+#         run_type=DagRunType.MANUAL,
+#         execution_date=info.logical_date,
+#         data_interval=info.data_interval,
+#         conf=conf
+#     )
+#     ti = TaskInstance(task, run_id=None)
+#     ti.dag_run = dr
+#
+#     if xcom_push_data is not None:
+#         for key, value in xcom_push_data.items():
+#             ti.xcom_push(key=str(key), value=value)
+#
+#     ti.run(
+#         mark_success=False,
+#         ignore_task_deps=True,
+#         ignore_depends_on_past=ignore_depends_on_past,
+#         ignore_ti_state=False,
+#         test_mode=True,
+#     )
+#     return ti