code: def test_ray_debugger_public(
      shutdown_only, call_ray_stop_only, ray_debugger_external
  ):
      redis_substring_prefix = "--address='"
      cmd = ["ray", "start", "--head", "--num-cpus=1"]
      if ray_debugger_external:
          cmd.append("--ray-debugger-external")
      out = ray._private.utils.decode(
          subprocess.check_output(cmd, stderr=subprocess.STDOUT)
      )
      # Get the redis address from the output.
      redis_substring_prefix = "--address='"
      address_location = out.find(redis_substring_prefix) + len(
          redis_substring_prefix
      )
      address = out[address_location:]
      address = address.split("'")[0]
  
      ray.init(address=address)
  
      @ray.remote
      def f():
          ray.util.pdb.set_trace()
          return 1
  
      result = f.remote()
  
      wait_for_condition(
          lambda: len(
              ray.experimental.internal_kv._internal_kv_list(
                  "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
              )
          )
          > 0
      )
  
      active_sessions = ray.experimental.internal_kv._internal_kv_list(
          "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
      )
      assert len(active_sessions) == 1
      session = json.loads(
          ray.experimental.internal_kv._internal_kv_get(
              active_sessions[0],
              namespace=ray_constants.KV_NAMESPACE_PDB,
          )
      )
  
      host, port = session["pdb_address"].split(":")
      if ray_debugger_external:
          assert host == services.get_node_ip_address(), host
      else:
          assert host == "localhost", host
  
      # Check that we can successfully connect to both breakpoints.
      tn = Telnet(host, int(port))
      tn.write(b"c\n")
  
      # The message above should cause this to return now.
      ray.get(result)
  
imports: import json
  import os
  import platform
  import subprocess
  import sys
  from telnetlib import Telnet
  import unittest
  import pexpect
  import pytest
  import ray
  from ray._private import ray_constants, services
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 6 for calling a global function with example usage: test_ray_debugger_public(shutdown_only, call_ray_stop_only, ray_debugger_external) and returns: int