8000 gh-133490: Fix syntax highlighting for remote PDB (#133494) · python/cpython@fd37f1a · GitHub
[go: up one dir, main page]
Skip to content

Commit fd37f1a

Browse files
godlygeekgodlygeek
authored
gh-133490: Fix syntax highlighting for remote PDB (#133494)
1 parent 120c9d4 commit fd37f1a

File tree

3 files changed

+158
-7
lines changed

3 files changed

+158
-7
lines changed

‎Lib/_pyrepl/utils.py

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,9 @@
2323
BUILTINS = {str(name) for name in dir(builtins) if not name.startswith('_')}
2424

2525

26-
def THEME():
26+
def THEME(**kwargs):
2727
# Not cached: the user can modify the theme inside the interactive session.
28-
return _colorize.get_theme().syntax
28+
return _colorize.get_theme(**kwargs).syntax
2929

3030

3131
class Span(NamedTuple):
@@ -254,7 +254,10 @@ def is_soft_keyword_used(*tokens: TI | None) -> bool:
254254

255255

256256
def disp_str(
257-
buffer: str, colors: list[ColorSpan] | None = None, start_index: int = 0
257+
buffer: str,
258+
colors: list[ColorSpan] | None = None,
259+
start_index: int = 0,
260+
force_color: bool = False,
258261
) -> tuple[CharBuffer, CharWidths]:
259262
r"""Decompose the input buffer into a printable variant with applied colors.
260263
@@ -295,7 +298,7 @@ def disp_str(
295298
# move past irrelevant spans
296299
colors.pop(0)
297300

298-
theme = THEME()
301+
theme = THEME(force_color=force_color)
299302
pre_color = ""
300303
post_color = ""
301304
if colors and colors[0].span.start < start_index:

‎Lib/pdb.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1069,7 +1069,7 @@ def handle_command_def(self, line):
10691069
def _colorize_code(self, code):
10701070
if self.colorize:
10711071
colors = list(_pyrepl.utils.gen_colors(code))
1072-
chars, _ = _pyrepl.utils.disp_str(code, colors=colors)
1072+
chars, _ = _pyrepl.utils.disp_str(code, colors=colors, force_color=True)
10731073
code = "".join(chars)
10741074
return code
10751075

‎Lib/test/test_remote_pdb.py

Lines changed: 150 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import itertools
44
import json
55
import os
6+
import re
67
import signal
78
import socket
89
import subprocess
@@ -12,9 +13,9 @@
1213
import threading
1314
import unittest
1415
import unittest.mock
15-
from contextlib import closing, contextmanager, redirect_stdout, ExitStack
16+
from contextlib import closing, contextmanager, redirect_stdout, redirect_stderr, ExitStack
1617
from pathlib import Path
17-
from test.support import is_wasi, os_helper, requires_subprocess, SHORT_TIMEOUT
18+
from test.support import is_wasi, cpython_only, force_color, requires_subprocess, SHORT_TIMEOUT
1819
from test.support.os_helper import temp_dir, TESTFN, unlink
1920
from typing import Dict, List, Optional, Tuple, Union, Any
2021

@@ -1431,5 +1432,152 @@ def test_multi_line_commands(self):
14311432
self.assertIn("Function returned: 42", stdout)
14321433
self.assertEqual(process.returncode, 0)
14331434

1435+
1436+
def _supports_remote_attaching():
1437+
from contextlib import suppress
1438+
PROCESS_VM_READV_SUPPORTED = False
1439+
1440+
try:
1441+
from _remote_debugging import PROCESS_VM_READV_SUPPORTED
1442+
except ImportError:
1443+
pass
1444+
1445+
return PROCESS_VM_READV_SUPPORTED
1446+
1447+
1448+
@unittest.skipIf(not sys.is_remote_debug_enabled(), "Remote debugging is not enabled")
1449+
@unittest.skipIf(sys.platform != "darwin" and sys.platform != "linux" and sys.platform != "win32",
1450+
"Test only runs on Linux, Windows and MacOS")
1451+
@unittest.skipIf(sys.platform == "linux" and not _supports_remote_attaching(),
1452+
"Testing on Linux requires process_vm_readv support")
1453+
@cpython_only
1454+
@requires_subprocess()
1455+
class PdbAttachTestCase(unittest.TestCase):
1456+
def setUp(self):
1457+
# Create a server socket that will wait for the debugger to connect
1458+
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1459+
self.sock.bind(('127.0.0.1', 0)) # Let OS assign port
1460+
self.sock.listen(1)
1461+
self.port = self.sock.getsockname()[1]
1462+
self._create_script()
1463+
1464+
def _create_script(self, script=None):
1465+
# Create a file for subprocess script
1466+
script = textwrap.dedent(
1467+
f"""
1468+
import socket
1469+
import time
1470+
1471+
def foo():
1472+
return bar()
1473+
1474+
def bar():
1475+
return baz()
1476+
1477+
def baz():
1478+
x = 1
1479+
# Trigger attach
1480+
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1481+
sock.connect(('127.0.0.1', {self.port}))
1482+
sock.close()
1483+
count = 0
1484+
while x == 1 and count < 100:
1485+ count += 1
1486+
time.sleep(0.1)
1487+
return x
1488+
1489+
result = foo()
1490+
print(f"Function returned: {{result}}")
1491+
"""
1492+
)
1493+
1494+
self.script_path = TESTFN + "_connect_test.py"
1495+
with open(self.script_path, 'w') as f:
1496+
f.write(script)
1497+
1498+
def tearDown(self):
1499+
self.sock.close()
1500+
try:
1501+
unlink(self.script_path)
1502+
except OSError:
1503+
pass
1504+
1505+
def do_integration_test(self, client_stdin):
1506+
process = subprocess.Popen(
1507+
[sys.executable, self.script_path],
1508+
stdout=subprocess.PIPE,
1509+
stderr=subprocess.PIPE,
1510+
text=True
1511+
)
1512+
self.addCleanup(process.stdout.close)
1513+
self.addCleanup(process.stderr.close)
1514+
1515+
# Wait for the process to reach our attachment point
1516+
self.sock.settimeout(10)
1517+
conn, _ = self.sock.accept()
1518+
conn.close()
1519+
1520+
client_stdin = io.StringIO(client_stdin)
1521+
client_stdout = io.StringIO()
1522+
client_stderr = io.StringIO()
1523+
1524+
self.addCleanup(client_stdin.close)
1525+
self.addCleanup(client_stdout.close)
1526+
self.addCleanup(client_stderr.close)
1527+
self.addCleanup(process.wait)
1528+
1529+
with (
1530+
unittest.mock.patch("sys.stdin", client_stdin),
1531+
redirect_stdout(client_stdout),
1532+
redirect_stderr(client_stderr),
1533+
unittest.mock.patch("sys.argv", ["pdb", "-p", str(process.pid)]),
1534+
):
1535+
try:
1536+
pdb.main()
1537+
except PermissionError:
1538+
self.skipTest("Insufficient permissions for remote execution")
1539+
1540+
process.wait()
1541+
server_stdout = process.stdout.read()
1542+
server_stderr = process.stderr.read()
1543+
1544+
if process.returncode != 0:
1545+
print("server failed")
1546+
print(f"server stdout:\n{server_stdout}")
1547+
print(f"server stderr:\n{server_stderr}")
1548+
1549+
self.assertEqual(process.returncode, 0)
1550+
return {
1551+
"client": {
1552+
"stdout": client_stdout.getvalue(),
1553+
"stderr": client_stderr.getvalue(),
1554+
},
1555+
"server": {
1556+
"stdout": server_stdout,
1557+
"stderr": server_stderr,
1558+
},
1559+
}
1560+
1561+
def test_attach_to_process_without_colors(self):
1562+
with force_color(False):
1563+
output = self.do_integration_test("ll\nx=42\n")
1564+
self.assertEqual(output["client"]["stderr"], "")
1565+
self.assertEqual(output["server"]["stderr"], "")
1566+
1567+
self.assertEqual(output["server"]["stdout"], "Function returned: 42\n")
1568+
self.assertIn("while x == 1", output["client"]["stdout"])
1569+
self.assertNotIn("\x1b", output["client"]["stdout"])
1570+
1571+
def test_attach_to_process_with_colors(self):
1572+
with force_color(True):
1573+
output = self.do_integration_test("ll\nx=42\n")
1574+
self.assertEqual(output["client"]["stderr"], "")
1575+
self.assertEqual(output["server"]["stderr"], "")
1576+
1577+
self.assertEqual(output["server"]["stdout"], "Function returned: 42\n")
1578+
self.assertIn("\x1b", output["client"]["stdout"])
1579+
self.assertNotIn("while x == 1", output["client"]["stdout"])
1580+
self.assertIn("while x == 1", re.sub("\x1b[^m]*m", "", output["client"]["stdout"]))
1581+
14341582
if __name__ == "__main__":
14351583
unittest.main()

0 commit comments

Comments
 (0)
0