code: def test_embedded_objectrefs(workflow_start_regular):
      from ray.workflow.tests.utils import skip_client_mode_test
  
      # This test uses low-level storage APIs and restarts the cluster,
      # so it is not for client mode test
      skip_client_mode_test()
  
      workflow_id = test_embedded_objectrefs.__name__
  
      class ObjectRefsWrapper:
          def __init__(self, refs):
              self.refs = refs
  
      from ray._private.storage import _storage_uri
  
      wrapped = ObjectRefsWrapper([ray.put(1), ray.put(2)])
  
      store = workflow_storage.WorkflowStorage(workflow_id)
      serialization.dump_to_storage("key", wrapped, workflow_id, store)
  
      # Be extremely explicit about shutting down. We want to make sure the
      # `_get` call deserializes the full object and puts it in the object store.
      # Shutting down the cluster should guarantee we don't accidently get the
      # old object and pass the test.
      ray.shutdown()
      subprocess.check_output("ray stop --force", shell=True)
  
      ray.init(storage=_storage_uri)
      workflow.init()
      storage2 = workflow_storage.WorkflowStorage(workflow_id)
  
      result = storage2._get("key")
      assert ray.get(result.refs) == [1, 2]
  
file_path: ray/python/ray/workflow/tests/test_serialization2.py
imports: import subprocess
  import time
  import pytest
  from filelock import FileLock
  import ray
  from ray import workflow
  from ray._private.test_utils import run_string_as_driver_nonblocking
  from ray.tests.conftest import *
  from ray.workflow import serialization, workflow_storage
label_desc: Use image node_id 3 for calling a global function with example usage: test_embedded_objectrefs(workflow_start_regular) without return types