diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1bdb7a8bbc..dc8284f172 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -301,6 +301,7 @@ jobs: run: ./ci.sh env: NO_TEST_REQUIREMENTS: '${{ matrix.no_test_requirements }}' + TERM: xterm - if: >- always() && matrix.check_formatting != '1' @@ -346,6 +347,8 @@ jobs: check-latest: true - name: Run tests run: ./ci.sh + env: + TERM: xterm - if: always() uses: codecov/codecov-action@v5 with: diff --git a/newsfragments/3007.bugfix.rst b/newsfragments/3007.bugfix.rst new file mode 100644 index 0000000000..98fa369ca7 --- /dev/null +++ b/newsfragments/3007.bugfix.rst @@ -0,0 +1,6 @@ +Switch ``python -m trio`` to use ``pyrepl``. This means that keyboard +interrupts should be processed more consistently, as well as matching +the builtin Python REPL more. One downside is that now using the trio +REPL requires PyPy, CPython 3.13+, or installing the ``pyrepl`` package +from PyPI. We recommend the first two options, as our support for the +version on PyPI is worse. diff --git a/src/trio/_repl.py b/src/trio/_repl.py index 6b5612e28f..496e56d462 100644 --- a/src/trio/_repl.py +++ b/src/trio/_repl.py @@ -1,48 +1,51 @@ from __future__ import annotations import ast -import contextlib import inspect import sys -import warnings from code import InteractiveConsole -from types import CodeType, FrameType, FunctionType -from typing import TYPE_CHECKING +from signal import SIGINT, raise_signal +from types import CodeType, FunctionType +from typing import cast import outcome import trio -import trio.lowlevel from trio._util import final -if TYPE_CHECKING: - from collections.abc import Callable - - -class SuppressDecorator(contextlib.ContextDecorator, contextlib.suppress): - pass - - -@SuppressDecorator(KeyboardInterrupt) -@trio.lowlevel.disable_ki_protection -def terminal_newline() -> None: # TODO: test this line - import fcntl - import termios - - # Fake up a newline char as if user had typed it at the terminal +try: + import pyrepl + from pyrepl import commands, reader as r, readline +except ImportError: try: - fcntl.ioctl(sys.stdin, termios.TIOCSTI, b"\n") # type: ignore[attr-defined, unused-ignore] - except OSError as e: - print(f"\nPress enter! Newline injection failed: {e}", end="", flush=True) + import _pyrepl as pyrepl + from _pyrepl import commands, reader as r, readline + except ImportError: + print( + "Trio's REPL requires CPython 3.13+, PyPy, or installing pyrepl from PyPI." + ) + exit(1) + +# there are differences between the CPython pyrepl and PyPI pyrepl +try: + # The following expression fails on PyPy, even though you can + # `import pyrepl`. This is important because PyPy simply vendors + # CPython pyrepl: https://github.com/pypy/pypy/issues/4990 + pyrepl.__version__ # noqa: B018 +except AttributeError: + CPYTHON_VENDOR = True +else: + CPYTHON_VENDOR = False @final class TrioInteractiveConsole(InteractiveConsole): - def __init__(self, repl_locals: dict[str, object] | None = None) -> None: + def __init__( + self, + repl_locals: dict[str, object] | None = None, + ) -> None: super().__init__(locals=repl_locals) - self.token: trio.lowlevel.TrioToken | None = None self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT - self.interrupted = False def runcode(self, code: CodeType) -> None: func = FunctionType(code, self.locals) @@ -71,79 +74,71 @@ def runcode(self, code: CodeType) -> None: # We always use sys.excepthook, unlike other implementations. # This means that overriding self.write also does nothing to tbs. sys.excepthook(sys.last_type, sys.last_value, sys.last_traceback) + # clear any residual KI trio.from_thread.run(trio.lowlevel.checkpoint_if_cancelled) # trio.from_thread.check_cancelled() has too long of a memory - if sys.platform == "win32": # TODO: test this line - def raw_input(self, prompt: str = "") -> str: - try: - return input(prompt) - except EOFError: - # check if trio has a pending KI - trio.from_thread.run(trio.lowlevel.checkpoint_if_cancelled) - raise - - else: - - def raw_input(self, prompt: str = "") -> str: - from signal import SIGINT, signal - - assert not self.interrupted - - def install_handler() -> ( - Callable[[int, FrameType | None], None] | int | None - ): - def handler( - sig: int, frame: FrameType | None - ) -> None: # TODO: test this line - self.interrupted = True - token.run_sync_soon(terminal_newline, idempotent=True) - - token = trio.lowlevel.current_trio_token() - - return signal(SIGINT, handler) +async def repl_input(reader: r.Reader | None, prompt: str) -> str: # type: ignore[no-any-unimported] + assert reader is not None + reader.ps1 = prompt + reader.prepare() + try: + reader.refresh() + while not reader.finished: + if not reader.handle1(block=False): + if sys.platform == "win32": # TODO: test this line + await trio.lowlevel.wait_readable(pyrepl.windows_console.InHandle) + else: + await trio.lowlevel.wait_readable(reader.console.input_fd) + + if CPYTHON_VENDOR: + return reader.get_unicode() # type: ignore[no-any-return] + else: + return reader.get_str() # type: ignore[no-any-return] + finally: + reader.restore() - prev_handler = trio.from_thread.run_sync(install_handler) - try: - return input(prompt) - finally: - trio.from_thread.run_sync(signal, SIGINT, prev_handler) - if self.interrupted: # TODO: test this line - raise KeyboardInterrupt - - def write(self, output: str) -> None: - if self.interrupted: # TODO: test this line - assert output == "\nKeyboardInterrupt\n" - sys.stderr.write(output[1:]) - self.interrupted = False - else: - sys.stderr.write(output) +async def run_repl(console: TrioInteractiveConsole, reader: r.Reader | None) -> None: # type: ignore[no-any-unimported] + # mostly copy-pasted from code.InteractiveConsole.interact + try: + sys.ps1 # noqa: B018 + except AttributeError: + sys.ps1 = ">>> " + try: + sys.ps2 # noqa: B018 + except AttributeError: + sys.ps2 = "... " -async def run_repl(console: TrioInteractiveConsole) -> None: banner = ( f"trio REPL {sys.version} on {sys.platform}\n" f'Use "await" directly instead of "trio.run()".\n' f'Type "help", "copyright", "credits" or "license" ' f"for more information.\n" - f'{getattr(sys, "ps1", ">>> ")}import trio' + f'{getattr(sys, "ps1", ">>> ")}import trio\n' ) - try: - await trio.to_thread.run_sync(console.interact, banner) - finally: - warnings.filterwarnings( - "ignore", - message=r"^coroutine .* was never awaited$", - category=RuntimeWarning, - ) + console.write(banner) + more = 0 + while True: + try: + prompt = cast("str", sys.ps2 if more else sys.ps1) + try: + line = await repl_input(reader, prompt) + except EOFError: + console.write("\n") + break + else: + more = await trio.to_thread.run_sync(console.push, line) + except KeyboardInterrupt: # TODO: test this line + console.write("\nKeyboardInterrupt\n") + console.resetbuffer() + more = 0 -def main(original_locals: dict[str, object]) -> None: - with contextlib.suppress(ImportError): - import readline # noqa: F401 +def main(original_locals: dict[str, object]) -> None: repl_locals: dict[str, object] = {"trio": trio} for key in { "__name__", @@ -155,5 +150,19 @@ def main(original_locals: dict[str, object]) -> None: }: repl_locals[key] = original_locals[key] + # This call also registers all necessary signal handlers. + # Otherwise, we would not be able to run `readline` in a child + # thread. + reader = readline._get_reader() + + if not CPYTHON_VENDOR: + # The default `interrupt` command finishes the console, which + # adds an extra newline. Unforgivable! + class interrupt(commands.FinishCommand): # type: ignore[misc,no-any-unimported] + def do(self) -> None: # TODO: test this line + raise_signal(SIGINT) + + reader.commands["interrupt"] = interrupt + console = TrioInteractiveConsole(repl_locals) - trio.run(run_repl, console) + trio.run(run_repl, console, reader) diff --git a/src/trio/_tests/test_repl.py b/src/trio/_tests/test_repl.py index 28075d551c..5da0003dd7 100644 --- a/src/trio/_tests/test_repl.py +++ b/src/trio/_tests/test_repl.py @@ -1,47 +1,59 @@ from __future__ import annotations +import errno import os -import pathlib +import re import signal -import subprocess import sys -from functools import partial -from typing import Protocol +from typing import TYPE_CHECKING, Protocol import pytest -import trio._repl +from trio._tests.pytest_plugin import SKIP_OPTIONAL_IMPORTS +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable -class RawInput(Protocol): +if sys.platform == "win32" and sys.version_info < (3, 13): + pytest.skip("PyPI pyrepl only supports unix", allow_module_level=True) + +try: + import trio._repl +except SystemExit: + assert SKIP_OPTIONAL_IMPORTS + pytest.skip( + "we exit out of the REPL quickly if no pyrepl is found", + allow_module_level=True, + ) + + +class ReplInput(Protocol): def __call__(self, prompt: str = "") -> str: ... -def build_raw_input(cmds: list[str]) -> RawInput: +def build_repl_input(cmds: list[str]) -> Callable[[object, str], Awaitable[str]]: """ Pass in a list of strings. Returns a callable that returns each string, each time its called When there are not more strings to return, raise EOFError """ cmds_iter = iter(cmds) - prompts = [] - def _raw_helper(prompt: str = "") -> str: - prompts.append(prompt) + async def _repl_input_helper(_reader: object, _prompt: str) -> str: try: return next(cmds_iter) except StopIteration: raise EOFError from None - return _raw_helper + return _repl_input_helper -def test_build_raw_input() -> None: +async def test_build_repl_input() -> None: """Quick test of our helper function.""" - raw_input = build_raw_input(["cmd1"]) - assert raw_input() == "cmd1" + repl_input = build_repl_input(["cmd1"]) + assert await repl_input(None, "x") == "cmd1" with pytest.raises(EOFError): - raw_input() + await repl_input(None, "x") async def test_basic_interaction( @@ -53,7 +65,7 @@ async def test_basic_interaction( Ensure that the interpreted prints the expected results. """ console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ # evaluate simple expression and recall the value "x = 1", @@ -75,22 +87,22 @@ async def test_basic_interaction( "sys.stdout.write('hello stdout\\n')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, _err = capsys.readouterr() assert out.splitlines() == ["x=1", "'hello'", "2", "4", "hello stdout", "13"] async def test_system_exits_quit_interpreter(monkeypatch: pytest.MonkeyPatch) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ "raise SystemExit", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) with pytest.raises(SystemExit): - await trio._repl.run_repl(console) + await trio._repl.run_repl(console, None) async def test_KI_interrupts( @@ -98,7 +110,7 @@ async def test_KI_interrupts( monkeypatch: pytest.MonkeyPatch, ) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ "import signal, trio, trio.lowlevel", "async def f():", @@ -113,8 +125,8 @@ async def test_KI_interrupts( "print('AFTER KeyboardInterrupt')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, err = capsys.readouterr() assert "KeyboardInterrupt" in err assert "should" not in out @@ -126,7 +138,7 @@ async def test_system_exits_in_exc_group( monkeypatch: pytest.MonkeyPatch, ) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ "import sys", "if sys.version_info < (3, 11):", @@ -136,8 +148,8 @@ async def test_system_exits_in_exc_group( "print('AFTER BaseExceptionGroup')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, _err = capsys.readouterr() # assert that raise SystemExit in an exception group # doesn't quit @@ -149,7 +161,7 @@ async def test_system_exits_in_nested_exc_group( monkeypatch: pytest.MonkeyPatch, ) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ "import sys", "if sys.version_info < (3, 11):", @@ -160,8 +172,8 @@ async def test_system_exits_in_nested_exc_group( "print('AFTER BaseExceptionGroup')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, _err = capsys.readouterr() # assert that raise SystemExit in an exception group # doesn't quit @@ -173,15 +185,15 @@ async def test_base_exception_captured( monkeypatch: pytest.MonkeyPatch, ) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ # The statement after raise should still get executed "raise BaseException", "print('AFTER BaseException')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, err = capsys.readouterr() assert "_threads.py" not in err assert "_repl.py" not in err @@ -193,15 +205,15 @@ async def test_exc_group_captured( monkeypatch: pytest.MonkeyPatch, ) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ # The statement after raise should still get executed "raise ExceptionGroup('', [KeyError()])", "print('AFTER ExceptionGroup')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, _err = capsys.readouterr() assert "AFTER ExceptionGroup" in out @@ -211,7 +223,7 @@ async def test_base_exception_capture_from_coroutine( monkeypatch: pytest.MonkeyPatch, ) -> None: console = trio._repl.TrioInteractiveConsole() - raw_input = build_raw_input( + repl_input = build_repl_input( [ "async def async_func_raises_base_exception():", " raise BaseException", @@ -222,207 +234,138 @@ async def test_base_exception_capture_from_coroutine( "print('AFTER BaseException')", ], ) - monkeypatch.setattr(console, "raw_input", raw_input) - await trio._repl.run_repl(console) + monkeypatch.setattr(trio._repl, "repl_input", repl_input) + await trio._repl.run_repl(console, None) out, err = capsys.readouterr() assert "_threads.py" not in err assert "_repl.py" not in err assert "AFTER BaseException" in out +def start_repl() -> tuple[int, int]: + assert sys.platform != "win32" + + import pty + + # NOTE: this cannot be subprocess.Popen because pty.fork + # does some magic to set the controlling terminal. + # (which I don't know how to replicate... so I copied this + # structure from pty.spawn...) + pid, pty_fd = pty.fork() # type: ignore[attr-defined,unused-ignore] + if pid == 0: # pragma: no cover + os.execlp(sys.executable, *[sys.executable, "-u", "-m", "trio"]) + + return pid, pty_fd + + +def read_until(buffer: bytearray, expected: bytes, fd: int) -> None: + while expected not in strip_some_escape_chars(buffer): + try: + res = os.read(fd, 4096) + except OSError as e: + # process died + assert e.errno == errno.EIO # noqa: PT017 + break + + if res == b"": + break + buffer += res + + if b"_pyrepl.unix_console.InvalidTerminal" in buffer: + pytest.skip("this PTY doesn't support the necessary operations") + + +def write_out(text: bytes, buffer: bytearray, fd: int) -> None: + # for some reason this is required for pypy :( + small_buffer = bytearray() + for i, char in enumerate(text): + os.write(fd, bytes([char])) + read_until(small_buffer, text[: i + 1], fd) + + buffer += small_buffer + + +def strip_some_escape_chars(buffer: bytearray) -> bytes: + # https://superuser.com/a/380778 + return re.sub(b"\x1b" + rb"\[.*?[\x40-\x7E]", b"", buffer) + + +@pytest.mark.skipif( + sys.platform == "win32", + reason="Python doesn't support ConPTY, so we can't make the right environment for the REPL.", +) def test_main_entrypoint() -> None: """ Basic smoke test when running via the package __main__ entrypoint. """ - repl = subprocess.run([sys.executable, "-m", "trio"], input=b"exit()") - assert repl.returncode == 0 + pid, fd = start_repl() + # setup: + buffer = bytearray() + read_until(buffer, b"import trio", fd) + + buffer = buffer.split(b"import trio")[-1] + read_until(buffer, b">>>", fd) -def should_try_newline_injection() -> bool: - if sys.platform != "linux": - return False + # just exit: + write_out(b"exit()", buffer, fd) + os.write(fd, b"\n") - sysctl = pathlib.Path("/proc/sys/dev/tty/legacy_tiocsti") - if not sysctl.exists(): # pragma: no cover - return True + # and flush any output: + buffer = bytearray() + read_until(buffer, b"something impossible", fd) - else: - return sysctl.read_text() == "1" + assert os.waitpid(pid, 0)[1] == 0 @pytest.mark.skipif( - not should_try_newline_injection(), - reason="the ioctl we use is disabled in CI", + sys.platform == "win32", + reason="Python doesn't support ConPTY, so we can't make the right environment for the REPL.", ) -def test_ki_newline_injection() -> None: # TODO: test this line - # TODO: we want to remove this functionality, eg by using vendored - # pyrepls. - assert sys.platform != "win32" - - import pty - - # NOTE: this cannot be subprocess.Popen because pty.fork - # does some magic to set the controlling terminal. - # (which I don't know how to replicate... so I copied this - # structure from pty.spawn...) - pid, pty_fd = pty.fork() # type: ignore[attr-defined,unused-ignore] - if pid == 0: - os.execlp(sys.executable, *[sys.executable, "-u", "-m", "trio"]) +def test_repl_ki() -> None: + pid, fd = start_repl() # setup: - buffer = b"" - while not buffer.endswith(b"import trio\r\n>>> "): - buffer += os.read(pty_fd, 4096) - - # sanity check: - print(buffer.decode()) - buffer = b"" - os.write(pty_fd, b'print("hello!")\n') - while not buffer.endswith(b">>> "): - buffer += os.read(pty_fd, 4096) + buffer = bytearray() + read_until(buffer, b"import trio", fd) - assert buffer.count(b"hello!") == 2 + buffer = buffer.split(b"import trio")[-1] + read_until(buffer, b">>>", fd) - # press ctrl+c + # sanity check: print(buffer.decode()) - buffer = b"" - os.kill(pid, signal.SIGINT) - while not buffer.endswith(b">>> "): - buffer += os.read(pty_fd, 4096) + buffer = bytearray() + write_out(b'print("hello!") # mark', buffer, fd) + os.write(fd, b"\n") + read_until(buffer, b"# mark", fd) - assert b"KeyboardInterrupt" in buffer + buffer = buffer.split(b"# mark")[-1] + read_until(buffer, b">>>", fd) - # press ctrl+c later - print(buffer.decode()) - buffer = b"" - os.write(pty_fd, b'print("hello!")') - os.kill(pid, signal.SIGINT) - while not buffer.endswith(b">>> "): - buffer += os.read(pty_fd, 4096) + assert buffer.count(b"hello!") == 1 - assert b"KeyboardInterrupt" in buffer + # press ctrl+c print(buffer.decode()) - os.close(pty_fd) + if trio._repl.CPYTHON_VENDOR: + # would you believe me if I told you pyrepl has a bug where + # `UnixConsole.get_event(block=False)`... blocks? well anyways, + # that means this part of the test doesn't work. + buffer = bytearray() + os.kill(pid, signal.SIGINT) + read_until(buffer, b">>>", fd) + + assert b"KeyboardInterrupt" in buffer + + # press ctrl+c later + print(buffer.decode()) + buffer = bytearray() + write_out(b'print("hello!") # mark', buffer, fd) + read_until(buffer, b"# mark", fd) + + buffer = bytearray() + os.kill(pid, signal.SIGINT) + read_until(buffer, b">>>", fd) + + assert b"KeyboardInterrupt" in buffer + os.close(fd) os.waitpid(pid, 0)[1] - - -async def test_ki_in_repl() -> None: - async with trio.open_nursery() as nursery: - proc = await nursery.start( - partial( - trio.run_process, - [sys.executable, "-u", "-m", "trio"], - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - stdin=subprocess.PIPE, - creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0, # type: ignore[attr-defined,unused-ignore] - ) - ) - - async with proc.stdout: - # setup - buffer = b"" - async for part in proc.stdout: # pragma: no branch - buffer += part - # TODO: consider making run_process stdout have some universal newlines thing - if buffer.replace(b"\r\n", b"\n").endswith(b"import trio\n>>> "): - break - - # ensure things work - print(buffer.decode()) - buffer = b"" - await proc.stdin.send_all(b'print("hello!")\n') - async for part in proc.stdout: # pragma: no branch - buffer += part - if buffer.endswith(b">>> "): - break - - assert b"hello!" in buffer - print(buffer.decode()) - - # this seems to be necessary on Windows for reasons - # (the parents of process groups ignore ctrl+c by default...) - if sys.platform == "win32": - buffer = b"" - await proc.stdin.send_all( - b"import ctypes; ctypes.windll.kernel32.SetConsoleCtrlHandler(None, False)\n" - ) - async for part in proc.stdout: # pragma: no branch - buffer += part - if buffer.endswith(b">>> "): - break - - print(buffer.decode()) - - # try to decrease flakiness... - buffer = b"" - await proc.stdin.send_all( - b"import coverage; trio.lowlevel.enable_ki_protection(coverage.pytracer.PyTracer._trace)\n" - ) - async for part in proc.stdout: # pragma: no branch - buffer += part - if buffer.endswith(b">>> "): - break - - print(buffer.decode()) - - # ensure that ctrl+c on a prompt works - # NOTE: for some reason, signal.SIGINT doesn't work for this test. - # Using CTRL_C_EVENT is also why we need subprocess.CREATE_NEW_PROCESS_GROUP - signal_sent = signal.CTRL_C_EVENT if sys.platform == "win32" else signal.SIGINT # type: ignore[attr-defined,unused-ignore] - os.kill(proc.pid, signal_sent) - if sys.platform == "win32": - # we rely on EOFError which... doesn't happen with pipes. - # I'm not sure how to fix it... - await proc.stdin.send_all(b"\n") - else: - # we test injection separately - await proc.stdin.send_all(b"\n") - - buffer = b"" - async for part in proc.stdout: # pragma: no branch - buffer += part - if buffer.endswith(b">>> "): - break - - assert b"KeyboardInterrupt" in buffer - - # ensure ctrl+c while a command runs works - print(buffer.decode()) - await proc.stdin.send_all(b'print("READY"); await trio.sleep_forever()\n') - killed = False - buffer = b"" - async for part in proc.stdout: # pragma: no branch - buffer += part - if buffer.replace(b"\r\n", b"\n").endswith(b"READY\n") and not killed: - os.kill(proc.pid, signal_sent) - killed = True - if buffer.endswith(b">>> "): - break - - assert b"trio" in buffer - assert b"KeyboardInterrupt" in buffer - - # make sure it works for sync commands too - # (though this would be hard to break) - print(buffer.decode()) - await proc.stdin.send_all( - b'import time; print("READY"); time.sleep(99999)\n' - ) - killed = False - buffer = b"" - async for part in proc.stdout: # pragma: no branch - buffer += part - if buffer.replace(b"\r\n", b"\n").endswith(b"READY\n") and not killed: - os.kill(proc.pid, signal_sent) - killed = True - if buffer.endswith(b">>> "): - break - - assert b"Traceback" in buffer - assert b"KeyboardInterrupt" in buffer - - print(buffer.decode()) - - # kill the process - nursery.cancel_scope.cancel() diff --git a/test-requirements.in b/test-requirements.in index 272da6a34c..e913c819c6 100644 --- a/test-requirements.in +++ b/test-requirements.in @@ -8,6 +8,7 @@ trustme # for the ssl + DTLS tests pylint # for pylint finding all symbols tests jedi; implementation_name == "cpython" # for jedi code completion tests cryptography>=41.0.0 # cryptography<41 segfaults on pypy3.10 +pyrepl; python_version < "3.13" and implementation_name != "pypy" # Tools black; implementation_name == "cpython" diff --git a/test-requirements.txt b/test-requirements.txt index 4b1aff1376..2551811f50 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -129,6 +129,8 @@ pylint==4.0.5 # via -r test-requirements.in pyopenssl==26.2.0 # via -r test-requirements.in +pyrepl==0.11.4 ; python_full_version < '3.13' and implementation_name != 'pypy' + # via -r test-requirements.in pyright==1.1.409 # via -r test-requirements.in pytest==9.0.3