code: def test_ray_debugger_public_multi_node(
      shutdown_only, ray_debugger_external
  ):
      c = Cluster(
          initialize_head=True,
          connect=True,
          head_node_args={
              "num_cpus": 0,
              "num_gpus": 1,
              "ray_debugger_external": ray_debugger_external,
          },
      )
      c.add_node(
          num_cpus=1, ray_debugger_external=ray_debugger_external
      )
  
      @ray.remote
      def f():
          ray.util.pdb.set_trace()
          return 1
  
      # num_gpus=1 forces the task onto the head node.
      head_node_result = f.options(num_cpus=0, num_gpus=1).remote()
  
      # num_cpus=1 forces the task onto the worker node.
      worker_node_result = f.options(num_cpus=1).remote()
  
      wait_for_condition(
          lambda: len(
              ray.experimental.internal_kv._internal_kv_list(
                  "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
              )
          )
          == 2
      )
  
      active_sessions = ray.experimental.internal_kv._internal_kv_list(
          "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
      )
      assert len(active_sessions) == 2
      session1 = json.loads(
          ray.experimental.internal_kv._internal_kv_get(
              active_sessions[0],
              namespace=ray_constants.KV_NAMESPACE_PDB,
          )
      )
      session2 = json.loads(
          ray.experimental.internal_kv._internal_kv_get(
              active_sessions[1],
              namespace=ray_constants.KV_NAMESPACE_PDB,
          )
      )
  
      host1, port1 = session1["pdb_address"].split(":")
      if ray_debugger_external:
          assert host1 == services.get_node_ip_address(), host1
      else:
          assert host1 == "localhost", host1
  
      host2, port2 = session2["pdb_address"].split(":")
      if ray_debugger_external:
          assert host2 == services.get_node_ip_address(), host2
      else:
          assert host2 == "localhost", host2
  
      # Check that we can successfully connect to both breakpoints.
      tn1 = Telnet(host1, int(port1))
      tn1.write(b"c\n")
  
      tn2 = Telnet(host2, int(port2))
      tn2.write(b"c\n")
  
      # The messages above should cause these to return now.
      ray.get([head_node_result, worker_node_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 7 for calling a global function with example usage: test_ray_debugger_public_multi_node(shutdown_only, ray_debugger_external) and returns: int