a09e215abf
Also explicitly fail the test if we timeout while waiting for output so that this failure mode is caught earlier.
1130 lines
36 KiB
Python
1130 lines
36 KiB
Python
import io
|
|
import itertools
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import rlcompleter
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from unittest import TestCase, skipUnless
|
|
from unittest.mock import patch
|
|
from test.support import force_not_colorized
|
|
from test.support import SHORT_TIMEOUT
|
|
from test.support.import_helper import import_module
|
|
from test.support.os_helper import unlink
|
|
|
|
from .support import (
|
|
FakeConsole,
|
|
handle_all_events,
|
|
handle_events_narrow_console,
|
|
more_lines,
|
|
multiline_input,
|
|
code_to_events,
|
|
clean_screen,
|
|
make_clean_env,
|
|
)
|
|
from _pyrepl.console import Event
|
|
from _pyrepl.readline import ReadlineAlikeReader, ReadlineConfig
|
|
from _pyrepl.readline import multiline_input as readline_multiline_input
|
|
|
|
try:
|
|
import pty
|
|
except ImportError:
|
|
pty = None
|
|
|
|
class TestCursorPosition(TestCase):
|
|
def prepare_reader(self, events):
|
|
console = FakeConsole(events)
|
|
config = ReadlineConfig(readline_completer=None)
|
|
reader = ReadlineAlikeReader(console=console, config=config)
|
|
return reader
|
|
|
|
def test_up_arrow_simple(self):
|
|
# fmt: off
|
|
code = (
|
|
"def f():\n"
|
|
" ...\n"
|
|
)
|
|
# fmt: on
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
],
|
|
)
|
|
|
|
reader, console = handle_all_events(events)
|
|
self.assertEqual(reader.cxy, (0, 1))
|
|
console.move_cursor.assert_called_once_with(0, 1)
|
|
|
|
def test_down_arrow_end_of_input(self):
|
|
# fmt: off
|
|
code = (
|
|
"def f():\n"
|
|
" ...\n"
|
|
)
|
|
# fmt: on
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
],
|
|
)
|
|
|
|
reader, console = handle_all_events(events)
|
|
self.assertEqual(reader.cxy, (0, 2))
|
|
console.move_cursor.assert_called_once_with(0, 2)
|
|
|
|
def test_left_arrow_simple(self):
|
|
events = itertools.chain(
|
|
code_to_events("11+11"),
|
|
[
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
],
|
|
)
|
|
|
|
reader, console = handle_all_events(events)
|
|
self.assertEqual(reader.cxy, (4, 0))
|
|
console.move_cursor.assert_called_once_with(4, 0)
|
|
|
|
def test_right_arrow_end_of_line(self):
|
|
events = itertools.chain(
|
|
code_to_events("11+11"),
|
|
[
|
|
Event(evt="key", data="right", raw=bytearray(b"\x1bOC")),
|
|
],
|
|
)
|
|
|
|
reader, console = handle_all_events(events)
|
|
self.assertEqual(reader.cxy, (5, 0))
|
|
console.move_cursor.assert_called_once_with(5, 0)
|
|
|
|
def test_cursor_position_simple_character(self):
|
|
events = itertools.chain(code_to_events("k"))
|
|
|
|
reader, _ = handle_all_events(events)
|
|
self.assertEqual(reader.pos, 1)
|
|
|
|
# 1 for simple character
|
|
self.assertEqual(reader.cxy, (1, 0))
|
|
|
|
def test_cursor_position_double_width_character(self):
|
|
events = itertools.chain(code_to_events("樂"))
|
|
|
|
reader, _ = handle_all_events(events)
|
|
self.assertEqual(reader.pos, 1)
|
|
|
|
# 2 for wide character
|
|
self.assertEqual(reader.cxy, (2, 0))
|
|
|
|
def test_cursor_position_double_width_character_move_left(self):
|
|
events = itertools.chain(
|
|
code_to_events("樂"),
|
|
[
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
self.assertEqual(reader.pos, 0)
|
|
self.assertEqual(reader.cxy, (0, 0))
|
|
|
|
def test_cursor_position_double_width_character_move_left_right(self):
|
|
events = itertools.chain(
|
|
code_to_events("樂"),
|
|
[
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="right", raw=bytearray(b"\x1bOC")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
self.assertEqual(reader.pos, 1)
|
|
|
|
# 2 for wide character
|
|
self.assertEqual(reader.cxy, (2, 0))
|
|
|
|
def test_cursor_position_double_width_characters_move_up(self):
|
|
for_loop = "for _ in _:"
|
|
|
|
# fmt: off
|
|
code = (
|
|
f"{for_loop}\n"
|
|
" ' 可口可乐; 可口可樂'"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
|
|
# cursor at end of first line
|
|
self.assertEqual(reader.pos, len(for_loop))
|
|
self.assertEqual(reader.cxy, (len(for_loop), 0))
|
|
|
|
def test_cursor_position_double_width_characters_move_up_down(self):
|
|
for_loop = "for _ in _:"
|
|
|
|
# fmt: off
|
|
code = (
|
|
f"{for_loop}\n"
|
|
" ' 可口可乐; 可口可樂'"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
|
|
# cursor here (showing 2nd line only):
|
|
# < ' 可口可乐; 可口可樂'>
|
|
# ^
|
|
self.assertEqual(reader.pos, 19)
|
|
self.assertEqual(reader.cxy, (10, 1))
|
|
|
|
def test_cursor_position_multiple_double_width_characters_move_left(self):
|
|
events = itertools.chain(
|
|
code_to_events("' 可口可乐; 可口可樂'"),
|
|
[
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
self.assertEqual(reader.pos, 10)
|
|
|
|
# 1 for quote, 1 for space, 2 per wide character,
|
|
# 1 for semicolon, 1 for space, 2 per wide character
|
|
self.assertEqual(reader.cxy, (16, 0))
|
|
|
|
def test_cursor_position_move_up_to_eol(self):
|
|
first_line = "for _ in _:"
|
|
second_line = " hello"
|
|
|
|
# fmt: off
|
|
code = (
|
|
f"{first_line}\n"
|
|
f"{second_line}\n"
|
|
" h\n"
|
|
" hel"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
|
|
# Cursor should be at end of line 1, even though line 2 is shorter
|
|
# for _ in _:
|
|
# hello
|
|
# h
|
|
# hel
|
|
self.assertEqual(
|
|
reader.pos, len(first_line) + len(second_line) + 1
|
|
) # +1 for newline
|
|
self.assertEqual(reader.cxy, (len(second_line), 1))
|
|
|
|
def test_cursor_position_move_down_to_eol(self):
|
|
last_line = " hel"
|
|
|
|
# fmt: off
|
|
code = (
|
|
"for _ in _:\n"
|
|
" hello\n"
|
|
" h\n"
|
|
f"{last_line}"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
|
|
# Cursor should be at end of line 3, even though line 2 is shorter
|
|
# for _ in _:
|
|
# hello
|
|
# h
|
|
# hel
|
|
self.assertEqual(reader.pos, len(code))
|
|
self.assertEqual(reader.cxy, (len(last_line), 3))
|
|
|
|
def test_cursor_position_multiple_mixed_lines_move_up(self):
|
|
# fmt: off
|
|
code = (
|
|
"def foo():\n"
|
|
" x = '可口可乐; 可口可樂'\n"
|
|
" y = 'abckdfjskldfjslkdjf'"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
13 * [Event(evt="key", data="left", raw=bytearray(b"\x1bOD"))],
|
|
[Event(evt="key", data="up", raw=bytearray(b"\x1bOA"))],
|
|
)
|
|
|
|
reader, _ = handle_all_events(events)
|
|
|
|
# By moving left, we're before the s:
|
|
# y = 'abckdfjskldfjslkdjf'
|
|
# ^
|
|
# And we should move before the semi-colon despite the different offset
|
|
# x = '可口可乐; 可口可樂'
|
|
# ^
|
|
self.assertEqual(reader.pos, 22)
|
|
self.assertEqual(reader.cxy, (15, 1))
|
|
|
|
def test_cursor_position_after_wrap_and_move_up(self):
|
|
# fmt: off
|
|
code = (
|
|
"def foo():\n"
|
|
" hello"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
],
|
|
)
|
|
reader, _ = handle_events_narrow_console(events)
|
|
|
|
# The code looks like this:
|
|
# def foo()\
|
|
# :
|
|
# hello
|
|
# After moving up we should be after the colon in line 2
|
|
self.assertEqual(reader.pos, 10)
|
|
self.assertEqual(reader.cxy, (1, 1))
|
|
|
|
|
|
class TestPyReplAutoindent(TestCase):
|
|
def prepare_reader(self, events):
|
|
console = FakeConsole(events)
|
|
config = ReadlineConfig(readline_completer=None)
|
|
reader = ReadlineAlikeReader(console=console, config=config)
|
|
return reader
|
|
|
|
def test_auto_indent_default(self):
|
|
# fmt: off
|
|
input_code = (
|
|
"def f():\n"
|
|
"pass\n\n"
|
|
)
|
|
|
|
output_code = (
|
|
"def f():\n"
|
|
" pass\n"
|
|
" "
|
|
)
|
|
# fmt: on
|
|
|
|
def test_auto_indent_continuation(self):
|
|
# auto indenting according to previous user indentation
|
|
# fmt: off
|
|
events = itertools.chain(
|
|
code_to_events("def f():\n"),
|
|
# add backspace to delete default auto-indent
|
|
[
|
|
Event(evt="key", data="backspace", raw=bytearray(b"\x7f")),
|
|
],
|
|
code_to_events(
|
|
" pass\n"
|
|
"pass\n\n"
|
|
),
|
|
)
|
|
|
|
output_code = (
|
|
"def f():\n"
|
|
" pass\n"
|
|
" pass\n"
|
|
" "
|
|
)
|
|
# fmt: on
|
|
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, output_code)
|
|
|
|
def test_auto_indent_prev_block(self):
|
|
# auto indenting according to indentation in different block
|
|
# fmt: off
|
|
events = itertools.chain(
|
|
code_to_events("def f():\n"),
|
|
# add backspace to delete default auto-indent
|
|
[
|
|
Event(evt="key", data="backspace", raw=bytearray(b"\x7f")),
|
|
],
|
|
code_to_events(
|
|
" pass\n"
|
|
"pass\n\n"
|
|
),
|
|
code_to_events(
|
|
"def g():\n"
|
|
"pass\n\n"
|
|
),
|
|
)
|
|
|
|
output_code = (
|
|
"def g():\n"
|
|
" pass\n"
|
|
" "
|
|
)
|
|
# fmt: on
|
|
|
|
reader = self.prepare_reader(events)
|
|
output1 = multiline_input(reader)
|
|
output2 = multiline_input(reader)
|
|
self.assertEqual(output2, output_code)
|
|
|
|
def test_auto_indent_multiline(self):
|
|
# fmt: off
|
|
events = itertools.chain(
|
|
code_to_events(
|
|
"def f():\n"
|
|
"pass"
|
|
),
|
|
[
|
|
# go to the end of the first line
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="\x05", raw=bytearray(b"\x1bO5")),
|
|
# new line should be autoindented
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
],
|
|
code_to_events(
|
|
"pass"
|
|
),
|
|
[
|
|
# go to end of last line
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
Event(evt="key", data="\x05", raw=bytearray(b"\x1bO5")),
|
|
# double newline to terminate the block
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
],
|
|
)
|
|
|
|
output_code = (
|
|
"def f():\n"
|
|
" pass\n"
|
|
" pass\n"
|
|
" "
|
|
)
|
|
# fmt: on
|
|
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, output_code)
|
|
|
|
def test_auto_indent_with_comment(self):
|
|
# fmt: off
|
|
events = code_to_events(
|
|
"def f(): # foo\n"
|
|
"pass\n\n"
|
|
)
|
|
|
|
output_code = (
|
|
"def f(): # foo\n"
|
|
" pass\n"
|
|
" "
|
|
)
|
|
# fmt: on
|
|
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, output_code)
|
|
|
|
def test_auto_indent_ignore_comments(self):
|
|
# fmt: off
|
|
events = code_to_events(
|
|
"pass #:\n"
|
|
)
|
|
|
|
output_code = (
|
|
"pass #:"
|
|
)
|
|
# fmt: on
|
|
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, output_code)
|
|
|
|
|
|
class TestPyReplOutput(TestCase):
|
|
def prepare_reader(self, events):
|
|
console = FakeConsole(events)
|
|
config = ReadlineConfig(readline_completer=None)
|
|
reader = ReadlineAlikeReader(console=console, config=config)
|
|
reader.can_colorize = False
|
|
return reader
|
|
|
|
def test_stdin_is_tty(self):
|
|
# Used during test log analysis to figure out if a TTY was available.
|
|
try:
|
|
if os.isatty(sys.stdin.fileno()):
|
|
return
|
|
except OSError as ose:
|
|
self.skipTest(f"stdin tty check failed: {ose}")
|
|
else:
|
|
self.skipTest("stdin is not a tty")
|
|
|
|
def test_stdout_is_tty(self):
|
|
# Used during test log analysis to figure out if a TTY was available.
|
|
try:
|
|
if os.isatty(sys.stdout.fileno()):
|
|
return
|
|
except OSError as ose:
|
|
self.skipTest(f"stdout tty check failed: {ose}")
|
|
else:
|
|
self.skipTest("stdout is not a tty")
|
|
|
|
def test_basic(self):
|
|
reader = self.prepare_reader(code_to_events("1+1\n"))
|
|
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "1+1")
|
|
self.assertEqual(clean_screen(reader.screen), "1+1")
|
|
|
|
def test_multiline_edit(self):
|
|
events = itertools.chain(
|
|
code_to_events("def f():\n...\n\n"),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
|
|
Event(evt="key", data="backspace", raw=bytearray(b"\x08")),
|
|
Event(evt="key", data="g", raw=bytearray(b"g")),
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
Event(evt="key", data="backspace", raw=bytearray(b"\x08")),
|
|
Event(evt="key", data="delete", raw=bytearray(b"\x7F")),
|
|
Event(evt="key", data="right", raw=bytearray(b"g")),
|
|
Event(evt="key", data="backspace", raw=bytearray(b"\x08")),
|
|
Event(evt="key", data="p", raw=bytearray(b"p")),
|
|
Event(evt="key", data="a", raw=bytearray(b"a")),
|
|
Event(evt="key", data="s", raw=bytearray(b"s")),
|
|
Event(evt="key", data="s", raw=bytearray(b"s")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
],
|
|
)
|
|
reader = self.prepare_reader(events)
|
|
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "def f():\n ...\n ")
|
|
self.assertEqual(clean_screen(reader.screen), "def f():\n ...")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "def g():\n pass\n ")
|
|
self.assertEqual(clean_screen(reader.screen), "def g():\n pass")
|
|
|
|
def test_history_navigation_with_up_arrow(self):
|
|
events = itertools.chain(
|
|
code_to_events("1+1\n2+2\n"),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
],
|
|
)
|
|
|
|
reader = self.prepare_reader(events)
|
|
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "1+1")
|
|
self.assertEqual(clean_screen(reader.screen), "1+1")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "2+2")
|
|
self.assertEqual(clean_screen(reader.screen), "2+2")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "2+2")
|
|
self.assertEqual(clean_screen(reader.screen), "2+2")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "1+1")
|
|
self.assertEqual(clean_screen(reader.screen), "1+1")
|
|
|
|
def test_history_with_multiline_entries(self):
|
|
code = "def foo():\nx = 1\ny = 2\nz = 3\n\ndef bar():\nreturn 42\n\n"
|
|
events = list(itertools.chain(
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
]
|
|
))
|
|
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
output = multiline_input(reader)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(
|
|
clean_screen(reader.screen),
|
|
'def foo():\n x = 1\n y = 2\n z = 3'
|
|
)
|
|
self.assertEqual(output, "def foo():\n x = 1\n y = 2\n z = 3\n ")
|
|
|
|
|
|
def test_history_navigation_with_down_arrow(self):
|
|
events = itertools.chain(
|
|
code_to_events("1+1\n2+2\n"),
|
|
[
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
|
|
],
|
|
)
|
|
|
|
reader = self.prepare_reader(events)
|
|
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "1+1")
|
|
self.assertEqual(clean_screen(reader.screen), "1+1")
|
|
|
|
def test_history_search(self):
|
|
events = itertools.chain(
|
|
code_to_events("1+1\n2+2\n3+3\n"),
|
|
[
|
|
Event(evt="key", data="\x12", raw=bytearray(b"\x12")),
|
|
Event(evt="key", data="1", raw=bytearray(b"1")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
Event(evt="key", data="\n", raw=bytearray(b"\n")),
|
|
],
|
|
)
|
|
|
|
reader = self.prepare_reader(events)
|
|
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "1+1")
|
|
self.assertEqual(clean_screen(reader.screen), "1+1")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "2+2")
|
|
self.assertEqual(clean_screen(reader.screen), "2+2")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "3+3")
|
|
self.assertEqual(clean_screen(reader.screen), "3+3")
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "1+1")
|
|
self.assertEqual(clean_screen(reader.screen), "1+1")
|
|
|
|
def test_control_character(self):
|
|
events = code_to_events("c\x1d\n")
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, "c\x1d")
|
|
self.assertEqual(clean_screen(reader.screen), "c")
|
|
|
|
|
|
class TestPyReplCompleter(TestCase):
|
|
def prepare_reader(self, events, namespace):
|
|
console = FakeConsole(events)
|
|
config = ReadlineConfig()
|
|
config.readline_completer = rlcompleter.Completer(namespace).complete
|
|
reader = ReadlineAlikeReader(console=console, config=config)
|
|
return reader
|
|
|
|
@patch("rlcompleter._readline_available", False)
|
|
def test_simple_completion(self):
|
|
events = code_to_events("os.getpid\t\n")
|
|
|
|
namespace = {"os": os}
|
|
reader = self.prepare_reader(events, namespace)
|
|
|
|
output = multiline_input(reader, namespace)
|
|
self.assertEqual(output, "os.getpid()")
|
|
|
|
def test_completion_with_many_options(self):
|
|
# Test with something that initially displays many options
|
|
# and then complete from one of them. The first time tab is
|
|
# pressed, the options are displayed (which corresponds to
|
|
# when the repl shows [ not unique ]) and the second completes
|
|
# from one of them.
|
|
events = code_to_events("os.\t\tO_AP\t\n")
|
|
|
|
namespace = {"os": os}
|
|
reader = self.prepare_reader(events, namespace)
|
|
|
|
output = multiline_input(reader, namespace)
|
|
self.assertEqual(output, "os.O_APPEND")
|
|
|
|
def test_empty_namespace_completion(self):
|
|
events = code_to_events("os.geten\t\n")
|
|
namespace = {}
|
|
reader = self.prepare_reader(events, namespace)
|
|
|
|
output = multiline_input(reader, namespace)
|
|
self.assertEqual(output, "os.geten")
|
|
|
|
def test_global_namespace_completion(self):
|
|
events = code_to_events("py\t\n")
|
|
namespace = {"python": None}
|
|
reader = self.prepare_reader(events, namespace)
|
|
output = multiline_input(reader, namespace)
|
|
self.assertEqual(output, "python")
|
|
|
|
def test_updown_arrow_with_completion_menu(self):
|
|
"""Up arrow in the middle of unfinished tab completion when the menu is displayed
|
|
should work and trigger going back in history. Down arrow should subsequently
|
|
get us back to the incomplete command."""
|
|
code = "import os\nos.\t\t"
|
|
namespace = {"os": os}
|
|
|
|
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")),
|
|
],
|
|
code_to_events("\n"),
|
|
)
|
|
reader = self.prepare_reader(events, namespace=namespace)
|
|
output = multiline_input(reader, namespace)
|
|
# This is the first line, nothing to see here
|
|
self.assertEqual(output, "import os")
|
|
# This is the second line. We pressed up and down arrows
|
|
# so we should end up where we were when we initiated tab completion.
|
|
output = multiline_input(reader, namespace)
|
|
self.assertEqual(output, "os.")
|
|
|
|
@patch("_pyrepl.readline._ReadlineWrapper.get_reader")
|
|
@patch("sys.stderr", new_callable=io.StringIO)
|
|
def test_completion_with_warnings(self, mock_stderr, mock_get_reader):
|
|
class Dummy:
|
|
@property
|
|
def test_func(self):
|
|
import warnings
|
|
|
|
warnings.warn("warnings\n")
|
|
return None
|
|
|
|
dummy = Dummy()
|
|
events = code_to_events("dummy.test_func.\t\n\n")
|
|
namespace = {"dummy": dummy}
|
|
reader = self.prepare_reader(events, namespace)
|
|
mock_get_reader.return_value = reader
|
|
output = readline_multiline_input(more_lines, ">>>", "...")
|
|
self.assertEqual(output, "dummy.test_func.__")
|
|
self.assertEqual(mock_stderr.getvalue(), "")
|
|
|
|
|
|
class TestPasteEvent(TestCase):
|
|
def prepare_reader(self, events):
|
|
console = FakeConsole(events)
|
|
config = ReadlineConfig(readline_completer=None)
|
|
reader = ReadlineAlikeReader(console=console, config=config)
|
|
return reader
|
|
|
|
def test_paste(self):
|
|
# fmt: off
|
|
code = (
|
|
"def a():\n"
|
|
" for x in range(10):\n"
|
|
" if x%2:\n"
|
|
" print(x)\n"
|
|
" else:\n"
|
|
" pass\n"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
[
|
|
Event(evt="key", data="f3", raw=bytearray(b"\x1bOR")),
|
|
],
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="f3", raw=bytearray(b"\x1bOR")),
|
|
],
|
|
code_to_events("\n"),
|
|
)
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, code)
|
|
|
|
def test_paste_mid_newlines(self):
|
|
# fmt: off
|
|
code = (
|
|
"def f():\n"
|
|
" x = y\n"
|
|
" \n"
|
|
" y = z\n"
|
|
)
|
|
# fmt: on
|
|
|
|
events = itertools.chain(
|
|
[
|
|
Event(evt="key", data="f3", raw=bytearray(b"\x1bOR")),
|
|
],
|
|
code_to_events(code),
|
|
[
|
|
Event(evt="key", data="f3", raw=bytearray(b"\x1bOR")),
|
|
],
|
|
code_to_events("\n"),
|
|
)
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, code)
|
|
|
|
def test_paste_mid_newlines_not_in_paste_mode(self):
|
|
# fmt: off
|
|
code = (
|
|
"def f():\n"
|
|
"x = y\n"
|
|
"\n"
|
|
"y = z\n\n"
|
|
)
|
|
|
|
expected = (
|
|
"def f():\n"
|
|
" x = y\n"
|
|
" "
|
|
)
|
|
# fmt: on
|
|
|
|
events = code_to_events(code)
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, expected)
|
|
|
|
def test_paste_not_in_paste_mode(self):
|
|
# fmt: off
|
|
input_code = (
|
|
"def a():\n"
|
|
"for x in range(10):\n"
|
|
"if x%2:\n"
|
|
"print(x)\n"
|
|
"else:\n"
|
|
"pass\n\n"
|
|
)
|
|
|
|
output_code = (
|
|
"def a():\n"
|
|
" for x in range(10):\n"
|
|
" if x%2:\n"
|
|
" print(x)\n"
|
|
" else:"
|
|
)
|
|
# fmt: on
|
|
|
|
events = code_to_events(input_code)
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, output_code)
|
|
|
|
def test_bracketed_paste(self):
|
|
"""Test that bracketed paste using \x1b[200~ and \x1b[201~ works."""
|
|
# fmt: off
|
|
input_code = (
|
|
"def a():\n"
|
|
" for x in range(10):\n"
|
|
"\n"
|
|
" if x%2:\n"
|
|
" print(x)\n"
|
|
"\n"
|
|
" else:\n"
|
|
" pass\n"
|
|
)
|
|
|
|
output_code = (
|
|
"def a():\n"
|
|
" for x in range(10):\n"
|
|
"\n"
|
|
" if x%2:\n"
|
|
" print(x)\n"
|
|
"\n"
|
|
" else:\n"
|
|
" pass\n"
|
|
)
|
|
# fmt: on
|
|
|
|
paste_start = "\x1b[200~"
|
|
paste_end = "\x1b[201~"
|
|
|
|
events = itertools.chain(
|
|
code_to_events(paste_start),
|
|
code_to_events(input_code),
|
|
code_to_events(paste_end),
|
|
code_to_events("\n"),
|
|
)
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, output_code)
|
|
|
|
def test_bracketed_paste_single_line(self):
|
|
input_code = "oneline"
|
|
|
|
paste_start = "\x1b[200~"
|
|
paste_end = "\x1b[201~"
|
|
|
|
events = itertools.chain(
|
|
code_to_events(paste_start),
|
|
code_to_events(input_code),
|
|
code_to_events(paste_end),
|
|
code_to_events("\n"),
|
|
)
|
|
reader = self.prepare_reader(events)
|
|
output = multiline_input(reader)
|
|
self.assertEqual(output, input_code)
|
|
|
|
|
|
@skipUnless(pty, "requires pty")
|
|
class TestMain(TestCase):
|
|
def setUp(self):
|
|
# Cleanup from PYTHON* variables to isolate from local
|
|
# user settings, see #121359. Such variables should be
|
|
# added later in test methods to patched os.environ.
|
|
patcher = patch('os.environ', new=make_clean_env())
|
|
self.addCleanup(patcher.stop)
|
|
patcher.start()
|
|
|
|
@force_not_colorized
|
|
def test_exposed_globals_in_repl(self):
|
|
pre = "['__annotations__', '__builtins__'"
|
|
post = "'__loader__', '__name__', '__package__', '__spec__']"
|
|
output, exit_code = self.run_repl(["sorted(dir())", "exit()"])
|
|
if "can't use pyrepl" in output:
|
|
self.skipTest("pyrepl not available")
|
|
self.assertEqual(exit_code, 0)
|
|
|
|
# if `__main__` is not a file (impossible with pyrepl)
|
|
case1 = f"{pre}, '__doc__', {post}" in output
|
|
|
|
# if `__main__` is an uncached .py file (no .pyc)
|
|
case2 = f"{pre}, '__doc__', '__file__', {post}" in output
|
|
|
|
# if `__main__` is a cached .pyc file and the .py source exists
|
|
case3 = f"{pre}, '__cached__', '__doc__', '__file__', {post}" in output
|
|
|
|
# if `__main__` is a cached .pyc file but there's no .py source file
|
|
case4 = f"{pre}, '__cached__', '__doc__', {post}" in output
|
|
|
|
self.assertTrue(case1 or case2 or case3 or case4, output)
|
|
|
|
def _assertMatchOK(
|
|
self, var: str, expected: str | re.Pattern, actual: str
|
|
) -> None:
|
|
if isinstance(expected, re.Pattern):
|
|
self.assertTrue(
|
|
expected.match(actual),
|
|
f"{var}={actual} does not match {expected.pattern}",
|
|
)
|
|
else:
|
|
self.assertEqual(
|
|
actual,
|
|
expected,
|
|
f"expected {var}={expected}, got {var}={actual}",
|
|
)
|
|
|
|
@force_not_colorized
|
|
def _run_repl_globals_test(self, expectations, *, as_file=False, as_module=False):
|
|
clean_env = make_clean_env()
|
|
clean_env["NO_COLOR"] = "1" # force_not_colorized doesn't touch subprocesses
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
blue = pathlib.Path(td) / "blue"
|
|
blue.mkdir()
|
|
mod = blue / "calx.py"
|
|
mod.write_text("FOO = 42", encoding="utf-8")
|
|
commands = [
|
|
"print(f'{" + var + "=}')" for var in expectations
|
|
] + ["exit()"]
|
|
if as_file and as_module:
|
|
self.fail("as_file and as_module are mutually exclusive")
|
|
elif as_file:
|
|
output, exit_code = self.run_repl(
|
|
commands,
|
|
cmdline_args=[str(mod)],
|
|
env=clean_env,
|
|
)
|
|
elif as_module:
|
|
output, exit_code = self.run_repl(
|
|
commands,
|
|
cmdline_args=["-m", "blue.calx"],
|
|
env=clean_env,
|
|
cwd=td,
|
|
)
|
|
else:
|
|
self.fail("Choose one of as_file or as_module")
|
|
|
|
if "can't use pyrepl" in output:
|
|
self.skipTest("pyrepl not available")
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
for var, expected in expectations.items():
|
|
with self.subTest(var=var, expected=expected):
|
|
if m := re.search(rf"[\r\n]{var}=(.+?)[\r\n]", output):
|
|
self._assertMatchOK(var, expected, actual=m.group(1))
|
|
else:
|
|
self.fail(f"{var}= not found in output")
|
|
|
|
self.assertNotIn("Exception", output)
|
|
self.assertNotIn("Traceback", output)
|
|
|
|
def test_inspect_keeps_globals_from_inspected_file(self):
|
|
expectations = {
|
|
"FOO": "42",
|
|
"__name__": "'__main__'",
|
|
"__package__": "None",
|
|
# "__file__" is missing in -i, like in the basic REPL
|
|
}
|
|
self._run_repl_globals_test(expectations, as_file=True)
|
|
|
|
def test_inspect_keeps_globals_from_inspected_module(self):
|
|
expectations = {
|
|
"FOO": "42",
|
|
"__name__": "'__main__'",
|
|
"__package__": "'blue'",
|
|
"__file__": re.compile(r"^'.*calx.py'$"),
|
|
}
|
|
self._run_repl_globals_test(expectations, as_module=True)
|
|
|
|
def test_dumb_terminal_exits_cleanly(self):
|
|
env = os.environ.copy()
|
|
env.update({"TERM": "dumb"})
|
|
output, exit_code = self.run_repl("exit()\n", env=env)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("warning: can\'t use pyrepl", output)
|
|
self.assertNotIn("Exception", output)
|
|
self.assertNotIn("Traceback", output)
|
|
|
|
@force_not_colorized
|
|
def test_python_basic_repl(self):
|
|
env = os.environ.copy()
|
|
commands = ("from test.support import initialized_with_pyrepl\n"
|
|
"initialized_with_pyrepl()\n"
|
|
"exit()\n")
|
|
|
|
env.pop("PYTHON_BASIC_REPL", None)
|
|
output, exit_code = self.run_repl(commands, env=env)
|
|
if "can\'t use pyrepl" in output:
|
|
self.skipTest("pyrepl not available")
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("True", output)
|
|
self.assertNotIn("False", output)
|
|
self.assertNotIn("Exception", output)
|
|
self.assertNotIn("Traceback", output)
|
|
|
|
env["PYTHON_BASIC_REPL"] = "1"
|
|
output, exit_code = self.run_repl(commands, env=env)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("False", output)
|
|
self.assertNotIn("True", output)
|
|
self.assertNotIn("Exception", output)
|
|
self.assertNotIn("Traceback", output)
|
|
|
|
def test_not_wiping_history_file(self):
|
|
# skip, if readline module is not available
|
|
import_module('readline')
|
|
|
|
hfile = tempfile.NamedTemporaryFile(delete=False)
|
|
self.addCleanup(unlink, hfile.name)
|
|
env = os.environ.copy()
|
|
env["PYTHON_HISTORY"] = hfile.name
|
|
commands = "123\nspam\nexit()\n"
|
|
|
|
env.pop("PYTHON_BASIC_REPL", None)
|
|
output, exit_code = self.run_repl(commands, env=env)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("123", output)
|
|
self.assertIn("spam", output)
|
|
self.assertNotEqual(pathlib.Path(hfile.name).stat().st_size, 0)
|
|
|
|
hfile.file.truncate()
|
|
hfile.close()
|
|
|
|
env["PYTHON_BASIC_REPL"] = "1"
|
|
output, exit_code = self.run_repl(commands, env=env)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("123", output)
|
|
self.assertIn("spam", output)
|
|
self.assertNotEqual(pathlib.Path(hfile.name).stat().st_size, 0)
|
|
|
|
def run_repl(
|
|
self,
|
|
repl_input: str | list[str],
|
|
env: dict | None = None,
|
|
*,
|
|
cmdline_args: list[str] | None = None,
|
|
cwd: str | None = None,
|
|
) -> tuple[str, int]:
|
|
assert pty
|
|
master_fd, slave_fd = pty.openpty()
|
|
cmd = [sys.executable, "-i", "-u"]
|
|
if env is None:
|
|
cmd.append("-I")
|
|
if cmdline_args is not None:
|
|
cmd.extend(cmdline_args)
|
|
process = subprocess.Popen(
|
|
cmd,
|
|
stdin=slave_fd,
|
|
stdout=slave_fd,
|
|
stderr=slave_fd,
|
|
cwd=cwd,
|
|
text=True,
|
|
close_fds=True,
|
|
env=env if env else os.environ,
|
|
)
|
|
os.close(slave_fd)
|
|
if isinstance(repl_input, list):
|
|
repl_input = "\n".join(repl_input) + "\n"
|
|
os.write(master_fd, repl_input.encode("utf-8"))
|
|
|
|
output = []
|
|
while select.select([master_fd], [], [], SHORT_TIMEOUT)[0]:
|
|
try:
|
|
data = os.read(master_fd, 1024).decode("utf-8")
|
|
if not data:
|
|
break
|
|
except OSError:
|
|
break
|
|
output.append(data)
|
|
else:
|
|
os.close(master_fd)
|
|
process.kill()
|
|
self.fail(f"Timeout while waiting for output, got: {''.join(output)}")
|
|
|
|
os.close(master_fd)
|
|
try:
|
|
exit_code = process.wait(timeout=SHORT_TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
exit_code = process.wait()
|
|
return "\n".join(output), exit_code
|