Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion Lib/_pyrepl/unix_console.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -731,7 +732,23 @@ 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):
# Input-hook callbacks (GUI toolkit event loops, and any warning,
# traceback or print they emit) expect normal cooked-mode output, but
# pyrepl runs with OPOST disabled so it can drive the cursor itself.
# Temporarily restore the terminal's saved output flags around the
# hook so that '\n' is translated to '\r\n' as it was under the
# readline REPL, then re-enter raw mode. Only oflag is toggled --
# re-enabling ECHO/ICANON at the prompt would be wrong. See gh-152907.
cooked = self.__rawtermstate.copy()
cooked.oflag = self.__svtermstate.oflag
self.__input_fd_set(cooked)
try:
posix._inputhook()
finally:
self.__input_fd_set(self.__rawtermstate)

def __enable_bracketed_paste(self) -> None:
os.write(self.output_fd, b"\x1b[?2004h")
Expand Down
91 changes: 91 additions & 0 deletions Lib/test/test_pyrepl/test_unix_console.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
import errno
import itertools
import os
import select
import signal
import sys
import threading
import time
import unittest
from functools import partial
from _colorize import ANSIColors
Expand Down Expand Up @@ -424,3 +426,92 @@ 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")
class TestUnixConsoleInputHook(TestCase):
# gh-152907: pyrepl runs with OPOST disabled so it can drive the cursor
# itself, but input-hook callbacks (GUI event loops, and any warning,
# traceback or print they emit) expect normal cooked-mode output. The
# console must restore cooked output around the hook call so that '\n' is
# translated back to '\r\n', then re-enter raw mode.

def test_input_hook_output_is_cooked(self):
master_fd, slave_fd = pty.openpty()

# Continuously drain the master side, like a real terminal. This is
# required because pyrepl switches the terminal mode with TCSADRAIN,
# which blocks until queued output has been consumed.
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)

# Establish a normal cooked terminal as the saved state so the fix has
# something to restore around the hook call.
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)
hook()

# 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)

# Give the reader a moment to drain the hook's output.
time.sleep(0.2)
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)
Original file line number Diff line number Diff line change
@@ -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.
Loading