Address comments and add a test

Author: gaogaotiantian

Lib/pdb.py

@@ -710,25 +710,22 @@ def _get_asyncio_task(self):
         return task
 
     def _get_pid_from_process(self, process):
-        """process could be a subprocess.Popen, multiprocessing.Process or a pid
+        """process could evaluate to any object with a `process` attribute or an integer
         """
-        # They are not used elsewhere so do a lazy import
-        from multiprocessing import Process
-        from subprocess import Popen
 
         try:
             process = self._getval(process)
         except:
             # Error message is already displayed
             return None
 
-        if isinstance(process, (Process, Popen)):
-            return process.pid
-        elif isinstance(process, int):
-            return process
+        pid = getattr(process, "pid", process)
 
-        self.error(f"Invalid process {process}")
-        return None
+        if not isinstance(pid, int):
+            self.error(f"Invalid process {process!r}")
+            return None
+
+        return pid 
 
     def interaction(self, frame, tb_or_exc):
         # Restore the previous signal handler at the Pdb prompt.
@@ -1982,13 +1979,13 @@ def do_debug(self, arg):
 
     complete_debug = _complete_expression
 
-    def do_attach(self, process):
+    def do_attach(self, arg):
         """attach process
 
-        Attach to process, which can be a subprocess.Popen,
-        multiprocessing.Process or a pid.
+        Attach to process, which can be any object that has a pid
+        attribute or a process ID.
         """
-        pid = self._get_pid_from_process(process)
+        pid = self._get_pid_from_process(arg)
 
         if pid is not None:
             self.message(f"Attaching to process {pid}")
@@ -2780,6 +2777,7 @@ def _ensure_valid_message(self, msg):
                 # needs to know it for multi-line editing.
                 pass
             case {"attach": int()}:
+                # Ask the client to attach to the given process ID.
                 pass
             case _:
                 raise AssertionError(
@@ -3441,6 +3439,12 @@ def _connect(
 
 def attach(pid, commands=()):
     """Attach to a running process with the given PID."""
+
+    if threading.current_thread() is not threading.main_thread():
+        raise RuntimeError(
+            "pdb.attach() must be called from the main thread"
+        )
+
     with ExitStack() as stack:
         server = stack.enter_context(
             closing(socket.create_server(("localhost", 0)))

Lib/test/test_remote_pdb.py

@@ -1600,6 +1600,17 @@ def test_attach_to_process_with_colors(self):
         self.assertNotIn("while x == 1", output["client"]["stdout"])
         self.assertIn("while x == 1", re.sub("\x1b[^m]*m", "", output["client"]["stdout"]))
 
+    def test_attach_from_worker_thread(self):
+        # Test attaching from a worker thread
+        def worker():
+            with self.assertRaises(RuntimeError):
+                # We are not allowed to attach from a thread that's not main
+                pdb.attach(1234)
+
+        thread = threading.Thread(target=worker)
+        thread.start()
+        thread.join()
+
 
 @unittest.skipIf(not sys.is_remote_debug_enabled(), "Remote debugging is not enabled")
 @unittest.skipIf(sys.platform != "darwin" and sys.platform != "linux" and sys.platform != "win32",