code: def test_task_summary(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=2)
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=2)
  
      @ray.remote
      def run_long_time_task():
          time.sleep(30)
          return True
  
      @ray.remote
      def task_wait_for_dep(dep):
          print(dep)
  
      a = task_wait_for_dep.remote(run_long_time_task.remote())  # noqa
      b = task_wait_for_dep.remote(run_long_time_task.remote())  # noqa
  
      def verify():
          # task_name -> states
          task_summary = summarize_tasks()
          task_summary = task_summary["cluster"]["summary"]
          assert "task_wait_for_dep" in task_summary
          assert "run_long_time_task" in task_summary
          assert (
              task_summary["task_wait_for_dep"]["state_counts"][
                  "PENDING_ARGS_AVAIL"
              ]
              == 2
          )
          assert (
              task_summary["run_long_time_task"]["state_counts"][
                  "RUNNING"
              ]
              == 2
          )
          assert (
              task_summary["task_wait_for_dep"]["type"] == "NORMAL_TASK"
          )
          return True
  
      wait_for_condition(verify)
  
      """
      Test CLI
      """
      runner = CliRunner()
      result = runner.invoke(summary_state_cli_group, ["tasks"])
      assert "task_wait_for_dep" in result.output
      assert result.exit_code == 0
  
imports: import time
  import json
  import pytest
  import ray
  from unittest.mock import MagicMock, AsyncMock
  import random
  import sys
  from dataclasses import asdict
  from ray.util.state import summarize_tasks, summarize_actors, summarize_objects
  from ray._private.test_utils import wait_for_condition
  from ray._raylet import ActorID, TaskID, ObjectID
  from ray.core.generated.common_pb2 import TaskStatus, TaskType, WorkerType
  from ray.core.generated.node_manager_pb2 import GetObjectsInfoReply
  from ray.tests.test_state_api import generate_task_data, generate_task_event, generate_actor_data, generate_object_info
  from ray.util.state.common import DEFAULT_RPC_TIMEOUT, SummaryApiOptions, Link, NestedTaskSummary, TaskSummaries, DRIVER_TASK_ID_PREFIX
  from ray.core.generated.gcs_service_pb2 import GetAllActorInfoReply
  from ray.core.generated.gcs_pb2 import ActorTableData
  from click.testing import CliRunner
  from ray.util.state.state_cli import summary_state_cli_group
  from ray.dashboard.state_aggregator import StateAPIManager
  from ray.util.state.state_manager import StateDataSourceClient
label_desc: Use image node_id 3 for calling a global function with example usage: test_task_summary(ray_start_cluster) and returns: True, True