Add new tests and update old test to suppport the changes made.

tests/decorators/test_task_group.py

@@ -333,3 +333,19 @@ def another_tg():
     assert test_task.retries == 1
     assert test_task.owner == "y"
     assert test_task.execution_timeout == timedelta(seconds=10)
+
+
+def test_task_group_display_name_used_as_label():
+    """Test that the group_display_name for TaskGroup is used as the label for display on the UI."""
+
+    @dag(schedule=None, start_date=pendulum.datetime(2022, 1, 1))
+    def pipeline():
+        @task_group(group_display_name="my_custom_name")
+        def tg():
+            pass
+
+        tg()
+
+    p = pipeline()
+
+    assert p.task_group_dict["tg"].label == "my_custom_name"

tests/serialization/test_dag_serialization.py

@@ -140,6 +140,7 @@
         },
         "task_group": {
             "_group_id": None,
+            "group_display_name": "",
             "prefix_group_id": True,
             "children": {
                 "bash_task": ("operator", "bash_task"),
@@ -2994,6 +2995,7 @@ def tg(a: str) -> None:
                 "type": "dict-of-lists",
                 "value": {"__type": "dict", "__var": {"a": [".", ".."]}},
             },
+            "group_display_name": "",
             "is_mapped": True,
             "prefix_group_id": True,
             "tooltip": "",

tests/utils/test_task_group.py

@@ -1744,3 +1744,32 @@ def test_task_group_with_invalid_arg_type_raises_error():
     with DAG(dag_id="dag_with_tg_invalid_arg_type", schedule=None):
         with pytest.raises(TypeError, match=error_msg):
             _ = TaskGroup("group_1", ui_color=123)
+
+
+def test_task_group_display_name_used_as_label():
+    """Test that the group_display_name for TaskGroup is used as the label for display on the UI."""
+
+    with DAG(dag_id="display_name", schedule=None, start_date=pendulum.datetime(2022, 1, 1)) as dag:
+        with TaskGroup(group_id="tg", group_display_name="my_custom_name") as tg:
+            task1 = BaseOperator(task_id="task1")
+            task2 = BaseOperator(task_id="task2")
+            task1 >> task2
+
+    assert tg.group_id == "tg"
+    assert tg.label == "my_custom_name"
+    expected_node_id = {
+        "id": None,
+        "label": None,
+        "children": [
+            {
+                "id": "tg",
+                "label": "my_custom_name",
+                "children": [
+                    {"id": "tg.task1", "label": "task1"},
+                    {"id": "tg.task2", "label": "task2"},
+                ],
+            },
+        ],
+    }
+
+    assert extract_node_id(task_group_to_dict(dag.task_group), include_label=True) == expected_node_id