[airflow] branch v2-2-test updated (c9dc0e4 -> b088b36)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a change to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git.


 discard c9dc0e4  Fix occasional external task sensor tests (#18853)
     new b088b36  Fix occasional external task sensor tests (#18853)

This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version.  This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:

 * -- * -- B -- O -- O -- O   (c9dc0e4)
            \
             N -- N -- N   refs/heads/v2-2-test (b088b36)

You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.

Any revisions marked "omit" are not gone; other references still
refer to them.  Any revisions marked "discard" are gone forever.

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.


Summary of changes:

[airflow] 01/01: Fix occasional external task sensor tests (#18853)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit b088b367fca1ff5990737d9b2a8b888605990401
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Tue Oct 12 20:27:25 2021 +0200

    Fix occasional external task sensor tests (#18853)
    
    Occassionally the sensor tests fail with assertion where
    state seems to be None. This might be caused by
    
    ```
          def assert_ti_state_equal(task_instance, state):
              """
              Assert state of task_instances equals the given state.
              """
              task_instance.refresh_from_db()
      >       assert task_instance.state == state
      E       AssertionError: assert None == <TaskInstanceState.SUCCESS: 'success'>
      E        +  where None = <TaskI$anstance: dag_1.task_b_1 manual__2015-01-01T00:00:00+00:00 [None]>.state
    ```
    
    Turned out it was because the task instance fields from
    dagrun.taskinstance relationship could be returned in different
    order so some of the dependencies were not met for some of the
    tasks when later task was returned before earlier one.
    
    Deterministic sorting according to task_id solved the problem.
    
    (cherry picked from commit 7a28ee370945de81fe8a16eac63197cbe93b3c3a)
---
 tests/sensors/test_external_task_sensor.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/tests/sensors/test_external_task_sensor.py b/tests/sensors/test_external_task_sensor.py
index 28018b9..86986a7 100644
--- a/tests/sensors/test_external_task_sensor.py
+++ b/tests/sensors/test_external_task_sensor.py
@@ -569,7 +569,12 @@ def run_tasks(dag_bag, execution_date=DEFAULT_DATE, session=None):
             run_type=DagRunType.MANUAL,
             session=session,
         )
-        for ti in dagrun.task_instances:
+        # we use sorting by task_id here because for the test DAG structure of ours
+        # this is equivalent to topological sort. It would not work in general case
+        # but it works for our case because we specifically constructed test DAGS
+        # in the way that those two sort methods are equivalent
+        tasks = sorted((ti for ti in dagrun.task_instances), key=lambda ti: ti.task_id)
+        for ti in tasks:
             ti.refresh_from_task(dag.get_task(ti.task_id))
             tis[ti.task_id] = ti
             ti.run(session=session)