code: def test_streaming_object_ref_generator_task_failed_unit(
      mocked_worker,
  ):
      """
      Verify when a task is failed by a system error,
      the generator ref is returned.
      """
      with patch("ray.get") as mocked_ray_get:
          with patch("ray.wait") as mocked_ray_wait:
              c = mocked_worker.core_worker
              generator_ref = ray.ObjectRef.from_random()
              generator = ObjectRefGenerator(
                  generator_ref, mocked_worker
              )
  
              # Simulate the worker failure happens.
              next_ref = ray.ObjectRef.from_random()
              c.peek_object_ref_stream.return_value = (next_ref, False)
              mocked_ray_wait.return_value = [next_ref], []
              mocked_ray_get.side_effect = WorkerCrashedError()
  
              c.try_read_next_object_ref_stream.side_effect = (
                  ObjectRefStreamEndOfStreamError("")
              )  # noqa
              ref = generator._next_sync(timeout_s=0)
              # If the generator task fails by a systsem error,
              # meaning the ref will raise an exception
              # it should be returned.
              assert ref == generator_ref
  
              # Once exception is raised, it should always
              # raise stopIteration regardless of what
              # the ref contains now.
              with pytest.raises(StopIteration):
                  ref = generator._next_sync(timeout_s=0)
  
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 3 for calling a global function with example usage: test_streaming_object_ref_generator_task_failed_unit(mocked_worker) without return types