This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 505681284c4 [v3-4-test] Fix stale-session assertion in mapped
placeholder promotion test (#75175) (#75181)
505681284c4 is described below
commit 505681284c43029c3fa74c065cabe7eaadcbb9ce
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 20 14:08:10 2026 +0530
[v3-4-test] Fix stale-session assertion in mapped placeholder promotion
test (#75175) (#75181)
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.
(cherry picked from commit 11ed2049e12c6662396fb2c482cd170035540ec0)
Co-authored-by: Jed Cunningham
<[email protected]>
---
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
