|
8 | 8 | import subprocess |
9 | 9 | import sys |
10 | 10 | import textwrap |
| 11 | +import threading |
11 | 12 | import unittest |
12 | 13 | import unittest.mock |
13 | 14 | from contextlib import closing, contextmanager, redirect_stdout, redirect_stderr, ExitStack |
|
18 | 19 | import pdb |
19 | 20 | from pdb import _PdbServer, _PdbClient |
20 | 21 |
|
| 22 | +try: |
| 23 | + import pty |
| 24 | +except ImportError: |
| 25 | + pty = None |
| 26 | + |
21 | 27 |
|
22 | 28 | if not sys.is_remote_debug_enabled(): |
23 | 29 | raise unittest.SkipTest('remote debugging is disabled') |
@@ -1090,6 +1096,45 @@ def _connect_and_get_client_file(self): |
1090 | 1096 |
|
1091 | 1097 | return process, client_file |
1092 | 1098 |
|
| 1099 | + def _connect_and_get_client_file_via_pty(self): |
| 1100 | + """Like _connect_and_get_client_file, but run the target under a pty. |
| 1101 | +
|
| 1102 | + With a real terminal on stdin/stdout, PyREPL is available *inside the |
| 1103 | + target process*, which is the condition that exercises pdb's PyREPL |
| 1104 | + input handling in the remote server. |
| 1105 | + """ |
| 1106 | + controller, worker = pty.openpty() |
| 1107 | + self.addCleanup(os.close, controller) |
| 1108 | + env = dict(os.environ, TERM="xterm-256color") |
| 1109 | + env.pop("PYTHON_BASIC_REPL", None) # don't opt out of PyREPL |
| 1110 | + process = subprocess.Popen( |
| 1111 | + [sys.executable, self.script_path], |
| 1112 | + stdin=worker, stdout=worker, stderr=worker, env=env, close_fds=True, |
| 1113 | + ) |
| 1114 | + os.close(worker) # only the child keeps the worker end open |
| 1115 | + |
| 1116 | + # Continuously drain the terminal so the child never blocks on a write. |
| 1117 | + drainer = threading.Thread( |
| 1118 | + target=self._drain_until_eof, args=(controller,), daemon=True |
| 1119 | + ) |
| 1120 | + drainer.start() |
| 1121 | + self.addCleanup(drainer.join, SHORT_TIMEOUT) |
| 1122 | + |
| 1123 | + client_sock, _ = self.server_sock.accept() |
| 1124 | + client_file = client_sock.makefile('rwb') |
| 1125 | + self.addCleanup(client_file.close) |
| 1126 | + self.addCleanup(client_sock.close) |
| 1127 | + |
| 1128 | + return process, client_file |
| 1129 | + |
| 1130 | + @staticmethod |
| 1131 | + def _drain_until_eof(fd): |
| 1132 | + try: |
| 1133 | + while os.read(fd, 1024): |
| 1134 | + pass |
| 1135 | + except OSError: |
| 1136 | + pass # controller closed, or the pty went away with the child |
| 1137 | + |
1093 | 1138 | def _read_until_prompt(self, client_file): |
1094 | 1139 | """Helper to read messages until a prompt is received.""" |
1095 | 1140 | messages = [] |
@@ -1292,6 +1337,32 @@ def test_handle_eof(self): |
1292 | 1337 | self.assertEqual(process.returncode, 0) |
1293 | 1338 | self.assertEqual(stderr, "") |
1294 | 1339 |
|
| 1340 | + @unittest.skipUnless(pty, "requires pty") |
| 1341 | + def test_prompt_with_interactive_terminal(self): |
| 1342 | + """The server must send "(Pdb) " even when the target owns a terminal. |
| 1343 | +
|
| 1344 | + The remote server transmits its prompt string to the client, which |
| 1345 | + displays it. When the target process has an interactive terminal, |
| 1346 | + PyREPL is enabled inside it; the base pdb machinery then blanks |
| 1347 | + ``self.prompt`` (a local PyREPL would draw the prompt itself). The |
| 1348 | + remote server has no local PyREPL -- it reads input from the socket -- |
| 1349 | + so it must keep the real prompt rather than transmit an empty one. |
| 1350 | + Regression test for gh-154467, where attaching with ``pdb -p`` to a |
| 1351 | + process running at an interactive prompt showed a blank prompt. |
| 1352 | + """ |
| 1353 | + self._create_script() |
| 1354 | + process, client_file = self._connect_and_get_client_file_via_pty() |
| 1355 | + |
| 1356 | + with kill_on_error(process): |
| 1357 | + messages = self._read_until_prompt(client_file) |
| 1358 | + # The message that ended the read is the prompt request. |
| 1359 | + self.assertEqual(messages[-1], {"prompt": "(Pdb) ", "state": "pdb"}) |
| 1360 | + |
| 1361 | + # Let the target run to completion so nothing is left attached. |
| 1362 | + self._send_command(client_file, "c") |
| 1363 | + process.wait(timeout=SHORT_TIMEOUT) |
| 1364 | + self.assertEqual(process.returncode, 0) |
| 1365 | + |
1295 | 1366 | def test_protocol_version(self): |
1296 | 1367 | """Test that incompatible protocol versions are properly detected.""" |
1297 | 1368 | # Create a script using an incompatible protocol version |
|
0 commit comments