code: def test_cluster_status_parser_cluster_resource_state():
      test_data = {
          "cluster_resource_state": {
              "node_states": [
                  {
                      "node_id": b"1" * 4,
                      "instance_id": "instance1",
                      "ray_node_type_name": "head_node",
                      "available_resources": {
                          "CPU": 0.5,
                          "GPU": 2.0,
                      },
                      "total_resources": {
                          "CPU": 1,
                          "GPU": 2.0,
                      },
                      "status": "RUNNING",
                      "node_ip_address": "10.10.10.10",
                      "instance_type_name": "m5.large",
                  },
                  {
                      "node_id": b"2" * 4,
                      "instance_id": "instance2",
                      "ray_node_type_name": "worker_node",
                      "available_resources": {},
                      "total_resources": {
                          "CPU": 1,
                          "GPU": 2.0,
                      },
                      "status": "DEAD",
                      "node_ip_address": "22.22.22.22",
                      "instance_type_name": "m5.large",
                  },
                  {
                      "node_id": b"3" * 4,
                      "instance_id": "instance3",
                      "ray_node_type_name": "worker_node",
                      "available_resources": {
                          "CPU": 1.0,
                          "GPU": 2.0,
                      },
                      "total_resources": {
                          "CPU": 1,
                          "GPU": 2.0,
                      },
                      "idle_duration_ms": 100,
                      "status": "IDLE",
                      "node_ip_address": "22.22.22.22",
                      "instance_type_name": "m5.large",
                  },
              ],
              "pending_gang_resource_requests": [
                  {
                      "requests": [
                          {
                              "resources_bundle": {"CPU": 1, "GPU": 1},
                              "placement_constraints": [
                                  {
                                      "anti_affinity": {
                                          "label_name": "_PG_1x1x",
                                          "label_value": "",
                                      }
                                  }
                              ],
                          },
                      ],
                      "details": "1x1x:STRICT_SPREAD|PENDING",
                  },
                  {
                      "requests": [
                          {
                              "resources_bundle": {"GPU": 2},
                              "placement_constraints": [
                                  {
                                      "affinity": {
                                          "label_name": "_PG_2x2x",
                                          "label_value": "",
                                      }
                                  }
                              ],
                          },
                      ],
                      "details": "2x2x:STRICT_PACK|PENDING",
                  },
              ],
              "pending_resource_requests": [
                  {
                      "request": {
                          "resources_bundle": {"CPU": 1, "GPU": 1},
                          "placement_constraints": [],
                      },
                      "count": 1,
                  },
              ],
              "cluster_resource_constraints": [
                  {
                      "min_bundles": [
                          {
                              "request": {
                                  "resources_bundle": {
                                      "GPU": 2,
                                      "CPU": 100,
                                  },
                                  "placement_constraints": [],
                              },
                              "count": 1,
                          },
                      ]
                  }
              ],
              "cluster_resource_state_version": 10,
          },
          "autoscaling_state": {},
      }
      reply = _gen_cluster_status_reply(test_data)
      stats = Stats(gcs_request_time_s=0.1)
      cluster_status = (
          ClusterStatusParser.from_get_cluster_status_reply(
              reply, stats
          )
      )
  
      # Assert on health nodes
      assert (
          len(cluster_status.idle_nodes)
          + len(cluster_status.active_nodes)
          == 2
      )
      assert cluster_status.active_nodes[0].instance_id == "instance1"
      assert (
          cluster_status.active_nodes[0].ray_node_type_name
          == "head_node"
      )
      cluster_status.active_nodes[0].resource_usage.usage.sort(
          key=lambda x: x.resource_name
      )
      assert cluster_status.active_nodes[0].resource_usage == NodeUsage(
          usage=[
              ResourceUsage(resource_name="CPU", total=1.0, used=0.5),
              ResourceUsage(resource_name="GPU", total=2.0, used=0.0),
          ],
          idle_time_ms=0,
      )
  
      assert cluster_status.idle_nodes[0].instance_id == "instance3"
      assert (
          cluster_status.idle_nodes[0].ray_node_type_name
          == "worker_node"
      )
      cluster_status.idle_nodes[0].resource_usage.usage.sort(
          key=lambda x: x.resource_name
      )
      assert cluster_status.idle_nodes[0].resource_usage == NodeUsage(
          usage=[
              ResourceUsage(resource_name="CPU", total=1.0, used=0.0),
              ResourceUsage(resource_name="GPU", total=2.0, used=0.0),
          ],
          idle_time_ms=100,
      )
  
      # Assert on dead nodes
      assert len(cluster_status.failed_nodes) == 1
      assert cluster_status.failed_nodes[0].instance_id == "instance2"
      assert (
          cluster_status.failed_nodes[0].ray_node_type_name
          == "worker_node"
      )
      assert cluster_status.failed_nodes[0].resource_usage is None
  
      # Assert on resource demands from tasks
      assert (
          len(cluster_status.resource_demands.ray_task_actor_demand)
          == 1
      )
      assert cluster_status.resource_demands.ray_task_actor_demand[
          0
      ].bundles_by_count == [
          ResourceRequestByCount(
              bundle={"CPU": 1, "GPU": 1},
              count=1,
          )
      ]
  
      # Assert on resource demands from placement groups
      assert (
          len(cluster_status.resource_demands.placement_group_demand)
          == 2
      )
      assert sorted(
          cluster_status.resource_demands.placement_group_demand,
          key=lambda x: x.pg_id,
      ) == [
          PlacementGroupResourceDemand(
              bundles_by_count=[
                  ResourceRequestByCount(
                      bundle={"CPU": 1, "GPU": 1}, count=1
                  )
              ],
              strategy="STRICT_SPREAD",
              pg_id="1x1x",
              state="PENDING",
              details="1x1x:STRICT_SPREAD|PENDING",
          ),
          PlacementGroupResourceDemand(
              bundles_by_count=[
                  ResourceRequestByCount(bundle={"GPU": 2}, count=1)
              ],
              strategy="STRICT_PACK",
              pg_id="2x2x",
              state="PENDING",
              details="2x2x:STRICT_PACK|PENDING",
          ),
      ]
  
      # Assert on resource constraints
      assert (
          len(cluster_status.resource_demands.cluster_constraint_demand)
          == 1
      )
      assert cluster_status.resource_demands.cluster_constraint_demand[
          0
      ].bundles_by_count == [
          ResourceRequestByCount(bundle={"GPU": 2, "CPU": 100}, count=1)
      ]
  
      # Assert on the cluster_resource_usage
      assert sorted(
          cluster_status.cluster_resource_usage,
          key=lambda x: x.resource_name,
      ) == [
          ResourceUsage(resource_name="CPU", total=2.0, used=0.5),
          ResourceUsage(resource_name="GPU", total=4.0, used=0.0),
      ]
  
      # Assert on the node stats
      assert cluster_status.stats.cluster_resource_state_version == "10"
      assert cluster_status.stats.gcs_request_time_s == 0.1
  
file_path: ray/python/ray/autoscaler/v2/tests/test_utils.py
imports: import os
  import sys
  from typing import Dict
  import pytest
  from google.protobuf.json_format import ParseDict
  from ray.autoscaler.v2.schema import ClusterConstraintDemand, ClusterStatus, LaunchRequest, NodeInfo, NodeUsage, PlacementGroupResourceDemand, RayTaskActorDemand, ResourceDemandSummary, ResourceRequestByCount, ResourceUsage, Stats
  from ray.autoscaler.v2.utils import ClusterStatusFormatter, ClusterStatusParser
  from ray.core.generated.autoscaler_pb2 import GetClusterStatusReply
label_desc: Use image node_id 2 for calling a global function with example usage: test_cluster_status_parser_cluster_resource_state() without return types