code: def test_generator_streaming(
      shutdown_only, use_actors, store_in_plasma
  ):
      """Verify the generator is working in a streaming fashion."""
      ray.init()
      remote_generator_fn = None
      if use_actors:
  
          @ray.remote
          class Generator:
              def __init__(self):
                  pass
  
              def generator(self, num_returns, store_in_plasma):
                  for i in range(num_returns):
                      if store_in_plasma:
                          yield np.ones(1_000_000, dtype=np.int8) * i
                      else:
                          yield [i]
  
          g = Generator.remote()
          remote_generator_fn = g.generator
      else:
  
          @ray.remote(max_retries=0)
          def generator(num_returns, store_in_plasma):
              for i in range(num_returns):
                  if store_in_plasma:
                      yield np.ones(1_000_000, dtype=np.int8) * i
                  else:
                      yield [i]
  
          remote_generator_fn = generator
  
      """Verify num_returns="streaming" is streaming"""
      gen = remote_generator_fn.remote(3, store_in_plasma)
      i = 0
      for ref in gen:
          id = ref.hex()
          if store_in_plasma:
              expected = np.ones(1_000_000, dtype=np.int8) * i
              assert np.array_equal(ray.get(ref), expected)
          else:
              expected = [i]
              assert ray.get(ref) == expected
  
          del ref
  
          wait_for_condition(
              lambda id=id: len(
                  list_objects(filters=[("object_id", "=", id)])
              )
              == 0
          )
          i += 1
  
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import threading
  import gc
  from unittest.mock import patch, Mock
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.experimental.state.api import list_objects
  from ray._raylet import ObjectRefGenerator, ObjectRefStreamEndOfStreamError
  from ray.cloudpickle import dumps
  from ray.exceptions import WorkerCrashedError
label_desc: Use image node_id 7 for calling a global function with example usage: test_generator_streaming(shutdown_only, use_actors, store_in_plasma) without return types