code: def test_get_cluster_status_resources(ray_start_cluster):
      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)
  
      @ray.remote(num_cpus=1)
      class Actor:
          def loop(self):
              while True:
                  pass
  
      # Schedule tasks to use all resources.
      @ray.remote(num_cpus=1)
      def loop():
          while True:
              pass
  
      [loop.remote() for _ in range(2)]
      actor = Actor.remote()
      actor.loop.remote()
  
      def verify_cpu_resources_all_used():
          cluster_status = get_cluster_status(cluster.address)
          total_cluster_resources = get_total_resources(
              cluster_status.cluster_resource_usage
          )
          assert total_cluster_resources["CPU"] == 3.0
  
          available_cluster_resources = get_available_resources(
              cluster_status.cluster_resource_usage
          )
          assert available_cluster_resources["CPU"] == 0.0
          return True
  
      wait_for_condition(verify_cpu_resources_all_used)
  
      # Schedule more tasks should show up as task demands
      [loop.remote() for _ in range(2)]
  
      def verify_task_demands():
          resource_demands = get_cluster_status(
              cluster.address
          ).resource_demands
          assert len(resource_demands.ray_task_actor_demand) == 1
          assert resource_demands.ray_task_actor_demand[
              0
          ].bundles_by_count == [
              ResourceRequestByCount(
                  bundle={"CPU": 1.0},
                  count=2,
              )
          ]
          return True
  
      wait_for_condition(verify_task_demands)
  
      # Request resources through SDK
      request_cluster_resources(
          gcs_address=cluster.address, to_request=[{"GPU": 1, "CPU": 2}]
      )
  
      def verify_cluster_constraint_demand():
          resource_demands = get_cluster_status(
              cluster.address
          ).resource_demands
          assert len(resource_demands.cluster_constraint_demand) == 1
          assert resource_demands.cluster_constraint_demand[
              0
          ].bundles_by_count == [
              ResourceRequestByCount(
                  bundle={"GPU": 1.0, "CPU": 2.0},
                  count=1,
              )
          ]
          return True
  
      wait_for_condition(verify_cluster_constraint_demand)
  
      # Try to schedule some PGs
      pg1 = ray.util.placement_group([{"CPU": 1}] * 3)
  
      def verify_pg_demands():
          resource_demands = get_cluster_status(
              cluster.address
          ).resource_demands
          assert len(resource_demands.placement_group_demand) == 1
          assert resource_demands.placement_group_demand[
              0
          ].bundles_by_count == [
              ResourceRequestByCount(
                  bundle={"CPU": 1.0},
                  count=3,
              )
          ]
          assert (
              resource_demands.placement_group_demand[0].pg_id
              == pg1.id.hex()
          )
          assert (
              resource_demands.placement_group_demand[0].strategy
              == "PACK"
          )
          assert (
              resource_demands.placement_group_demand[0].state
              == "PENDING"
          )
          return True
  
      wait_for_condition(verify_pg_demands)
  
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 15 for calling a global function with example usage: test_get_cluster_status_resources(ray_start_cluster) and returns: True, True, True, True