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
15 changes: 14 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,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")
Expand Down
100 changes: 100 additions & 0 deletions Lib/test/test_pyrepl/test_unix_console.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
import errno
import itertools
import os
import select
import signal
import sys
import threading
import unittest
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
Expand Down Expand Up @@ -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)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit, feel free to ignore: relying on a background reader plus time.sleep(0.2) to observe the output tends to be a source of buildbot flakiness. Since the TCSADRAIN in __run_input_hook already guarantees the bytes reach the master before the hook returns, can we drain master_fd directly after hook() instead of sleeping?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Tried that first but it hangs on macOS — TCSADRAIN there doesn't return until the master side actually reads the output, so the mode switch blocks forever without a reader. Kept the reader thread but dropped the sleep: once the hook returns the output is already drained, so joining the reader is enough.


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