code: def test_ray_debugger_breakpoint(shutdown_only):
      ray.init(num_cpus=1)
  
      @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
  
      # Now continue execution:
      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(":")
      assert host == "localhost"  # Should be private by default.
  
      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 1 for calling a global function with example usage: test_ray_debugger_breakpoint(shutdown_only) and returns: int