code: def test_fusion_objects(
      fs_only_object_spilling_config, shutdown_only
  ):
      # Limit our object store to 75 MiB of memory.
      (
          object_spilling_config,
          temp_folder,
      ) = fs_only_object_spilling_config
      min_spilling_size = 10 * 1024 * 1024
      address = ray.init(
          object_store_memory=75 * 1024 * 1024,
          _system_config={
              "max_io_workers": 3,
              "automatic_object_spilling_enabled": True,
              "object_store_full_delay_ms": 100,
              "object_spilling_config": object_spilling_config,
              "min_spilling_size": min_spilling_size,
          },
      )
      replay_buffer = []
      solution_buffer = []
      buffer_length = 100
  
      # Create objects of more than 800 MiB.
      for _ in range(buffer_length):
          ref = None
          while ref is None:
              multiplier = random.choice([1, 2, 3])
              arr = np.random.rand(multiplier * 1024 * 1024)
              ref = ray.put(arr)
              replay_buffer.append(ref)
              solution_buffer.append(arr)
  
      print("-----------------------------------")
      # randomly sample objects
      for _ in range(1000):
          index = random.choice(list(range(buffer_length)))
          ref = replay_buffer[index]
          solution = solution_buffer[index]
          sample = ray.get(ref, timeout=None)
          assert np.array_equal(sample, solution)
  
      is_test_passing = False
      # Since we'd like to see the temp directory that stores the files,
      # we need to append this directory.
      temp_folder = (
          temp_folder / ray._private.ray_constants.DEFAULT_OBJECT_PREFIX
      )
      for path in temp_folder.iterdir():
          file_size = path.stat().st_size
          # Make sure there are at least one
          # file_size that exceeds the min_spilling_size.
          # If we don't fusion correctly, this cannot happen.
          if file_size >= min_spilling_size:
              is_test_passing = True
      assert is_test_passing
      assert_no_thrashing(address["address"])
  
imports: import os
  import platform
  import random
  import subprocess
  import sys
  import tempfile
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.tests.test_object_spilling import assert_no_thrashing, is_dir_empty
  from ray._private.external_storage import FileSystemStorage, ExternalStorageRayStorageImpl
label_desc: Use image node_id 6 for calling a global function with example usage: test_fusion_objects(fs_only_object_spilling_config, shutdown_only) without return types