code: def test_dag_edges():
      execution_date = pendulum.parse("20200101")
      with DAG("test_dag_edges", start_date=execution_date) as dag:
          task1 = EmptyOperator(task_id="task1")
          with TaskGroup("group_a") as group_a:
              with TaskGroup("group_b") as group_b:
                  task2 = EmptyOperator(task_id="task2")
                  task3 = EmptyOperator(task_id="task3")
                  task4 = EmptyOperator(task_id="task4")
                  task2 >> [task3, task4]
  
              task5 = EmptyOperator(task_id="task5")
  
              task5 << group_b
  
          task1 >> group_a
  
          with TaskGroup("group_c") as group_c:
              task6 = EmptyOperator(task_id="task6")
              task7 = EmptyOperator(task_id="task7")
              task8 = EmptyOperator(task_id="task8")
              [task6, task7] >> task8
              group_a >> group_c
  
          task5 >> task8
  
          task9 = EmptyOperator(task_id="task9")
          task10 = EmptyOperator(task_id="task10")
  
          group_c >> [task9, task10]
  
          with TaskGroup("group_d") as group_d:
              task11 = EmptyOperator(task_id="task11")
              task12 = EmptyOperator(task_id="task12")
              task11 >> task12
  
          group_d << group_c
  
      nodes = task_group_to_dict(dag.task_group)
      edges = dag_edges(dag)
  
      assert extract_node_id(nodes) == {
          "id": None,
          "children": [
              {
                  "id": "group_a",
                  "children": [
                      {
                          "id": "group_a.group_b",
                          "children": [
                              {"id": "group_a.group_b.task2"},
                              {"id": "group_a.group_b.task3"},
                              {"id": "group_a.group_b.task4"},
                              {
                                  "id": "group_a.group_b.downstream_join_id"
                              },
                          ],
                      },
                      {"id": "group_a.task5"},
                      {"id": "group_a.upstream_join_id"},
                      {"id": "group_a.downstream_join_id"},
                  ],
              },
              {
                  "id": "group_c",
                  "children": [
                      {"id": "group_c.task6"},
                      {"id": "group_c.task7"},
                      {"id": "group_c.task8"},
                      {"id": "group_c.upstream_join_id"},
                      {"id": "group_c.downstream_join_id"},
                  ],
              },
              {
                  "id": "group_d",
                  "children": [
                      {"id": "group_d.task11"},
                      {"id": "group_d.task12"},
                      {"id": "group_d.upstream_join_id"},
                  ],
              },
              {"id": "task1"},
              {"id": "task10"},
              {"id": "task9"},
          ],
      }
  
      assert sorted(
          (e["source_id"], e["target_id"]) for e in edges
      ) == [
          ("group_a.downstream_join_id", "group_c.upstream_join_id"),
          ("group_a.group_b.downstream_join_id", "group_a.task5"),
          ("group_a.group_b.task2", "group_a.group_b.task3"),
          ("group_a.group_b.task2", "group_a.group_b.task4"),
          (
              "group_a.group_b.task3",
              "group_a.group_b.downstream_join_id",
          ),
          (
              "group_a.group_b.task4",
              "group_a.group_b.downstream_join_id",
          ),
          ("group_a.task5", "group_a.downstream_join_id"),
          ("group_a.task5", "group_c.task8"),
          ("group_a.upstream_join_id", "group_a.group_b.task2"),
          ("group_c.downstream_join_id", "group_d.upstream_join_id"),
          ("group_c.downstream_join_id", "task10"),
          ("group_c.downstream_join_id", "task9"),
          ("group_c.task6", "group_c.task8"),
          ("group_c.task7", "group_c.task8"),
          ("group_c.task8", "group_c.downstream_join_id"),
          ("group_c.upstream_join_id", "group_c.task6"),
          ("group_c.upstream_join_id", "group_c.task7"),
          ("group_d.task11", "group_d.task12"),
          ("group_d.upstream_join_id", "group_d.task11"),
          ("task1", "group_a.upstream_join_id"),
      ]
  
imports: from __future__ import annotations
  from datetime import timedelta
  import pendulum
  import pytest
  from airflow.decorators import dag, setup, task, task_group, teardown
  from airflow.exceptions import TaskAlreadyInTaskGroup
  from airflow.models.baseoperator import BaseOperator
  from airflow.models.dag import DAG
  from airflow.models.xcom_arg import XComArg
  from airflow.operators.bash import BashOperator
  from airflow.operators.empty import EmptyOperator
  from airflow.operators.python import PythonOperator
  from airflow.utils.dag_edges import dag_edges
  from airflow.utils.task_group import TaskGroup, task_group_to_dict
  from tests.models import DEFAULT_DATE
label_desc: Use image node_id 8 for calling a global function with example usage: test_dag_edges() without return types