diff --git a/CHANGELOG.md b/CHANGELOG.md index 45bff09ed..37acbd5a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,23 @@ -## 4.2.4 (TBD) +## 4.3.0 (TBD) - Bug Fixes - Fixed the right prompt being redrawn beside every accepted command line in the scrollback. prompt-toolkit includes it in the final frame of each prompt, which is the frame left on the terminal; it is now hidden there, as the bottom toolbar already was, and stays on the live prompt line only + - Fixed an intermittent bottom toolbar flicker when running a command. The toolbar's + prompt-toolkit application now stays running across both the prompt and command execution + instead of stopping and restarting for each one, since prompt-toolkit drops the bottom toolbar + from the final frame of every application run. +- Enhancements + - `enable_bottom_toolbar=True` now keeps the toolbar visible and refreshing during command + execution + - Added an embedded pager which `Cmd.ppaged()` uses while the bottom toolbar is running, so the + toolbar stays visible and refreshing instead of the terminal being handed to an external + pager. It supports vertical and horizontal scrolling, incremental search, and chopped lines on + every platform, including Windows where the default external pager (`more`) always wraps. + Output which already fits on the screen is printed directly rather than paged. Set + `self.use_builtin_pager = False` to keep using the external `pager`/`pager_chop` commands. ## 4.2.3 (September 2, 2026) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index c4a21c831..be12509e2 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -48,6 +48,7 @@ from collections.abc import ( Callable, Iterable, + Iterator, Mapping, Sequence, ) @@ -108,6 +109,7 @@ from . import ( argparse_completer, argparse_utils, + command_toolbar, constants, plugin, utils, @@ -418,7 +420,7 @@ def __init__( This allows CommandSets with custom constructor parameters to be loaded. This also allows the a set of CommandSets to be provided when `auto_load_commands` is set to False - :param enable_bottom_toolbar: if ``True``, enables a bottom toolbar while at the main prompt. + :param enable_bottom_toolbar: if ``True``, enables a bottom toolbar at the main prompt and during commands. Override ``get_bottom_toolbar()`` to define its content. :param enable_rprompt: if ``True``, enables a right prompt while at the main prompt. Override ``get_rprompt()`` to define its content. @@ -561,6 +563,9 @@ def __init__( # custom prompt). Completion and UI logic should reference this variable # to ensure they modify the correct session state. self.active_session = self.main_session + self._command_toolbar: command_toolbar.CommandToolbar | None = None + # Set once a toolbar fails to start, so the failure is reported only once + self._command_toolbar_disabled = False # Commands to exclude from the history command self.exclude_from_history = ["_eof", "history"] @@ -646,6 +651,10 @@ def __init__( if callargs: self._startup_commands.extend(callargs) + # The embedded pager shares the main toolbar. Applications can opt back + # into their configured external pager by setting this to False. + self.use_builtin_pager = enable_bottom_toolbar + # Set the pager(s) for use when displaying output using a pager if sys.platform.startswith("win"): self.pager = self.pager_chop = "more" @@ -1917,12 +1926,17 @@ def ppaged( fits on the screen. A pager is not used inside a script (Python or text) or when output is redirected or piped, and in these cases, output is sent to `poutput`. + While the bottom toolbar is running, the built-in pager keeps it visible and refreshing. + Set ``use_builtin_pager=False`` to use the configured external ``pager`` or ``pager_chop`` + command instead; external pagers temporarily hide the toolbar. Where no toolbar is + running, such as outside the command loop, the external pager is used regardless. + :param chop: True -> causes lines longer than the screen width to be chopped (truncated) rather than wrapped - truncated text is still accessible by scrolling with the right & left arrow keys - chopping is ideal for displaying wide tabular data as is done in utilities like pgcli False -> causes lines longer than the screen width to wrap to the next line - wrapping is ideal when you want to keep users from having to use horizontal scrolling - WARNING: On Windows, the text always wraps regardless of what the chop argument is set to + WARNING: The default external pager on Windows always wraps; the built-in pager supports chopping. :param soft_wrap: Enable soft wrap mode. If True, lines of text will not be word-wrapped or cropped to fit the terminal width. Defaults to True. @@ -1965,11 +1979,20 @@ def ppaged( soft_wrap=soft_wrap, **(rich_print_kwargs if rich_print_kwargs is not None else {}), ) - output_bytes = capture.get().encode("utf-8", "replace") + output = capture.get() + + # Page inside the toolbar's display only when the command loop is already + # running one. Starting one here would seize the terminal for commands run + # outside that loop, which the toolbar is documented not to do. + if self.use_builtin_pager and self._command_toolbar is not None and self._command_toolbar.is_active: + self._command_toolbar.page(output, chop=chop) + return + + output_bytes = output.encode("utf-8", "replace") # Prevent KeyboardInterrupts while in the pager. The pager application will # still receive the SIGINT since it is in the same process group as us. - with self.sigint_protection: + with self.suspend_bottom_toolbar(), self.sigint_protection: import subprocess pipe_proc = subprocess.Popen( # noqa: S602 @@ -2070,7 +2093,7 @@ def ppretty( def get_bottom_toolbar(self) -> AnyFormattedText: """Get the bottom toolbar content. - This method is called by prompt-toolkit while at the main prompt if ``enable_bottom_toolbar`` + This method is called by prompt-toolkit at the main prompt and during commands if ``enable_bottom_toolbar`` was set to ``True`` during initialization. Because prompt-toolkit executes this callback on every UI refresh (such as on every keypress or at scheduled refresh intervals), keeping this function highly optimized is critical to ensuring the CLI remains responsive. @@ -2079,10 +2102,75 @@ def get_bottom_toolbar(self) -> AnyFormattedText: your application. This could be information like the application name, current state, or even a real-time clock. + Inside the command loop this callback runs in a background UI thread, at the prompt as + well as during command execution, because a single application renders both. Protect + shared state with a lock when necessary. The built-in pager shares this toolbar. It is + suspended while another prompt, external pager, or interactive shell owns the terminal. + :return: Content to populate the bottom toolbar. """ return None + @contextlib.contextmanager + def suspend_bottom_toolbar(self) -> Iterator[None]: + """Temporarily hide the command toolbar and give exclusive access to the terminal. + + Use this context manager around application-specific calls to ``input()``, other + terminal UIs, or subprocesses that inherit the terminal. cmd2 automatically suspends + its toolbar for its own input prompts, external pagers, and shell commands. + """ + if self._command_toolbar is None: + yield + else: + with self._command_toolbar.suspend(): + yield + + @contextlib.contextmanager + def _command_toolbar_context(self) -> Iterator[None]: + """Display the toolbar around commands launched by the interactive command loop.""" + if ( + self._command_toolbar is not None + or self._command_toolbar_disabled + or self.main_session.bottom_toolbar is None + or not self._is_tty_session(self.main_session) + ): + yield + return + + try: + with self.sigint_protection: + toolbar = command_toolbar.CommandToolbar(self) + toolbar.start() + self._command_toolbar = toolbar + except Exception as exc: # noqa: BLE001 + # The toolbar is cosmetic, so a display that cannot start must not take + # the command down, nor escape cmdloop() and leave its signal handlers + # installed. Report it once and run without it for the rest of the session. + self._command_toolbar_disabled = True + self.perror(f"Disabling the bottom toolbar during commands: {exc!r}") + yield + return + + try: + yield + finally: + with self.sigint_protection: + try: + toolbar.stop() + finally: + # Always forget a toolbar that has been torn down. Keeping a failed + # one would disable the toolbar for the rest of the session. + self._command_toolbar = None + + @contextlib.contextmanager + def _command_mode_context(self) -> Iterator[None]: + """Show the command display while a command runs, if a toolbar is running.""" + if self._command_toolbar is None: + yield + else: + with self._command_toolbar.command_mode(): + yield + def get_rprompt(self) -> AnyFormattedText: """Provide text to populate the prompt-toolkit right prompt. @@ -3107,6 +3195,7 @@ def onecmd_plus_hooks( return stop + @command_toolbar.suspend_toolbar def _run_cmdfinalization_hooks(self, stop: bool, statement: Statement | None) -> bool: """Run the command finalization hooks.""" if self._initial_termios_settings is not None and self.stdin.isatty(): # type: ignore[unreachable] @@ -3342,32 +3431,50 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState: if shell: kwargs["executable"] = shell - # For any stream that is a StdSim, we will use a pipe so we can capture its output - proc = subprocess.Popen( # noqa: S602 - statement.redirect_to, - stdin=subproc_stdin, - stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout, # type: ignore[unreachable] - stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr, - shell=True, - **kwargs, + # Hand the pipe process the real terminal when there is one to inherit, since it + # may be interactive. Otherwise capture its output so it can pass through a Python + # stream, including the toolbar proxy which prints above the running display. + pipe_stdout = ( + None + if isinstance(self.stdout, utils.StdSim) # type: ignore[unreachable] + else command_toolbar.pipe_target(self.stdout) ) + pipe_stderr = None if isinstance(sys.stderr, utils.StdSim) else command_toolbar.pipe_target(sys.stderr) + + with contextlib.ExitStack() as terminal_stack: + # The toolbar can neither draw nor hold the keyboard while a pipe process owns + # the terminal, so step aside until that process has finished. + if pipe_stdout is not None or pipe_stderr is not None: + terminal_stack.enter_context(self.suspend_bottom_toolbar()) + + proc = subprocess.Popen( # noqa: S602 + statement.redirect_to, + stdin=subproc_stdin, + stdout=subprocess.PIPE if pipe_stdout is None else pipe_stdout, + stderr=subprocess.PIPE if pipe_stderr is None else pipe_stderr, + shell=True, + **kwargs, + ) - # Popen was called with shell=True so the user can chain pipe commands and redirect their output - # like: !ls -l | grep user | wc -l > out.txt. But this makes it difficult to know if the pipe process - # started OK, since the shell itself always starts. Therefore, we will wait a short time and check - # if the pipe process is still running. - with contextlib.suppress(subprocess.TimeoutExpired): - proc.wait(0.2) + # Popen was called with shell=True so the user can chain pipe commands and redirect their output + # like: !ls -l | grep user | wc -l > out.txt. But this makes it difficult to know if the pipe process + # started OK, since the shell itself always starts. Therefore, we will wait a short time and check + # if the pipe process is still running. + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(0.2) + + # Check if the pipe process already exited + if proc.returncode is not None: + subproc_stdin.close() + new_stdout.close() + raise RedirectionError(f"Pipe process exited with code {proc.returncode} before command could run") + redir_saved_state.redirecting = True + cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr) - # Check if the pipe process already exited - if proc.returncode is not None: - subproc_stdin.close() - new_stdout.close() - raise RedirectionError(f"Pipe process exited with code {proc.returncode} before command could run") - redir_saved_state.redirecting = True - cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr) + self.stdout = new_stdout - self.stdout = new_stdout + # Hold the suspension open until _restore_output() reaps the pipe process. + redir_saved_state.toolbar_suspension = terminal_stack.pop_all() elif statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND): if statement.redirect_to: @@ -3419,29 +3526,35 @@ def _restore_output(self, statement: Statement, saved_redir_state: utils.Redirec :param statement: Statement object which contains the parsed input from the user :param saved_redir_state: contains information needed to restore state data """ - if saved_redir_state.redirecting: - # If we redirected output to the clipboard - if ( - statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND) - and not statement.redirect_to - ): - self.stdout.seek(0) - write_to_paste_buffer(self.stdout.read()) + # The toolbar gets the terminal back once the pipe process is done with it. + with contextlib.ExitStack() as terminal_stack: + if saved_redir_state.toolbar_suspension is not None: + terminal_stack.callback(saved_redir_state.toolbar_suspension.close) + saved_redir_state.toolbar_suspension = None + + if saved_redir_state.redirecting: + # If we redirected output to the clipboard + if ( + statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND) + and not statement.redirect_to + ): + self.stdout.seek(0) + write_to_paste_buffer(self.stdout.read()) - with contextlib.suppress(BrokenPipeError): - # Close the file or pipe that stdout was redirected to - self.stdout.close() + with contextlib.suppress(BrokenPipeError): + # Close the file or pipe that stdout was redirected to + self.stdout.close() - # Restore self.stdout - self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout) + # Restore self.stdout + self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout) - # Check if we need to wait for the process being piped to - if self._cur_pipe_proc_reader is not None: - self._cur_pipe_proc_reader.wait() + # Check if we need to wait for the process being piped to + if self._cur_pipe_proc_reader is not None: + self._cur_pipe_proc_reader.wait() - # These are restored regardless of whether the command redirected - self._cur_pipe_proc_reader = saved_redir_state.saved_pipe_proc_reader - self._redirecting = saved_redir_state.saved_redirecting + # These are restored regardless of whether the command redirected + self._cur_pipe_proc_reader = saved_redir_state.saved_pipe_proc_reader + self._redirecting = saved_redir_state.saved_redirecting def get_command_func(self, command: str) -> BoundCommandFunc[...] | None: """Get the bound command function for a command. @@ -3543,6 +3656,7 @@ def _is_tty_session(session: PromptSession[str]) -> bool: # a DummyOutput. return not isinstance(session.input, DummyInput) + @command_toolbar.suspend_toolbar def _read_raw_input( self, prompt: Callable[[], ANSI | str] | ANSI | str, @@ -3804,6 +3918,11 @@ def _pre_prompt() -> None: self._alert_condition.notify_all() try: + # A running command display already owns the terminal and the application. + # Reading through it keeps that single run alive across the prompt, so the + # toolbar is never erased by an end-of-run frame. + if self._command_toolbar is not None and self._command_toolbar.is_active: + return self._command_toolbar.read_line(prompt_to_use, pre_run=_pre_prompt) return self._read_raw_input( prompt=prompt_to_use, session=self.main_session, @@ -3823,22 +3942,31 @@ def _cmdloop(self) -> None: This serves the same role as cmd.cmdloop(). """ try: - # Run startup commands - stop = self.runcmds_plus_hooks(self._startup_commands) - self._startup_commands.clear() + # Hold one command display open for the whole loop. Starting and stopping + # it per command would end its application each time, and prompt-toolkit + # drops the bottom toolbar from the final frame of every run. + with self._command_toolbar_context(): + # Run startup commands + if self._startup_commands: + with self._command_mode_context(): + stop = self.runcmds_plus_hooks(self._startup_commands) + else: + stop = False + self._startup_commands.clear() - while not stop: - # Get commands from user - try: - line = self._read_command_line(self.prompt) - except KeyboardInterrupt: - self.poutput("^C") - line = "" - except EOFError: - line = "_eof" + while not stop: + # Get commands from user + try: + line = self._read_command_line(self.prompt) + except KeyboardInterrupt: + self.poutput("^C") + line = "" + except EOFError: + line = "_eof" - # Run the command along with all associated pre and post hooks - stop = self.onecmd_plus_hooks(line) + # Run the command along with all associated pre and post hooks + with self._command_mode_context(): + stop = self.onecmd_plus_hooks(line) finally: with self.sigint_protection: # Shut down the alert thread. @@ -4668,6 +4796,7 @@ def do_quit(self, _: argparse.Namespace) -> bool | None: self.last_result = True return True + @command_toolbar.suspend_toolbar def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], prompt: str = "Your choice? ") -> Any: """Present a menu to the user. @@ -4876,6 +5005,7 @@ def _build_shell_parser(cls) -> Cmd2ArgumentParser: # Preserve quotes since we are passing these strings to the shell @with_argparser(_build_shell_parser, preserve_quotes=True) + @command_toolbar.suspend_toolbar def do_shell(self, args: argparse.Namespace) -> None: """Execute a command as if at the OS prompt.""" import signal @@ -4994,6 +5124,7 @@ def _restore_cmd2_env(self, cmd2_env: _SavedCmd2Env) -> None: readline.set_completer(cmd2_env.completer) + @command_toolbar.suspend_toolbar def _run_python(self, *, pyscript: str | None = None) -> bool | None: """Run an interactive Python shell or execute a pyscript file. @@ -5205,6 +5336,7 @@ def _build_ipython_parser() -> Cmd2ArgumentParser: return argparse_utils.DEFAULT_ARGUMENT_PARSER(description="Run an interactive IPython shell.") @with_argparser(_build_ipython_parser) + @command_toolbar.suspend_toolbar def do_ipy(self, _: argparse.Namespace) -> bool | None: # pragma: no cover """Run an interactive IPython shell. @@ -5918,20 +6050,22 @@ def cmdloop(self, intro: RenderableType = "") -> int: self.poutput(self.intro) # And then call _cmdloop() to enter the main loop - self._cmdloop() + try: + self._cmdloop() + finally: + # Restore original signal handlers however the loop ended. Leaving cmd2's + # handlers installed would outlive the application in its host process. + signal.signal(signal.SIGINT, original_sigint_handler) + + if not sys.platform.startswith("win"): + signal.signal(signal.SIGHUP, original_sighup_handler) + signal.signal(signal.SIGTERM, original_sigterm_handler) # Run the postloop() no matter what for func in self._postloop_hooks: func() self.postloop() - # Restore original signal handlers - signal.signal(signal.SIGINT, original_sigint_handler) - - if not sys.platform.startswith("win"): - signal.signal(signal.SIGHUP, original_sighup_handler) - signal.signal(signal.SIGTERM, original_sigterm_handler) - return self.exit_code ### diff --git a/cmd2/command_toolbar.py b/cmd2/command_toolbar.py new file mode 100644 index 000000000..925160d77 --- /dev/null +++ b/cmd2/command_toolbar.py @@ -0,0 +1,603 @@ +"""Internal support for displaying a toolbar during synchronous commands.""" + +import codecs +import contextlib +import contextvars +import functools +import os +import signal +import sys +import threading +from collections.abc import Callable, Iterator +from concurrent.futures import Future +from concurrent.futures import TimeoutError as FutureTimeoutError +from typing import TYPE_CHECKING, Any, TextIO, TypeVar, cast + +from prompt_toolkit.application import Application, create_app_session +from prompt_toolkit.enums import EditingMode +from prompt_toolkit.filters import Condition, to_filter +from prompt_toolkit.formatted_text import ANSI +from prompt_toolkit.input.typeahead import get_typeahead, store_typeahead +from prompt_toolkit.key_binding import KeyBindings, merge_key_bindings +from prompt_toolkit.key_binding.key_processor import KeyPress, KeyPressEvent +from prompt_toolkit.layout import HSplit, Layout, Window +from prompt_toolkit.layout.containers import ConditionalContainer +from prompt_toolkit.patch_stdout import StdoutProxy +from prompt_toolkit.utils import suspend_to_background_supported + +from .pager import Pager, output_fits + +if TYPE_CHECKING: + from prompt_toolkit.buffer import Buffer + from prompt_toolkit.key_binding.key_bindings import KeyBindingsBase + + from .cmd2 import Cmd + + #: The prompt text, or a callable returning it for a dynamic prompt. + PromptMessage = Callable[[], ANSI | str] | ANSI | str + +_F = TypeVar("_F", bound=Callable[..., Any]) +_R = TypeVar("_R") + + +def suspend_toolbar(func: _F) -> _F: + """Give a method exclusive access to the terminal.""" + + @functools.wraps(func) + def wrapped(self: "Cmd", *args: Any, **kwargs: Any) -> Any: + with self.suspend_bottom_toolbar(): + return func(self, *args, **kwargs) + + return cast(_F, wrapped) + + +def pipe_target(stream: Any) -> Any: + """Return the stream a pipe process can inherit, or ``None`` if its output must be captured. + + A pipe process may be interactive, such as ``less`` or ``fzf``, so it needs the real + terminal rather than a stream this process reads on its behalf. Look through a + :class:`ToolbarStream` wrapper, but only hand back a stream owning a file descriptor. + """ + if isinstance(stream, ToolbarStream): + stream = stream.original + try: + stream.fileno() + except (AttributeError, OSError): + # io.UnsupportedOperation, raised by streams like io.StringIO, subclasses OSError. + return None + return stream + + +class _ContextStdoutProxy(StdoutProxy): + """Keep stdout's flush worker in the toolbar's isolated application session.""" + + def _start_write_thread(self) -> threading.Thread: + context = contextvars.copy_context() + thread = threading.Thread(target=context.run, args=(self._write_thread,), daemon=True) + thread.start() + return thread + + +class ToolbarStream: + """Keep a stable stream identity across suspensions and cmd2 redirections.""" + + def __init__(self, original: TextIO, lock: "threading.RLock") -> None: + """Wrap a terminal stream while preserving its ordinary file attributes.""" + self.original = original + self.proxy: StdoutProxy | None = None + # Shared with the toolbar so a write from another thread cannot land on a proxy + # that is being closed. Such a write is accepted by the dead proxy and discarded. + self._lock = lock + self.buffer = _ToolbarBuffer(self) + + def write(self, data: str) -> int: + """Write above the toolbar, or directly while the toolbar is suspended.""" + with self._lock: + return (self.proxy or self.original).write(data) + + def flush(self) -> None: + """Flush the currently active output stream.""" + with self._lock: + (self.proxy or self.original).flush() + + def __getattr__(self, name: str) -> Any: + """Delegate file attributes to the original terminal stream.""" + return getattr(self.original, name) + + +class _ToolbarBuffer: + """Decode subprocess output incrementally, including split Unicode characters.""" + + def __init__(self, stream: ToolbarStream) -> None: + self.stream = stream + self._decoder = codecs.getincrementaldecoder(stream.original.encoding or "utf-8")(errors="replace") + self._lock = threading.Lock() + + def write(self, data: bytes) -> int: + with self._lock: + self.stream.write(self._decoder.decode(data)) + return len(data) + + def flush(self) -> None: + self.stream.flush() + + def finish(self) -> None: + with self._lock: + self.stream.write(self._decoder.decode(b"", final=True)) + self._decoder.reset() + + +class CommandToolbar: + """Borrow the main prompt's application while a command runs on the main thread. + + The display owns terminal input so it can receive cursor position reports. Keys + typed during execution are saved for the next prompt; Ctrl-C is sent to cmd2's + normal signal handler. Terminal output goes through prompt-toolkit's stdout + proxy so that it appears above the toolbar. + """ + + def __init__(self, cmd: "Cmd") -> None: + """Configure a command display using the main prompt's terminal and settings.""" + self.cmd = cmd + self._stack: contextlib.ExitStack | None = None + self._keys: list[KeyPress] = [] + self._error: BaseException | None = None + self._ready = threading.Event() + self._thread: threading.Thread | None = None + self._streams: list[ToolbarStream] = [] + self._proxy: StdoutProxy | None = None + self._lock = threading.RLock() + self._pausing = False + + session = cmd.main_session + self.app = session.app + # PromptSession has no public hook for replacing just its input area. + # Keep this small dependency on its layout shape in one place, and fail + # explicitly if upstream changes it. Reuse the actual toolbar container, + # including its visibility filter and support for multiline toolbars. + root = session.layout.container + if not isinstance(root, HSplit): + raise TypeError("Unsupported PromptSession layout") + self.toolbar = root.children[-1] + if not ( + isinstance(self.toolbar, ConditionalContainer) + and isinstance(self.toolbar.content, Window) + and self.toolbar.content.style == "class:bottom-toolbar" + ): + raise RuntimeError("Cannot locate PromptSession bottom toolbar") + self._layout = Layout(HSplit([Window(height=0), Window(), self.toolbar])) + self._display_stack: contextlib.ExitStack | None = None + bindings = KeyBindings() + + @bindings.add("") + def save_key(event: KeyPressEvent) -> None: + self._keys.extend(event.key_sequence) + + @bindings.add("c-c") + def interrupt(event: KeyPressEvent) -> None: # noqa: ARG001 + # Match the terminal's normal Ctrl-C input flush: cancelled typeahead + # must not become a command when the main prompt resumes. + self._keys.clear() + if sys.platform == "win32": + # os.kill(..., SIGINT) terminates the process on Windows instead + # of dispatching Python's signal handler. This reaches only this + # process, so a console subprocess started by a command keeps + # running until it is waited on. + import _thread + + _thread.interrupt_main() + else: + # Raw mode clears ISIG, so no signal is generated for us. Signal the + # foreground process group the way the terminal driver would, since a + # command may be waiting on a subprocess that shares this group. Pipe + # processes are excluded because cmd2 starts them in their own session + # and forwards to them from sigint_handler(). + os.killpg(os.getpgrp(), signal.SIGINT) + + @bindings.add( + "c-z", + filter=Condition(lambda: suspend_to_background_supported() and to_filter(session.enable_suspend)()), + ) + def suspend(event: KeyPressEvent) -> None: + # This restores cooked mode before stopping the process group and + # redraws the toolbar after the process resumes. + event.app.suspend_to_background() + + self._bindings = bindings + self._suspend_binding = suspend + + # Prompt mode reuses the PromptSession's own layout, so completion menus, + # auto-suggestions, and the rprompt all keep working. Only the accept path + # changes: it hands the line to the command thread instead of ending the run. + self._prompt_layout = session.layout + + prompt_bindings = KeyBindings() + + @prompt_bindings.add("c-c") + def prompt_interrupt(event: KeyPressEvent) -> None: # noqa: ARG001 + # PromptSession aborts by calling app.exit(), which would end the run that + # draws the toolbar. Fail the pending read instead and keep rendering. + self._fail_line(KeyboardInterrupt()) + + @prompt_bindings.add("c-d", filter=Condition(lambda: not self.app.current_buffer.text)) + def prompt_eof(event: KeyPressEvent) -> None: # noqa: ARG001 + self._fail_line(EOFError()) + + # session.app.key_bindings is captured before _resume() swaps in the command + # display's set, so it is the PromptSession's own: completion, history, and + # cmd2's bindings. Merging the overrides last lets them win, because + # KeyProcessor calls matches[-1]. + session_bindings = session.app.key_bindings + self._prompt_bindings: KeyBindingsBase = ( + merge_key_bindings([session_bindings, prompt_bindings]) if session_bindings is not None else prompt_bindings + ) + self._line: Future[str] | None = None + # Only swapped in for the duration of read_line(). session.prompt() still runs + # nested prompts and the no-toolbar fallback, and those rely on the stock + # handler's app.exit() to return at all. + self._session_accept = session.default_buffer.accept_handler + + def _after_render(self, app: Application[str]) -> None: # noqa: ARG002 + self._ready.set() + + def start(self) -> None: + """Start rendering and protect terminal output.""" + stack = contextlib.ExitStack() + self._stack = stack + self._ready.clear() + self._error = None + try: + stack.enter_context(create_app_session(input=self.app.input, output=self.app.output)) + # Only replace terminal streams. In particular, preserve redirected stderr + # and self.stdout when a nested command has redirected its output to a file. + for obj, name in ((self.cmd, "stdout"), (sys, "stdout"), (sys, "stderr")): + stream = getattr(obj, name) + if stream.isatty(): + wrapper = ToolbarStream(stream, self._lock) + self._streams.append(wrapper) + setattr(obj, name, cast(TextIO, wrapper)) + stack.callback(self._restore_stream, obj, name, wrapper) + self._resume() + except BaseException: + self.stop() + raise + + @staticmethod + def _restore_stream(obj: Any, name: str, stream: ToolbarStream) -> None: + if getattr(obj, name) is stream: + setattr(obj, name, stream.original) + + def _resume(self) -> None: + self._ready.clear() + self._error = None + stack = self._display_stack = contextlib.ExitStack() + for name, value in (("layout", self._layout), ("key_bindings", self._bindings), ("erase_when_done", True)): + stack.callback(setattr, self.app, name, getattr(self.app, name)) + setattr(self.app, name, value) + self.app.after_render += self._after_render + stack.callback(self.app.after_render.remove_handler, self._after_render) + context = contextvars.copy_context() + + def run() -> None: + try: + self.app.run(handle_sigint=False, set_exception_handler=False) + except EOFError: + pass + except BaseException as exc: # noqa: BLE001 + # Propagate startup/render failures to the command thread. + self._error = exc + finally: + self._ready.set() + self._app_exited() + + self._thread = threading.Thread(target=context.run, args=(run,), name="cmd2-toolbar", daemon=True) + self._thread.start() + self._ready.wait() + if self._error is not None: + raise self._error + # The worker already combines queued writes. A batching sleep would also + # delay close(), which runs at each command finalization boundary. + proxy = _ContextStdoutProxy(raw=True, sleep_between_writes=0) + with self._lock: + self._proxy = proxy + for stream in self._streams: + stream.proxy = proxy + + def _app_exited(self) -> None: + """Give the terminal back to the streams when the display stops on its own. + + ``_ready`` is set as soon as the first frame renders, so a failure after that is + never seen by the command thread waiting in ``_resume()``. The display is gone at + that point and its stdout proxy can no longer reach the terminal, so anything + written through it would be discarded without a trace. + """ + if self._pausing: + # A deliberate pause restores the streams itself, in the right order. + return + + with self._lock: + # Leave self._proxy set so that the next _pause() still drains and closes + # it. With the display gone, its worker writes to the terminal directly. + started = self._proxy is not None + for stream in self._streams: + stream.proxy = None + + # A proxy exists only once _resume() has handed startup failures to the command + # thread, so reporting here does not duplicate the exception it raises. + if started and self._error is not None: + self.cmd.perror(f"Bottom toolbar stopped after an error: {self._error!r}") + + def _exit(self) -> None: + """Stop the display unless it has already stopped on its own. + + Application.exit() raises once the result is set, and this runs later than the + check that scheduled it. Any exception here would reach the loop's default + handler, which prints a traceback over the terminal. + + Output queued before this does not need draining: Application.run_async() waits + for cursor position reports and for run_in_terminal() calls still in flight + before its loop closes. + """ + if self.app.is_running and not self.app.is_done: + self.app.exit() + + def _pause(self) -> None: + self._pausing = True + try: + try: + # Hold off other threads while the proxy drains so their output is never + # handed to a proxy whose worker has already stopped. Writes that arrive + # after this go straight to the terminal, still in order. + with self._lock: + try: + if self._proxy is not None: + self._proxy.flush() + self._proxy.close() + finally: + self._proxy = None + for stream in self._streams: + stream.proxy = None + finally: + # The lock is released before joining, since the toolbar thread may be + # blocked writing through a stream that is waiting on it. + if self.app.is_running and self.app.loop is not None: + self.app.loop.call_soon_threadsafe(self._exit) + if self._thread is not None: + self._thread.join() + self._thread = None + # Return the borrowed application to the main prompt, including on + # proxy failures. The upstream toolbar owned a separate application. + if self._display_stack is not None: + self._display_stack.close() + self._display_stack = None + # Application.run() saves its unprocessed queue before the thread + # exits. Those keys arrived after the ones handled by save_key(). + pending_keys = get_typeahead(self.app.input) + store_typeahead(self.app.input, self._keys + pending_keys) + self._keys.clear() + finally: + self._pausing = False + + def stop(self) -> None: + """Flush output, stop rendering, and restore the terminal and its streams.""" + try: + for stream in self._streams: + stream.buffer.finish() + self._pause() + finally: + if self._stack is not None: + self._stack.close() + self._stack = None + + @property + def is_active(self) -> bool: + """Whether the display currently owns the terminal.""" + return self._proxy is not None and self.app.is_running + + def _call_in_ui(self, func: Callable[[], _R]) -> _R: + """Change UI state on its event loop, propagating failures to the command.""" + result: Future[_R] = Future() + + def call() -> None: + try: + value = func() + except BaseException as exc: # noqa: BLE001 + result.set_exception(exc) + else: + result.set_result(value) + + if self.app.loop is None: + raise RuntimeError("Toolbar is not running") + self.app.loop.call_soon_threadsafe(call) + while True: + try: + value = result.result(timeout=0.1) + except FutureTimeoutError: + if result.done(): + raise + self._check_running() + else: + return value + + def _check_running(self) -> None: + if self._thread is None or not self._thread.is_alive(): + if self._error is not None: + raise self._error + raise EOFError + + def _accept_line(self, buff: "Buffer") -> bool: + """Hand an accepted line to the waiting command thread without ending the run.""" + if self._line is not None and not self._line.done(): + self._line.set_result(buff.document.text) + # Discard the text so the live prompt region stops showing the accepted line. + # _echo_accepted() is what commits it, and leaving it here too would draw it + # twice. validate_and_handle() still records history before resetting. + return False + + def _fail_line(self, error: BaseException) -> None: + """Abort the pending read_line() with Ctrl-C's or Ctrl-D's exception.""" + if self._line is not None and not self._line.done(): + self._line.set_exception(error) + + def read_line( + self, + message: "PromptMessage", + *, + pre_run: Callable[[], None] | None = None, + ) -> str: + """Read one line at the prompt while the toolbar's application keeps running. + + :param message: the prompt text, or a callable returning it for a dynamic prompt + :param pre_run: optional callback run on the UI thread once the prompt is shown + :return: the accepted line + :raises KeyboardInterrupt: if the user pressed Ctrl-C + :raises EOFError: if the user pressed Ctrl-D on an empty line + """ + line: Future[str] = Future() + self._line = line + session = self.cmd.main_session + + def enter() -> None: + session.message = message + session.default_buffer.reset() + session.default_buffer.accept_handler = self._accept_line + self.app.layout = self._prompt_layout + self.app.layout.focus(session.default_buffer) + self.app.key_bindings = self._prompt_bindings + if pre_run is not None: + pre_run() + self.app.invalidate() + + # Keys typed while the command ran were swallowed by the save_key binding + # so they could not become commands mid-execution. Replay them now that a + # prompt is on screen and its bindings are active, in arrival order. + if self._keys: + pending, self._keys = self._keys, [] + self.app.key_processor.feed_multiple(pending) + self.app.key_processor.process_keys() + + try: + self._call_in_ui(enter) + while True: + try: + text = line.result(timeout=0.1) + except FutureTimeoutError: + if line.done(): + raise + self._check_running() + else: + self._echo_accepted(message, text) + return text + finally: + self._line = None + session.default_buffer.accept_handler = self._session_accept + + @staticmethod + def _resolve_message(message: "PromptMessage") -> str: + """Render the prompt to text for the scrollback echo.""" + resolved = message if isinstance(message, (ANSI, str)) else message() + return resolved.value if isinstance(resolved, ANSI) else resolved + + def _echo_accepted(self, message: "PromptMessage", text: str) -> None: + """Commit the accepted line above the toolbar. + + prompt-toolkit normally leaves the prompt line in the scrollback when its run + ends. Prompt mode never ends a run, so write the line through the same stdout + proxy that keeps command output above the toolbar. + """ + self.cmd.stdout.write(f"{self._resolve_message(message)}{text}\n") + self.cmd.stdout.flush() + + @contextlib.contextmanager + def command_mode(self) -> Iterator[None]: + """Swap to the toolbar-only layout while a command runs on the main thread.""" + + def enter() -> None: + self.app.layout = self._layout + self.app.key_bindings = self._bindings + self.app.invalidate() + + self._call_in_ui(enter) + try: + yield + finally: + # Restore the command display if a nested prompt or the pager swapped the + # layout out. When the display has already stopped, _pause() has restored + # the borrowed application's own state and there is nothing to put back. + if self._thread is not None and self._thread.is_alive(): + self._call_in_ui(enter) + + def page(self, text: str, *, chop: bool) -> None: + """Show a pager above the same toolbar without starting another input reader.""" + size = self.app.output.get_size() + # Measuring the toolbar can invoke its callback; keep that work on the + # UI thread along with rendering and layout changes. + toolbar_height = self._call_in_ui(lambda: self.toolbar.preferred_height(size.columns, size.rows).preferred) + if output_fits(text, size.columns, max(0, size.rows - toolbar_height), chop=chop): + self.cmd.stdout.write(text) + self.cmd.stdout.flush() + return + + pager = Pager(text, chop=chop) + pager.bindings.add( + "c-z", + filter=Condition(lambda: suspend_to_background_supported() and to_filter(self.cmd.main_session.enable_suspend)()), + )(self._suspend_binding) + layout = Layout(HSplit([pager.container, self.toolbar]), focused_element=pager.text) + previous = (self.app.layout, self.app.key_bindings, self.app.editing_mode, self.app.full_screen) + entered = False + + def enter() -> None: + nonlocal entered + entered = True + self.app.renderer.erase() + self.app.layout = layout + self.app.key_bindings = pager.bindings + self.app.editing_mode = EditingMode.EMACS + self.app.full_screen = self.app.renderer.full_screen = True + self.app.invalidate() + + def leave() -> None: + nonlocal entered + if not entered: + return + entered = False + self.app.renderer.erase() + self.app.layout, self.app.key_bindings, self.app.editing_mode, self.app.full_screen = previous + self.app.renderer.full_screen = self.app.full_screen + self.app.renderer.request_absolute_cursor_position() + self.app.invalidate() + + def close() -> None: + # Switch bindings before the next key is processed, preserving + # typeahead sent in the same terminal read as the pager's quit key. + leave() + pager.closed.set() + + pager.on_close = close + + try: + self._call_in_ui(enter) + while not pager.closed.wait(0.1): + self._check_running() + finally: + if self._thread is not None and self._thread.is_alive(): + self._call_in_ui(leave) + else: + # The application's shutdown already reset the renderer. + self.app.layout, self.app.key_bindings, self.app.editing_mode, self.app.full_screen = previous + self.app.renderer.full_screen = self.app.full_screen + + @contextlib.contextmanager + def suspend(self) -> Iterator[None]: + """Temporarily restore ordinary terminal access, including nested suspensions.""" + if self._proxy is None: + yield + return + with self.cmd.sigint_protection: + self._pause() + try: + yield + finally: + with self.cmd.sigint_protection: + self._resume() diff --git a/cmd2/pager.py b/cmd2/pager.py new file mode 100644 index 000000000..438dd4e48 --- /dev/null +++ b/cmd2/pager.py @@ -0,0 +1,229 @@ +"""A pager view hosted by the main prompt-toolkit application.""" + +import re +import threading +from collections.abc import Callable +from functools import partial + +from prompt_toolkit.document import Document +from prompt_toolkit.filters import has_focus, is_searching, to_filter +from prompt_toolkit.formatted_text import ANSI, fragment_list_to_text, to_formatted_text +from prompt_toolkit.formatted_text.base import StyleAndTextTuples +from prompt_toolkit.formatted_text.utils import split_lines +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.key_binding.bindings import search +from prompt_toolkit.key_binding.key_processor import KeyPressEvent +from prompt_toolkit.layout import HSplit, Window +from prompt_toolkit.layout.containers import ConditionalContainer +from prompt_toolkit.layout.controls import FormattedTextControl, UIContent +from prompt_toolkit.lexers import Lexer +from prompt_toolkit.search import SearchDirection, start_search +from prompt_toolkit.utils import get_cwidth +from prompt_toolkit.widgets import SearchToolbar, TextArea + +_OSC_RE = re.compile(r"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)") + + +def _fragments(text: str) -> StyleAndTextTuples: + """Parse captured output, dropping the trailing newline that ends its last line.""" + # prompt-toolkit's ANSI parser does not handle OSC sequences, including Rich + # hyperlinks. Remove their metadata while retaining visible text and SGR styles. + text = _OSC_RE.sub("", text) + return to_formatted_text(ANSI(text.removesuffix("\n"))) + + +def output_fits(text: str, columns: int, rows: int, *, chop: bool) -> bool: + """Check rendered line heights, including wrapping and wide Unicode characters. + + Measuring the text itself keeps output that needs no scrolling from paying for a + Pager's widgets and key bindings, which would only be built to be thrown away. + """ + lines = fragment_list_to_text(_fragments(text)).split("\n") + if chop: + # Even one wide line needs a pager so its hidden columns remain + # accessible through horizontal scrolling. + return len(lines) <= rows and all(get_cwidth(line) <= columns for line in lines) + # Reuse prompt-toolkit's own wrapping arithmetic so this matches what a Pager + # would render, without building a control to ask on the UI thread. + content = UIContent(get_line=lambda number: [("", lines[number])], line_count=len(lines)) + height = 0 + for line in range(content.line_count): + height += content.get_height_for_line(line, columns, None) + if height > rows: + return False + return True + + +class _AnsiLexer(Lexer): + """Preserve captured Rich styles while searching and scrolling plain text.""" + + def __init__(self, fragments: StyleAndTextTuples) -> None: + self.lines = list(split_lines(fragments)) + + def lex_document(self, document: Document) -> Callable[[int], StyleAndTextTuples]: # noqa: ARG002 + def get_line(number: int) -> StyleAndTextTuples: + return self.lines[number] if number < len(self.lines) else [] + + return get_line + + +class Pager: + """Scrollable, searchable output; the host supplies the persistent toolbar.""" + + def __init__(self, text: str, *, chop: bool) -> None: + """Build an independent view without creating an application or input reader.""" + self.closed = threading.Event() + self.on_close: Callable[[], None] = self.closed.set + self.chop = chop + fragments = _fragments(text) + self.search = SearchToolbar() + self.text = TextArea( + text=fragment_list_to_text(fragments), + lexer=_AnsiLexer(fragments), + read_only=True, + wrap_lines=not chop, + search_field=self.search, + ) + self.text.window.always_hide_cursor = to_filter(True) + self.container = HSplit( + [ + self.text, + self.search, + ConditionalContainer( + Window( + FormattedTextControl(" Space/PgDn: next b/PgUp: back /: search n: next match q: quit"), + height=1, + style="class:bottom-toolbar", + ), + filter=~is_searching, + ), + ] + ) + self.bindings = bindings = KeyBindings() + focused = has_focus(self.text) + + @bindings.add("", filter=focused) + def ignore(event: KeyPressEvent) -> None: + # Pager keystrokes must not become commands at the next prompt. + pass + + @bindings.add("q", filter=focused) + @bindings.add("escape", filter=focused, eager=True) + @bindings.add("c-c", filter=focused) + def close(event: KeyPressEvent) -> None: # noqa: ARG001 + self.on_close() + + for keys, pages in ( + ((" ", "pagedown", "f", "c-f"), 1.0), + (("b", "pageup", "c-b"), -1.0), + (("d", "c-d"), 0.5), + (("u", "c-u"), -0.5), + ): + for key in keys: + bindings.add(key, filter=focused)(partial(self._scroll_page, pages=pages)) + + @bindings.add("j", filter=focused) + @bindings.add("down", filter=focused) + @bindings.add("enter", filter=focused) + def down(event: KeyPressEvent) -> None: + self._scroll(event, 1) + + @bindings.add("k", filter=focused) + @bindings.add("up", filter=focused) + def up(event: KeyPressEvent) -> None: + self._scroll(event, -1) + + @bindings.add("right", filter=focused) + @bindings.add("l", filter=focused) + def right(event: KeyPressEvent) -> None: + self._scroll_horizontal(event, 1) + + @bindings.add("left", filter=focused) + @bindings.add("h", filter=focused) + def left(event: KeyPressEvent) -> None: + self._scroll_horizontal(event, -1) + + @bindings.add("g", filter=focused) + @bindings.add("home", filter=focused) + def first(event: KeyPressEvent) -> None: + event.current_buffer.cursor_position = 0 + + @bindings.add("G", filter=focused) + @bindings.add("end", filter=focused) + def last(event: KeyPressEvent) -> None: + event.current_buffer.cursor_position = len(event.current_buffer.text) + + @bindings.add("/", filter=focused) + def find(event: KeyPressEvent) -> None: # noqa: ARG001 + start_search(direction=SearchDirection.FORWARD) + + @bindings.add("?", filter=focused) + def find_backwards(event: KeyPressEvent) -> None: # noqa: ARG001 + start_search(direction=SearchDirection.BACKWARD) + + @bindings.add("n", filter=focused) + def next_match(event: KeyPressEvent) -> None: + event.current_buffer.apply_search(event.app.current_search_state, include_current_position=False) + + @bindings.add("N", filter=focused) + def previous_match(event: KeyPressEvent) -> None: + event.current_buffer.apply_search(~event.app.current_search_state, include_current_position=False) + + # Explicit search bindings also work when the main prompt uses Vi mode. + bindings.add("enter", filter=is_searching)(search.accept_search) + bindings.add("escape", filter=is_searching, eager=True)(search.abort_search) + bindings.add("c-c", filter=is_searching)(search.abort_search) + + @staticmethod + def _column_at_width(line: str, width: int) -> int: + used = 0 + for column, char in enumerate(line): + if used >= width: + return column + used += get_cwidth(char) + return len(line) + + def _scroll_page(self, event: KeyPressEvent, *, pages: float) -> None: + info = self.text.window.render_info + if info is not None: + amount = max(1, int(max(1, info.window_height - 1) * abs(pages))) + self._scroll(event, amount if pages > 0 else -amount) + + def _scroll(self, event: KeyPressEvent, rows: int) -> None: + """Move by display rows, including within lines taller than the viewport.""" + info = self.text.window.render_info + if info is None or info.window_width == 0: + return + document = event.current_buffer.document + line = document.cursor_position_row + wrapped_row = 0 if self.chop else get_cwidth(document.current_line_before_cursor) // info.window_width + target = wrapped_row + rows + + def height(number: int) -> int: + return 1 if self.chop else info.get_height_for_line(number) + + while target < 0 and line > 0: + line -= 1 + target += height(line) + while target >= height(line) and line < document.line_count - 1: + target -= height(line) + line += 1 + target = max(0, min(target, height(line) - 1)) + # Chopped lines never wrap, so leave the cursor at the column the reader + # scrolled to. Moving it to the start of the line would drag the view back + # with it, since the window scrolls horizontally to keep the cursor visible. + width = self.text.window.horizontal_scroll if self.chop else target * info.window_width + column = self._column_at_width(document.lines[line], width) + event.current_buffer.cursor_position = document.translate_row_col_to_index(line, column) + self.text.window.vertical_scroll = line + self.text.window.vertical_scroll_2 = target if height(line) > info.window_height else 0 + + def _scroll_horizontal(self, event: KeyPressEvent, direction: int) -> None: + info = self.text.window.render_info + if info is None or not self.chop: + return + document = event.current_buffer.document + target = max(0, self.text.window.horizontal_scroll + direction * max(1, info.window_width // 2)) + column = self._column_at_width(document.current_line, target) + event.current_buffer.cursor_position = document.translate_row_col_to_index(document.cursor_position_row, column) + self.text.window.horizontal_scroll = get_cwidth(document.current_line[:column]) diff --git a/cmd2/utils.py b/cmd2/utils.py index f88a46f20..f9c449346 100644 --- a/cmd2/utils.py +++ b/cmd2/utils.py @@ -690,6 +690,9 @@ def __init__( self.saved_pipe_proc_reader = pipe_proc_reader self.saved_redirecting = saved_redirecting + # Holds the bottom toolbar's suspension while a pipe process owns the terminal + self.toolbar_suspension: contextlib.ExitStack | None = None + def categorize(func: Callable[..., Any] | Iterable[Callable[..., Any]], category: str) -> None: """Categorize a function. diff --git a/docs/features/initialization.md b/docs/features/initialization.md index fc084ff3d..88e6d79d7 100644 --- a/docs/features/initialization.md +++ b/docs/features/initialization.md @@ -52,6 +52,7 @@ Here are instance attributes of `cmd2.Cmd` which developers might wish to overri - **max_completion_table_items**: The maximum number of completion results allowed for a completion table to appear (Default: 50) - **pager**: sets the pager command used by the `Cmd.ppaged()` method for displaying wrapped output using a pager - **pager_chop**: sets the pager command used by the `Cmd.ppaged()` method for displaying chopped/truncated output using a pager +- **use_builtin_pager**: defaults to `enable_bottom_toolbar`. While the command toolbar is running, `Cmd.ppaged()` uses an embedded pager that keeps that toolbar visible. Set to `False` to always use the external `pager`/`pager_chop` commands. This is an opt-*out* only: setting it to `True` does nothing unless `enable_bottom_toolbar` is also set, because the embedded pager shares the toolbar's display. - **py_bridge_name**: name by which embedded Python environments and scripts refer to the `cmd2` application by in order to call commands (Default: `app`) - **py_locals**: dictionary that defines specific variables/functions available in Python shells and scripts (provides more fine-grained control than making everything available with **self_in_py**) - **quiet**: if `True`, then completely suppress nonessential output (Default: `False`) diff --git a/docs/features/os.md b/docs/features/os.md index 444822fd6..0706c7a69 100644 --- a/docs/features/os.md +++ b/docs/features/os.md @@ -42,7 +42,48 @@ system. ## Terminal pagers -Output of any command can be displayed one page at a time using the [cmd2.Cmd.ppaged][] method. +Output of any command can be displayed one page at a time using the [cmd2.Cmd.ppaged][] method. A +pager is only used when the terminal is interactive. Inside a script, or when the command's output +is redirected or piped, `ppaged()` falls back to `poutput()`. + +### Embedded pager + +While the [bottom toolbar](./prompt.md#bottom-toolbar) is running, `ppaged()` displays its output in +an embedded pager rather than handing the terminal to an external program. The toolbar stays visible +and keeps refreshing beneath the paged output, and `cmd2` never gives up control of the terminal. + +Output which already fits in the space above the toolbar is printed directly, so there is no pager +to dismiss. + +| Keys | Action | +| ---------------------------------- | ------------------------------------------------- | +| `Space`, `PageDown`, `f`, `Ctrl-F` | Forward one page | +| `b`, `PageUp`, `Ctrl-B` | Back one page | +| `d` / `u`, `Ctrl-D` / `Ctrl-U` | Forward / back half a page | +| `j` / `k`, `Down` / `Up`, `Enter` | Down / up one line | +| `h` / `l`, `Left` / `Right` | Scroll left / right (chopped output only) | +| `g` / `G`, `Home` / `End` | Jump to the beginning / end | +| `/` / `?` | Search forward / backward | +| `n` / `N` | Jump to the next / previous match | +| `q`, `Escape`, `Ctrl-C` | Close the pager and return to the running command | +| `Ctrl-Z` | Suspend to the background, where supported | + +Passing `chop=True` truncates lines longer than the screen width instead of wrapping them, and the +hidden columns remain reachable with the left and right arrow keys. Unlike the default external +pager on Windows (`more`), the embedded pager supports chopping on every platform. + +Rich styles in the paged output are preserved: colors are rendered while scrolling and searching, +and the visible text of a hyperlink is shown without its escape sequence metadata. + +### External pagers + +Set `self.use_builtin_pager = False` to always use the external commands configured in `self.pager` +and `self.pager_chop`. This is the opt-out for applications which need `less` features the embedded +pager does not provide. External pagers temporarily hide the toolbar and restore it on exit. + +The setting only ever turns the embedded pager off. Because the embedded pager shares the toolbar's +display, it is unavailable anywhere a toolbar is not running, such as a command invoked outside the +command loop, and `ppaged()` uses the external pager there regardless of this setting. Alternatively, a terminal pager can be invoked directly using the ability to run shell commands with the `!` shortcut like so: @@ -51,7 +92,7 @@ the `!` shortcut like so: !!! warning - Once you are in a terminal pager, that program temporarily has control of your terminal, + Once you are in an external terminal pager, that program temporarily has control of your terminal, **NOT** `cmd2`. Typically you can use either the arrow keys or ``/`` keys to scroll around or type `q` to quit the pager and return control to your `cmd2` application. diff --git a/docs/features/prompt.md b/docs/features/prompt.md index 0f8a09082..3c2307ae3 100644 --- a/docs/features/prompt.md +++ b/docs/features/prompt.md @@ -61,7 +61,8 @@ example for a demonstration. ## Bottom Toolbar `cmd2` supports an optional, persistent bottom toolbar that is always visible at the bottom of the -terminal window while the application is idle and waiting for input. +terminal window while the application is waiting for input and while commands execute. Command +output appears above the toolbar. ### Enabling the Toolbar @@ -92,10 +93,10 @@ def get_bottom_toolbar(self) -> AnyFormattedText: ### Refreshing the Toolbar -Since the toolbar is rendered by `prompt-toolkit` as part of the prompt, it is naturally redrawn -whenever the prompt is refreshed. If you want the toolbar to update automatically (for example, to -display a clock), you can set `refresh_interval` in the [cmd2.Cmd.__init__][] constructor to a value -greater than 0.0. +The toolbar is rendered by `prompt-toolkit` and is naturally redrawn whenever the prompt is +refreshed. If you want the toolbar to update automatically during input and command execution (for +example, to display a clock), you can set `refresh_interval` in the [cmd2.Cmd.__init__][] +constructor to a value greater than 0.0. ```py class App(cmd2.Cmd): @@ -105,4 +106,39 @@ class App(cmd2.Cmd): See the [getting_started.py](https://github.com/python-cmd2/cmd2/blob/main/examples/getting_started.py) -example for a demonstration of this technique. +example for a demonstration of this technique. Run its `work 5` command to see the clock update +while the command prints output. + +Inside the command loop, `get_bottom_toolbar()` runs in a background UI thread, at the prompt as +well as during command execution, because a single prompt-toolkit application renders both. Keep the +callback fast and use a lock when reading state that a command or another thread modifies. + +### Commands That Take Over the Terminal + +While the toolbar is running, `ppaged()` uses an embedded pager which keeps the same toolbar visible +and refreshing instead of handing the terminal to an external pager. Short output is printed +directly above the toolbar, and longer output becomes a scrollable, searchable view. Set +`self.use_builtin_pager = False` to opt out and use your configured external `pager`/`pager_chop` +commands. See [Embedded pager](./os.md#embedded-pager) for the key bindings and full details. + +cmd2 temporarily hides the toolbar for its input prompts, external pagers, Python environments, and +shell commands. It also hides it while a command's output is piped to another process, since that +process may be interactive, as `less` and `fzf` are. It restores the toolbar when those operations +finish. + +For custom terminal UIs, calls to `input()`, or subprocesses your own command code starts, use +[cmd2.Cmd.suspend_bottom_toolbar][]: + +```py +with self.suspend_bottom_toolbar(): + answer = input("Continue? ") +``` + +Suspending matters for subprocesses even when they only print. While the toolbar is displayed, the +terminal is in raw mode, so the kernel generates no signals from keystrokes. cmd2 sends `Ctrl-C` on +to its own process group, which reaches a subprocess your command started and waits on, but `Ctrl-\` +(`SIGQUIT`) does nothing, exactly as at the main prompt. + +The command toolbar is used by the interactive command loop, including startup commands and scripts +launched from that loop. It is disabled for non-interactive input. Calls to `onecmd_plus_hooks()` +outside the command loop do not start a toolbar. diff --git a/examples/getting_started.py b/examples/getting_started.py index a1573c853..1c2e3ed83 100755 --- a/examples/getting_started.py +++ b/examples/getting_started.py @@ -25,6 +25,7 @@ import pathlib import sys import threading +import time from typing import Annotated from prompt_toolkit.application import get_app @@ -37,7 +38,7 @@ Color, stylize, ) -from cmd2.annotated import Option +from cmd2.annotated import Argument, Option class BasicApp(cmd2.Cmd): @@ -164,6 +165,18 @@ def get_rprompt(self) -> AnyFormattedText: text = f"cwd={current_working_directory}" return [(style, text)] + @cmd2.with_annotated + def do_work( + self, + seconds: Annotated[ # Optional positional argument, so both "work" and "work 5" are valid + int, Argument(nargs="?", help_text="number of seconds to work for") + ] = 5, + ) -> None: + """Simulate work while the bottom toolbar clock keeps updating.""" + for second in range(seconds): + self.poutput(f"Working: {second + 1}/{seconds}") + time.sleep(1) + @cmd2.with_annotated def do_cat( self, diff --git a/mkdocs.yml b/mkdocs.yml index d928fdcb0..f20c2aa07 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -75,6 +75,7 @@ plugins: filters: - "!^_" - "get_bottom_toolbar" + - "suspend_bottom_toolbar" merge_init_into_class: true docstring_style: sphinx docstring_section_style: spacy diff --git a/tests/conftest.py b/tests/conftest.py index 3a37e9856..746c9f2a1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ """Cmd2 unit/functional testing""" +import io import sys from collections.abc import Callable from contextlib import redirect_stderr @@ -12,6 +13,11 @@ ) import pytest +from prompt_toolkit.application import create_app_session +from prompt_toolkit.data_structures import Size +from prompt_toolkit.input import create_pipe_input +from prompt_toolkit.output import DummyOutput +from prompt_toolkit.shortcuts import PromptSession import cmd2 from cmd2 import rich_utils as ru @@ -192,3 +198,47 @@ def autoload_command_sets_app(): @pytest.fixture def manual_command_sets_app(): return WithCommandSets(auto_load_commands=False) + + +class Terminal(io.StringIO): + """An in-memory stream that claims to be a terminal.""" + + def isatty(self) -> bool: + return True + + +class RecordingOutput(DummyOutput): + """A prompt-toolkit output that records everything written to a Terminal.""" + + def __init__(self, stream: Terminal) -> None: + self.stdout = stream + self.size = Size(rows=24, columns=80) + + def get_size(self) -> Size: + return self.size + + def write(self, data: str) -> None: + self.stdout.write(data) + + def write_raw(self, data: str) -> None: + self.stdout.write(data) + + +@pytest.fixture +def toolbar_app(): + app = cmd2.Cmd(allow_cli_args=False) + output = Terminal() + app.stdout = output + with create_pipe_input() as pipe: + terminal = RecordingOutput(output) + # Bind the ambient app session to this terminal. Without it, prompt-toolkit + # builds a real one on demand for calls such as patch_stdout() in + # _read_raw_input(), which needs a console that Windows CI does not provide. + with create_app_session(input=pipe, output=terminal): + app.main_session = PromptSession( + input=pipe, + output=terminal, + bottom_toolbar="STATUS", + refresh_interval=0.01, + ) + yield app, pipe, output diff --git a/tests/test_command_toolbar.py b/tests/test_command_toolbar.py new file mode 100644 index 000000000..770eecfb9 --- /dev/null +++ b/tests/test_command_toolbar.py @@ -0,0 +1,894 @@ +"""Command toolbar lifecycle and terminal integration tests.""" + +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from unittest import mock + +import pytest +from prompt_toolkit.application import get_app +from prompt_toolkit.formatted_text import ANSI +from prompt_toolkit.input import create_pipe_input +from prompt_toolkit.input.typeahead import get_typeahead +from prompt_toolkit.keys import Keys +from prompt_toolkit.layout import HSplit, Layout, Window +from prompt_toolkit.shortcuts import PromptSession + +from cmd2 import Cmd, command_toolbar + +from .conftest import RecordingOutput, Terminal + + +def test_command_toolbar_refresh_and_output(toolbar_app, monkeypatch) -> None: + app, _, output = toolbar_app + refreshed = threading.Event() + state = ["BEFORE"] + threads = [] + + def toolbar(): + threads.append(threading.current_thread()) + if state[0] == "AFTER": + refreshed.set() + return state[0] + + app.main_session.bottom_toolbar = toolbar + monkeypatch.setattr(sys, "stdout", output) + original_stderr = sys.stderr + with app._command_toolbar_context(): + assert threading.current_thread() is threading.main_thread() + assert get_app() is app._command_toolbar.app + assert sys.stderr is original_stderr # Keep redirected stderr separate. + app.poutput("command output") + print("standard output", end="") # Flush unterminated output on exit. + state[0] = "AFTER" + assert refreshed.wait(2) + + assert "command output\n" in output.getvalue() + assert "standard output" in output.getvalue() + assert "AFTER" in output.getvalue() + assert all(thread is not threading.main_thread() and not thread.is_alive() for thread in threads) + assert app.stdout is output + assert sys.stdout is output + assert app._command_toolbar is None + + +def test_command_toolbar_redirected_output(toolbar_app, tmp_path) -> None: + app, _, output = toolbar_app + destination = tmp_path / "help.txt" + with app._command_toolbar_context(): + app.onecmd_plus_hooks(f'help > "{destination}"') + text = destination.read_text() + assert "Cmd2 Commands" in text + assert "STATUS" not in text + assert "Cmd2 Commands" not in output.getvalue() + + +def test_command_toolbar_redirection_survives_suspension(toolbar_app, tmp_path) -> None: + app, _, output = toolbar_app + destination = tmp_path / "output.txt" + + def command(statement, **kwargs): + app.poutput("before") + with app.suspend_bottom_toolbar(): + app.poutput("during") + app.poutput("after") + return False + + with mock.patch.object(app, "onecmd", side_effect=command), app._command_toolbar_context(): + app.onecmd_plus_hooks(f'custom > "{destination}"') + app.poutput("terminal output") + + assert destination.read_text() == "before\nduring\nafter\n" + assert "before" not in output.getvalue() + assert "during" not in output.getvalue() + assert "after" not in output.getvalue() + assert "terminal output" in output.getvalue() + assert app.stdout is output + + +def test_command_toolbar_pipe_output(toolbar_app) -> None: + app, _, output = toolbar_app + with app._command_toolbar_context(): + app.onecmd_plus_hooks(f'help | "{sys.executable}" -c "import sys; print(sys.stdin.read().upper())"') + assert "CMD2 COMMANDS" in output.getvalue() + + +class FileTerminal: + """A real file that claims to be a terminal, so it owns a descriptor a subprocess can inherit.""" + + def __init__(self, file) -> None: + self.file = file + + def isatty(self) -> bool: + return True + + def __getattr__(self, name): + return getattr(self.file, name) + + +@pytest.mark.parametrize("builtin_pager", [False, True]) +def test_command_toolbar_pipe_process_inherits_terminal(toolbar_app, tmp_path, builtin_pager) -> None: + app, _, _ = toolbar_app + app.use_builtin_pager = builtin_pager + destination = tmp_path / "terminal.txt" + running = [] + readers = [] + + def command(statement, **kwargs): + # The pipe process owns the terminal, so the toolbar must have stepped aside. + running.append(app._command_toolbar.app.is_running) + assert app.main_session.app.layout is app.main_session.layout + readers.append(app._cur_pipe_proc_reader) + app.ppaged("piped") + return False + + with destination.open("w+") as handle: + app.stdout = FileTerminal(handle) + with mock.patch.object(app, "onecmd", side_effect=command), app._command_toolbar_context(): + app.onecmd_plus_hooks(f'custom | "{sys.executable}" -c "import sys; sys.stdout.write(sys.stdin.read().upper())"') + # The terminal goes back to the toolbar once the pipe process has exited. + assert app._command_toolbar.app.is_running + assert app.stdout.proxy is not None + + assert running == [False] + # A process given the terminal writes to it directly instead of through a captured pipe. + assert readers[0]._proc.stdout is None + assert "PIPED" in destination.read_text() + + +def test_command_toolbar_binary_output(toolbar_app) -> None: + app, _, output = toolbar_app + data = "Unicode: 😇\n".encode() + with app._command_toolbar_context(): + for byte in data: + app.stdout.buffer.write(bytes([byte])) + app.stdout.buffer.flush() + assert "Unicode: 😇\n" in output.getvalue() + + +def test_command_toolbar_interrupt_uses_signal_handler(toolbar_app) -> None: + app, pipe, _ = toolbar_app + interrupted = threading.Event() + signal_target = "_thread.interrupt_main" if sys.platform == "win32" else "cmd2.command_toolbar.os.killpg" + with mock.patch(signal_target, side_effect=lambda *_: interrupted.set()) as interrupt: + with app._command_toolbar_context(): + pipe.send_text("\x03") + assert interrupted.wait(2) + interrupt.assert_called_once() + if sys.platform != "win32": + # Reach subprocesses a command started, as the terminal driver would. + import os + import signal + + assert interrupt.call_args.args == (os.getpgrp(), signal.SIGINT) + + +def test_command_toolbar_interrupt_discards_cancelled_typeahead(toolbar_app) -> None: + app, pipe, _ = toolbar_app + interrupted = threading.Event() + received = threading.Event() + signal_target = "_thread.interrupt_main" if sys.platform == "win32" else "cmd2.command_toolbar.os.killpg" + with mock.patch(signal_target, side_effect=lambda *_: interrupted.set()), app._command_toolbar_context(): + toolbar = app._command_toolbar + + def key_processed(_): + if "".join(key.data for key in toolbar._keys).endswith("kept\n"): + received.set() + + toolbar.app.key_processor.after_key_press += key_processed + pipe.send_text("cancelled\n\x03") + assert interrupted.wait(2) + # Input entered after the interrupt should still reach the next prompt. + pipe.send_text("kept\n") + assert received.wait(2) + + assert app._read_raw_input("Next: ", app.main_session) == "kept" + + +@pytest.mark.parametrize(("supported", "enabled"), [(True, True), (True, False), (False, True)]) +def test_command_toolbar_ctrl_z(toolbar_app, supported, enabled) -> None: + app, pipe, _ = toolbar_app + app.main_session.enable_suspend = enabled + processed = threading.Event() + with ( + mock.patch("cmd2.command_toolbar.suspend_to_background_supported", return_value=supported), + app._command_toolbar_context(), + ): + toolbar = app._command_toolbar + toolbar.app.key_processor.after_key_press += lambda _: processed.set() + with mock.patch.object(toolbar.app, "suspend_to_background") as suspend: + pipe.send_text("\x1a") + assert processed.wait(2) + if supported and enabled: + suspend.assert_called_once_with() + else: + suspend.assert_not_called() + + keys = get_typeahead(pipe) + assert [key.key for key in keys] == ([] if supported and enabled else [Keys.ControlZ]) + + +def test_command_toolbar_script_output_has_no_batching_delay(toolbar_app) -> None: + app, _, output = toolbar_app + sleep = mock.Mock() + + def command(statement, **kwargs): + app.poutput("script output") + return False + + # Observe requested sleeps instead of depending on the machine's execution speed. + with ( + mock.patch("prompt_toolkit.patch_stdout.time", SimpleNamespace(sleep=sleep)), + mock.patch.object(app, "onecmd", side_effect=command), + app._command_toolbar_context(), + ): + app.runcmds_plus_hooks(["custom"] * 10) + + assert output.getvalue().count("script output\n") == 10 + assert all(call.args[0] == 0 for call in sleep.call_args_list) + + +def test_command_toolbar_suspension_and_nested_input(toolbar_app) -> None: + app, pipe, output = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + with app.suspend_bottom_toolbar(): + assert not toolbar.app.is_running + assert app.stdout.original is output + assert app.stdout.proxy is None + with app.suspend_bottom_toolbar(): + assert not toolbar.app.is_running + assert toolbar.app.is_running + + # Feed the nested prompt only after it has taken ownership of input. + result = app._read_raw_input("Value: ", app.main_session, pre_run=lambda: pipe.send_text("answer\n")) + assert result == "answer" + assert toolbar.app.is_running + + +class CprOutput(RecordingOutput): + """A terminal that asks for cursor position reports and never answers them.""" + + def get_rows_below_cursor_position(self) -> int: + raise NotImplementedError + + @property + def responds_to_cpr(self) -> bool: + return True + + +def test_command_toolbar_flushes_writes_waiting_on_cursor_reports() -> None: + app = Cmd(allow_cli_args=False) + output = Terminal() + app.stdout = output + + with create_pipe_input() as pipe: + app.main_session = PromptSession( + input=pipe, + output=CprOutput(output), + bottom_toolbar="STATUS", + refresh_interval=0.01, + ) + # Terminal writes wait for a pending cursor position report, so stopping the + # display must not cancel them out from under the text. + with app._command_toolbar_context(): + app.poutput("last words") + + assert "last words\n" in output.getvalue() + + +def test_command_toolbar_suspension_waits_for_in_flight_writes(toolbar_app) -> None: + app, _, output = toolbar_app + writing = threading.Event() + + with app._command_toolbar_context(): + proxy = app._command_toolbar._proxy + proxy_write = proxy.write + + def slow_write(data: str) -> int: + # Widen the window in which suspending could close this proxy. A closed + # proxy accepts writes and discards them, so the output would vanish. + writing.set() + time.sleep(0.1) + return proxy_write(data) + + proxy.write = slow_write + thread = threading.Thread(target=lambda: app.poutput("in flight")) + thread.start() + assert writing.wait(2) + + # A command reaches this at every finalization boundary while its own + # threads are still printing. + with app.suspend_bottom_toolbar(): + pass + thread.join() + + assert "in flight\n" in output.getvalue() + + +def test_command_toolbar_typeahead(toolbar_app) -> None: + app, pipe, _ = toolbar_app + received = threading.Event() + with app._command_toolbar_context(): + toolbar = app._command_toolbar + + def key_processed(_): + if len(toolbar._keys) == len("next\n"): + received.set() + + toolbar.app.key_processor.after_key_press += key_processed + pipe.send_text("next\n") + assert received.wait(2) + + assert app._read_raw_input("Next: ", app.main_session) == "next" + + +def test_command_toolbar_typeahead_preserves_pending_input_order(toolbar_app) -> None: + app, pipe, _ = toolbar_app + exiting = threading.Event() + with app._command_toolbar_context(): + toolbar = app._command_toolbar + + def exit_after_first_key(_): + # Leave the remaining keys in prompt-toolkit's queue, as happens when + # input arrives just as a command finishes. + toolbar.app.exit() + exiting.set() + + toolbar.app.key_processor.after_key_press += exit_after_first_key + pipe.send_text("ab\n") + assert exiting.wait(2) + toolbar._thread.join(timeout=2) + assert not toolbar._thread.is_alive() + toolbar.app.key_processor.after_key_press -= exit_after_first_key + + assert app._read_raw_input("Next: ", app.main_session) == "ab" + + +@pytest.mark.parametrize("exception", [RuntimeError, KeyboardInterrupt, SystemExit]) +def test_command_toolbar_cleanup_on_exception(toolbar_app, exception) -> None: + app, _, output = toolbar_app + threads = [] + + def run_command(): + with app._command_toolbar_context(): + threads.append(app._command_toolbar._thread) + raise exception + + with pytest.raises(exception): + run_command() + assert len(threads) == 1 + assert not threads[0].is_alive() + assert app.stdout is output + assert app._command_toolbar is None + + +def test_command_toolbar_recovers_from_stop_failure(toolbar_app) -> None: + app, pipe, _ = toolbar_app + bindings = app.main_session.app.key_bindings + + context = app._command_toolbar_context() + context.__enter__() + toolbar = app._command_toolbar + real_stop = toolbar.stop + + def failing_stop() -> None: + # Tear down for real, then fail the way a broken stream close would. + real_stop() + raise RuntimeError("broken stop") + + toolbar.stop = failing_stop + with pytest.raises(RuntimeError, match="broken stop"): + context.__exit__(None, None, None) + + # A later command still gets a toolbar instead of being locked out by the dead one. + assert app._command_toolbar is None + assert app.main_session.app.layout is app.main_session.layout + assert app.main_session.app.key_bindings is bindings + with app._command_toolbar_context(): + assert app._command_toolbar is not None + assert app._command_toolbar is not toolbar + assert app._command_toolbar is None + assert app._read_raw_input("Next: ", app.main_session, pre_run=lambda: pipe.send_text("recovered\n")) == "recovered" + + +def test_command_toolbar_exit_after_result_is_set(toolbar_app) -> None: + app, _, _ = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + + def already_exiting(): + toolbar.app.exit() + # The result is set before run_async() has finished its cleanup. + assert toolbar.app.is_running + toolbar._exit() + + toolbar._call_in_ui(already_exiting) + assert app.main_session.app.layout is app.main_session.layout + + +def test_command_toolbar_ui_call_propagates_failures(toolbar_app) -> None: + app, _, _ = toolbar_app + + def fail(exception: BaseException) -> None: + raise exception + + with app._command_toolbar_context(): + toolbar = app._command_toolbar + + # A UI callback runs on the display's loop, so its failure has to be carried + # back to the command thread rather than reaching the loop's error handler. + with pytest.raises(ValueError, match="broken ui call"): + toolbar._call_in_ui(lambda: fail(ValueError("broken ui call"))) + + # A TimeoutError raised by the callback is the same class the pending future + # reports itself with, and must not be mistaken for one. + with pytest.raises(TimeoutError, match="slow ui call"): + toolbar._call_in_ui(lambda: fail(TimeoutError("slow ui call"))) + + # A callback that outlives the poll interval keeps waiting instead of giving up. + assert toolbar._call_in_ui(lambda: time.sleep(0.2) or "finished") == "finished" + + +def test_command_toolbar_ui_call_after_display_stopped(toolbar_app) -> None: + app, pipe, _ = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + pipe.close() + toolbar._thread.join(timeout=2) + assert not toolbar._thread.is_alive() + + # There is no loop left to run UI work on, so asking must fail rather than + # queue a callback onto a closed loop. + assert toolbar.app.loop is None + with pytest.raises(RuntimeError, match="Toolbar is not running"): + toolbar._call_in_ui(lambda: None) + + +def test_command_toolbar_ui_call_reports_display_failure(toolbar_app, capsys) -> None: + app, _, _ = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + loop = toolbar.app.loop + schedule = loop.call_soon_threadsafe + failed = threading.Event() + + def die(*args, **kwargs): + # The display dies instead of running the queued callback, so the future + # the command is waiting on never resolves. Only drop that one request, + # made here on the command thread. asyncio uses call_soon_threadsafe from + # its own threads, and on Windows the default executor's join is reported + # through it while asyncio.run() shuts the loop down. Swallowing that + # report strands the toolbar thread for 300 seconds, or forever before + # Python 3.12, where shutdown_default_executor() has no timeout. + if not failed.is_set() and threading.current_thread() is threading.main_thread(): + failed.set() + schedule(lambda: toolbar.app.exit(exception=ValueError("broken display"))) + return None + return schedule(*args, **kwargs) + + with ( + mock.patch.object(loop, "call_soon_threadsafe", side_effect=die), + pytest.raises(ValueError, match="broken display"), + ): + toolbar._call_in_ui(lambda: None) + + assert "broken display" in capsys.readouterr().err + + +def test_command_toolbar_failure_after_startup_is_reported(toolbar_app, capsys) -> None: + app, _, output = toolbar_app + + with app._command_toolbar_context(): + toolbar = app._command_toolbar + # Stop the display the way an unhandled error in its own thread would, after + # _resume() has already returned and can no longer raise for the command. + toolbar.app.loop.call_soon_threadsafe(lambda: toolbar.app.exit(exception=ValueError("broken display"))) + toolbar._thread.join(timeout=2) + assert not toolbar._thread.is_alive() + + # Output must still reach the terminal rather than a proxy nothing is draining. + assert all(stream.proxy is None for stream in toolbar._streams) + app.poutput("after failure") + + assert "broken display" in capsys.readouterr().err + assert "after failure\n" in output.getvalue() + + +def test_command_toolbar_input_eof_is_not_reported(toolbar_app, capsys) -> None: + app, pipe, output = toolbar_app + + with app._command_toolbar_context(): + toolbar = app._command_toolbar + # Losing the terminal's input ends the display with EOFError. That is an + # ordinary shutdown, not a failure the running command should hear about. + pipe.close() + toolbar._thread.join(timeout=2) + assert not toolbar._thread.is_alive() + assert toolbar._error is None + + # Output must still reach the terminal rather than a proxy nothing is draining. + assert all(stream.proxy is None for stream in toolbar._streams) + app.poutput("after eof") + + assert capsys.readouterr().err == "" + assert "after eof\n" in output.getvalue() + assert app.stdout is output + assert app._command_toolbar is None + + +def test_command_toolbar_startup_failure_still_runs_the_command(toolbar_app, capsys) -> None: + app, _, output = toolbar_app + + def broken_toolbar(): + raise ValueError("broken toolbar") + + app.main_session.bottom_toolbar = broken_toolbar + ran = [] + # The toolbar is cosmetic. A display that cannot start must not take the command + # with it, and must not escape cmdloop() and leave signal handlers installed. + with app._command_toolbar_context(): + ran.append(True) + app.poutput("command output") + + assert ran == [True] + assert "broken toolbar" in capsys.readouterr().err + assert "command output\n" in output.getvalue() + assert app.stdout is output + assert app._command_toolbar is None + + +def test_command_toolbar_is_not_retried_after_a_startup_failure(toolbar_app, capsys) -> None: + app, _, _ = toolbar_app + + def broken_toolbar(): + raise ValueError("broken toolbar") + + app.main_session.bottom_toolbar = broken_toolbar + with app._command_toolbar_context(): + pass + assert "broken toolbar" in capsys.readouterr().err + + # Without this, every later command repeats the same failure and the same message. + app.main_session.bottom_toolbar = "STATUS" + with mock.patch("cmd2.command_toolbar.CommandToolbar") as toolbar, app._command_toolbar_context(): + toolbar.assert_not_called() + assert capsys.readouterr().err == "" + + +def test_cmdloop_restores_signal_handlers_when_the_loop_fails(toolbar_app, monkeypatch) -> None: + import signal + + app, _, _ = toolbar_app + original = signal.getsignal(signal.SIGINT) + monkeypatch.setattr(app, "_cmdloop", mock.Mock(side_effect=RuntimeError("loop failed"))) + + with pytest.raises(RuntimeError, match="loop failed"): + app.cmdloop() + + # cmd2's handlers must not outlive the loop in the host process. + assert signal.getsignal(signal.SIGINT) is original + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_command_toolbar_headless(enabled) -> None: + app = Cmd(allow_cli_args=False, enable_bottom_toolbar=enabled) + with mock.patch("cmd2.command_toolbar.CommandToolbar") as toolbar, app._command_toolbar_context(): + toolbar.assert_not_called() + + +def test_cmdloop_runs_commands_with_toolbar(toolbar_app, monkeypatch) -> None: + app, _, _ = toolbar_app + monkeypatch.setattr(app, "_read_command_line", lambda _: "quit") + commands = [] + + def command(line, **kwargs): + assert threading.current_thread() is threading.main_thread() + assert app._command_toolbar.app.is_running + commands.append(line) + return line == "quit" + + app._startup_commands = ["startup"] + monkeypatch.setattr(app, "onecmd_plus_hooks", command) + app._cmdloop() + assert commands == ["startup", "quit"] + + +@pytest.mark.parametrize( + ("layout", "message"), + [ + (Layout(Window()), "Unsupported PromptSession layout"), + (Layout(HSplit([Window()])), "Cannot locate PromptSession bottom toolbar"), + ], +) +def test_command_toolbar_requires_the_prompt_toolbar(toolbar_app, capsys, layout, message) -> None: + app, _, output = toolbar_app + # The display reuses the prompt's own toolbar container. If a future prompt-toolkit + # release moves it, say so and keep running without a toolbar. + with mock.patch.object(app.main_session, "layout", layout), app._command_toolbar_context(): + app.poutput("command output") + assert message in capsys.readouterr().err + assert app._command_toolbar is None + assert "command output\n" in output.getvalue() + assert app.stdout is output + + +def test_command_toolbar_reuses_prompt_application(toolbar_app) -> None: + app, pipe, _ = toolbar_app + session = app.main_session + layout, bindings, erase = session.app.layout, session.app.key_bindings, session.app.erase_when_done + with app._command_toolbar_context(): + toolbar = app._command_toolbar + assert toolbar.app is session.app + assert toolbar.toolbar is session.layout.container.children[-1] + with app.suspend_bottom_toolbar(): + assert session.app.layout is layout + assert session.app.key_bindings is bindings + assert session.app.erase_when_done is erase + assert session.app.layout is toolbar._layout + assert session.app.layout is layout + assert session.app.key_bindings is bindings + assert session.app.erase_when_done is erase + assert app._read_raw_input("Next: ", session, pre_run=lambda: pipe.send_text("answer\n")) == "answer" + + +@pytest.mark.parametrize("quit_key", ["q", "\x03"]) +def test_builtin_pager_keeps_toolbar_live(toolbar_app, monkeypatch, quit_key) -> None: + app, pipe, _ = toolbar_app + app.use_builtin_pager = True + monkeypatch.setattr(app, "stdin", Terminal()) + monkeypatch.setenv("TERM", "xterm") + entered, refreshed, moved, found = (threading.Event() for _ in range(4)) + state = ["BEFORE"] + + def toolbar_text(): + assert threading.current_thread() is not threading.main_thread() + if app.main_session.app.full_screen and state[0] == "AFTER": + refreshed.set() + return state[0] + + app.main_session.bottom_toolbar = toolbar_text + prompt_layout = app.main_session.app.layout + + def observe(ui): + if not ui.full_screen: + return + assert ui.layout.container.children[-1] is app._command_toolbar.toolbar + entered.set() + row = ui.layout.current_buffer.document.cursor_position_row + if row > 0: + moved.set() + if row == 80: + found.set() + + def interact(): + try: + assert entered.wait(2) + state[0] = "AFTER" + assert refreshed.wait(2) + pipe.send_text(" ") + assert moved.wait(2) + pipe.send_text("/row 080\n") + assert found.wait(2) + finally: + pipe.send_text(quit_key) + + app.main_session.app.after_render += observe + with mock.patch("subprocess.Popen") as external, ThreadPoolExecutor() as executor: + interaction = executor.submit(interact) + with app._command_toolbar_context(): + thread = app._command_toolbar._thread + app.ppaged("\n".join(f"row {i:03d}" for i in range(100))) + assert app._command_toolbar._thread is thread + assert app._command_toolbar.is_active + assert not app.main_session.app.full_screen + assert not app.main_session.app.renderer.full_screen + interaction.result(timeout=2) + external.assert_not_called() + assert app.main_session.app.layout is prompt_layout + assert refreshed.is_set() + assert get_typeahead(pipe) == [] + + +def test_builtin_pager_short_output(toolbar_app, monkeypatch) -> None: + app, _, output = toolbar_app + app.use_builtin_pager = True + monkeypatch.setattr(app, "stdin", Terminal()) + monkeypatch.setenv("TERM", "xterm") + with mock.patch("subprocess.Popen") as external, app._command_toolbar_context(): + app.ppaged("short output") + assert app._command_toolbar.is_active + external.assert_not_called() + assert "short output\n" in output.getvalue() + + +def test_builtin_pager_short_output_builds_no_pager(toolbar_app, monkeypatch) -> None: + app, _, output = toolbar_app + app.use_builtin_pager = True + monkeypatch.setattr(app, "stdin", Terminal()) + monkeypatch.setenv("TERM", "xterm") + # Output that fits is written directly, so none of the pager's widgets are needed. + with mock.patch("cmd2.command_toolbar.Pager") as pager, app._command_toolbar_context(): + app.ppaged("short output") + pager.assert_not_called() + assert "short output\n" in output.getvalue() + + +def test_builtin_pager_needs_an_already_running_toolbar(toolbar_app, monkeypatch) -> None: + app, _, _ = toolbar_app + app.use_builtin_pager = True + monkeypatch.setattr(app, "stdin", Terminal()) + monkeypatch.setenv("TERM", "xterm") + # Outside the command loop there is no toolbar to page inside. Starting one here + # would wrap the terminal streams and enter raw mode where the docs promise not to. + with ( + mock.patch.object(command_toolbar.CommandToolbar, "start") as start, + mock.patch("subprocess.Popen") as external, + ): + app.ppaged("row\n" * 200) + start.assert_not_called() + external.assert_called_once() + + +def test_external_pager_suspends_shared_application(toolbar_app, monkeypatch) -> None: + app, _, _ = toolbar_app + app.use_builtin_pager = False + monkeypatch.setattr(app, "stdin", Terminal()) + monkeypatch.setenv("TERM", "xterm") + + def external(*args, **kwargs): + assert not app.main_session.app.is_running + assert app.main_session.app.layout is app.main_session.layout + return mock.Mock() + + with mock.patch("subprocess.Popen", side_effect=external), app._command_toolbar_context(): + app.ppaged("external pager") + assert app._command_toolbar.is_active + + +def test_builtin_pager_eof_restores_prompt(toolbar_app) -> None: + app, pipe, output = toolbar_app + layout = app.main_session.app.layout + + def close_input(ui): + if ui.full_screen: + pipe.close() + + app.main_session.app.after_render += close_input + with pytest.raises(EOFError), app._command_toolbar_context(): + app._command_toolbar.page("line\n" * 100, chop=False) + assert app.main_session.app.layout is layout + assert not app.main_session.app.full_screen + assert not app.main_session.app.renderer.full_screen + assert app.stdout is output + + +def test_builtin_pager_does_not_capture_redirected_output(toolbar_app, monkeypatch, tmp_path) -> None: + app, _, output = toolbar_app + app.use_builtin_pager = True + monkeypatch.setattr(app, "stdin", Terminal()) + monkeypatch.setenv("TERM", "xterm") + target = tmp_path / "help.txt" + with mock.patch("cmd2.command_toolbar.Pager") as pager, app._command_toolbar_context(): + app.onecmd_plus_hooks(f'help > "{target}"') + pager.assert_not_called() + assert "Cmd2 Commands" in target.read_text() + assert "Cmd2 Commands" not in output.getvalue() + + +def test_prompt_ctrl_c_raises_keyboard_interrupt(toolbar_app) -> None: + """Ctrl-C at the main prompt aborts the line rather than returning text.""" + app, pipe, _ = toolbar_app + pipe.send_text("partial\x03") + with pytest.raises(KeyboardInterrupt): + app._read_command_line(app.prompt) + + +def test_prompt_ctrl_d_raises_eof(toolbar_app) -> None: + """Ctrl-D on an empty line signals end of input, which cmdloop turns into _eof.""" + app, pipe, _ = toolbar_app + pipe.send_text("\x04") + with pytest.raises(EOFError): + app._read_command_line(app.prompt) + + +def test_accepted_line_appears_once_in_output(toolbar_app) -> None: + """The accepted command line reaches the terminal on the no-toolbar fallback path. + + Reading outside a running display falls through to session.prompt(), where + prompt-toolkit's final `is_done` frame is what leaves the line in the scrollback. + Only its presence is asserted: how many times the string appears in the rendered + byte stream depends on screen diffing and on how many frames were drawn while the + text arrived, so an exact count is a property of the renderer, not of cmd2. + read_line()'s own echo is pinned exactly by + test_read_line_echoes_the_accepted_line. + """ + app, pipe, output = toolbar_app + app.do_probe = lambda _: None + pipe.send_text("probe\n") + line = app._read_command_line(app.prompt) + with app._command_toolbar_context(): + app.onecmd_plus_hooks(line) + assert f"{app.prompt}probe" in output.getvalue() + + +def test_read_line_keeps_the_application_running(toolbar_app) -> None: + """read_line() returns a line without ending the run that draws the toolbar.""" + app, pipe, _ = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + pipe.send_text("hello\n") + assert toolbar.read_line(ANSI("> ")) == "hello" + assert toolbar.is_active, "the application stopped when the line was accepted" + + +def test_read_line_ctrl_c_raises_without_stopping_the_app(toolbar_app) -> None: + """Ctrl-C aborts the line but leaves the toolbar's application running.""" + app, pipe, _ = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + pipe.send_text("partial\x03") + with pytest.raises(KeyboardInterrupt): + toolbar.read_line(ANSI("> ")) + assert toolbar.is_active + + +def test_read_line_ctrl_d_raises_without_stopping_the_app(toolbar_app) -> None: + """Ctrl-D on an empty line signals EOF but leaves the application running.""" + app, pipe, _ = toolbar_app + with app._command_toolbar_context(): + toolbar = app._command_toolbar + pipe.send_text("\x04") + with pytest.raises(EOFError): + toolbar.read_line(ANSI("> ")) + assert toolbar.is_active + + +def test_read_line_echoes_the_accepted_line(toolbar_app) -> None: + """The accepted line is committed to the terminal exactly once. + + prompt-toolkit's is_done frame used to do this. A continuous application never + renders one, so read_line() has to write the line above the toolbar itself. + """ + app, pipe, _ = toolbar_app + written: list[str] = [] + with app._command_toolbar_context(): + toolbar = app._command_toolbar + stream_write = app.stdout.write + + def spy(data: str) -> int: + written.append(data) + return stream_write(data) + + app.stdout.write = spy + pipe.send_text("hello\n") + assert toolbar.read_line(ANSI("myapp> ")) == "hello" + + # The input area must not still be displaying the line that was just + # committed, or the reader sees it twice: once live and once in the + # scrollback. This is what accepting with keep_text=False buys. + assert app.main_session.default_buffer.text == "" + + # Count what cmd2 writes rather than what reaches the terminal. The rendered + # byte stream also contains the prompt line from the live input area, and + # whether it lands there as one contiguous string depends on prompt-toolkit's + # screen diffing and on how many frames it drew while the text arrived. + assert "".join(written).count("myapp> hello") == 1 + + +def test_typeahead_during_a_command_reaches_the_next_prompt(toolbar_app) -> None: + """Keys typed while a command runs are handled by the prompt that follows it.""" + app, pipe, _ = toolbar_app + + def slow(_) -> None: + pipe.send_text("typed\n") + time.sleep(0.2) + + app.do_slow = slow + with app._command_toolbar_context(): + toolbar = app._command_toolbar + with app._command_mode_context(): + app.onecmd_plus_hooks("slow") + assert toolbar.read_line(ANSI("> ")) == "typed" diff --git a/tests/test_pager.py b/tests/test_pager.py new file mode 100644 index 000000000..a2a4283ba --- /dev/null +++ b/tests/test_pager.py @@ -0,0 +1,321 @@ +"""Tests for the built-in pager view.""" + +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from unittest import mock + +import pytest +from prompt_toolkit.data_structures import Size +from prompt_toolkit.input.typeahead import get_typeahead +from rich.console import Console + +from cmd2.pager import Pager, output_fits + + +@pytest.mark.parametrize("chop", [False, True]) +def test_pager_styles_and_wrapping(chop) -> None: + pager = Pager("\x1b[31m" + "界" * 40 + "\x1b[0m\n", chop=chop) + assert pager.text.text == "界" * 40 + assert pager.text.read_only + lexer = pager.text.lexer.lex_document(pager.text.document) + assert "ansired" in lexer(0)[0][0] + + +@pytest.mark.parametrize("chop", [False, True]) +@pytest.mark.parametrize("terminator", ["\x1b\\", "\x07"], ids=["ST", "BEL"]) +def test_pager_rich_hyperlinks(chop, terminator) -> None: + # legacy_windows=False keeps rich from suppressing OSC 8 hyperlinks when the + # tests run against a legacy Windows console. + console = Console(force_terminal=True, color_system="standard", no_color=False, legacy_windows=False) + with console.capture() as capture: + console.print("[link=https://example.com][red]click here[/red][/link] after") + captured = capture.get() + assert "\x1b]8;" in captured + captured = captured.replace("\x1b\\", terminator) + + pager = Pager(captured, chop=chop) + assert pager.text.text == "click here after" + lexer = pager.text.lexer.lex_document(pager.text.document) + assert "ansired" in lexer(0)[0][0] + assert output_fits(captured, len("click here after"), 1, chop=chop) + assert not output_fits(captured, len("click here after") - 1, 1, chop=chop) + + +@pytest.mark.parametrize("chop", [False, True]) +def test_output_fits_measures_styled_and_wide_text(chop) -> None: + # Forty double-width characters occupy eighty columns, and styling them adds + # escape sequences that must not count towards the measurement. + text = "\x1b[31m" + "界" * 40 + "\x1b[0m\n" + assert not output_fits(text, 20, 1, chop=chop) + # Wrapping the line onto four rows fits; chopping keeps it one wide row that does not. + assert output_fits(text, 20, 5, chop=chop) is (not chop) + assert output_fits(text, 100, 1, chop=chop) + + +@pytest.mark.parametrize("chop", [False, True]) +def test_pager_long_line_navigation_resize_and_typeahead(toolbar_app, chop) -> None: + app, pipe, _ = toolbar_app + app.main_session.bottom_toolbar = "STATUS ONE\nSTATUS TWO" + entered, scrolled, resized = (threading.Event() for _ in range(3)) + + def observe(ui): + if not ui.full_screen: + return + # Check the rendered frame, not the stream of incremental terminal + # writes, to verify both toolbar rows survive navigation and resizing. + screen = ui.renderer._last_screen + size = ui.output.get_size() + bottom = "".join(screen.data_buffer[size.rows - 1][x].char for x in range(size.columns)) + assert bottom.startswith("STATUS TWO") + entered.set() + if ui.current_buffer.cursor_position > 0: + scrolled.set() + if size.columns == 60: + resized.set() + + def interact(): + try: + assert entered.wait(2) + pipe.send_text("\x1b[C" if chop else " ") + assert scrolled.wait(2) + app.main_session.output.size = Size(rows=20, columns=60) + app.main_session.app.invalidate() + assert resized.wait(2) + finally: + pipe.send_text("qnext\n") + + app.main_session.app.after_render += observe + with ThreadPoolExecutor() as executor: + interaction = executor.submit(interact) + with app._command_toolbar_context(): + app._command_toolbar.page("界" * 4000, chop=chop) + interaction.result(timeout=2) + assert app._read_raw_input("Next: ", app.main_session) == "next" + + +class PagerKeys: + """Send keys to the built-in pager and wait for the frame that reflects them.""" + + def __init__(self, app, pipe, pager) -> None: + self.pipe = pipe + self.pager = pager + self.presses = 0 + self.states = [] + self.updated = threading.Condition() + ui = app.main_session.app + ui.key_processor.after_key_press += self._count + ui.after_render += self._record + + def _count(self, _) -> None: + with self.updated: + self.presses += 1 + + def _record(self, _) -> None: + # Read the pager's own buffer, not the focused one, which is the search field + # while a search is being typed. Read it after rendering, because + # prompt-toolkit settles the window's scroll offsets while it draws. + with self.updated: + self.states.append( + ( + self.presses, + self.pager.text.buffer.document.cursor_position_row, + self.pager.text.window.horizontal_scroll, + ) + ) + self.updated.notify_all() + + def press(self, keys, row, column=0) -> None: + """Send keys and wait for a drawn frame that shows the expected position.""" + with self.updated: + handled = self.presses + self.pipe.send_text(keys) + deadline = time.monotonic() + 5 + index = 0 + with self.updated: + while True: + while index < len(self.states): + presses, *position = self.states[index] + index += 1 + if presses > handled and position == [row, column]: + return + remaining = deadline - time.monotonic() + notified = remaining > 0 and self.updated.wait(remaining) + assert notified, f"pager ignored {keys!r}: wanted {(row, column)}, saw {self.states[-3:]}" + + +def drive_pager(app, pipe, text, *, chop, script) -> None: + """Page text and run script against its keys while the pager is displayed.""" + created = [] + entered = threading.Event() + + def make_pager(*args, **kwargs): + pager = Pager(*args, **kwargs) + created.append(pager) + return pager + + def observe(ui): + if created and ui.full_screen and ui.layout.current_buffer is created[0].text.buffer: + entered.set() + + def interact(): + try: + assert entered.wait(5) + script(PagerKeys(app, pipe, created[0])) + finally: + pipe.send_text("q") + + app.main_session.app.after_render += observe + with mock.patch("cmd2.command_toolbar.Pager", side_effect=make_pager), ThreadPoolExecutor() as executor: + interaction = executor.submit(interact) + with app._command_toolbar_context(): + app._command_toolbar.page(text, chop=chop) + interaction.result(timeout=10) + assert get_typeahead(pipe) == [] + + +def test_pager_vertical_scrolling_keeps_horizontal_position(toolbar_app) -> None: + app, pipe, _ = toolbar_app + # Wide rows are what chopped output is for. Scrolling right to read a column and + # then moving down a row must not throw that column away. + text = "\n".join(f"row {index:03d} " + "col " * 40 for index in range(100)) + + def script(keys) -> None: + keys.press("\x1b[C", row=0, column=40) + keys.press("j", row=1, column=40) + keys.press("k", row=0, column=40) + + drive_pager(app, pipe, text, chop=True, script=script) + + +@pytest.mark.parametrize("chop", [False, True]) +def test_pager_navigation_keys(toolbar_app, chop) -> None: + app, pipe, _ = toolbar_app + lines = [f"row {index:03d}" for index in range(100)] + lines[3] = "" # A line with no columns for a scroll target to be clamped against. + + def script(keys) -> None: + keys.press("\x1b[B", row=1) # Down arrow. + keys.press("\x1b[A", row=0) # Up arrow. + page_rows = keys.pager.text.window.render_info.window_height - 1 + keys.press("\x1b[6~", row=page_rows) # Page Down. + keys.press("\x1b[5~", row=0) # Page Up. + keys.press("\r", row=1) + keys.press("\x1b[A", row=0) + keys.press("j", row=1) + keys.press("j", row=2) + keys.press("j", row=3) # Land on the empty line. + keys.press("k", row=2) # Moving up off it re-enters the line above. + keys.press("G", row=99) + keys.press("g", row=0) + keys.press("/row 05\n", row=50) + keys.press("n", row=51) + keys.press("N", row=50) + keys.press("?row 01\n", row=19) + keys.press("x", row=19) # Unbound keys are swallowed, not queued for the prompt. + # Horizontal scrolling applies only to chopped output, and stops at the + # end of a line shorter than the requested column. + keys.press("\x1b[C", row=19, column=len("row 019") if chop else 0) + keys.press("\x1b[D", row=19, column=0) + + drive_pager(app, pipe, "\n".join(lines), chop=chop, script=script) + + +@pytest.mark.parametrize("chop", [False, True]) +def test_pager_scrolling_before_first_render(chop) -> None: + pager = Pager("row\n" * 100, chop=chop) + # Keys can arrive before the first frame is drawn, when the window still has no + # rendered geometry to scroll against. + assert pager.text.window.render_info is None + event = SimpleNamespace(current_buffer=pager.text.buffer) + pager._scroll_page(event, pages=1.0) + pager._scroll(event, 1) + pager._scroll_horizontal(event, 1) + assert pager.text.buffer.cursor_position == 0 + assert pager.text.window.horizontal_scroll == 0 + + +@pytest.mark.parametrize("chop", [False, True]) +def test_pager_half_page_scrolling(toolbar_app, chop) -> None: + """Half-page keys move half of what the full-page keys move, and never zero rows.""" + app, pipe, _ = toolbar_app + lines = [f"row {index:03d}" for index in range(200)] + + def script(keys) -> None: + page_rows = max(1, keys.pager.text.window.render_info.window_height - 1) + half_rows = max(1, int(page_rows * 0.5)) + # A half page has to be a distinct, smaller step for this test to mean + # anything. The pager's own arithmetic is what decides the row it lands on. + assert 0 < half_rows < page_rows + keys.press("d", row=half_rows) + keys.press("d", row=2 * half_rows) + keys.press("u", row=half_rows) + keys.press("\x04", row=2 * half_rows) # Ctrl-D. + keys.press("\x15", row=half_rows) # Ctrl-U. + # Half-page steps stay half a page next to a full one taken from the same row. + keys.press("g", row=0) + keys.press("\x1b[6~", row=page_rows) # Page Down. + + drive_pager(app, pipe, "\n".join(lines), chop=chop, script=script) + + +@pytest.mark.parametrize("key", ["\x1b", "\x03"], ids=["escape", "ctrl-c"]) +def test_pager_search_abort_keys(toolbar_app, key) -> None: + """Escape and Ctrl-C abort a search rather than closing the pager. + + While the search field has focus the pager's own close bindings are filtered out, + so these keys have to reach prompt-toolkit's search bindings instead. Those are + registered explicitly so that they still work when the main prompt uses Vi mode. + """ + app, pipe, _ = toolbar_app + lines = [f"row {index:03d}" for index in range(100)] + + def script(keys) -> None: + keys.press("j", row=1) + # Typing a search previews it without moving the pager's own cursor. + keys.press("/row 05", row=1) + # Aborting restores the row the search started from instead of quitting. + keys.press(key, row=1) + # Focus is back on the pager, so ordinary navigation works again. + keys.press("j", row=2) + + drive_pager(app, pipe, "\n".join(lines), chop=False, script=script) + + +@pytest.mark.parametrize("key", ["\x1b", "\x03", "q"], ids=["escape", "ctrl-c", "q"]) +def test_pager_close_keys(toolbar_app, key) -> None: + """Each close key ends the pager without leaving the keystroke for the next prompt. + + Escape is bound eagerly, and it is also the first byte of every arrow and page + key. Closing on a bare Escape must therefore not come at the cost of the escape + sequences that arrive with more bytes behind them. + """ + app, pipe, _ = toolbar_app + entered = threading.Event() + closed = threading.Event() + + def observe(ui) -> None: + if ui.full_screen: + entered.set() + + def interact() -> None: + assert entered.wait(5), "pager never opened" + # Scroll first, so the pager is known to be reading keys before the close key. + pipe.send_text("j") + pipe.send_text(key) + if not closed.wait(5): + # Rescue the blocked main thread so this fails as an assertion, not a hang. + pipe.send_text("q") + raise AssertionError(f"{key!r} did not close the pager") + + app.main_session.app.after_render += observe + with ThreadPoolExecutor() as executor: + interaction = executor.submit(interact) + try: + with app._command_toolbar_context(): + app._command_toolbar.page("\n".join(f"row {index:03d}" for index in range(200)), chop=False) + finally: + closed.set() + interaction.result(timeout=10) + assert get_typeahead(pipe) == [] diff --git a/tests/test_toolbar_flicker.py b/tests/test_toolbar_flicker.py new file mode 100644 index 000000000..5623f3b7b --- /dev/null +++ b/tests/test_toolbar_flicker.py @@ -0,0 +1,88 @@ +"""Regression gate: one application must serve both the prompt and command execution.""" + +import threading +from concurrent.futures import Future + +from prompt_toolkit.buffer import Buffer + + +def test_display_does_not_restart_between_prompt_and_command(toolbar_app) -> None: + """Reading a line must not stop and restart the application that draws the toolbar. + + prompt-toolkit ends every Application.run() with a render whose `is_done` filter + drops the bottom toolbar, so a run that ends between the prompt and the command + erases the toolbar and the next run paints it again. That erase-and-repaint is the + flicker. The fix is to never end a run there, which is what this asserts. + + This checks the run's lifecycle rather than counting rendered frames. Frame + sampling proved unreliable in this harness: the borrowed application sometimes + finishes early over a pipe input, producing teardown frames with no visible + windows that are indistinguishable from a real gap. + """ + app, pipe, _ = toolbar_app + app.do_probe = lambda _: None + pipe.send_text("probe\n") + + # Mirror _cmdloop: one display held open across both the prompt and the command, + # rather than started and stopped around each command. + with app._command_toolbar_context(): + toolbar = app._command_toolbar + assert toolbar is not None + run_thread = toolbar._thread + assert toolbar.is_active + + line = app._read_command_line(app.prompt) + + assert line == "probe" + assert toolbar.is_active, "the display stopped in order to read the line" + assert toolbar._thread is run_thread, "the application was restarted after the prompt" + + # Entering command mode only swaps the layout, so the same run continues. + with app._command_mode_context(): + assert toolbar.is_active + assert toolbar._thread is run_thread, "the application was restarted for the command" + app.onecmd_plus_hooks(line) + + # Nothing is asserted past this point: _run_cmdfinalization_hooks is decorated + # with @suspend_toolbar, so the display is deliberately stopped and restarted + # once per command at finalization. That boundary predates this change and is + # a separate opportunity for the toolbar to blink. + + +def test_spike_accept_without_exit(toolbar_app) -> None: + """Replacing the accept handler yields the line while the application keeps running. + + PromptSession's own accept handler calls app.exit(result=...), which is what ends + the run and produces the toolbar-erasing is_done frame. If a replacement handler + can hand the text to another thread instead, a single long-lived Application can + serve both the prompt and command execution. + """ + app, pipe, _ = toolbar_app + session = app.main_session + ui = session.app + result: Future[str] = Future() + still_running = threading.Event() + + def accept(buff: Buffer) -> bool: + if not result.done(): + result.set_result(buff.document.text) + # Report whether the app is still running at accept time. + if ui.is_running and not ui.is_done: + still_running.set() + return True # Keep the text; the caller resets the buffer. + + session.default_buffer.accept_handler = accept + + def drive() -> None: + assert result.result(timeout=5) == "hello" + # Now end the run explicitly so session.prompt() returns. + ui.loop.call_soon_threadsafe(lambda: ui.exit(result="")) + + worker = threading.Thread(target=drive, daemon=True) + worker.start() + pipe.send_text("hello\n") + session.prompt("> ") + worker.join(timeout=5) + + assert result.done(), "accept handler never fired" + assert still_running.is_set(), "the application had already exited at accept time"