Skip to content
Merged
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
16 changes: 15 additions & 1 deletion Lib/_pyrepl/unix_console.py
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,7 @@ def prepare(self):
raw.cc[termios.VMIN] = 1
raw.cc[termios.VTIME] = 0
self.__input_fd_set(raw)
self.__rawtermstate = raw

# In macOS terminal we need to deactivate line wrap via ANSI escape code
if self.is_apple_terminal:
Expand Down Expand Up @@ -599,7 +600,20 @@ def input_hook(self):
except ImportError:
return None
if posix._is_inputhook_installed():
return posix._inputhook
return self.__run_input_hook

def __run_input_hook(self):
import posix
# 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
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,12 +1,14 @@
import errno
import itertools
import os
import select
import signal
import sys
import threading
import unittest
from functools import partial
from test.support import os_helper
from test.support import is_android, is_apple_mobile, is_emscripten, is_wasi
from test.support import threading_helper

from unittest import TestCase
Expand Down Expand Up @@ -392,3 +394,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")
@unittest.skipIf(is_android or is_apple_mobile or is_emscripten or is_wasi,
"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("posix._is_inputhook_installed", return_value=True),
patch("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