diff --git a/CHANGELOG.md b/CHANGELOG.md index 45bff09ed..48e92d434 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,28 @@ -## 4.2.4 (TBD) +## 4.3.0 (TBD) - Bug Fixes + - Fixed output redirection and piping raising `UnicodeEncodeError` and leaving an empty file + behind. Command output is rendered by Rich and contains box-drawing characters, which no + `cp125x` Windows ANSI code page can represent, so redirecting or piping it failed. This + affects current Windows 11 with default settings on Python 3.11 through 3.14, not only older + systems. Redirection targets and pipes now use UTF-8 explicitly - 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 +- Enhancements + - `enable_bottom_toolbar=True` now keeps the toolbar visible and refreshing during command + execution + - `Cmd.read_input()` and `Cmd.read_secret()` now keep the bottom toolbar visible while they wait + for input, refreshing at the same `refresh_interval` as the main prompt, instead of the + toolbar disappearing for the duration of the nested prompt. `Cmd.select()` is unchanged, since + prompt-toolkit's `choice()` offers no way to configure a refresh interval + - 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..7d4b466c2 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,65 @@ 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 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 + def get_rprompt(self) -> AnyFormattedText: """Provide text to populate the prompt-toolkit right prompt. @@ -3107,6 +3185,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] @@ -3324,9 +3403,11 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState: # Create a pipe with read and write sides read_fd, write_fd = os.pipe() - # Open each side of the pipe - subproc_stdin = open(read_fd) # noqa: SIM115 - new_stdout: TextIO = cast(TextIO, open(write_fd, "w")) # noqa: SIM115 + # Open each side of the pipe. Both ends are given an explicit encoding: + # command output is rendered by Rich and routinely contains non-ASCII, which + # the locale encoding cannot always represent. + subproc_stdin = open(read_fd, encoding="utf-8") # noqa: SIM115 + new_stdout: TextIO = cast(TextIO, open(write_fd, "w", encoding="utf-8")) # noqa: SIM115 # Create pipe process in a separate group to isolate our signals from it. If a Ctrl-C event occurs, # our sigint handler will forward it only to the most recent pipe process. This makes sure pipe @@ -3342,32 +3423,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: @@ -3375,8 +3474,15 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState: # statement.output can only contain REDIRECTION_APPEND or REDIRECTION_OUTPUT mode = "a" if statement.redirector == constants.REDIRECTION_APPEND else "w" try: - # Use line buffering - new_stdout = cast(TextIO, open(su.strip_quotes(statement.redirect_to), mode=mode, buffering=1)) # noqa: SIM115 + # Use line buffering. The encoding is explicit rather than the + # locale's: command output is rendered by Rich and routinely contains + # non-ASCII, so on a non-UTF-8 system -- a default Windows console, + # for instance -- redirection would otherwise fail and leave an empty + # file behind. + new_stdout = cast( + TextIO, + open(su.strip_quotes(statement.redirect_to), mode=mode, buffering=1, encoding="utf-8"), # noqa: SIM115 + ) except OSError as ex: raise RedirectionError("Failed to redirect output") from ex @@ -3419,29 +3525,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 +3655,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, @@ -3672,6 +3785,7 @@ def read_input( temp_session: PromptSession[str] = PromptSession( auto_suggest=self.main_session.auto_suggest, + bottom_toolbar=self.get_bottom_toolbar if self.main_session.bottom_toolbar is not None else None, color_depth=self.main_session.color_depth, complete_style=self.main_session.complete_style, complete_in_thread=self.main_session.complete_in_thread, @@ -3682,6 +3796,7 @@ def read_input( key_bindings=self.main_session.key_bindings, input=self.main_session.input, output=self.main_session.output, + refresh_interval=self.main_session.refresh_interval, style=self.main_session.style, ) @@ -3699,10 +3814,12 @@ def read_secret( :raises Exception: any other exceptions raised by prompt() """ temp_session: PromptSession[str] = PromptSession( + bottom_toolbar=self.get_bottom_toolbar if self.main_session.bottom_toolbar is not None else None, color_depth=self.main_session.color_depth, enable_suspend=self.main_session.enable_suspend, input=self.main_session.input, output=self.main_session.output, + refresh_interval=self.main_session.refresh_interval, style=self.main_session.style, ) @@ -3824,7 +3941,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: @@ -3838,7 +3959,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. @@ -4668,6 +4790,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 +4999,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 +5118,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 +5330,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 +6044,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..96faacb31 --- /dev/null +++ b/cmd2/command_toolbar.py @@ -0,0 +1,468 @@ +"""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.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.patch_stdout import StdoutProxy +from prompt_toolkit.utils import suspend_to_background_supported + +from .pager import Pager, output_fits + +if TYPE_CHECKING: + from .cmd2 import Cmd + +_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 + + 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(): + # The callback finished while this poll was expiring, or raised a + # TimeoutError of its own -- indistinguishable here, because + # concurrent.futures.TimeoutError is TimeoutError on Python 3.11+. + # Ask the future for its outcome rather than re-raising this poll's + # timeout, which would report a failure for a call that succeeded. + return result.result() + 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 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..2368ab180 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,43 @@ 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 + +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 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. + +[cmd2.Cmd.read_input][] and [cmd2.Cmd.read_secret][] keep the toolbar visible, and it continues to +refresh at the same `refresh_interval` as the main prompt, so a clock or status display does not go +stale while your command waits for input. [cmd2.Cmd.select][] does not yet show one, because +prompt-toolkit's `choice()` has no way to configure a refresh interval and its toolbar would go +stale while the selection sits idle. + +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..db1af0b63 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 @@ -80,6 +86,23 @@ def run_cmd(app: cmd2.Cmd, cmd: str) -> tuple[list[str], list[str]]: return normalize(out), normalize(err) +#: Environment variables that change how Rich and cmd2 render output. Left inherited, +#: they make unrelated tests fail depending on who runs the suite: NO_COLOR fails 15 +#: tests, FORCE_COLOR and TTY_COMPATIBLE 53 each. +COLOR_ENVIRONMENT = ("NO_COLOR", "FORCE_COLOR", "TTY_COMPATIBLE", "TTY_INTERACTIVE") + + +@pytest.fixture(autouse=True) +def neutral_color_environment(monkeypatch: pytest.MonkeyPatch) -> None: + """Render output the same way regardless of the caller's environment. + + Tests that exercise these variables set them explicitly, which still works because + a test's own monkeypatching runs after this fixture. + """ + for name in COLOR_ENVIRONMENT: + monkeypatch.delenv(name, raising=False) + + @pytest.fixture def base_app() -> cmd2.Cmd: return cmd2.Cmd(include_py=True, include_ipy=True) @@ -192,3 +215,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_cmd2.py b/tests/test_cmd2.py index 6f4ae3656..1aa0dca81 100644 --- a/tests/test_cmd2.py +++ b/tests/test_cmd2.py @@ -2391,6 +2391,72 @@ def test_read_secret_eof(base_app, monkeypatch): base_app.read_secret("Secret: ") +def _capture_nested_session(app, call): + """Run `call` and return the real PromptSession cmd2 built for the nested prompt.""" + captured = [] + + def fake_read_raw_input(prompt, session, **kwargs): + captured.append(session) + return "typed" + + with mock.patch.object(app, "_read_raw_input", side_effect=fake_read_raw_input): + call() + assert len(captured) == 1 + return captured[0] + + +def test_read_input_session_shows_the_bottom_toolbar(base_app) -> None: + """A nested prompt keeps the persistent toolbar instead of dropping it. + + read_input() builds its own PromptSession, so without being told about the + toolbar the bar simply vanishes for as long as the nested prompt is up. + """ + base_app.main_session.bottom_toolbar = base_app.get_bottom_toolbar + + session = _capture_nested_session(base_app, lambda: base_app.read_input("Prompt> ")) + + assert session.bottom_toolbar == base_app.get_bottom_toolbar + + +def test_read_input_session_inherits_refresh_interval(base_app) -> None: + """A toolbar clock keeps ticking while the nested prompt waits for input.""" + base_app.main_session.bottom_toolbar = base_app.get_bottom_toolbar + base_app.main_session.refresh_interval = 0.25 + + session = _capture_nested_session(base_app, lambda: base_app.read_input("Prompt> ")) + + assert session.refresh_interval == 0.25 + + +def test_read_secret_session_shows_the_bottom_toolbar(base_app) -> None: + """Reading a secret keeps the persistent toolbar; the bar is not the secret.""" + base_app.main_session.bottom_toolbar = base_app.get_bottom_toolbar + + session = _capture_nested_session(base_app, lambda: base_app.read_secret("Secret: ")) + + assert session.bottom_toolbar == base_app.get_bottom_toolbar + + +def test_read_secret_session_inherits_refresh_interval(base_app) -> None: + """A toolbar clock keeps ticking while a secret is being entered.""" + base_app.main_session.bottom_toolbar = base_app.get_bottom_toolbar + base_app.main_session.refresh_interval = 0.25 + + session = _capture_nested_session(base_app, lambda: base_app.read_secret("Secret: ")) + + assert session.refresh_interval == 0.25 + + +@pytest.mark.parametrize("method", ["read_input", "read_secret"]) +def test_nested_prompt_has_no_toolbar_when_the_app_has_none(base_app, method) -> None: + """An app without a bottom toolbar must not sprout one at a nested prompt.""" + assert base_app.main_session.bottom_toolbar is None + + session = _capture_nested_session(base_app, lambda: getattr(base_app, method)("Prompt> ")) + + assert session.bottom_toolbar is None + + def test_read_input_passes_all_arguments_to_resolver(base_app): mock_choices = ["choice1", "choice2"] mock_provider = mock.MagicMock(name="provider") diff --git a/tests/test_command_toolbar.py b/tests/test_command_toolbar.py new file mode 100644 index 000000000..b2cf2d099 --- /dev/null +++ b/tests/test_command_toolbar.py @@ -0,0 +1,808 @@ +"""Command toolbar lifecycle and terminal integration tests.""" + +import sys +import threading +import time +from concurrent.futures import Future, ThreadPoolExecutor +from concurrent.futures import TimeoutError as FutureTimeoutError +from types import SimpleNamespace +from unittest import mock + +import pytest +from prompt_toolkit.application import 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.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(encoding="utf-8") + 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(encoding="utf-8") == "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(encoding="utf-8") + + +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_returns_a_result_that_lands_during_the_poll(toolbar_app, monkeypatch) -> None: + """A callback finishing while the poll expires must return its value, not a timeout. + + `concurrent.futures.TimeoutError` is `TimeoutError` on Python 3.11+, so the poll + expiring and the callback raising a timeout of its own are indistinguishable by type. + Re-raising the caught exception once the future is done therefore reports a timeout for + a call that actually succeeded. + """ + app, _, _ = toolbar_app + + class RacyFuture(Future): + """Completes, and only then reports the poll as having expired.""" + + def __init__(self) -> None: + super().__init__() + self._polled = False + + def result(self, timeout=None): # type: ignore[no-untyped-def] + if timeout is not None and not self._polled: + self._polled = True + super().result(timeout=5) # let the callback finish first + raise FutureTimeoutError # then act as though the poll had expired + return super().result(timeout) + + monkeypatch.setattr(command_toolbar, "Future", RacyFuture) + with app._command_toolbar_context(): + toolbar = app._command_toolbar + assert toolbar._call_in_ui(lambda: "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(encoding="utf-8") + assert "Cmd2 Commands" not in output.getvalue() 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_run_pyscript.py b/tests/test_run_pyscript.py index 69b335fca..e4030ebde 100644 --- a/tests/test_run_pyscript.py +++ b/tests/test_run_pyscript.py @@ -246,7 +246,7 @@ def test_run_pyscript_print_redirection(base_app, request, tmp_path, capsys) -> out, err = capsys.readouterr() # Verify the output file contains what we expect from print() - content = pathlib.Path(out_file).read_text() + content = pathlib.Path(out_file).read_text(encoding="utf-8") # Look for everything written to self.stdout assert len(content.splitlines()) == 4 diff --git a/tests/test_suite_environment.py b/tests/test_suite_environment.py new file mode 100644 index 000000000..bb9cb8d1f --- /dev/null +++ b/tests/test_suite_environment.py @@ -0,0 +1,59 @@ +"""Guards that the suite renders output independently of the developer's environment. + +Rich and cmd2 both consult environment variables when deciding whether to emit styling. +Inheriting them makes large numbers of unrelated tests fail depending on who runs them, +which is expensive to diagnose because the failures look like product regressions. +""" + +import os +import sys + +import pytest + +import cmd2 + +#: Variables that change how output is rendered. Rich reads all of these; cmd2 reads +#: NO_COLOR directly. Tests that exercise them set them explicitly instead. +COLOR_ENVIRONMENT = ("NO_COLOR", "FORCE_COLOR", "TTY_COMPATIBLE", "TTY_INTERACTIVE") + + +@pytest.mark.parametrize("name", COLOR_ENVIRONMENT) +def test_color_environment_does_not_leak_into_tests(name: str) -> None: + """A developer exporting any of these must not change the suite's results.""" + assert name not in os.environ, ( + f"{name} leaked into the test environment; output-rendering assertions would depend on who is running the suite" + ) + + +class EncodingProbe(cmd2.Cmd): + """Reports the encoding of whatever stream output is currently going to.""" + + def do_show_encoding(self, _: str) -> None: + """Print the current output stream's encoding.""" + self.poutput(f"ENCODING={getattr(self.stdout, 'encoding', None)}") + + +def test_redirection_to_a_file_uses_utf8(tmp_path) -> None: + """cmd2 renders non-ASCII, so a redirect target must not use the locale encoding. + + Opened with the locale encoding, redirecting styled output raises UnicodeEncodeError + on any non-UTF-8 system -- which includes a default Windows console -- leaving the + user an empty file and an error. + """ + app = EncodingProbe(allow_cli_args=False) + target = tmp_path / "out.txt" + app.onecmd_plus_hooks(f'show_encoding > "{target}"') + assert "ENCODING=utf-8" in target.read_text(encoding="utf-8") + + +#: A pass-through filter, run with this interpreter so the test does not depend on Unix +#: utilities being installed. `cmd.exe` has no `cat`, and this fix exists for Windows. +PASS_THROUGH = "import sys; sys.stdin.reconfigure(encoding='utf-8'); sys.stdout.write(sys.stdin.read())" + + +def test_piping_uses_utf8(tmp_path) -> None: + """The same applies to the pipe the subprocess reads from.""" + app = EncodingProbe(allow_cli_args=False) + target = tmp_path / "piped.txt" + app.onecmd_plus_hooks(f'show_encoding | "{sys.executable}" -c "{PASS_THROUGH}" > "{target}"') + assert "ENCODING=utf-8" in target.read_text(encoding="utf-8")