code: def test_object_directory_basic(ray_start_cluster_with_resource):
      cluster, num_nodes = ray_start_cluster_with_resource
  
      @ray.remote
      def task(x):
          pass
  
      # Test a single task.
      x_id = ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
      ray.get(
          task.options(resources={str(3): 1}).remote(x_id), timeout=10
      )
  
      # Test multiple tasks on all nodes can find locations properly.
      object_refs = []
      for _ in range(num_nodes):
          object_refs.append(
              ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
          )
      ray.get(
          [
              task.options(resources={str(i): 1}).remote(object_refs[i])
              for i in range(num_nodes)
          ]
      )
      del object_refs
  
      @ray.remote
      class ObjectHolder:
          def __init__(self):
              self.x = ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
  
          def get_obj(self):
              return self.x
  
          def ready(self):
              return True
  
      # Test if tasks can find object location properly
      # when there are multiple owners
      object_holders = [
          ObjectHolder.options(
              num_cpus=0.01, resources={str(i): 1}
          ).remote()
          for i in range(num_nodes)
      ]
      ray.get([o.ready.remote() for o in object_holders])
  
      object_refs = []
      for i in range(num_nodes):
          object_refs.append(
              object_holders[(i + 1) % num_nodes].get_obj.remote()
          )
      ray.get(
          [
              task.options(num_cpus=0.01, resources={str(i): 1}).remote(
                  object_refs[i]
              )
              for i in range(num_nodes)
          ]
      )
  
      # Test a stressful scenario.
      object_refs = []
      repeat = 10
      for _ in range(num_nodes):
          for _ in range(repeat):
              object_refs.append(
                  ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
              )
      tasks = []
      for i in range(num_nodes):
          for r in range(repeat):
              tasks.append(
                  task.options(
                      num_cpus=0.01, resources={str(i): 0.1}
                  ).remote(object_refs[i * r])
              )
      ray.get(tasks)
  
      object_refs = []
      for i in range(num_nodes):
          object_refs.append(
              object_holders[(i + 1) % num_nodes].get_obj.remote()
          )
      tasks = []
      for i in range(num_nodes):
          for _ in range(10):
              tasks.append(
                  task.options(
                      num_cpus=0.01, resources={str(i): 0.1}
                  ).remote(object_refs[(i + 1) % num_nodes])
              )
  
imports: import multiprocessing
  import time
  import warnings
  from collections import defaultdict
  import numpy
  import pytest
  import ray
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 13 for calling a global function with example usage: test_object_directory_basic(ray_start_cluster_with_resource) and returns: self, True