This is an automated email from the ASF dual-hosted git repository.
dheerajturaga 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 ef9feefd83f Fix Grid and Graph 500 errors for cyclic TaskGroup
dependencies (#73724)
ef9feefd83f is described below
commit ef9feefd83fcd1cf7b18c245d424fd2eb5634211
Author: Dheeraj Turaga <[email protected]>
AuthorDate: Thu Oct 8 06:59:43 2026 -0500
Fix Grid and Graph 500 errors for cyclic TaskGroup dependencies (#73724)
---
.../airflow/serialization/definitions/taskgroup.py | 42 +------
.../api_fastapi/core_api/routes/ui/test_grid.py | 26 ++++
.../core_api/routes/ui/test_structure.py | 22 ++++
airflow-core/tests/unit/utils/test_task_group.py | 134 +++++++++++++++++++++
shared/dagnode/src/airflow_shared/dagnode/node.py | 113 ++++++++++++++++-
shared/dagnode/tests/dagnode/test_node.py | 36 +++++-
task-sdk/src/airflow/sdk/definitions/taskgroup.py | 45 +------
task-sdk/tests/task_sdk/definitions/test_dag.py | 12 ++
8 files changed, 353 insertions(+), 77 deletions(-)
diff --git a/airflow-core/src/airflow/serialization/definitions/taskgroup.py
b/airflow-core/src/airflow/serialization/definitions/taskgroup.py
index b82e1c0b933..c250363d837 100644
--- a/airflow-core/src/airflow/serialization/definitions/taskgroup.py
+++ b/airflow-core/src/airflow/serialization/definitions/taskgroup.py
@@ -236,10 +236,8 @@ class SerializedTaskGroup(TaskGroupMixin, DAGNode):
"""
Sort children topologically — a task always comes after its upstream
dependencies.
- See ``TaskGroup.topological_sort`` in task-sdk for the algorithm.
Cycles are
- treated as corrupt input: ``DAG.check_cycle`` rejects cyclic Dags
before
- serialization, so a cycle reaching this code indicates malformed
serialized data,
- and we raise ``ValueError`` rather than silently looping forever.
+ See ``TaskGroup.topological_sort`` in task-sdk for the algorithm,
including how siblings
+ that depend on each other in a cycle are ordered.
"""
children = self.children
if not children:
@@ -329,44 +327,16 @@ class SerializedTaskGroup(TaskGroupMixin, DAGNode):
emitted[i] = 1
order_append(nodes[i])
if len(next_pending) == len(pending):
- raise ValueError(f"A cyclic dependency occurred in dag:
{self.dag_id}")
+ return self._sort_cyclic_projection(nodes, projected)
pending = next_pending
return order
def _sort_via_pass_numbering(
self, nodes: list[DAGNode], projected: list[tuple[int, ...]]
) -> list[DAGNode]:
- n = len(nodes)
- in_degree = [len(deps) for deps in projected]
- successors: list[list[int]] = [[] for _ in range(n)]
- for i, deps in enumerate(projected):
- for d in deps:
- successors[d].append(i)
-
- pass_of = [0] * n
- queue: deque[int] = deque(i for i in range(n) if in_degree[i] == 0)
- processed = 0
- while queue:
- i = queue.popleft()
- my_pass = 1
- for d in projected[i]:
- d_pass = pass_of[d]
- if d < i:
- if d_pass > my_pass:
- my_pass = d_pass
- elif d_pass + 1 > my_pass:
- my_pass = d_pass + 1
- pass_of[i] = my_pass
- processed += 1
- for s in successors[i]:
- in_degree[s] -= 1
- if in_degree[s] == 0:
- queue.append(s)
-
- if processed != n:
- raise ValueError(f"A cyclic dependency occurred in dag:
{self.dag_id}")
-
- sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i))
+ sorted_indices = self._compute_pass_order(projected)
+ if len(sorted_indices) != len(nodes):
+ return self._sort_cyclic_projection(nodes, projected)
return [nodes[i] for i in sorted_indices]
def add(self, node: DAGNode) -> DAGNode:
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
index ad4450903df..dabf34b9d84 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
@@ -1063,6 +1063,32 @@ class TestGetGridDataEndpoint:
{"id": "task_b", "label": "task_b"},
]
+ def test_structure_filter_creating_task_group_cycle(self, dag_maker,
test_client, session):
+ """Dropping ``g.guard`` makes ``g.second`` a root of ``g``, so ``g``
and ``bridge`` depend on each other."""
+ with dag_maker(dag_id="filtered_task_group_cycle", serialized=True,
session=session):
+ with TaskGroup(group_id="g"):
+ first = EmptyOperator(task_id="first")
+ second = EmptyOperator(task_id="second")
+ guard = EmptyOperator(task_id="guard")
+ guard >> second
+ bridge = EmptyOperator(task_id="bridge")
+ first >> bridge >> second
+ dag_maker.sync_dagbag_to_db()
+
+ response = test_client.get(
+
"/grid/structure/filtered_task_group_cycle?root=g.first&include_downstream=true"
+ )
+
+ assert response.status_code == 200
+ assert response.json() == [
+ {"id": "bridge", "label": "bridge"},
+ {
+ "id": "g",
+ "label": "g",
+ "children": [{"id": "g.first", "label": "first"}, {"id":
"g.second", "label": "second"}],
+ },
+ ]
+
# Tests for root, include_upstream, and include_downstream parameters
@pytest.mark.parametrize(
("params", "expected_task_ids", "description"),
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_structure.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_structure.py
index 61b2eed0b3d..8010203dd69 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_structure.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_structure.py
@@ -393,6 +393,28 @@ class TestStructureDataEndpoint:
assert mapped_in_group["is_mapped"] is True
assert mapped_in_group["operator"] == "PythonOperator"
+ def test_filter_creating_task_group_cycle(self, dag_maker, test_client,
session):
+ """Dropping ``g.guard`` makes ``g.second`` a root of ``g``, so ``g``
and ``bridge`` depend on each other."""
+ with dag_maker(dag_id="filtered_task_group_cycle", serialized=True,
session=session):
+ with TaskGroup(group_id="g"):
+ first = EmptyOperator(task_id="first")
+ second = EmptyOperator(task_id="second")
+ guard = EmptyOperator(task_id="guard")
+ guard >> second
+ bridge = EmptyOperator(task_id="bridge")
+ first >> bridge >> second
+ dag_maker.sync_dagbag_to_db()
+
+ response = test_client.get(
+ "/structure/structure_data",
+ params={"dag_id": "filtered_task_group_cycle", "root": "g.first",
"include_downstream": True},
+ )
+
+ assert response.status_code == 200
+ nodes = response.json()["nodes"]
+ assert [node["id"] for node in nodes] == ["bridge", "g"]
+ assert [child["id"] for child in nodes[1]["children"]] == ["g.first",
"g.second"]
+
def test_ui_colors_passed_through_to_graph(self, dag_maker, test_client,
session):
"""Both raw hex colors and Chakra palette tokens reach the graph
unchanged, for operators and groups."""
diff --git a/airflow-core/tests/unit/utils/test_task_group.py
b/airflow-core/tests/unit/utils/test_task_group.py
index dc9a306ccdc..62f15d9a223 100644
--- a/airflow-core/tests/unit/utils/test_task_group.py
+++ b/airflow-core/tests/unit/utils/test_task_group.py
@@ -1351,6 +1351,140 @@ def
test_topological_sort_serialized_padded_reverse_chain_uses_pass_numbering(mo
assert position[f"r{i}"] < position[f"r{i + 1}"]
+# Deserialization orders a group's children by label, so these Dags declare
children in label order
+# to give the Task SDK sort the same input as the serialized one.
+
+
+def _make_sibling_groups_cycle():
+ with DAG("sibling_groups_cycle", schedule=None, start_date=DEFAULT_DATE)
as dag:
+ end = EmptyOperator(task_id="end")
+ with TaskGroup("group1"):
+ a1 = EmptyOperator(task_id="a1")
+ a2 = EmptyOperator(task_id="a2")
+ with TaskGroup("group2"):
+ b1 = EmptyOperator(task_id="b1")
+ b2 = EmptyOperator(task_id="b2")
+ start = EmptyOperator(task_id="start")
+ start >> [a1, b2]
+ a1 >> b1
+ b2 >> a2
+ [a2, b1] >> end
+ return dag
+
+
+def _make_group_bridged_by_outside_task():
+ with DAG("group_bridged_by_outside_task", schedule=None,
start_date=DEFAULT_DATE) as dag:
+ bridge = EmptyOperator(task_id="bridge")
+ with TaskGroup("group"):
+ a = EmptyOperator(task_id="a")
+ b = EmptyOperator(task_id="b")
+ a >> bridge >> b
+ return dag
+
+
+def _make_bridged_group_among_siblings():
+ with DAG("bridged_group_among_siblings", schedule=None,
start_date=DEFAULT_DATE) as dag:
+ after = EmptyOperator(task_id="after")
+ bridge = EmptyOperator(task_id="bridge")
+ with TaskGroup("group"):
+ a = EmptyOperator(task_id="a")
+ b = EmptyOperator(task_id="b")
+ for i in range(3):
+ EmptyOperator(task_id=f"x{i}")
+ a >> bridge >> b >> after
+ return dag
+
+
+def _make_nested_bridged_group():
+ with DAG("nested_bridged_group", schedule=None, start_date=DEFAULT_DATE)
as dag:
+ with TaskGroup("outer"):
+ after = EmptyOperator(task_id="after")
+ bridge = EmptyOperator(task_id="bridge")
+ with TaskGroup("inner"):
+ a = EmptyOperator(task_id="a")
+ b = EmptyOperator(task_id="b")
+ a >> bridge >> b >> after
+ return dag
+
+
+def _make_three_group_ring():
+ with DAG("three_group_ring", schedule=None, start_date=DEFAULT_DATE) as
dag:
+ groups = {}
+ for group_id in ("g0", "g1", "g2"):
+ with TaskGroup(group_id):
+ groups[group_id] = (EmptyOperator(task_id="first"),
EmptyOperator(task_id="second"))
+ groups["g1"][0] >> groups["g0"][1]
+ groups["g2"][0] >> groups["g1"][1]
+ groups["g0"][0] >> groups["g2"][1]
+ return dag
+
+
+def _get_topological_orders(group_dict):
+ return {
+ group_id: [node.node_id for node in
group.topological_sort(group_dict=group_dict)]
+ for group_id, group in group_dict.items()
+ }
+
+
[email protected](
+ ("make_dag", "expected"),
+ [
+ pytest.param(
+ _make_sibling_groups_cycle,
+ {
+ None: ["start", "group1", "group2", "end"],
+ "group1": ["group1.a1", "group1.a2"],
+ "group2": ["group2.b1", "group2.b2"],
+ },
+ id="sibling-groups",
+ ),
+ pytest.param(
+ _make_group_bridged_by_outside_task,
+ {None: ["bridge", "group"], "group": ["group.a", "group.b"]},
+ id="group-bridged-by-outside-task",
+ ),
+ # Only two of six siblings depend on a later sibling, which keeps the
sweep.
+ pytest.param(
+ _make_bridged_group_among_siblings,
+ {
+ None: ["bridge", "group", "x0", "x1", "x2", "after"],
+ "group": ["group.a", "group.b"],
+ },
+ id="bridged-group-among-siblings",
+ ),
+ # Two of outer's three children depend on a later sibling, which
selects pass numbering.
+ pytest.param(
+ _make_nested_bridged_group,
+ {
+ None: ["outer"],
+ "outer": ["outer.bridge", "outer.inner", "outer.after"],
+ "outer.inner": ["outer.inner.a", "outer.inner.b"],
+ },
+ id="nested-bridged-group",
+ ),
+ # Two of the three siblings depend on a later sibling, which selects
pass numbering.
+ pytest.param(
+ _make_three_group_ring,
+ {
+ None: ["g0", "g1", "g2"],
+ "g0": ["g0.first", "g0.second"],
+ "g1": ["g1.first", "g1.second"],
+ "g2": ["g2.first", "g2.second"],
+ },
+ id="three-group-ring",
+ ),
+ ],
+)
+def test_topological_sort_task_group_cycle(make_dag, expected):
+ """Siblings that depend on each other only at the group level are ordered
instead of raising."""
+ dag = make_dag()
+ dag.check_cycle()
+ serialized = create_scheduler_dag(dag)
+
+ assert
_get_topological_orders(serialized.task_group.get_task_group_dict()) == expected
+ assert _get_topological_orders(dag.task_group.get_task_group_dict()) ==
expected
+
+
def test_task_group_arrow_with_setup_group():
with DAG(dag_id="setup_group_teardown_group") as dag:
with TaskGroup("group_1") as g1:
diff --git a/shared/dagnode/src/airflow_shared/dagnode/node.py
b/shared/dagnode/src/airflow_shared/dagnode/node.py
index 9bc3b91f60a..15f060d25a5 100644
--- a/shared/dagnode/src/airflow_shared/dagnode/node.py
+++ b/shared/dagnode/src/airflow_shared/dagnode/node.py
@@ -17,12 +17,13 @@
from __future__ import annotations
+from collections import deque
from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar
import structlog
if TYPE_CHECKING:
- from collections.abc import Collection, Iterable
+ from collections.abc import Collection, Iterable, Iterator
from typing import Self
from ..logging.types import Logger
@@ -59,6 +60,7 @@ class TaskGroupProtocol(Protocol):
Dag = TypeVar("Dag", bound=DagProtocol)
Task = TypeVar("Task", bound=TaskProtocol)
TaskGroup = TypeVar("TaskGroup", bound=TaskGroupProtocol)
+Node = TypeVar("Node")
class GenericDAGNode(Generic[Dag, Task, TaskGroup]):
@@ -294,4 +296,113 @@ class TaskGroupMixin:
(t for root in self.get_roots() for t in root.upstream_task_ids),
)
+ @staticmethod
+ def _compute_pass_order(projected: list[tuple[int, ...]]) -> list[int]:
+ """
+ Order child indices by (pass number, insertion index).
+
+ pass(X) is the earliest pass at which a greedy sweep in insertion
order would emit X:
+ ``max over deps d of (pass(d) if idx(d) < idx(X) else pass(d) + 1)``.
A dep declared
+ before X can be emitted in the same pass; a dep declared after X
forces X into the next
+ pass. Computed via Kahn's traversal in O(V + E).
+
+ :param projected: for each child, the indices of the siblings it
depends on
+ :return: the ordered indices, leaving out every child on or downstream
of a cycle
+ """
+ n = len(projected)
+ in_degree = [len(deps) for deps in projected]
+ successors: list[list[int]] = [[] for _ in range(n)]
+ for i, deps in enumerate(projected):
+ for d in deps:
+ successors[d].append(i)
+
+ pass_of = [0] * n
+ queue: deque[int] = deque(i for i in range(n) if in_degree[i] == 0)
+ processed: list[int] = []
+ while queue:
+ i = queue.popleft()
+ my_pass = 1
+ for d in projected[i]:
+ d_pass = pass_of[d]
+ if d < i:
+ if d_pass > my_pass:
+ my_pass = d_pass
+ elif d_pass + 1 > my_pass:
+ my_pass = d_pass + 1
+ pass_of[i] = my_pass
+ processed.append(i)
+ for s in successors[i]:
+ in_degree[s] -= 1
+ if in_degree[s] == 0:
+ queue.append(s)
+
+ return sorted(processed, key=lambda i: (pass_of[i], i))
+
+ def _sort_cyclic_projection(self, nodes: list[Node], projected:
list[tuple[int, ...]]) -> list[Node]:
+ """
+ Order children whose projection has a cycle, ordering the siblings on
each cycle as one unit.
+
+ ``DAG.check_cycle`` rejects task-level cycles, but a task with no
upstream inside its own
+ group counts as a root of that group, so edges routed through tasks
outside the group can
+ still make siblings depend on each other. ``partial_subset`` can also
create such a cycle
+ by dropping a task's in-group upstream. Grid and Graph must still
render these Dags.
+
+ Each unit is placed by its first child and follows the same pass
ordering as the acyclic
+ path; a unit's children keep insertion order.
+ """
+ component_of = self._find_projection_components(projected)
+ count = max(component_of) + 1
+ members: list[list[int]] = [[] for _ in range(count)]
+ component_deps: list[set[int]] = [set() for _ in range(count)]
+ for i, deps in enumerate(projected):
+ c = component_of[i]
+ members[c].append(i)
+ component_deps[c].update(component_of[d] for d in deps if
component_of[d] != c)
+ component_order = self._compute_pass_order([tuple(deps) for deps in
component_deps])
+ return [nodes[i] for c in component_order for i in members[c]]
+
+ @staticmethod
+ def _find_projection_components(projected: list[tuple[int, ...]]) ->
list[int]:
+ """Return each child's strongly connected component, numbered in order
of its first child."""
+ n = len(projected)
+ successors: list[list[int]] = [[] for _ in range(n)]
+ for i, deps in enumerate(projected):
+ for d in deps:
+ successors[d].append(i)
+
+ # Kosaraju: finish order along successors, then collect components
along dependencies.
+ visited = bytearray(n)
+ finish_order: list[int] = []
+ for start in range(n):
+ if visited[start]:
+ continue
+ visited[start] = 1
+ stack: list[tuple[int, Iterator[int]]] = [(start,
iter(successors[start]))]
+ while stack:
+ node, remaining = stack[-1]
+ for s in remaining:
+ if not visited[s]:
+ visited[s] = 1
+ stack.append((s, iter(successors[s])))
+ break
+ else:
+ stack.pop()
+ finish_order.append(node)
+
+ root_of = [-1] * n
+ for start in reversed(finish_order):
+ if root_of[start] != -1:
+ continue
+ root_of[start] = start
+ to_visit = [start]
+ while to_visit:
+ node = to_visit.pop()
+ for d in projected[node]:
+ if root_of[d] == -1:
+ root_of[d] = start
+ to_visit.append(d)
+
+ numbering: dict[int, int] = {}
+ return [numbering.setdefault(root, len(numbering)) for root in root_of]
+
# TODO: Move more duplicated logic between Core and SDK task group types.
diff --git a/shared/dagnode/tests/dagnode/test_node.py
b/shared/dagnode/tests/dagnode/test_node.py
index 2166074d5db..6855b861730 100644
--- a/shared/dagnode/tests/dagnode/test_node.py
+++ b/shared/dagnode/tests/dagnode/test_node.py
@@ -22,7 +22,7 @@ from unittest import mock
import pytest
-from airflow_shared.dagnode.node import GenericDAGNode
+from airflow_shared.dagnode.node import GenericDAGNode, TaskGroupMixin
class Task:
@@ -81,3 +81,37 @@ class TestDAGNode:
assert node.label == "test_group_id.test_node_id"
node.task_group = TaskGroup(prefix_group_id)
assert node.label == expected_label
+
+
+class TestTaskGroupMixin:
+ @pytest.mark.parametrize(
+ ("projected", "expected"),
+ [
+ pytest.param([(), (0,), (1,)], [0, 1, 2], id="chain"),
+ pytest.param([(1,), (2,), (0,)], [0, 0, 0], id="ring"),
+ pytest.param([(), (2,), (1,), (1,)], [0, 1, 1, 2],
id="cycle-between-acyclic-children"),
+ pytest.param([(1,), (0,), (3,), (2,)], [0, 0, 1, 1],
id="two-cycles"),
+ ],
+ )
+ def test_find_projection_components(self, projected, expected):
+ assert TaskGroupMixin._find_projection_components(projected) ==
expected
+
+ @pytest.mark.parametrize(
+ ("nodes", "projected", "expected"),
+ [
+ pytest.param(
+ ["after", "bridge", "group", "x0"],
+ [(2,), (2,), (1,), ()],
+ ["bridge", "group", "x0", "after"],
+ id="cycle-before-its-downstream",
+ ),
+ pytest.param(
+ ["end", "group1", "group2", "start"],
+ [(1, 2), (2, 3), (1, 3), ()],
+ ["start", "group1", "group2", "end"],
+ id="cycle-after-its-upstream",
+ ),
+ ],
+ )
+ def test_sort_cyclic_projection(self, nodes, projected, expected):
+ assert TaskGroupMixin()._sort_cyclic_projection(nodes, projected) ==
expected
diff --git a/task-sdk/src/airflow/sdk/definitions/taskgroup.py
b/task-sdk/src/airflow/sdk/definitions/taskgroup.py
index 89bc37d7127..21417e75b01 100644
--- a/task-sdk/src/airflow/sdk/definitions/taskgroup.py
+++ b/task-sdk/src/airflow/sdk/definitions/taskgroup.py
@@ -32,7 +32,6 @@ from airflow.sdk import TriggerRule
from airflow.sdk._shared.dagnode.node import TaskGroupMixin
from airflow.sdk.definitions._internal.node import DAGNode, validate_group_key
from airflow.sdk.exceptions import (
- AirflowDagCycleException,
DuplicateTaskIdFound,
NodeNotFound,
TaskAlreadyInTaskGroup,
@@ -561,7 +560,8 @@ class TaskGroup(TaskGroupMixin, DAGNode):
traversal, O((V + E) log V), avoids the O(N²) blowup the sweep would
hit.
Both branches produce the same emission order: level-by-legacy-pass,
ties broken by
- children insertion order.
+ children insertion order. Siblings that depend on each other in a
cycle, which
+ ``DAG.check_cycle`` allows, are ordered as one unit (see
``_sort_cyclic_projection``).
"""
children = self.children
if not children:
@@ -657,49 +657,16 @@ class TaskGroup(TaskGroupMixin, DAGNode):
emitted[i] = 1
order_append(nodes[i])
if len(next_pending) == len(pending):
- raise AirflowDagCycleException(f"A cyclic dependency occurred
in dag: {self.dag_id}")
+ return self._sort_cyclic_projection(nodes, projected)
pending = next_pending
return order
def _sort_via_pass_numbering(
self, nodes: list[DAGNode], projected: list[tuple[int, ...]]
) -> list[DAGNode]:
- # Sort by (pass_number, insertion_index). pass_number(X) is the
earliest pass at
- # which a greedy-sweep emission of X would occur:
- # pass(X) = max over deps d of (pass(d) if idx(d) < idx(X) else
pass(d)+1)
- # A dep declared before X can be emitted in the same pass; a dep
declared after X
- # forces X into the next pass. Computed via Kahn's traversal in O(V +
E).
- n = len(nodes)
- in_degree = [len(deps) for deps in projected]
- successors: list[list[int]] = [[] for _ in range(n)]
- for i, deps in enumerate(projected):
- for d in deps:
- successors[d].append(i)
-
- pass_of = [0] * n
- queue: deque[int] = deque(i for i in range(n) if in_degree[i] == 0)
- processed = 0
- while queue:
- i = queue.popleft()
- my_pass = 1
- for d in projected[i]:
- d_pass = pass_of[d]
- if d < i:
- if d_pass > my_pass:
- my_pass = d_pass
- elif d_pass + 1 > my_pass:
- my_pass = d_pass + 1
- pass_of[i] = my_pass
- processed += 1
- for s in successors[i]:
- in_degree[s] -= 1
- if in_degree[s] == 0:
- queue.append(s)
-
- if processed != n:
- raise AirflowDagCycleException(f"A cyclic dependency occurred in
dag: {self.dag_id}")
-
- sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i))
+ sorted_indices = self._compute_pass_order(projected)
+ if len(sorted_indices) != len(nodes):
+ return self._sort_cyclic_projection(nodes, projected)
return [nodes[i] for i in sorted_indices]
def iter_mapped_task_groups(self) -> Iterator[MappedTaskGroup]:
diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py
b/task-sdk/tests/task_sdk/definitions/test_dag.py
index 8de4f0b80f0..604c212d5da 100644
--- a/task-sdk/tests/task_sdk/definitions/test_dag.py
+++ b/task-sdk/tests/task_sdk/definitions/test_dag.py
@@ -80,6 +80,18 @@ class TestDag:
assert dag.topological_sort() == ()
+ def test_dag_topological_sort_task_group_cycle(self):
+ """A TaskGroup that depends on a sibling in a cycle, although
check_cycle passes, still sorts."""
+ with DAG("dag", schedule=None, start_date=DEFAULT_DATE) as dag:
+ bridge = BaseOperator(task_id="bridge")
+ with TaskGroup("group"):
+ a = BaseOperator(task_id="a")
+ b = BaseOperator(task_id="b")
+ a >> bridge >> b
+ dag.check_cycle()
+
+ assert [task.task_id for task in dag.topological_sort()] == ["bridge",
"group.a", "group.b"]
+
def test_dag_naive_start_date_string(self):
DAG("DAG", schedule=None, default_args={"start_date": "2019-06-01"})
