diff --git a/src/prompt_toolkit/application/application.py b/src/prompt_toolkit/application/application.py index ecf52d176..439c09b49 100644 --- a/src/prompt_toolkit/application/application.py +++ b/src/prompt_toolkit/application/application.py @@ -973,10 +973,13 @@ def _called_from_ipython() -> bool: # In Python 3.12, we can use asyncio.run(loop_factory=...) # For now, use `run_until_complete()`. loop = new_eventloop_with_inputhook(inputhook) - result = loop.run_until_complete(coro) - loop.run_until_complete(loop.shutdown_asyncgens()) - loop.close() - return result + try: + return loop.run_until_complete(coro) + finally: + try: + loop.run_until_complete(loop.shutdown_asyncgens()) + finally: + loop.close() elif _called_from_ipython(): # workaround to make input hooks work for IPython until diff --git a/src/prompt_toolkit/eventloop/inputhook.py b/src/prompt_toolkit/eventloop/inputhook.py index aaeb0cade..95aff892b 100644 --- a/src/prompt_toolkit/eventloop/inputhook.py +++ b/src/prompt_toolkit/eventloop/inputhook.py @@ -27,9 +27,7 @@ import asyncio import os -import select import selectors -import sys import threading from asyncio import AbstractEventLoop, get_running_loop from collections.abc import Callable, Mapping @@ -151,21 +149,11 @@ def input_is_ready() -> bool: # own loop, or by checking the `input_is_ready` function regularly. self.inputhook(InputHookContext(self._r, input_is_ready)) - # Flush the read end of the pipe. - try: - # Before calling 'os.read', call select.select. This is required - # when the gevent monkey patch has been applied. 'os.read' is never - # monkey patched and won't be cooperative, so that would block all - # other select() calls otherwise. - # See: http://www.gevent.org/gevent.os.html - - # Note: On Windows, this is apparently not an issue. - # However, if we would ever want to add a select call, it - # should use `windll.kernel32.WaitForMultipleObjects`, - # because `select.select` can't wait for a pipe on Windows. - if sys.platform != "win32": - select.select([self._r], [], [], None) + # Wait for the real selector to be done before reading from the pipe. + # The selector thread writes to the pipe before setting `ready`. + th.join() + try: os.read(self._r, 1024) except OSError: # This happens when the window resizes and a SIGWINCH was received. @@ -173,8 +161,6 @@ def input_is_ready() -> bool: # Just ignore. pass - # Wait for the real selector to be done. - th.join() assert result is not None return result @@ -182,7 +168,7 @@ def close(self) -> None: """ Clean up resources. """ - if self._r: + if self._r >= 0: os.close(self._r) os.close(self._w) diff --git a/tests/test_inputhook.py b/tests/test_inputhook.py new file mode 100644 index 000000000..4e31e878c --- /dev/null +++ b/tests/test_inputhook.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import time +from types import SimpleNamespace + +import pytest + +import prompt_toolkit.application.application as application_module +import prompt_toolkit.eventloop.inputhook as inputhook_module +from prompt_toolkit.application.application import Application +from prompt_toolkit.eventloop.inputhook import InputHookContext, InputHookSelector +from prompt_toolkit.output import DummyOutput + + +class StubSelector: + def __init__(self) -> None: + self.closed = False + + def select(self, timeout: float | None = None) -> list[tuple[object, int]]: + return [] + + def close(self) -> None: + self.closed = True + + def get_map(self) -> dict[object, object]: + return {} + + +def test_inputhook_selector_handles_large_pipe_descriptors(monkeypatch): + writes = [] + reads = [] + + def read(fd, size): + reads.append((fd, size)) + return b"x" + + fake_os = SimpleNamespace( + pipe=lambda: (1024, 1025), + write=lambda fd, data: writes.append((fd, data)) or len(data), + read=read, + close=lambda fd: None, + ) + monkeypatch.setattr(inputhook_module, "os", fake_os) + monkeypatch.setattr( + inputhook_module, + "select", + SimpleNamespace(select=lambda *args: pytest.fail("select.select was called")), + raising=False, + ) + monkeypatch.setattr( + inputhook_module, "get_running_loop", lambda: SimpleNamespace(_ready=[]) + ) + + def inputhook(context: InputHookContext) -> None: + deadline = time.monotonic() + 1 + while not context.input_is_ready(): + if time.monotonic() >= deadline: + raise AssertionError("selector did not become ready") + time.sleep(0.001) + + selector = InputHookSelector(StubSelector(), inputhook) + + try: + assert selector.select(timeout=0) == [] + assert writes == [(1025, b"x")] + assert reads == [(1024, 1024)] + finally: + selector.close() + + +@pytest.mark.parametrize("pipe_fds", [(0, 1), (1024, 1025)]) +def test_inputhook_selector_close_handles_pipe_descriptors(monkeypatch, pipe_fds): + closed = [] + fake_os = SimpleNamespace(pipe=lambda: pipe_fds, close=closed.append) + monkeypatch.setattr(inputhook_module, "os", fake_os) + selector = InputHookSelector(StubSelector(), lambda context: None) + + selector.close() + selector.close() + + assert closed == list(pipe_fds) + + +@pytest.mark.parametrize("error", [KeyboardInterrupt, EOFError]) +def test_application_run_closes_inputhook_loop_after_interrupt(monkeypatch, error): + run_coro = object() + shutdown_coro = object() + calls = [] + + class StubLoop: + def run_until_complete(self, coro): + if coro is run_coro: + raise error + assert coro is shutdown_coro + calls.append("shutdown") + + def shutdown_asyncgens(self): + return shutdown_coro + + def close(self): + calls.append("close") + + loop = StubLoop() + app = Application(output=DummyOutput()) + app.run_async = lambda **kwargs: run_coro + monkeypatch.setattr( + application_module, "new_eventloop_with_inputhook", lambda inputhook: loop + ) + + with pytest.raises(error): + app.run(inputhook=lambda context: None) + + assert calls == ["shutdown", "close"]