mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-29 11:14:38 +08:00
Code Activity
Files
main
RustPython/Lib/test/test_pyrepl/test_unix_console.py
T
youknowdotandCodex e30f005142 Update Lib stream concurrency tests and pyrepl decorators to Python 3.15 (#9434) 
* Update Lib stream concurrency tests and pyrepl decorators to Python 3.15

Sync canonical bz2, lzma, and pickle free-threading tests and retire obsolete class-decorator symbol-table workarounds while preserving the Android marker.

Assisted-by: OpenAI Codex: model version unavailable

* Fix LZMA bounded-output stream progress

Keep processing header notifications while output space remains, drain buffered output with empty input, and retain capped-output state for needs_input.

Assisted-by: OpenAI Codex: model version unavailable

* Retire passing bounded LZMA decoding markers

Regenerate the canonical LZMA tests with update_lib, retaining all other existing markers. Both complete affected methods pass with the native buffered-output fix.

Assisted-by: OpenAI Codex: model version unavailable

---------

Co-authored-by: Codex <codex@openai.com>
2026-10-29 09:39:42 +09:00

520 lines
17 KiB
Python
Vendored
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()
@unittest.expectedFailureIf(sys.platform == "android", "TODO: RUSTPYTHON; TypeError: signal handler must be signal.SIG_IGN, signal.SIG_DFL, or a callable object")
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)
View Git Blame Copy Permalink