code: def test_cluster_storage_init(workflow_start_cluster, tmp_path):
      address, storage_uri = workflow_start_cluster
  
      err_msg = "When connecting to an existing cluster, "
      "storage must not be provided."
  
      with pytest.raises(ValueError, match=err_msg):
          ray.init(address=address, storage=str(tmp_path))
  
      with pytest.raises(ValueError, match=err_msg):
          ray.init(address=address, storage=storage_uri)
  
      ray.init(address=address)
  
      @ray.remote
      def f():
          return 10
  
      assert workflow.run(f.bind()) == 10
  
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 6 for calling a global function with example usage: test_cluster_storage_init(workflow_start_cluster, tmp_path) and returns: int