code: def test_get_cluster_status(ray_start_cluster):
      # This test is to make sure the grpc stub is working.
      # TODO(rickyx): Add e2e tests for the autoscaler state service in a separate PR
      # to validate the data content.
      cluster = ray_start_cluster
      # Head node
      cluster.add_node(
          num_cpus=1, _system_config={"enable_autoscaler_v2": True}
      )
      ray.init(address=cluster.address)
      # Worker node
      cluster.add_node(num_cpus=2)
  
      head_node_id, worker_node_ids = get_node_ids()
  
      def verify_nodes():
          cluster_status = get_cluster_status(cluster.address)
          assert_nodes(
              cluster_status.idle_nodes,
              [
                  ExpectedNodeInfo(
                      worker_node_ids[0],
                      "IDLE",
                      lambda idle_ms: idle_ms > 0,
                      total_resources={"CPU": 2.0},
                      available_resources={"CPU": 2.0},
                  ),
                  ExpectedNodeInfo(
                      head_node_id,
                      "IDLE",
                      lambda idle_ms: idle_ms > 0,
                      total_resources={"CPU": 1.0},
                      available_resources={"CPU": 1.0},
                  ),
              ],
          )
          return True
  
      wait_for_condition(verify_nodes)
  
      # Schedule a task running
      @ray.remote(num_cpus=2)
      def f():
          while True:
              pass
  
      f.remote()
  
      def verify_nodes_busy():
          cluster_status = get_cluster_status(cluster.address)
          assert_nodes(
              cluster_status.idle_nodes,
              [
                  ExpectedNodeInfo(
                      head_node_id, "IDLE", lambda idle_ms: idle_ms > 0
                  ),
              ],
          )
          assert_nodes(
              cluster_status.active_nodes,
              [
                  ExpectedNodeInfo(
                      worker_node_ids[0],
                      "RUNNING",
                      lambda idle_ms: idle_ms == 0,
                      total_resources={"CPU": 2.0},
                      available_resources={"CPU": 0.0},
                  ),
              ],
          )
          return True
  
      wait_for_condition(verify_nodes_busy)
  
      stub = _autoscaler_state_service_stub()
      state = autoscaler_pb2.AutoscalingState(
          last_seen_cluster_resource_state_version=0,
          autoscaler_state_version=1,
          pending_instance_requests=[
              autoscaler_pb2.PendingInstanceRequest(
                  instance_type_name="m5.large",
                  ray_node_type_name="worker",
                  count=2,
                  request_ts=1000,
              )
          ],
          failed_instance_requests=[
              autoscaler_pb2.FailedInstanceRequest(
                  instance_type_name="m5.large",
                  ray_node_type_name="worker",
                  count=2,
                  start_ts=1000,
                  failed_ts=2000,
                  reason="insufficient quota",
              )
          ],
          pending_instances=[
              autoscaler_pb2.PendingInstance(
                  instance_id="instance-id",
                  instance_type_name="m5.large",
                  ray_node_type_name="worker",
                  ip_address="10.10.10.10",
                  details="launching",
              )
          ],
      )
      report_autoscaling_state(stub, autoscaling_state=state)
  
      def verify_autoscaler_state():
          # TODO(rickyx): Add infeasible asserts.
  
          cluster_status = get_cluster_status(cluster.address)
          assert len(cluster_status.pending_launches) == 1
          assert_launches(
              cluster_status,
              expected_pending_launches=[
                  LaunchRequest(
                      instance_type_name="m5.large",
                      ray_node_type_name="worker",
                      count=2,
                      state=LaunchRequest.Status.PENDING,
                      request_ts_s=1000,
                  )
              ],
              expected_failed_launches=[
                  LaunchRequest(
                      instance_type_name="m5.large",
                      ray_node_type_name="worker",
                      count=2,
                      state=LaunchRequest.Status.FAILED,
                      request_ts_s=1000,
                      failed_ts_s=2000,
                      details="insufficient quota",
                  )
              ],
          )
  
          assert_nodes(
              cluster_status.pending_nodes,
              [
                  ExpectedNodeInfo(
                      instance_id="instance-id",
                      ray_node_type_name="worker",
                      details="launching",
                      ip_address="10.10.10.10",
                  )
              ],
          )
          return True
  
      wait_for_condition(verify_autoscaler_state)
  
imports: import os
  import sys
  import time
  from dataclasses import dataclass
  from typing import Callable, List, Optional, Tuple
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import wait_for_condition
  from ray.autoscaler.v2.schema import ClusterStatus, LaunchRequest, NodeInfo, ResourceRequestByCount
  from ray.autoscaler.v2.sdk import get_cluster_status, request_cluster_resources
  from ray.autoscaler.v2.tests.util import get_available_resources, get_cluster_resource_state, get_total_resources, report_autoscaling_state
  from ray.core.generated import autoscaler_pb2, autoscaler_pb2_grpc
  from ray.core.generated.autoscaler_pb2 import ClusterResourceState, NodeStatus
  from ray.util.state.api import list_nodes
label_desc: Use image node_id 16 for calling a global function with example usage: test_get_cluster_status(ray_start_cluster) and returns: True, True, True