code: def test_streaming_object_ref_generator_basic_unit(mocked_worker):
      """
      Verify the basic case:
      create a generator -> read values -> nothing more to read -> delete.
      """
      with patch("ray.wait") as mocked_ray_wait:
          with patch("ray.get") as mocked_ray_get:
              c = mocked_worker.core_worker
              generator_ref = ray.ObjectRef.from_random()
              generator = ObjectRefGenerator(
                  generator_ref, mocked_worker
              )
  
              # Test when there's no new ref, it returns a nil.
              new_ref = ray.ObjectRef.from_random()
              c.peek_object_ref_stream.return_value = (new_ref, False)
              mocked_ray_wait.return_value = [], [new_ref]
              ref = generator._next_sync(timeout_s=0)
              assert ref.is_nil()
  
              # When the new ref is available, next should return it.
              # When peek_object_ref_stream returns is_ready = True,
              # it shouldn't call ray.wait.
              new_ref = ray.ObjectRef.from_random()
              c.peek_object_ref_stream.return_value = (new_ref, True)
              c.try_read_next_object_ref_stream.return_value = new_ref
              ref = generator._next_sync(timeout_s=0)
              assert new_ref == ref
  
              # When the new ref is available, next should return it.
              # When peek_object_ref_stream returns is_ready = False,
              # it should wait until ray.wait returns.
              for _ in range(3):
                  new_ref = ray.ObjectRef.from_random()
                  c.peek_object_ref_stream.return_value = (
                      new_ref,
                      False,
                  )
                  mocked_ray_wait.return_value = [new_ref], []
                  c.try_read_next_object_ref_stream.return_value = (
                      new_ref
                  )
                  ref = generator._next_sync(timeout_s=0)
                  assert new_ref == ref
  
              # When try_read_next_object_ref_stream raises a
              # ObjectRefStreamEndOfStreamError, it should raise a stop iteration.
              new_ref = ray.ObjectRef.from_random()
              c.peek_object_ref_stream.return_value = (new_ref, True)
              c.try_read_next_object_ref_stream.side_effect = (
                  ObjectRefStreamEndOfStreamError("")
              )  # noqa
              mocked_ray_get.return_value = None
              with pytest.raises(StopIteration):
                  ref = generator._next_sync(timeout_s=0)
              # Make sure we cannot serialize the generator.
              with pytest.raises(TypeError):
                  dumps(generator)
  
              del generator
              c.delete_object_ref_stream.assert_called()
  
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 2 for calling a global function with example usage: test_streaming_object_ref_generator_basic_unit(mocked_worker) without return types