code: def test_idle_termination(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(resources={"head": 1})
      ray.init(address=cluster.address)
      cluster.add_node(resources={"worker": 1})
      cluster.wait_for_nodes()
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      head_node_id = ray.get(
          get_node_id.options(resources={"head": 1}).remote()
      )
      worker_node_id = ray.get(
          get_node_id.options(resources={"worker": 1}).remote()
      )
  
      wait_for_condition(
          lambda: {
              node["NodeID"] for node in ray.nodes() if (node["Alive"])
          }
          == {head_node_id, worker_node_id}
      )
  
      @ray.remote(num_cpus=1, resources={"worker": 1})
      class Actor:
          def ping(self):
              pass
  
      actor = Actor.remote()
      ray.get(actor.ping.remote())
  
      gcs_client = GcsClient(
          address=ray.get_runtime_context().gcs_address
      )
  
      # The worker node is not idle so the drain request should be rejected.
      is_accepted = gcs_client.drain_node(
          worker_node_id,
          autoscaler_pb2.DrainNodeReason.Value(
              "DRAIN_NODE_REASON_IDLE_TERMINATION"
          ),
          "idle for long enough",
      )
      assert not is_accepted
  
      ray.kill(actor)
  
      def drain_until_accept():
          # The worker node is idle now so the drain request should be accepted.
          is_accepted = gcs_client.drain_node(
              worker_node_id,
              autoscaler_pb2.DrainNodeReason.Value(
                  "DRAIN_NODE_REASON_IDLE_TERMINATION"
              ),
              "idle for long enough",
          )
          return is_accepted
  
      wait_for_condition(drain_until_accept)
  
      wait_for_condition(
          lambda: {
              node["NodeID"] for node in ray.nodes() if (node["Alive"])
          }
          == {head_node_id}
      )
  
      # Draining a dead node is always accepted.
      is_accepted = gcs_client.drain_node(
          worker_node_id,
          autoscaler_pb2.DrainNodeReason.Value(
              "DRAIN_NODE_REASON_IDLE_TERMINATION"
          ),
          "idle for long enough",
      )
      assert is_accepted
  
file_path: ray/python/ray/tests/test_draining.py
imports: import sys
  import pytest
  import ray
  import time
  from ray._raylet import GcsClient
  from ray.core.generated import autoscaler_pb2
  from ray._private.test_utils import wait_for_condition
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, PlacementGroupSchedulingStrategy
label_desc: Use image node_id 1 for calling a global function with example usage: test_idle_termination(ray_start_cluster) and returns: ray, is_accepted