code: def test_generator_basic(shutdown_only):
      ray.init(num_cpus=1)
  
      """Basic cases"""
      print("Test basic case")
  
      @ray.remote
      def f():
          for i in range(5):
              yield i
  
      gen = f.remote()
      i = 0
      for ref in gen:
          print(ray.get(ref))
          assert i == ray.get(ref)
          del ref
          i += 1
  
      """Exceptions"""
      print("Test exceptions")
  
      @ray.remote
      def f():
          for i in range(5):
              if i == 2:
                  raise ValueError
              yield i
  
      gen = f.remote()
      print(ray.get(next(gen)))
      print(ray.get(next(gen)))
      with pytest.raises(ray.exceptions.RayTaskError) as e:
          print(ray.get(next(gen)))
      with pytest.raises(StopIteration):
          ray.get(next(gen))
      with pytest.raises(StopIteration):
          ray.get(next(gen))
  
      """Generator Task failure"""
      print("Test task failures")
  
      @ray.remote
      class A:
          def getpid(self):
              import os
  
              return os.getpid()
  
          def f(self):
              for i in range(5):
                  time.sleep(1)
                  yield i
  
      a = A.remote()
      gen = a.f.remote()
      i = 0
      for ref in gen:
          if i == 2:
              ray.kill(a)
          if i == 3:
              with pytest.raises(ray.exceptions.RayActorError) as e:
                  ray.get(ref)
              assert (
                  "The actor is dead because it was killed by `ray.kill`"
                  in str(e.value)
              )
              break
          assert i == ray.get(ref)
          del ref
          i += 1
      for _ in range(10):
          with pytest.raises(StopIteration):
              next(gen)
  
      """Retry exceptions"""
      print("Test retry exceptions")
  
      @ray.remote
      class Actor:
          def __init__(self):
              self.should_kill = True
  
          def should_kill(self):
              return self.should_kill
  
          async def set(self, wait_s):
              await asyncio.sleep(wait_s)
              self.should_kill = False
  
      @ray.remote(retry_exceptions=[ValueError], max_retries=10)
      def f(a):
          for i in range(5):
              should_kill = ray.get(a.should_kill.remote())
              if i == 3 and should_kill:
                  raise ValueError
              yield i
  
      a = Actor.remote()
      gen = f.remote(a)
      assert ray.get(next(gen)) == 0
      assert ray.get(next(gen)) == 1
      assert ray.get(next(gen)) == 2
      a.set.remote(3)
      assert ray.get(next(gen)) == 3
      assert ray.get(next(gen)) == 4
      with pytest.raises(StopIteration):
          ray.get(next(gen))
  
      """Cancel"""
      print("Test cancel")
  
      @ray.remote
      def f():
          for i in range(5):
              time.sleep(5)
              yield i
  
      gen = f.remote()
      assert ray.get(next(gen)) == 0
      ray.cancel(gen)
      with pytest.raises(ray.exceptions.RayTaskError) as e:
          assert ray.get(next(gen)) == 1
      assert "was cancelled" in str(e.value)
      with pytest.raises(StopIteration):
          next(gen)
  
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 4 for calling a global function with example usage: test_generator_basic(shutdown_only) and returns: os, self