code: def test_workflow_storage(workflow_start_regular):
      from ray.workflow.tests.utils import skip_client_mode_test
  
      # This test depends on raw storage, so we cannot test under client mode.
      skip_client_mode_test()
  
      workflow_id = test_workflow_storage.__name__
      wf_storage = workflow_storage.WorkflowStorage(workflow_id)
      task_id = "some_task"
      task_options = WorkflowTaskRuntimeOptions(
          task_type=TaskType.FUNCTION,
          catch_exceptions=False,
          retry_exceptions=True,
          max_retries=0,
          checkpoint=False,
          ray_options={},
      )
      input_metadata = {
          "name": "test_basic_workflows.append1",
          "workflow_refs": ["some_ref"],
          "task_options": task_options.to_dict(),
      }
      output_metadata = {
          "output_task_id": "a12423",
          "dynamic_output_task_id": "b1234",
      }
      root_output_metadata = {"output_task_id": "c123"}
      flattened_args = [
          signature.DUMMY_TYPE,
          1,
          signature.DUMMY_TYPE,
          "2",
          "k",
          b"543",
      ]
      args = signature.recover_args(flattened_args)
      output = ["the_answer"]
      object_resolved = 42
      obj_ref = ray.put(object_resolved)
  
      # test basics
      wf_storage._put(
          wf_storage._key_task_input_metadata(task_id),
          input_metadata,
          True,
      )
  
      wf_storage._put(
          wf_storage._key_task_function_body(task_id), some_func
      )
      wf_storage._put(
          wf_storage._key_task_args(task_id), flattened_args
      )
  
      wf_storage._put(
          wf_storage._key_obj_id(obj_ref.hex()), ray.get(obj_ref)
      )
      wf_storage._put(
          wf_storage._key_task_output_metadata(task_id),
          output_metadata,
          True,
      )
      wf_storage._put(
          wf_storage._key_task_output_metadata(""),
          root_output_metadata,
          True,
      )
      wf_storage._put(wf_storage._key_task_output(task_id), output)
  
      assert wf_storage.load_task_output(task_id) == output
  
      with serialization_context.workflow_args_resolving_context([]):
          assert (
              signature.recover_args(
                  ray.get(wf_storage.load_task_args(task_id))
              )
              == args
          )
      assert wf_storage.load_task_func_body(task_id)(33) == 34
      assert (
          ray.get(wf_storage.load_object_ref(obj_ref.hex()))
          == object_resolved
      )
  
      # test s3 path
      # here we hardcode the path to make sure s3 path is parsed correctly
      from ray._private.storage import _storage_uri
  
      if _storage_uri.startswith("s3://"):
          assert (
              wf_storage._get("tasks/outputs.json", True)
              == root_output_metadata
          )
  
      # test "inspect_task"
      inspect_result = wf_storage.inspect_task(task_id)
      assert inspect_result == workflow_storage.TaskInspectResult(
          output_object_valid=True
      )
      assert inspect_result.is_recoverable()
  
      task_id = "some_task2"
      wf_storage._put(
          wf_storage._key_task_input_metadata(task_id),
          input_metadata,
          True,
      )
      wf_storage._put(
          wf_storage._key_task_function_body(task_id), some_func
      )
      wf_storage._put(wf_storage._key_task_args(task_id), args)
      wf_storage._put(
          wf_storage._key_task_output_metadata(task_id),
          output_metadata,
          True,
      )
  
      inspect_result = wf_storage.inspect_task(task_id)
      assert inspect_result == workflow_storage.TaskInspectResult(
          output_task_id=output_metadata["dynamic_output_task_id"]
      )
      assert inspect_result.is_recoverable()
  
      task_id = "some_task3"
      wf_storage._put(
          wf_storage._key_task_input_metadata(task_id),
          input_metadata,
          True,
      )
      wf_storage._put(
          wf_storage._key_task_function_body(task_id), some_func
      )
      wf_storage._put(wf_storage._key_task_args(task_id), args)
      inspect_result = wf_storage.inspect_task(task_id)
      assert inspect_result == workflow_storage.TaskInspectResult(
          args_valid=True,
          func_body_valid=True,
          workflow_refs=input_metadata["workflow_refs"],
          task_options=task_options,
      )
      assert inspect_result.is_recoverable()
  
      task_id = "some_task4"
      wf_storage._put(
          wf_storage._key_task_input_metadata(task_id),
          input_metadata,
          True,
      )
  
      wf_storage._put(
          wf_storage._key_task_function_body(task_id), some_func
      )
      inspect_result = wf_storage.inspect_task(task_id)
      assert inspect_result == workflow_storage.TaskInspectResult(
          func_body_valid=True,
          workflow_refs=input_metadata["workflow_refs"],
          task_options=task_options,
      )
      assert not inspect_result.is_recoverable()
  
      task_id = "some_task5"
      wf_storage._put(
          wf_storage._key_task_input_metadata(task_id),
          input_metadata,
          True,
      )
  
      inspect_result = wf_storage.inspect_task(task_id)
      assert inspect_result == workflow_storage.TaskInspectResult(
          workflow_refs=input_metadata["workflow_refs"],
          task_options=task_options,
      )
      assert not inspect_result.is_recoverable()
  
      task_id = "some_task6"
      inspect_result = wf_storage.inspect_task(task_id)
      print(inspect_result)
      assert inspect_result == workflow_storage.TaskInspectResult()
      assert not inspect_result.is_recoverable()
  
file_path: ray/python/ray/workflow/tests/test_storage.py
imports: import subprocess
  import time
  import pytest
  import ray
  from ray import workflow
  from ray._private import signature
  from ray.tests.conftest import *
  from ray.workflow import workflow_storage
  from ray.workflow.common import TaskType, WorkflowTaskRuntimeOptions
  from ray.workflow.exceptions import WorkflowNotFoundError
  from ray.workflow import serialization_context
  from ray.workflow.tests import utils
label_desc: Use image node_id 5 for calling a global function with example usage: test_workflow_storage(workflow_start_regular) without return types