diff --git a/CHANGELOG.md b/CHANGELOG.md index 552217630..7058dcfcc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## 4.3.0 (TBD) + +- Enhancements + - `enable_bottom_toolbar=True` now keeps the toolbar visible and refreshing during command + execution + ## 4.2.4 (September 8, 2026) - Bug Fixes diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index fcd26b2e4..0d8032c99 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,7 @@ 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 # Commands to exclude from the history command self.exclude_from_history = ["_eof", "history"] @@ -1896,6 +1899,7 @@ def pfeedback( rich_print_kwargs=rich_print_kwargs, ) + @command_toolbar.suspend_toolbar def ppaged( self, *objects: Any, @@ -2070,7 +2074,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 +2083,54 @@ def get_bottom_toolbar(self) -> AnyFormattedText: your application. This could be information like the application name, current state, or even a real-time clock. + During command execution this callback runs in a background UI thread. Protect shared + state with a lock when necessary. The toolbar is suspended while another prompt, 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, 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.main_session.bottom_toolbar is None + or not self._is_tty_session(self.main_session) + ): + yield + return + + toolbar = command_toolbar.CommandToolbar(self) + try: + with self.sigint_protection: + toolbar.start() + self._command_toolbar = toolbar + 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 + def get_rprompt(self) -> AnyFormattedText: """Provide text to populate the prompt-toolkit right prompt. @@ -3107,6 +3155,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] @@ -3344,32 +3393,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: @@ -3428,29 +3495,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. @@ -3552,6 +3625,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, @@ -3833,7 +3907,11 @@ def _cmdloop(self) -> None: """ try: # Run startup commands - stop = self.runcmds_plus_hooks(self._startup_commands) + if self._startup_commands: + with self._command_toolbar_context(): + stop = self.runcmds_plus_hooks(self._startup_commands) + else: + stop = False self._startup_commands.clear() while not stop: @@ -3847,7 +3925,8 @@ def _cmdloop(self) -> None: line = "_eof" # Run the command along with all associated pre and post hooks - stop = self.onecmd_plus_hooks(line) + with self._command_toolbar_context(): + stop = self.onecmd_plus_hooks(line) finally: with self.sigint_protection: # Shut down the alert thread. @@ -4677,6 +4756,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. @@ -4885,6 +4965,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 @@ -5003,6 +5084,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. @@ -5214,6 +5296,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. diff --git a/cmd2/command_toolbar.py b/cmd2/command_toolbar.py new file mode 100644 index 000000000..4db580f53 --- /dev/null +++ b/cmd2/command_toolbar.py @@ -0,0 +1,355 @@ +"""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 typing import TYPE_CHECKING, Any, TextIO, TypeVar, cast + +from prompt_toolkit.application import Application, create_app_session +from prompt_toolkit.filters import Condition, is_done, renderer_height_is_known, to_filter +from prompt_toolkit.input.typeahead import get_typeahead, store_typeahead +from prompt_toolkit.key_binding import KeyBindings +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.layout.controls import FormattedTextControl +from prompt_toolkit.patch_stdout import StdoutProxy +from prompt_toolkit.utils import suspend_to_background_supported + +if TYPE_CHECKING: + from .cmd2 import Cmd + +_F = TypeVar("_F", bound=Callable[..., Any]) + + +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: + """Run a prompt-toolkit display in a thread 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 + 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.app: Application[None] = Application( + layout=Layout( + HSplit( + [ + Window(height=0), + Window(), + ConditionalContainer( + Window( + FormattedTextControl(lambda: session.bottom_toolbar, style="class:bottom-toolbar.text"), + style="class:bottom-toolbar", + height=1, + always_hide_cursor=True, + ), + filter=~is_done & renderer_height_is_known, + ), + ] + ) + ), + input=session.input, + output=session.output, + style=session.style, + color_depth=session.color_depth, + refresh_interval=session.refresh_interval, + key_bindings=bindings, + erase_when_done=True, + after_render=lambda _: 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 + 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: + 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 + # 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 + + @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/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/prompt.md b/docs/features/prompt.md index 0f8a09082..6a40912f0 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,31 @@ 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. + +During command execution, `get_bottom_toolbar()` runs in a background UI thread. Keep the callback +fast and use a lock when reading state that a command or another thread modifies. + +### Commands That Take Over the Terminal + +cmd2 temporarily hides the toolbar for its input prompts, 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/test_command_toolbar.py b/tests/test_command_toolbar.py new file mode 100644 index 000000000..c3c8a99d8 --- /dev/null +++ b/tests/test_command_toolbar.py @@ -0,0 +1,497 @@ +"""Command toolbar lifecycle and terminal integration tests.""" + +import io +import sys +import threading +import time +from types import SimpleNamespace +from unittest import mock + +import pytest +from prompt_toolkit.application import create_app_session, get_app +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.output import DummyOutput +from prompt_toolkit.shortcuts import PromptSession + +from cmd2 import Cmd + + +class Terminal(io.StringIO): + def isatty(self) -> bool: + return True + + +class RecordingOutput(DummyOutput): + def __init__(self, stream: Terminal) -> None: + self.stdout = stream + + 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 = 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 + + +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) + + +def test_command_toolbar_pipe_process_inherits_terminal(toolbar_app, tmp_path) -> None: + app, _, _ = toolbar_app + 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) + readers.append(app._cur_pipe_proc_reader) + app.poutput("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() + + 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, _, _ = toolbar_app + + 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 + 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 + + +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_render_failure(toolbar_app) -> None: + app, _, output = toolbar_app + + def broken_toolbar(): + raise ValueError("broken toolbar") + + app.main_session.bottom_toolbar = broken_toolbar + with pytest.raises(ValueError, match="broken toolbar"), app._command_toolbar_context(): + pytest.fail("Command should not run after a toolbar startup failure") + assert app.stdout is output + assert app._command_toolbar is None + + +@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"]