xBis7 commented on code in PR #71737:
URL: https://github.com/apache/airflow/pull/71737#discussion_r4380902491
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -5041,6 +5056,35 @@ def my_teardown():
}
+def test_scheduling_decisions_mark_orphaned_finished_tis_removed(dag_maker,
session):
+ """A finished ti whose task no longer exists in the dag is marked REMOVED
and excluded."""
+ with dag_maker(session=session):
+ EmptyOperator(task_id="stays")
+ EmptyOperator(task_id="vanishes")
+ dr = dag_maker.create_dagrun()
+ for ti in dr.get_task_instances(session=session):
+ ti.state = TaskInstanceState.SUCCESS
+ session.flush()
+
+ del dr.dag.task_dict["vanishes"]
+ decision = dr.task_instance_scheduling_decisions(session=session)
+
+ assert len(decision.finished_tis) == 1
+ finished_ti = decision.finished_tis[0]
+ assert isinstance(finished_ti, FinishedTI)
+ assert finished_ti.task_id == "stays"
+ assert finished_ti.state == TaskInstanceState.SUCCESS
+ assert finished_ti.task is dr.dag.task_dict["stays"]
+ # The view carries the row's identity and coordinates so that trigger-rule
evaluation
+ # and the dag-callback re-fetch never need the full TaskInstance.
+ stays = dr.get_task_instance("stays", session=session)
+ assert (finished_ti.id, finished_ti.dag_id, finished_ti.run_id) ==
(stays.id, dr.dag_id, dr.run_id)
+ assert (finished_ti.region_id, finished_ti.region_index) ==
(stays.region_id, stays.region_index)
+ assert finished_ti.dag_version_id == stays.dag_version_id
+ session.expire_all()
+ assert dr.get_task_instance("vanishes", session=session).state ==
TaskInstanceState.REMOVED
Review Comment:
Added the missing tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
