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
92 changes: 92 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,93 @@ 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()
self.addCleanup(os.close, master_fd)

# tcsetattr(TCSADRAIN) blocks on some platforms (e.g. macOS) while the
# master still holds unread output, so empty it before each mode switch.
def drain():
out = b""
while select.select([master_fd], [], [], 0)[0]:
try:
data = os.read(master_fd, 4096)
except OSError:
break
if not data:
break
out += data
return out

# 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:
drain() # discard prepare()'s own setup sequences
# 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")
observed["output"] = drain()
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)
# The tty translated the hook's bare '\n' into '\r\n'.
self.assertEqual(observed["output"], b"line1\r\nline2\r\n")
finally:
# restore() writes and only then switches modes, so there is no
# point left to drain from here; keep the master empty elsewhere.
stop = threading.Event()

def pump():
while not stop.is_set():
if select.select([master_fd], [], [], 0.05)[0]:
try:
if not os.read(master_fd, 4096):
break
except OSError:
break

pump_thread = threading.Thread(target=pump)
pump_thread.start()
try:
console.restore()
finally:
stop.set()
pump_thread.join()
os.close(slave_fd)
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