code: def test_ray_init_existing_instance(call_ray_start, address):
      ray_address = call_ray_start
      # If no address is specified, we will default to an existing cluster.
      res = ray.init(address=address)
      assert res.address_info["gcs_address"] == ray_address
      ray.shutdown()
  
      # Start a second local Ray instance.
      try:
          subprocess.check_output("ray start --head", shell=True)
          # If there are multiple local instances, connect to the latest.
          res = ray.init(address=address)
          assert res.address_info["gcs_address"] != ray_address
          ray.shutdown()
  
          # If there are multiple local instances and we specify an address
          # explicitly, it works.
          with unittest.mock.patch.dict(
              os.environ, {"RAY_ADDRESS": ray_address}
          ):
              res = ray.init(address=address)
              assert res.address_info["gcs_address"] == ray_address
      finally:
          ray.shutdown()
          subprocess.check_output("ray stop --force", shell=True)
  
imports: import os
  import sys
  import unittest.mock
  import signal
  import subprocess
  import grpc
  import pytest
  import ray
  import ray._private.services
  from ray.client_builder import ClientContext
  from ray.cluster_utils import Cluster
  from ray.util.client.common import ClientObjectRef
  from ray.util.client.ray_client_helpers import ray_start_client_server
  from ray.util.client.worker import Worker
  from ray._private.test_utils import wait_for_condition, enable_external_redis
label_desc: Use image node_id 3 for calling a global function with example usage: test_ray_init_existing_instance(call_ray_start, address) without return types