diff --git a/Lib/_pyrepl/unix_console.py b/Lib/_pyrepl/unix_console.py index 8edd1c36a7232d..8796cb19d05820 100644 --- a/Lib/_pyrepl/unix_console.py +++ b/Lib/_pyrepl/unix_console.py @@ -491,6 +491,7 @@ def prepare(self) -> None: raw.cc[termios.VMIN] = b"\x01" raw.cc[termios.VTIME] = b"\x00" self.__input_fd_set(raw) + self.__rawtermstate = raw # Apple Terminal will re-wrap lines for us unless we preempt the # damage. @@ -731,7 +732,19 @@ def input_hook(self): # avoid inline imports here so the repl doesn't get flooded # with import logging from -X importtime=2 if posix is not None and posix._is_inputhook_installed(): - return posix._inputhook + return self.__run_input_hook + + def __run_input_hook(self): + # gh-152907: input hooks expect cooked output, but pyrepl runs with + # OPOST disabled. Restore the saved output flags around the hook + # (only oflag; input must stay raw at the prompt). + cooked = self.__rawtermstate.copy() + cooked.oflag = self.__svtermstate.oflag + self.__input_fd_set(cooked) + try: + return posix._inputhook() + finally: + self.__input_fd_set(self.__rawtermstate) def __enable_bracketed_paste(self) -> None: os.write(self.output_fd, b"\x1b[?2004h") diff --git a/Lib/test/test_pyrepl/test_unix_console.py b/Lib/test/test_pyrepl/test_unix_console.py index 2fc8398923cbf3..1897c735671785 100644 --- a/Lib/test/test_pyrepl/test_unix_console.py +++ b/Lib/test/test_pyrepl/test_unix_console.py @@ -1,6 +1,7 @@ import errno import itertools import os +import select import signal import sys import threading @@ -8,6 +9,7 @@ from functools import partial from _colorize import ANSIColors from test.support import force_color, os_helper, force_not_colorized_test_class +from test.support import is_android, is_apple_mobile, is_wasm32 from test.support import threading_helper from unittest import TestCase @@ -424,3 +426,101 @@ def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr): # EIO error should be handled gracefully in restore() console.restore() + + +try: + import pty + import termios as _termios +except ImportError: + pty = None + + +@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows") +@unittest.skipUnless(pty, "requires pty") +@unittest.skipIf(is_android or is_apple_mobile or is_wasm32, + "pty is not available on this platform") +class TestUnixConsoleInputHook(TestCase): + # gh-152907: the console must restore cooked output (OPOST) around + # input-hook calls, then re-enter raw mode. + + def test_input_hook_output_is_cooked(self): + master_fd, slave_fd = pty.openpty() + + # Drain the master continuously: on some platforms (e.g. macOS) + # tcsetattr(TCSADRAIN) blocks until the master side is read, so an + # undrained pty would deadlock the mode switch. + chunks = [] + reading = True + + def reader(): + while reading: + r, _, _ = select.select([master_fd], [], [], 0.1) + if master_fd in r: + try: + data = os.read(master_fd, 4096) + except OSError: + break + if not data: + break + chunks.append(data) + + reader_thread = threading.Thread(target=reader) + reader_thread.start() + + def cleanup(): + nonlocal reading + reading = False + reader_thread.join() + os.close(master_fd) + self.addCleanup(cleanup) + + # Start from a cooked terminal so there are saved flags to restore. + attr = _termios.tcgetattr(slave_fd) + attr[1] |= _termios.OPOST | _termios.ONLCR + _termios.tcsetattr(slave_fd, _termios.TCSANOW, attr) + + console = UnixConsole(slave_fd, slave_fd, term="xterm") + console.prepare() + try: + # pyrepl's own rendering runs with OPOST cleared. + self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST) + + observed = {} + + def fake_hook(): + observed["oflag"] = _termios.tcgetattr(slave_fd)[1] + os.write(slave_fd, b"line1\nline2\n") + return 0 + + with patch("_pyrepl.unix_console.posix") as mock_posix: + mock_posix._is_inputhook_installed.return_value = True + mock_posix._inputhook.side_effect = fake_hook + hook = console.input_hook + self.assertIsNotNone(hook) + self.assertEqual(hook(), 0) + + # The hook ran with cooked output (OPOST on)... + self.assertTrue(observed["oflag"] & _termios.OPOST) + # ...and raw mode was restored afterwards. + self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST) + finally: + console.restore() + os.close(slave_fd) + + # The switch back to raw mode already drained the hook's output, so + # joining the reader is enough -- no sleep needed. + reading = False + reader_thread.join() + while select.select([master_fd], [], [], 0)[0]: + try: + extra = os.read(master_fd, 4096) + except OSError: + break + if not extra: + break + chunks.append(extra) + + data = b"".join(chunks) + # The tty translated the hook's bare '\n' into '\r\n'. + self.assertIn(b"line1\r\nline2\r\n", data) + self.assertNotIn(b"line1\nline2\n", data) diff --git a/Misc/NEWS.d/next/Library/2026-07-08-23-20-00.gh-issue-152907.oyPV9Y.rst b/Misc/NEWS.d/next/Library/2026-07-08-23-20-00.gh-issue-152907.oyPV9Y.rst new file mode 100644 index 00000000000000..33247f0f3a3b96 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-07-08-23-20-00.gh-issue-152907.oyPV9Y.rst @@ -0,0 +1,4 @@ +Restore cooked-mode terminal output flags around :c:data:`PyOS_InputHook` +callbacks in the new :term:`REPL` (:mod:`!_pyrepl`), so that output written +by an input hook (for example a GUI toolkit event loop) is no longer emitted +with ``OPOST`` disabled and keeps its carriage returns.