code: def get_test_run(dag):
      def callback(context: Context):
          ti = context["dag_run"].get_task_instances()
          if not ti:
              logging.warning(
                  "could not retrieve tasks that ran in the DAG, cannot display a summary"
              )
              return
  
          ti.sort(key=lambda x: x.end_date)
  
          headers = ["Task ID", "Status", "Duration (s)"]
          results = []
          # computing time using previous task's end time
          # because task.duration only counts last poke for sensors.
          # Only produces meaningful results when running sequentially, which is the case for system tests
          prev_time = ti[0].end_date - timedelta(seconds=ti[0].duration)
          for t in ti:
              results.append(
                  [
                      t.task_id,
                      t.state,
                      f"{(t.end_date - prev_time).total_seconds():.1f}",
                  ]
              )
              prev_time = t.end_date
  
          logging.info(
              "EXECUTION SUMMARY:\n"
              + tabulate(
                  results, headers=headers, tablefmt="fancy_grid"
              )
          )
  
      def add_callback(
          current: list[Callable] | Callable | None, new: Callable
      ) -> list[Callable] | Callable:
          if not current:
              return new
          elif isinstance(current, list):
              current.append(new)
              return current
          else:
              return [current, new]
  
      def test_run():
          dag.on_failure_callback = add_callback(
              dag.on_failure_callback, callback
          )
          dag.on_success_callback = add_callback(
              dag.on_success_callback, callback
          )
          dag.clear(dag_run_state=State.QUEUED)
          dag.run()
  
      return test_run
  
file_path: airflow/tests/system/utils/__init__.py
imports: from __future__ import annotations
  import logging
  import os
  from datetime import timedelta
  from typing import TYPE_CHECKING, Callable
  from tabulate import tabulate
  from airflow.utils.state import State
label_desc: Use image node_id 1 for calling a global function with example usage: get_test_run(dag) and returns: test_run, new, current, list