Skip to content

Commit 8207539

Browse files
pablogsalharjothkharaclaude
authored
[3.14] gh-152907: Restore cooked output flags around the input hook in the new REPL (GH-153389) (#158117)
(cherry picked from commit 46ee358) Co-authored-by: Harjoth Khara <harjoth.khara@gmail.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 23052e5 commit 8207539

3 files changed

Lines changed: 110 additions & 1 deletion

File tree

‎Lib/_pyrepl/unix_console.py‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -358,6 +358,7 @@ def prepare(self):
358358
raw.cc[termios.VMIN] = 1
359359
raw.cc[termios.VTIME] = 0
360360
self.__input_fd_set(raw)
361+
self.__rawtermstate = raw
361362

362363
# In macOS terminal we need to deactivate line wrap via ANSI escape code
363364
if self.is_apple_terminal:
@@ -595,7 +596,19 @@ def input_hook(self):
595596
# avoid inline imports here so the repl doesn't get flooded
596597
# with import logging from -X importtime=2
597598
if posix is not None and posix._is_inputhook_installed():
598-
return posix._inputhook
599+
return self.__run_input_hook
600+
601+
def __run_input_hook(self):
602+
# gh-152907: input hooks expect cooked output, but pyrepl runs with
603+
# OPOST disabled. Restore the saved output flags around the hook
604+
# (only oflag; input must stay raw at the prompt).
605+
cooked = self.__rawtermstate.copy()
606+
cooked.oflag = self.__svtermstate.oflag
607+
self.__input_fd_set(cooked)
608+
try:
609+
return posix._inputhook()
610+
finally:
611+
self.__input_fd_set(self.__rawtermstate)
599612

600613
def __enable_bracketed_paste(self) -> None:
601614
os.write(self.output_fd, b"\x1b[?2004h")

‎Lib/test/test_pyrepl/test_unix_console.py‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,15 @@
11
import errno
22
import itertools
33
import os
4+
import select
45
import signal
56
import sys
67
import threading
78
import unittest
89
from functools import partial
910
from _colorize import ANSIColors
1011
from test.support import os_helper, force_not_colorized_test_class
12+
from test.support import is_android, is_apple_mobile, is_wasm32
1113
from test.support import threading_helper
1214

1315
from unittest import TestCase
@@ -384,3 +386,93 @@ def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr):
384386

385387
# EIO error should be handled gracefully in restore()
386388
console.restore()
389+
390+
391+
try:
392+
import pty
393+
import termios as _termios
394+
except ImportError:
395+
pty = None
396+
397+
398+
@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows")
399+
@unittest.skipUnless(pty, "requires pty")
400+
@unittest.skipIf(is_android or is_apple_mobile or is_wasm32,
401+
"pty is not available on this platform")
402+
class TestUnixConsoleInputHook(TestCase):
403+
# gh-152907: the console must restore cooked output (OPOST) around
404+
# input-hook calls, then re-enter raw mode.
405+
406+
def test_input_hook_output_is_cooked(self):
407+
master_fd, slave_fd = pty.openpty()
408+
self.addCleanup(os.close, master_fd)
409+
410+
# tcsetattr(TCSADRAIN) blocks on some platforms (e.g. macOS) while the
411+
# master still holds unread output, so empty it before each mode switch.
412+
def drain():
413+
out = b""
414+
while select.select([master_fd], [], [], 0)[0]:
415+
try:
416+
data = os.read(master_fd, 4096)
417+
except OSError:
418+
break
419+
if not data:
420+
break
421+
out += data
422+
return out
423+
424+
# Start from a cooked terminal so there are saved flags to restore.
425+
attr = _termios.tcgetattr(slave_fd)
426+
attr[1] |= _termios.OPOST | _termios.ONLCR
427+
_termios.tcsetattr(slave_fd, _termios.TCSANOW, attr)
428+
429+
console = UnixConsole(slave_fd, slave_fd, term="xterm")
430+
console.prepare()
431+
try:
432+
drain() # discard prepare()'s own setup sequences
433+
# pyrepl's own rendering runs with OPOST cleared.
434+
self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST)
435+
436+
observed = {}
437+
438+
def fake_hook():
439+
observed["oflag"] = _termios.tcgetattr(slave_fd)[1]
440+
os.write(slave_fd, b"line1\nline2\n")
441+
observed["output"] = drain()
442+
return 0
443+
444+
with patch("_pyrepl.unix_console.posix") as mock_posix:
445+
mock_posix._is_inputhook_installed.return_value = True
446+
mock_posix._inputhook.side_effect = fake_hook
447+
hook = console.input_hook
448+
self.assertIsNotNone(hook)
449+
self.assertEqual(hook(), 0)
450+
451+
# The hook ran with cooked output (OPOST on)...
452+
self.assertTrue(observed["oflag"] & _termios.OPOST)
453+
# ...and raw mode was restored afterwards.
454+
self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST)
455+
# The tty translated the hook's bare '\n' into '\r\n'.
456+
self.assertEqual(observed["output"], b"line1\r\nline2\r\n")
457+
finally:
458+
# restore() writes and only then switches modes, so there is no
459+
# point left to drain from here; keep the master empty elsewhere.
460+
stop = threading.Event()
461+
462+
def pump():
463+
while not stop.is_set():
464+
if select.select([master_fd], [], [], 0.05)[0]:
465+
try:
466+
if not os.read(master_fd, 4096):
467+
break
468+
except OSError:
469+
break
470+
471+
pump_thread = threading.Thread(target=pump)
472+
pump_thread.start()
473+
try:
474+
console.restore()
475+
finally:
476+
stop.set()
477+
pump_thread.join()
478+
os.close(slave_fd)
Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
Restore cooked-mode terminal output flags around :c:data:`PyOS_InputHook`
2+
callbacks in the new :term:`REPL` (:mod:`!_pyrepl`), so that output written
3+
by an input hook (for example a GUI toolkit event loop) is no longer emitted
4+
with ``OPOST`` disabled and keeps its carriage returns.

0 commit comments

Comments
 (0)