fix: Compatibility with older airflow versions in dag tests

Author: tomasfarias

tests/dags/test_dbt_dags.py

@@ -34,10 +34,12 @@
 else:
     from airflow.providers.common.compat.sdk import DAG
 
-if AIRFLOW_V_3_0_PLUS:
+from airflow.serialization.serialized_objects import SerializedDAG
+
+try:
     from airflow.serialization.serialized_objects import DagSerialization
-else:
-    from airflow.serialization.serialized_objects import SerializedDAG
+except ImportError:
+    DagSerialization = SerializedDAG
 
 DATA_INTERVAL_START = pendulum.datetime(2022, 1, 1, tz="UTC")
 DATA_INTERVAL_END = DATA_INTERVAL_START + dt.timedelta(hours=1)
@@ -62,20 +64,12 @@ def sync_dag_to_db(
 
         def _write_dag(dag: DAG) -> SerializedDAG:
             if not SerializedDagModel.has_dag(dag.dag_id):
-                data = (
-                    DagSerialization.to_dict(dag)
-                    if AIRFLOW_V_3_0_PLUS
-                    else SerializedDAG.to_dict(dag)
-                )
+                data = DagSerialization.to_dict(dag)
                 SerializedDagModel.write_dag(
                     LazyDeserializedDAG(data=data), bundle_name, session=session
                 )
                 session.flush()
-            return (
-                DagSerialization.from_dict(data)
-                if AIRFLOW_V_3_0_PLUS
-                else SerializedDAG.from_dict(data)
-            )
+            return DagSerialization.from_dict(data)
 
         SerializedDAG.bulk_write_to_db(bundle_name, None, [dag], session=session)
         _ = _write_dag(dag)
@@ -123,6 +117,13 @@ def _run_task_instance(ti):
     if AIRFLOW_V_3_1_PLUS:
         return
 
+    if AIRFLOW_V_3_0:
+        # Airflow 3.0.6's TaskInstance runner fails in-process with a 422.
+        if isinstance(ti.task, DbtBaseOperator):
+            ti.task.execute({"ti": ti})
+        ti.state = TaskInstanceState.SUCCESS
+        return
+
     ti.run(ignore_ti_state=True)
 
 
@@ -365,6 +366,12 @@ def test_dbt_operators_in_taskflow_dag(
     else:
         dag = taskflow_dag
 
+    if AIRFLOW_V_3_0:
+        for task in dag.tasks:
+            if isinstance(task, DbtBaseOperator):
+                task.profiles_dir = str(profiles_file.parent)
+                task.project_dir = str(dbt_project_file.parent)
+
     dagrun = _create_dagrun(
         dag,
         state=DagRunState.RUNNING,