fix tests

Author: dstandish

airflow-core/src/airflow/models/taskinstance.py

@@ -305,6 +305,7 @@ def clear_task_instances(
             session.merge(ti)
 
     if dag_run_state is not False and tis:
+        from airflow.models.dag import DagModel
         from airflow.models.dagrun import DagRun  # Avoid circular import
 
         run_ids_by_dag_id = defaultdict(set)
@@ -326,8 +327,19 @@ def clear_task_instances(
             if dr.state in State.finished_dr_states:
                 dr.state = dag_run_state
                 dr.start_date = timezone.utcnow()
+                if TYPE_CHECKING:
+                    assert dag  # todo: change signature so this is required
                 if not dag.disable_bundle_versioning:
-                    dr.bundle_version = dr.dag_model.bundle_version
+                    if dr.dag_model:
+                        bundle_version = dr.dag_model.bundle_version
+                    else:
+                        bundle_version = session.scalar(
+                            select(DagModel.bundle_version).where(
+                                DagModel.dag_id == dag.dag_id,
+                            )
+                        )
+                    if bundle_version is not None:
+                        dr.bundle_version = bundle_version
                 if dag_run_state == DagRunState.QUEUED:
                     dr.last_scheduling_decision = None
                     dr.start_date = None

airflow-core/tests/unit/models/test_dag.py

@@ -1426,7 +1426,6 @@ def test_clear_set_dagrun_state(self, dag_run_state):
             dag_run_state=dag_run_state,
             session=session,
         )
-
         dagruns = session.query(DagRun).filter(DagRun.dag_id == dag_id).all()
 
         assert len(dagruns) == 1

airflow-core/tests/unit/models/test_dagrun.py

@@ -160,7 +160,7 @@ def test_clear_task_instances_for_backfill_finished_dagrun(self, dag_maker, stat
         self.create_dag_run(dag, logical_date=now, is_backfill=True, state=state, session=session)
 
         qry = session.query(TI).filter(TI.dag_id == dag.dag_id).all()
-        clear_task_instances(qry, session)
+        clear_task_instances(qry, session, dag=dag)
         session.flush()
         dr0 = session.query(DagRun).filter(DagRun.dag_id == dag_id, DagRun.logical_date == now).first()
         assert dr0.state == DagRunState.QUEUED