code: def test_serve_num_replica_idle_node():
      # Test that nodes become idle after serve scaling down.
      cluster = AutoscalingCluster(
          head_resources={"CPU": 0},
          worker_node_types={
              "type-1": {
                  "resources": {"CPU": 4},
                  "node_config": {},
                  "min_workers": 0,
                  "max_workers": 30,
              },
          },
          idle_timeout_minutes=999,
      )
  
      from ray import serve
  
      @serve.deployment(ray_actor_options={"num_cpus": 2})
      class Deployment:
          def __call__(self):
              return "hello"
  
      try:
          cluster.start(
              override_env={
                  "RAY_SERVE_PROXY_MIN_DRAINING_PERIOD_S": "2"
              }
          )
          # 5 workers nodes should be busy and have 2(replicas) * 2(cpus per replicas)
          # = 4 cpus used
          serve.run(Deployment.options(num_replicas=10).bind())
  
          gcs_address = ray.get_runtime_context().gcs_address
          expected_num_workers = 5
  
          def verify():
              cluster_state = get_cluster_status(gcs_address)
  
              # Verify that nodes are busy.
              assert (
                  len(cluster_state.active_nodes)
                  == expected_num_workers + 1
              )
              for node in cluster_state.active_nodes:
                  assert node.node_status == "RUNNING"
                  if not is_head_node_from_resource_usage(
                      node.total_resources()
                  ):
                      available = node.available_resources()
                      assert available["CPU"] == 0
              assert len(cluster_state.idle_nodes) == 0
              return True
  
          wait_for_condition(verify)
  
          # Downscale to 1 replicas, 4 workers nodes should be idle.
          serve.run(Deployment.options(num_replicas=1).bind())
  
          def verify():
              cluster_state = get_cluster_status(gcs_address)
              # We should only have 1 running worker for the 1 replica, the rest idle.
              expected_idle_workers = expected_num_workers - 1
  
              assert (
                  len(cluster_state.idle_nodes)
                  + len(cluster_state.active_nodes)
                  == expected_num_workers + 1
              )
              idle_nodes = []
  
              for node in cluster_state.idle_nodes:
                  if not is_head_node_from_resource_usage(
                      node.total_resources()
                  ):
                      available = node.available_resources()
                      if node.node_status == "IDLE":
                          assert available["CPU"] == 4
                          idle_nodes.append(node)
              assert (
                  len(cluster_state.idle_nodes) == expected_idle_workers
              )
              return True
  
          # A long sleep is needed for serve proxy to be removed.
          wait_for_condition(verify, timeout=15, retry_interval_ms=1000)
  
      finally:
          ray.shutdown()
          cluster.shutdown()
  
file_path: ray/python/ray/autoscaler/v2/tests/test_e2e.py
imports: import os
  import sys
  import time
  from typing import Dict
  import pytest
  import ray
  from ray._private.resource_spec import HEAD_NODE_RESOURCE_NAME
  from ray._private.test_utils import run_string_as_driver_nonblocking, wait_for_condition
  from ray.autoscaler.v2.sdk import get_cluster_status
  from ray.cluster_utils import AutoscalingCluster
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.state.api import list_placement_groups, list_tasks
label_desc: Use image node_id 7 for calling a global function with example usage: test_serve_num_replica_idle_node() and returns: str, True, True