code: def test_draining_reason(ray_start_cluster, graceful):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1, resources={"node1": 1})
      ray.init(
          address=cluster.address,
      )
      n = cluster.add_node(num_cpus=1, resources={"node2": 1})
  
      @ray.remote
      class Actor:
          def ping(self):
              pass
  
      gcs_client = GcsClient(
          address=ray.get_runtime_context().gcs_address
      )
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      node2_id = ray.get(
          get_node_id.options(resources={"node2": 1}).remote()
      )
  
      # Schedule actor
      actor = Actor.options(num_cpus=0, resources={"node2": 1}).remote()
      ray.get(actor.ping.remote())
  
      # Preemption is always accepted.
      is_accepted = gcs_client.drain_node(
          node2_id,
          autoscaler_pb2.DrainNodeReason.Value(
              "DRAIN_NODE_REASON_PREEMPTION"
          ),
          "preemption",
      )
      assert is_accepted
  
      cluster.remove_node(n, graceful)
      try:
          ray.get(actor.ping.remote())
          raise
      except ray.exceptions.RayActorError as e:
          assert e.preempted
  
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 5 for calling a global function with example usage: test_draining_reason(ray_start_cluster, graceful) and returns: ray