ilusha/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython synced 2026-11-13 20:26:34 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions 1
cpython/Lib/test/test_pyrepl/test_unix_console.py
Miss Islington (bot) 53fbec93ac
[3.15] gh-152907: Restore cooked output flags around the input hook in the new REPL (GH-153389) (#158112) 
gh-152907: Restore cooked output flags around the input hook in the new REPL (GH-153389)

* gh-152907: Restore cooked output flags around the input hook in the new REPL

pyrepl clears OPOST for its own cursor rendering but calls PyOS_InputHook
from inside the raw-mode read loop, so output written by an input hook
(GUI toolkit event loops, and any warning/traceback/print they emit) is
emitted with bare '\n' and no '\r'.  Restore the terminal's saved output
flags around the hook call and re-enter raw mode afterwards; only oflag is
toggled so ECHO/ICANON stay off at the prompt.

* Skip the input-hook test on platforms without pty devices

The Emscripten buildbot has the pty module but no pty devices, so
pty.openpty() raises OSError("out of pty devices").  Guard the test
class the same way Lib/test/test_pty.py does.

* Propagate the input hook's return value and drop the sleep from the test



* Trim comments



* Observe the hook's output synchronously instead of via the reader thread

The reader thread no longer feeds any assertion: the hook drains the pty
master itself, so the check is an exact comparison. A drainer is still
needed for restore(), which writes before switching modes.

---------
(cherry picked from commit 46ee3580c0)

Co-authored-by: Harjoth Khara <harjoth.khara@gmail.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-09-24 16:35:00 +00:00

518 lines
17 KiB
Python
Raw Permalink Blame History

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
from unittest.mock import MagicMock, call, patch, ANY, Mock
from .support import handle_all_events, code_to_events, more_lines
try:
from _pyrepl.console import Event
from _pyrepl.unix_console import UnixConsole
except ImportError:
pass
from _pyrepl.terminfo import _TERMINAL_CAPABILITIES
TERM_CAPABILITIES = _TERMINAL_CAPABILITIES["ansi"]
def unix_console(events, **kwargs):
console = UnixConsole(term="xterm")
console.get_event = MagicMock(side_effect=events)
console.getpending = MagicMock(return_value=Event("key", ""))
height = kwargs.get("height", 25)
width = kwargs.get("width", 80)
console.getheightwidth = MagicMock(side_effect=lambda: (height, width))
console.wait = MagicMock()
console.prepare()
for key, val in kwargs.items():
setattr(console, key, val)
return console
handle_events_unix_console = partial(
handle_all_events,
prepare_console=unix_console,
)
handle_events_narrow_unix_console = partial(
handle_all_events,
prepare_console=partial(unix_console, width=5),
)
handle_events_short_unix_console = partial(
handle_all_events,
prepare_console=partial(unix_console, height=1),
)
handle_events_unix_console_height_3 = partial(
handle_all_events, prepare_console=partial(unix_console, height=3)
)
@unittest.skipIf(sys.platform == "win32", "No Unix event queue on Windows")
@patch(
"_pyrepl.terminfo.tparm",
lambda s, *args: s + b":" + b",".join(str(i).encode() for i in args),
)
@patch(
"termios.tcgetattr",
lambda _: [
27394,
3,
19200,
536872399,
38400,
38400,
[
b"\x04",
b"\xff",
b"\xff",
b"\x7f",
b"\x17",
b"\x15",
b"\x12",
b"\x00",
b"\x03",
b"\x1c",
b"\x1a",
b"\x19",
b"\x11",
b"\x13",
b"\x16",
b"\x0f",
b"\x01",
b"\x00",
b"\x14",
b"\x00",
],
],
)
@patch("termios.tcsetattr", lambda a, b, c: None)
@patch("os.write")
@force_not_colorized_test_class
class TestConsole(TestCase):
@staticmethod
def _prepare_reader_with_prompts(console, **kwargs):
from _pyrepl.readline import ReadlineAlikeReader, ReadlineConfig
config = ReadlineConfig(
readline_completer=kwargs.pop("readline_completer", None)
)
reader = ReadlineAlikeReader(console=console, config=config)
reader.paste_mode = False
for key, val in kwargs.items():
setattr(reader, key, val)
return reader
def test_colorized_multiline_typing_does_not_redraw_previous_line(self, _os_write):
def prepare_reader_with_prompts(console, **kwargs):
reader = self._prepare_reader_with_prompts(console, **kwargs)
reader.more_lines = partial(more_lines, namespace=None)
return reader
with force_color(True):
events = itertools.chain(
code_to_events("def foo():"),
[Event(evt="key", data="\n", raw=bytearray(b"\n"))],
code_to_events("x = 1"),
[Event(evt="key", data="\n", raw=bytearray(b"\n"))],
code_to_events("y"),
)
_, con = handle_all_events(
events,
prepare_console=unix_console,
prepare_reader=prepare_reader_with_prompts,
)
con.restore()
self.assertNotIn(
call(ANY, b" \x1b[0m x \x1b[0m=\x1b[0m "),
_os_write.mock_calls,
)
self.assertIn(call(ANY, b"y"), _os_write.mock_calls)
def test_no_newline(self, _os_write):
code = "1"
events = code_to_events(code)
_, con = handle_events_unix_console(events)
self.assertNotIn(call(ANY, b'\n'), _os_write.mock_calls)
con.restore()
def test_reset_on_finish(self, _os_write):
# gh-152068: finish() must emit the ANSI reset sequence so any
# active color does not leak past the prompt.
code = "1"
events = code_to_events(code)
_, con = handle_events_unix_console(events)
con.finish()
_os_write.assert_any_call(ANY, ANSIColors.RESET.encode(con.encoding))
con.restore()
def test_reset_on_restore(self, _os_write):
# gh-152068: restore() must emit the ANSI reset sequence.
code = "1"
events = code_to_events(code)
_, con = handle_events_unix_console(events)
con.restore()
_os_write.assert_any_call(ANY, ANSIColors.RESET.encode(con.encoding))
def test_newline(self, _os_write):
code = "\n"
events = code_to_events(code)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, b"\n")
con.restore()
def test_simple_addition(self, _os_write):
code = "12+34"
events = code_to_events(code)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, b"1")
_os_write.assert_any_call(ANY, b"2")
_os_write.assert_any_call(ANY, b"+")
_os_write.assert_any_call(ANY, b"3")
_os_write.assert_any_call(ANY, b"4")
con.restore()
def test_wrap(self, _os_write):
code = "12+34"
events = code_to_events(code)
_, con = handle_events_narrow_unix_console(events)
_os_write.assert_any_call(ANY, b"1")
_os_write.assert_any_call(ANY, b"2")
_os_write.assert_any_call(ANY, b"+")
_os_write.assert_any_call(ANY, b"3")
_os_write.assert_any_call(ANY, b"\\")
_os_write.assert_any_call(ANY, b"\n")
_os_write.assert_any_call(ANY, b"4")
con.restore()
def test_cursor_left(self, _os_write):
code = "1"
events = itertools.chain(
code_to_events(code),
[Event(evt="key", data="left", raw=bytearray(b"\x1bOD"))],
)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cub"] + b":1")
con.restore()
def test_cursor_left_right(self, _os_write):
code = "1"
events = itertools.chain(
code_to_events(code),
[
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
Event(evt="key", data="right", raw=bytearray(b"\x1bOC")),
],
)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cub"] + b":1")
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cuf"] + b":1")
con.restore()
def test_cursor_up(self, _os_write):
code = "1\n2+3"
events = itertools.chain(
code_to_events(code),
[Event(evt="key", data="up", raw=bytearray(b"\x1bOA"))],
)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cuu"] + b":1")
con.restore()
def test_cursor_up_down(self, _os_write):
code = "1\n2+3"
events = itertools.chain(
code_to_events(code),
[
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
],
)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cuu"] + b":1")
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cud"] + b":1")
con.restore()
def test_cursor_back_write(self, _os_write):
events = itertools.chain(
code_to_events("1"),
[Event(evt="key", data="left", raw=bytearray(b"\x1bOD"))],
code_to_events("2"),
)
_, con = handle_events_unix_console(events)
_os_write.assert_any_call(ANY, b"1")
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["cub"] + b":1")
_os_write.assert_any_call(ANY, b"2")
con.restore()
def test_multiline_function_move_up_short_terminal(self, _os_write):
# fmt: off
code = (
"def f():\n"
" foo"
)
# fmt: on
events = itertools.chain(
code_to_events(code),
[
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
Event(evt="scroll", data=None),
],
)
_, con = handle_events_short_unix_console(events)
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["ri"] + b":")
con.restore()
def test_multiline_function_move_up_down_short_terminal(self, _os_write):
# fmt: off
code = (
"def f():\n"
" foo"
)
# fmt: on
events = itertools.chain(
code_to_events(code),
[
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
Event(evt="scroll", data=None),
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
Event(evt="scroll", data=None),
],
)
_, con = handle_events_short_unix_console(events)
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["ri"] + b":")
_os_write.assert_any_call(ANY, TERM_CAPABILITIES["ind"] + b":")
con.restore()
def test_resize_bigger_on_multiline_function(self, _os_write):
# fmt: off
code = (
"def f():\n"
" foo"
)
# fmt: on
events = itertools.chain(code_to_events(code))
reader, console = handle_events_short_unix_console(events)
console.getheightwidth = MagicMock(side_effect=lambda: (2, 80))
def same_reader(_):
return reader
def same_console(events):
console.get_event = MagicMock(side_effect=events)
return console
_, con = handle_all_events(
[Event(evt="resize", data=None)],
prepare_reader=same_reader,
prepare_console=same_console,
)
_os_write.assert_has_calls(
[
call(ANY, TERM_CAPABILITIES["ri"] + b":"),
call(ANY, TERM_CAPABILITIES["cup"] + b":0,0"),
call(ANY, b"def f():"),
]
)
console.restore()
con.restore()
def test_resize_smaller_on_multiline_function(self, _os_write):
# fmt: off
code = (
"def f():\n"
" foo"
)
# fmt: on
events = itertools.chain(code_to_events(code))
reader, console = handle_events_unix_console_height_3(events)
console.getheightwidth = MagicMock(side_effect=lambda: (1, 80))
def same_reader(_):
return reader
def same_console(events):
console.get_event = MagicMock(side_effect=events)
return console
_, con = handle_all_events(
[Event(evt="resize", data=None)],
prepare_reader=same_reader,
prepare_console=same_console,
)
_os_write.assert_has_calls(
[
call(ANY, TERM_CAPABILITIES["ind"] + b":"),
call(ANY, TERM_CAPABILITIES["cup"] + b":0,0"),
call(ANY, b" foo"),
]
)
console.restore()
con.restore()
def test_getheightwidth_with_invalid_environ(self, _os_write):
# gh-128636
console = UnixConsole(term="xterm")
with os_helper.EnvironmentVarGuard() as env:
env["LINES"] = ""
self.assertIsInstance(console.getheightwidth(), tuple)
env["COLUMNS"] = ""
self.assertIsInstance(console.getheightwidth(), tuple)
os.environ = []
self.assertIsInstance(console.getheightwidth(), tuple)
@unittest.skipUnless(sys.platform == "darwin", "requires macOS")
def test_restore_with_invalid_environ_on_macos(self, _os_write):
# gh-128636 for macOS
console = UnixConsole(term="xterm")
with os_helper.EnvironmentVarGuard():
os.environ = []
console.prepare() # needed to call restore()
console.restore() # this should succeed
@threading_helper.reap_threads
@threading_helper.requires_working_threading()
def test_restore_in_thread(self, _os_write):
# gh-139391: ensure that console.restore() silently suppresses
# exceptions when calling signal.signal() from a non-main thread.
console = unix_console([])
console.old_sigwinch = signal.SIG_DFL
thread = threading.Thread(target=console.restore)
thread.start()
thread.join() # this should not raise
@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows")
class TestUnixConsoleEIOHandling(TestCase):
@patch('_pyrepl.unix_console.tcsetattr')
@patch('_pyrepl.unix_console.tcgetattr')
def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr):
import termios
mock_termios = Mock()
mock_termios.iflag = 0
mock_termios.oflag = 0
mock_termios.cflag = 0
mock_termios.lflag = 0
mock_termios.cc = [0] * 32
mock_termios.copy.return_value = mock_termios
mock_tcgetattr.return_value = mock_termios
console = UnixConsole(term="xterm")
console.prepare()
mock_tcsetattr.side_effect = termios.error(errno.EIO, "Input/output error")
# 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)
Reference in a new issue View git blame Copy permalink