code: def test_streaming_generator_exception(shutdown_only):
      # Verify the exceptions are correctly raised.
      # Also verify the followup next will raise StopIteration.
      ray.init()
  
      @ray.remote
      class Actor:
          def f(self):
              raise ValueError
              yield 1  # noqa
  
          async def async_f(self):
              raise ValueError
              yield 1  # noqa
  
      a = Actor.remote()
      g = a.f.remote()
      with pytest.raises(ValueError):
          ray.get(next(g))
  
      with pytest.raises(StopIteration):
          ray.get(next(g))
  
      with pytest.raises(StopIteration):
          ray.get(next(g))
  
      g = a.async_f.remote()
      with pytest.raises(ValueError):
          ray.get(next(g))
  
      with pytest.raises(StopIteration):
          ray.get(next(g))
  
      with pytest.raises(StopIteration):
          ray.get(next(g))
  
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 11 for calling a global function with example usage: test_streaming_generator_exception(shutdown_only) without return types