code: def test_delete_2(workflow_start_regular):
      from ray._private.storage import _storage_uri
      from ray.workflow.tests.utils import skip_client_mode_test
  
      # This test restarts the cluster, so we cannot test under client mode.
      skip_client_mode_test()
  
      # Delete a workflow that has not finished and is not running.
      @ray.remote
      def never_ends(x):
          utils.set_global_mark()
          time.sleep(1000000)
          return x
  
      workflow.run_async(
          never_ends.bind("hello world"), workflow_id="never_finishes"
      )
  
      # Make sure the task is actualy executing before killing the cluster
      while not utils.check_global_mark():
          time.sleep(0.1)
  
      # Restart
      ray.shutdown()
      subprocess.check_output("ray stop --force", shell=True)
      ray.init(storage=_storage_uri)
      workflow.init()
  
      with pytest.raises(ValueError):
          workflow.get_output("never_finishes")
  
      workflow.delete("never_finishes")
  
      with pytest.raises(ValueError):
          # TODO(suquark): we should raise "ValueError" without
          #  been blocking over the result.
          workflow.get_output("never_finishes")
  
      # TODO(Alex): Uncomment after
      # https://github.com/ray-project/ray/issues/19481.
      # with pytest.raises(WorkflowNotFoundError):
      #     workflow.resume("never_finishes")
  
      with pytest.raises(WorkflowNotFoundError):
          workflow.delete(workflow_id="never_finishes")
  
      # Delete a workflow which has finished.
      @ray.remote
      def basic_task(arg):
          return arg
  
      result = workflow.run(
          basic_task.bind("hello world"), workflow_id="finishes"
      )
      assert result == "hello world"
      assert workflow.get_output("finishes") == "hello world"
  
      workflow.delete(workflow_id="finishes")
  
      with pytest.raises(ValueError):
          # TODO(suquark): we should raise "ValueError" without
          #  blocking over the result.
          workflow.get_output("finishes")
  
      # TODO(Alex): Uncomment after
      # https://github.com/ray-project/ray/issues/19481.
      # with pytest.raises(ValueError):
      #     workflow.resume("finishes")
  
      with pytest.raises(WorkflowNotFoundError):
          workflow.delete(workflow_id="finishes")
  
      assert workflow.list_all() == []
  
      # The workflow can be re-run as if it was never run before.
      assert (
          workflow.run(basic_task.bind("123"), workflow_id="finishes")
          == "123"
      )
  
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 4 for calling a global function with example usage: test_delete_2(workflow_start_regular) and returns: x, arg