code: def _init_cluster(storage_url, **params):
      init_kwargs = get_default_fixture_ray_kwargs()
      init_kwargs.update(**params)
      init_kwargs["storage"] = storage_url
  
      # Sometimes pytest does not cleanup all global variables.
      # we have to manually reset the workflow storage. This
      # should not be an issue for normal use cases, because global variables
      # are freed after the driver exits.
      ray.shutdown()
      subprocess.check_call(["ray", "stop", "--force"])
      init_kwargs["ray_client_server_port"] = 10001
      cluster = Cluster()
      init_kwargs.pop(
          "namespace"
      )  # we do not need namespace in workflow tests
      cluster.add_node(**init_kwargs)
      utils.clear_marks()
      yield cluster
      ray.shutdown()
      cluster.shutdown()
  
file_path: ray/python/ray/workflow/tests/conftest.py
imports: from contextlib import contextmanager
  import subprocess
  import pytest
  import ray
  from ray.tests.conftest import get_default_fixture_ray_kwargs
  from ray._private.test_utils import simulate_storage
  from ray.cluster_utils import Cluster
  from ray.tests.conftest import pytest_runtest_makereport
  from ray.workflow.tests import utils
label_desc: Use image node_id 1 for calling a global function with example usage: _init_cluster(storage_url) without return types