code: def test_spilled_locations(ray_start_cluster_enabled):
      cluster = ray_start_cluster_enabled
      cluster.add_node(num_cpus=1, object_store_memory=75 * 1024 * 1024)
      ray.init(cluster.address)
      cluster.wait_for_nodes()
  
      node_id = ray.get_runtime_context().get_node_id()
  
      @ray.remote
      def task():
          arr = np.random.rand(5 * 1024 * 1024)  # 40 MB
          refs = []
          refs.extend([ray.put(arr) for _ in range(2)])
          ray.get(ray.put(arr))
          ray.get(ray.put(arr))
          return refs
  
      object_refs = ray.get(task.remote())
      ray.wait(object_refs)
      locations = ray.experimental.get_object_locations(object_refs)
      for obj_ref in object_refs:
          location = locations[obj_ref]
          assert location["node_ids"] == [node_id]
          assert location["object_size"] > 0
  
file_path: ray/python/ray/tests/test_get_locations.py
imports: import time
  import numpy
  import pytest
  import ray
label_desc: Use image node_id 6 for calling a global function with example usage: test_spilled_locations(ray_start_cluster_enabled) and returns: refs