This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 11ed2049e12 Fix stale-session assertion in mapped placeholder
promotion test (#75175)
11ed2049e12 is described below
commit 11ed2049e12c6662396fb2c482cd170035540ec0
Author: Jed Cunningham <[email protected]>
AuthorDate: Mon Oct 19 21:47:21 2026 -0600
Fix stale-session assertion in mapped placeholder promotion test (#75175)
test_promotion_leaves_earlier_placeholder_task_instances_at_their_coordinates
asserted that the archived -1 attempt of a mapped task survives expansion.
Since #74407 expansion deletes those attempts, but the delete runs with
synchronize_session=False, so session.get() kept returning the stale
identity-map object and the test passed while asserting the opposite of
what the database holds.
Expire the session and query the database, asserting the archived
placeholder attempt is gone, and rename the test to match.
---
airflow-core/tests/unit/models/test_mappedoperator.py | 16 +++++++++++-----
1 file changed, 11 insertions(+), 5 deletions(-)
diff --git a/airflow-core/tests/unit/models/test_mappedoperator.py
b/airflow-core/tests/unit/models/test_mappedoperator.py
index 1d908059894..a6b89e03978 100644
--- a/airflow-core/tests/unit/models/test_mappedoperator.py
+++ b/airflow-core/tests/unit/models/test_mappedoperator.py
@@ -2235,10 +2235,9 @@ def
test_zero_length_expansion_leaves_a_region_holding_only_its_skipped_placehol
assert placeholder.state == TaskInstanceState.SKIPPED
-def
test_promotion_leaves_earlier_placeholder_task_instances_at_their_coordinates(mapped_dag,
session):
+def test_promotion_deletes_earlier_placeholder_task_instances(mapped_dag,
session):
dr, serialized, upstream_id = mapped_dag
placeholder = _live_tis(session, dr, serialized.task_id)[0]
- region_id = placeholder.region_id
placeholder.task = serialized
placeholder.try_number = 1
placeholder.state = TaskInstanceState.FAILED
@@ -2249,9 +2248,16 @@ def
test_promotion_leaves_earlier_placeholder_task_instances_at_their_coordinate
expand_mapped_task(serialized, dr.run_id, upstream_id, 2, session)
- archived = session.get(TaskInstance, archived_id)
- assert (archived.region_id, archived.region_index, archived.try_number) ==
(region_id, -1, 1)
- assert archived.working_set is None
+ # The delete doesn't synchronize the session, so read the database rather
than the identity map.
+ session.expire_all()
+ assert (
+ session.scalar(
+ select(TaskInstance.id)
+ .where(TaskInstance.id == archived_id)
+ .execution_options(include_all_attempts=True)
+ )
+ is None
+ )
live = _live_tis(session, dr, serialized.task_id)
assert [ti.region_index for ti in live] == [0, 1]
assert live[0].id == current.id
