code: def test_spread_scheduling_overrides_locality_aware_scheduling(
      ray_start_cluster,
  ):
      # This test ensures that explicit spread scheduling strategy has higher
      # priority than locality aware scheduling which means the lease request
      # will be sent to local raylet instead of locality favored raylet.
      cluster = ray_start_cluster
      local_node = cluster.add_node(
          num_cpus=8,
          _system_config={
              "worker_lease_timeout_milliseconds": 0,
              "max_direct_call_object_size": 0,
          },
      )
      ray.init(address=cluster.address)
      remote_node = cluster.add_node(num_cpus=8, resources={"pin": 1})
      cluster.wait_for_nodes()
  
      @ray.remote(resources={"pin": 1})
      def non_local():
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(scheduling_strategy="SPREAD")
      def f(x):
          return ray._private.worker.global_worker.node.unique_id
  
      # Test that task f() runs on the local node as well
      # even though remote node has the dependencies.
      obj1 = non_local.remote()
      obj2 = non_local.remote()
      assert {ray.get(f.remote(obj1)), ray.get(f.remote(obj2))} == {
          local_node.unique_id,
          remote_node.unique_id,
      }
  
imports: import collections
  import logging
  import platform
  import subprocess
  import sys
  import time
  import unittest
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.internal_api import memory_summary
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray._private.test_utils import Semaphore, SignalActor, object_memory_usage, get_metric_check_condition, wait_for_condition, MetricSamplePattern
label_desc: Use image node_id 8 for calling a global function with example usage: test_spread_scheduling_overrides_locality_aware_scheduling(ray_start_cluster) and returns: ray, ray