code: def test_scheduling_tasks_and_actors_during_draining(
      ray_start_cluster,
  ):
      """Test that the draining node is unschedulable for new tasks and actors."""
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1, resources={"head": 1})
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=1, 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()
      )
  
      @ray.remote
      class Actor:
          def ping(self):
              pass
  
      actor = Actor.options(
          num_cpus=0, resources={"worker": 1}
      ).remote()
      ray.get(actor.ping.remote())
  
      gcs_client = GcsClient(
          address=ray.get_runtime_context().gcs_address
      )
  
      # The worker node is not idle but the drain request should be still accepted.
      is_accepted = gcs_client.drain_node(
          worker_node_id,
          autoscaler_pb2.DrainNodeReason.Value(
              "DRAIN_NODE_REASON_PREEMPTION"
          ),
          "preemption",
      )
      assert is_accepted
  
      assert (
          ray.get(
              get_node_id.options(scheduling_strategy="SPREAD").remote()
          )
          == head_node_id
      )
      assert (
          ray.get(
              get_node_id.options(scheduling_strategy="SPREAD").remote()
          )
          == head_node_id
      )
  
      assert (
          ray.get(
              get_node_id.options(
                  scheduling_strategy=NodeAffinitySchedulingStrategy(
                      worker_node_id, soft=True
                  )
              ).remote()
          )
          == head_node_id
      )
  
      with pytest.raises(ray.exceptions.TaskUnschedulableError):
          ray.get(
              get_node_id.options(
                  scheduling_strategy=NodeAffinitySchedulingStrategy(
                      worker_node_id, soft=False
                  )
              ).remote()
          )
  
      head_actor = Actor.options(
          num_cpus=1, resources={"head": 1}
      ).remote()
      ray.get(head_actor.ping.remote())
  
      obj = get_node_id.remote()
  
      # Cannot run on the draining worker node even though it has resources.
      with pytest.raises(ray.exceptions.GetTimeoutError):
          ray.get(obj, timeout=2)
  
      ray.kill(head_actor)
      ray.get(obj, timeout=2) == head_node_id
  
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 4 for calling a global function with example usage: test_scheduling_tasks_and_actors_during_draining(ray_start_cluster) and returns: ray