| 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) |