code: def test_object_summary(monkeypatch, ray_start_cluster):
      with monkeypatch.context() as m:
          m.setenv("RAY_record_ref_creation_sites", "1")
          cluster = ray_start_cluster
          cluster.add_node(num_cpus=4)
          ray.init(address=cluster.address)
  
          dep = ray.put(1)  # noqa
  
          @ray.remote
          def task_wait_for_dep(dep):
              time.sleep(30)
  
          a = [task_wait_for_dep.remote(dep) for _ in range(2)]  # noqa
  
          def verify():
              summary = summarize_objects()
              assert "cluster" in summary
              assert summary["cluster"]["callsite_enabled"] is True
              summary = summary["cluster"]["summary"]
  
              deserialized_task_arg_summary = None
              put_obj_summary = None
              return_ref_summary = None
  
              for k, v in summary.items():
                  if "(deserialize task arg)" in k:
                      deserialized_task_arg_summary = v
                  elif "(put object)" in k:
                      put_obj_summary = v
                  elif "(task call)" in k:
                      return_ref_summary = v
  
              assert deserialized_task_arg_summary["total_objects"] == 2
              assert (
                  deserialized_task_arg_summary["total_num_workers"]
                  == 2
              )
              assert (
                  deserialized_task_arg_summary["total_num_nodes"] == 1
              )
              assert (
                  deserialized_task_arg_summary["task_state_counts"][
                      "NIL"
                  ]
                  == 2
              )
              assert (
                  deserialized_task_arg_summary["ref_type_counts"][
                      "PINNED_IN_MEMORY"
                  ]
                  == 2
              )
  
              assert put_obj_summary["total_objects"] == 1
              assert put_obj_summary["total_num_workers"] == 1
              assert put_obj_summary["total_num_nodes"] == 1
              assert (
                  put_obj_summary["task_state_counts"]["FINISHED"] == 1
              )
              assert (
                  put_obj_summary["ref_type_counts"][
                      "USED_BY_PENDING_TASK"
                  ]
                  == 1
              )
  
              assert return_ref_summary["total_objects"] == 2
              assert return_ref_summary["total_num_workers"] == 1
              assert return_ref_summary["total_num_nodes"] == 1
              assert (
                  return_ref_summary["task_state_counts"][
                      "SUBMITTED_TO_WORKER"
                  ]
                  == 2
              )
              assert (
                  return_ref_summary["ref_type_counts"][
                      "LOCAL_REFERENCE"
                  ]
                  == 2
              )
  
              return True
  
          wait_for_condition(verify)
  
          """
          Test CLI
          """
          runner = CliRunner()
          result = runner.invoke(summary_state_cli_group, ["objects"])
          assert "(deserialize task arg)" 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 5 for calling a global function with example usage: test_object_summary(monkeypatch, ray_start_cluster) and returns: True