diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index 22f1d5d8c..907843ba9 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -86,6 +86,7 @@ from prompt_toolkit.output import DummyOutput, create_output from prompt_toolkit.patch_stdout import patch_stdout from prompt_toolkit.shortcuts import CompleteStyle, PromptSession, choice, set_title +from prompt_toolkit.shortcuts.choice_input import ChoiceInput from prompt_toolkit.styles import DynamicStyle from rich.console import ( Group, @@ -2165,13 +2166,41 @@ def _require_terminal_ownership(self) -> None: return raise RuntimeError("the bottom toolbar's display has not released the terminal") + def _await_terminal_ownership(self) -> bool: + """Wait for a display that would not stop to release the terminal before the main prompt. + + Prompting while its thread is still inside the application would put two readers on one + terminal, so the command loop waits rather than failing. The thread usually finishes once + its toolbar callback returns. If it never does, Ctrl-C gives up on it and ends the loop. + + :return: whether the terminal is ours to prompt on + """ + display = self._display_holding_terminal + if display is None: + return True + try: + # The display left the terminal in raw mode, where Ctrl-C is only a key, and the binding + # that would signal it runs on the event loop the stuck callback is blocking. Cook the + # terminal so its driver sends SIGINT again. This is not undone: a display that finishes + # restores the modes it found, and one given up on should leave the terminal cooked. + # From here on Ctrl-C can arrive at any point, including while the notice is printing. + self.main_session.app.input.cooked_mode().__enter__() + self.perror("Waiting for the bottom toolbar to release the terminal. Press Ctrl-C to quit.") + while display.thread_is_alive: + time.sleep(0.05) + except KeyboardInterrupt: + return False + self._require_terminal_ownership() + return True + @contextlib.contextmanager def suspend_bottom_toolbar(self) -> Generator[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. + its toolbar for external pagers and shell commands. Managed input prompts borrow + the reservation without releasing the rows. In reserved mode this also gives the reserved rows back, because a program that inherits the terminal knows nothing about a scroll region and would find its output @@ -2272,6 +2301,11 @@ def _command_toolbar_context(self) -> Generator[None]: with self.sigint_protection: try: toolbar.stop() + except command_toolbar._DisplayStillRunningError as exc: + # The display has already disabled itself and kept hold of the terminal, + # which the next prompt waits for. Like a display that cannot start, it + # must not escape cmdloop() over a toolbar callback that will not return. + self.perror(f"Disabling the bottom toolbar during commands: {exc}") finally: # Always forget a toolbar that has been torn down. Keeping a failed # one would disable the toolbar for the rest of the session. @@ -3860,9 +3894,9 @@ def _read_raw_input( The command display is stopped either way, but only some prompts give the terminal away with it. The main prompt is the one the reservation exists for: it renders through the reserved output, and the toolbar has to still be there while the user is - typing -- that is what "stable across ordinary commands" means. Any other session is - an application prompt cmd2 has not bound to the reservation, so it gets the terminal - to itself, rows included. + typing. Managed nested prompts on the same terminal borrow that reservation with + their own bridge after the command input reader stops. Other terminals and prompts + inside an external handoff retain the exclusive-terminal path. :param prompt: the prompt text or a callable that returns the prompt. :param session: the PromptSession instance to use for reading. @@ -3870,11 +3904,13 @@ def _read_raw_input( :return: the stripped input string. :raises EOFError: if the input stream is closed or the user signals EOF (e.g., Ctrl+D) """ - owns_the_reservation = session is self.main_session + reserved = self._reserved_toolbar + owns_the_reservation = session is self.main_session or (reserved is not None and reserved.can_manage(session)) with self._quiesce_bottom_toolbar() if owns_the_reservation else self.suspend_bottom_toolbar(): - reserved = self._reserved_toolbar if owns_the_reservation and reserved is not None and reserved.bridge is not None: reserved.bridge.finish_command_output() + with reserved.prompt_session(session): + return self._read_raw_input_now(prompt, session, **prompt_kwargs) return self._read_raw_input_now(prompt, session, **prompt_kwargs) def _read_raw_input_now( @@ -4002,7 +4038,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, + bottom_toolbar=self.main_session.bottom_toolbar, color_depth=self.main_session.color_depth, complete_style=self.main_session.complete_style, complete_in_thread=self.main_session.complete_in_thread, @@ -4031,7 +4067,7 @@ 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, + bottom_toolbar=self.main_session.bottom_toolbar, color_depth=self.main_session.color_depth, enable_suspend=self.main_session.enable_suspend, input=self.main_session.input, @@ -4166,6 +4202,9 @@ def _cmdloop(self) -> None: self._startup_commands.clear() while not stop: + if not self._await_terminal_ownership(): + break + # Get commands from user try: line = self._read_command_line(self.prompt) @@ -5007,7 +5046,7 @@ def do_quit(self, _: argparse.Namespace) -> bool | None: self.last_result = True return True - @command_toolbar.suspend_toolbar + @command_toolbar.quiesce_toolbar def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], prompt: str = "Your choice? ") -> Any: """Present a menu to the user. @@ -5042,7 +5081,22 @@ def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], p try: while True: with create_app_session(input=self.main_session.input, output=self.main_session.output): - result = choice(message=prompt, options=fulloptions) + reserved = self._reserved_toolbar + if reserved is not None and reserved.can_manage(self.main_session): + if reserved.bridge is not None: + reserved.bridge.finish_command_output() + # ChoiceInput.prompt() constructs and runs in one step. Binding + # its application first gives the initial frame reserved geometry. + selection = ChoiceInput( + message=prompt, + options=fulloptions, + style=self.main_session.style, + enable_suspend=self.main_session.enable_suspend, + )._create_application() + with reserved.prompt_application(selection): + result = selection.run() + else: + result = choice(message=prompt, options=fulloptions) if result is not None: return result except KeyboardInterrupt: diff --git a/cmd2/command_toolbar.py b/cmd2/command_toolbar.py index 5d70bad12..8489361a9 100644 --- a/cmd2/command_toolbar.py +++ b/cmd2/command_toolbar.py @@ -8,12 +8,13 @@ import signal import sys import threading +import time from collections.abc import Callable, Generator 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.application import Application, create_app_session, run_in_terminal from prompt_toolkit.application.current import _current_app_session, get_app_session from prompt_toolkit.enums import EditingMode from prompt_toolkit.filters import Condition, to_filter @@ -43,6 +44,33 @@ _R = TypeVar("_R") +def suspend_process_group(suspend_group: bool = True) -> None: + """Stop this process, or its whole process group, with SIGTSTP from any thread. + + A process-directed stop signal can be taken by a thread other than the sender. Sent + from the display thread, it lands on the main thread, and the display thread carries + on for a few milliseconds: it puts the terminal back into raw mode and redraws before + the job has stopped. When the shell resumes the job it restores the modes it saved + beforehand, and nothing is left to undo that -- the terminal stays cooked, keys are + held until Enter, and every cursor-position query is echoed as ``^[[row;colR``. + + So after sending the signal this thread waits for the stop to take it too. A stop + shows as time passing while asleep; a signal that was ignored, or discarded for an + orphaned process group, shows as nothing, and the wait ends on its own. + + :param suspend_group: stop the whole process group, as a shell's Ctrl-Z would, rather + than only this process + """ + if not hasattr(signal, "SIGTSTP"): # pragma: no cover - POSIX only + return + os.kill(0 if suspend_group else os.getpid(), signal.SIGTSTP) + deadline = time.monotonic() + 0.25 + while (before := time.monotonic()) < deadline: + time.sleep(0.01) + if time.monotonic() - before > 0.1: + return + + def suspend_toolbar(func: _F) -> _F: """Give a method exclusive access to the terminal, reserved rows included. @@ -217,6 +245,12 @@ def __init__(self, cmd: "Cmd") -> None: session = cmd.main_session self.app = session.app + # Ctrl-Z during a command is handled on the display's thread, where upstream's + # version would not hold the thread until the job has stopped. The reserved toolbar + # layers its row release over whatever is installed, so one it put there first is + # left in place; it releases the rows and then stops the process the same way. + if "suspend_to_background" not in vars(self.app): + cast("Any", self.app).suspend_to_background = self._suspend_to_background # 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, @@ -282,6 +316,11 @@ def suspend(event: KeyPressEvent) -> None: self._bindings = bindings self._suspend_binding = suspend + @staticmethod + def _suspend_to_background(suspend_group: bool = True) -> None: + """Suspend like upstream's ``Application.suspend_to_background()``, from any thread.""" + run_in_terminal(functools.partial(suspend_process_group, suspend_group)) + def _display_started(self, app: Application[str]) -> None: # noqa: ARG002 """Report that the display is up and has finished its first frame.""" self._ready.set() @@ -721,7 +760,9 @@ def page(self, text: str, *, chop: bool) -> None: # 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): + reserved = self.cmd.reserved_toolbar + available_rows = size.rows if reserved is not None and reserved.is_active else max(0, size.rows - toolbar_height) + if output_fits(text, size.columns, available_rows, chop=chop): self.cmd.stdout.write(text) self.cmd.stdout.flush() return @@ -736,6 +777,7 @@ def page(self, text: str, *, chop: bool) -> None: entered = False restored = False close_error: BaseException | None = None + handoff = contextlib.ExitStack() def restore() -> None: """Give the application back to the display, whichever thread is doing it. @@ -763,6 +805,10 @@ def enter() -> None: # duration -- otherwise the pager swaps in its layout and nothing is ever painted. self._set_render_suppressed(False) self.app.renderer.erase() + if reserved is not None: + # The very first pager layout must see the physical size. Releasing from + # enter_alternate_screen during replay is too late: that frame was measured. + handoff.enter_context(reserved.suspended(defer_band_clear=True)) self.app.layout = layout self.app.key_bindings = pager.bindings self.app.editing_mode = EditingMode.EMACS @@ -776,8 +822,20 @@ def leave() -> None: entered = False try: self.app.renderer.erase() + except BaseException: + # An erase can fail before quitting the alternate screen. Complete upstream's + # mode/buffer cleanup before returning the main-screen reservation; never + # retry the erase itself, which may already have changed visible output. + with contextlib.suppress(Exception): + transaction = ( + reserved.lock.transaction("pager cleanup") if reserved is not None else contextlib.nullcontext() + ) + with transaction: + self.app.renderer.reset() + raise finally: restore() + handoff.__exit__(*sys.exc_info()) self.app.renderer.request_absolute_cursor_position() self.app.invalidate() @@ -813,6 +871,7 @@ def close() -> None: # the full-screen flag and editing mode would carry into the next prompt. if not restored: restore() + handoff.__exit__(*sys.exc_info()) # A reservation abandoned while the pager was open deferred its legacy fallback until # the pager closed; now that it has, on the main screen, finish it. if self._legacy_fallback_pending and self.thread_is_alive: diff --git a/cmd2/managed_output.py b/cmd2/managed_output.py index b0425f598..812b2aa38 100644 --- a/cmd2/managed_output.py +++ b/cmd2/managed_output.py @@ -63,6 +63,9 @@ def write(self, data: str) -> int: """ with self._lock.transaction("managed write"): try: + before_write = getattr(self.bridge, "before_managed_write", None) + if before_write is not None: + before_write() if self._output is None: written = self._stream.write(data) else: diff --git a/cmd2/prompt_toolkit_bridge.py b/cmd2/prompt_toolkit_bridge.py index c4ebccae9..8590695b6 100644 --- a/cmd2/prompt_toolkit_bridge.py +++ b/cmd2/prompt_toolkit_bridge.py @@ -300,6 +300,13 @@ def forget_unfinished_command_output(self) -> None: """ self._unfinished_command_output = False + def before_managed_write(self) -> None: + """Remove retained pager-startup cells before output can scroll them into history. + + Called inside the writer's transaction, before emitting any command text. + """ + self._display.clear_deferred_band() + def finish_command_output(self) -> None: """Start the next prompt on a fresh line if command output left one unfinished. @@ -831,6 +838,9 @@ def commit(self, prepared: PreparedRender) -> bool: self.require_resynchronization("the frame was laid out for a different size") return False try: + # Preparation may be expensive for a large pager. Its main-screen bar is + # retained until this validated batch is ready to replace the screen. + self._display.clear_deferred_band() prepared.batch.replay(self._display.output) self._display.output.flush() except Exception as error: # noqa: BLE001 - the terminal's state is now unknown diff --git a/cmd2/reserved_output.py b/cmd2/reserved_output.py index 58bc742d3..1f28831bc 100644 --- a/cmd2/reserved_output.py +++ b/cmd2/reserved_output.py @@ -55,6 +55,7 @@ def __init__(self, wrapped: Output, display: "TerminalDisplay") -> None: """ self._wrapped = wrapped self._display = display + self._alternate_handoff_owned = False # A plain attribute rather than a property: Output declares stdout as writable, and # code that reaches for the real stream must find the backend's, not a copy of it. self.stdout = getattr(wrapped, "stdout", None) @@ -158,13 +159,17 @@ def enter_alternate_screen(self) -> None: nothing about a reservation. Margins are restored before the switch so the main buffer is left in the state the shell expects if the switch is never undone. """ - self._display.release_region_for_handoff() + self._alternate_handoff_owned = not self._display.handoff_active + if self._alternate_handoff_owned: + self._display.release_region_for_handoff() self._wrapped.enter_alternate_screen() def quit_alternate_screen(self) -> None: """Return to the main buffer and re-establish the reservation for its geometry.""" self._wrapped.quit_alternate_screen() - self._display.reacquire_region_after_handoff() + if self._alternate_handoff_owned: + self._alternate_handoff_owned = False + self._display.reacquire_region_after_handoff() def scroll_buffer_to_prompt(self) -> None: """Scroll the Windows viewport to the prompt, then re-check the viewport origin. diff --git a/cmd2/reserved_toolbar.py b/cmd2/reserved_toolbar.py index aeb477b6d..0b654eb82 100644 --- a/cmd2/reserved_toolbar.py +++ b/cmd2/reserved_toolbar.py @@ -17,15 +17,19 @@ new ``bottom_toolbar`` to the session still reaches the band. """ -from contextlib import contextmanager, suppress +import signal +from contextlib import ExitStack, contextmanager, suppress from types import TracebackType from typing import TYPE_CHECKING, Any, Self -from prompt_toolkit.filters import Condition +from prompt_toolkit.application import run_in_terminal +from prompt_toolkit.filters import Condition, Never from prompt_toolkit.layout import HSplit, Window from prompt_toolkit.layout.containers import ConditionalContainer from prompt_toolkit.styles import DynamicStyle +from prompt_toolkit.utils import suspend_to_background_supported +from .command_toolbar import suspend_process_group from .prompt_toolkit_bridge import PromptToolkitBridge from .reserved_output import ReservedOutput from .terminal_display import TerminalDisplay @@ -107,6 +111,9 @@ def __init__( self._original_filter: Any = None self._installed_filter: Any = None self.stopped_handler: Callable[[], None] | None = None + self._job_control_stack = ExitStack() + self._nested_stacks: list[ExitStack] = [] + self._nested_bridges: list[PromptToolkitBridge] = [] @property def is_active(self) -> bool: @@ -132,7 +139,7 @@ def display(self) -> TerminalDisplay: @property def bridge(self) -> PromptToolkitBridge | None: """The renderer bridge while active, else ``None``.""" - return self._bridge + return self._nested_bridges[-1] if self._nested_bridges else self._bridge @property def painter(self) -> ToolbarPainter | None: @@ -195,6 +202,7 @@ def start(self) -> bool: # From here the application's own renders go through prepare and commit, which is # what puts them in the same queue as command output and toolbar paints. self._bridge.bind(app) + self._job_control_stack.enter_context(self._job_control(app)) # When the bridge abandons reserved rendering it cannot resume anything itself: # the rows are still withheld and the renderer is still routed through it. Giving # them back is this object's job, and it is what lets compatibility rendering @@ -240,7 +248,102 @@ def take_pending_error(self) -> BaseException | None: return error @contextmanager - def suspended(self) -> "Generator[None]": + def _job_control(self, app: Any) -> "Generator[None]": + """Release the physical reservation inside upstream's cooked-mode handoff.""" + original = app.suspend_to_background + # Whether anything had replaced upstream's method. The command display installs its own + # only where nothing has, so writing the bound method back would shut it out for good. + replaced = "suspend_to_background" in vars(app) + + def suspend_to_background(suspend_group: bool = True) -> None: + suspend_signal = getattr(signal, "SIGTSTP", None) + if suspend_signal is None or not suspend_to_background_supported(): + return + + def suspend_process() -> None: + # run_in_terminal has stopped rendering and detached input before this runs. + # A signal callback itself must never acquire the terminal transaction. + with self.suspended(): + suspend_process_group(suspend_group) + + run_in_terminal(suspend_process) + + app.suspend_to_background = suspend_to_background + try: + yield + finally: + if app.suspend_to_background is suspend_to_background: + if replaced: + app.suspend_to_background = original + else: + del app.suspend_to_background + + def can_manage(self, session: "PromptSession[Any]") -> bool: + """Whether a temporary prompt uses this terminal and its managed input reader.""" + return ( + self._display is not None + and not self._display.handoff_active + and session.input is self._session.input + and session.app.output in (self._bound_output, self._display.terminal.output) + and native_toolbar_container(session) is not None + ) + + @contextmanager + def prompt_session(self, session: "PromptSession[Any]") -> "Generator[None]": + """Lend the reservation to a temporary prompt after the command reader has stopped.""" + if session is self._session: + yield + return + native = native_toolbar_container(session) + if native is None: + raise RuntimeError("cannot locate the nested session's bottom toolbar window") + with self.prompt_application(session.app, native): + yield + + @contextmanager + def prompt_application(self, app: Any, native: ConditionalContainer | None = None) -> "Generator[None]": + """Bind a managed input application while retaining the physical reservation.""" + previous_output, previous_renderer_output = app.output, app.renderer.output + previous_filter = native.filter if native is not None else Never() + output = self._bound_output + installed_filter = previous_filter & Condition(lambda: not self.is_active) + bridge = PromptToolkitBridge(renderer=app.renderer, display=self.display, lock=self._lock) + previous_bridge = self.bridge + if previous_bridge is not None: + previous_bridge.forget_prompt_anchor() + previous_bridge.note_owner_change() + + def restore() -> None: + bridge.unbind() + if bridge in self._nested_bridges: + self._nested_bridges.remove(bridge) + if app.output is output: + app.output = previous_output + if app.renderer.output is output: + app.renderer.output = previous_renderer_output + if native is not None and native.filter is installed_filter: + native.filter = previous_filter + if previous_bridge is not None: + previous_bridge.forget_prompt_anchor() + previous_bridge.require_resynchronization("a nested prompt returned the terminal") + + with ExitStack() as stack: + stack.callback(restore) + # Abandonment restores the guest immediately, before native rendering resumes. + self._nested_stacks.append(stack) + stack.callback(self._nested_stacks.remove, stack) + app.output = app.renderer.output = output + if native is not None: + native.filter = installed_filter + self._nested_bridges.append(bridge) + bridge.bind(app) + bridge.set_frame_committed_handler(self.refresh) + bridge.set_emission_stopped_handler(self._emission_stopped) + stack.enter_context(self._job_control(app)) + yield + + @contextmanager + def suspended(self, *, defer_band_clear: bool = False) -> "Generator[None]": """Give the rows back for the duration of the block, and take them again after. A program that inherits the terminal -- a shell command, an editor, an external pager @@ -261,31 +364,50 @@ def suspended(self) -> "Generator[None]": says nothing about whose terminal it is: the guest the outer block handed it to still has it, and reinstalling margins or painting a band over their screen would be the same mistake as never releasing at all. + + :param defer_band_clear: keep the main-screen bar visible during managed pager + preparation; external terminal users must retain the default immediate clear. """ display = self._display if display is None: yield return - outermost = self._suspend_depth == 0 + outermost = self._suspend_depth == 0 and not display.handoff_active self._suspend_depth += 1 + body_failed = False try: + if not defer_band_clear: + with self._lock.transaction("clear retained toolbar"): + display.clear_deferred_band() if outermost: with self._lock.transaction("suspend"): - display.release_region_for_handoff() + display.release_region_for_handoff(defer_band_clear=defer_band_clear) # Forgotten as the terminal changes hands, not after the guest has # finished with it. From this moment the remembered row describes a screen # someone else is writing on, and anything that rendered against it would # paint over their output. self._invalidate_ownership("the terminal was handed to another program") yield + except BaseException: + body_failed = True + raise finally: self._suspend_depth -= 1 if outermost: - with self._lock.transaction("resume"): - display.reacquire_region_after_handoff() - self._invalidate_ownership("the terminal came back from another program") - self.refresh() + try: + with self._lock.transaction("resume"): + display.reacquire_region_after_handoff() + self._invalidate_ownership("the terminal came back from another program") + self.refresh() + except Exception as error: + self._pending_error = error + with suppress(Exception): + self.stop() + # A failed guest (including Ctrl-C/termination) remains the reason for + # unwinding; cleanup failure is available through take_pending_error(). + if not body_failed: + raise def _invalidate_ownership(self, reason: str) -> None: """Discard everything that described the screen before ownership changed. @@ -298,6 +420,9 @@ def _invalidate_ownership(self, reason: str) -> None: self._bridge.forget_prompt_anchor() self._bridge.forget_unfinished_command_output() self._bridge.require_resynchronization(reason) + for bridge in self._nested_bridges: + bridge.forget_prompt_anchor() + bridge.require_resynchronization(reason) def refresh(self) -> bool: """Evaluate the toolbar's content and paint whatever changed. @@ -335,8 +460,8 @@ def _emission_stopped(self) -> None: The error the bridge is holding is taken here rather than left with it: the bridge is dropped a moment later, and an error the user never sees is the same as none. """ - if self._bridge is not None and self._pending_error is None: - self._pending_error = self._bridge.take_pending_error() + if self.bridge is not None and self._pending_error is None: + self._pending_error = self.bridge.take_pending_error() with suppress(Exception): self.stop() @@ -357,8 +482,8 @@ def _paint_failed(self, error: BaseException) -> None: """ self._pending_error = error self._consecutive_paint_failures += 1 - if self._bridge is not None: - self._bridge.require_resynchronization("a toolbar paint failed; the cursor's position is unknown") + if self.bridge is not None: + self.bridge.require_resynchronization("a toolbar paint failed; the cursor's position is unknown") if self._consecutive_paint_failures >= _MAX_CONSECUTIVE_PAINT_FAILURES: with suppress(Exception): self.stop() @@ -371,6 +496,9 @@ def stop(self) -> None: other. """ display, self._display = self._display, None + for stack in reversed(tuple(self._nested_stacks)): + stack.close() + self._job_control_stack.close() if self._bridge is not None: self._bridge.unbind() self._bridge = None diff --git a/cmd2/terminal_display.py b/cmd2/terminal_display.py index 50184b192..b9d8e885e 100644 --- a/cmd2/terminal_display.py +++ b/cmd2/terminal_display.py @@ -330,6 +330,7 @@ def __init__(self, output: "Output", reserved_rows: int = 1) -> None: self._depth = 0 self._adapter: Any = None self._handoff_active = False + self._deferred_band: Geometry | None = None @property def terminal(self) -> PhysicalTerminal: @@ -346,6 +347,11 @@ def is_reserved(self) -> bool: """Whether a reservation is currently installed.""" return self._geometry is not None + @property + def handoff_active(self) -> bool: + """Whether another terminal owner holds the screen, even below the height floor.""" + return self._handoff_active + @property def output(self) -> "Output": """The output callers should render through: the adapter while reserved, else the backend.""" @@ -428,6 +434,7 @@ def release(self) -> None: def _teardown(self) -> None: """Restore full-screen margins and drop the adapter.""" + self.clear_deferred_band() self._adapter = None self._handoff_active = False if self._geometry is None: @@ -488,7 +495,7 @@ def reconfigure(self) -> bool: self._adapter = self._make_adapter() return True - def release_region_for_handoff(self) -> None: + def release_region_for_handoff(self, *, defer_band_clear: bool = False) -> None: """Restore full-screen margins for a program taking the terminal over. The reservation is remembered rather than dropped: the lease is still held, and @@ -500,6 +507,9 @@ def release_region_for_handoff(self) -> None: that had shrunk below the floor is released but still ours, and a guest can take it from that state just as readily; tying the record to the geometry snapshot would let a later resize reinstall margins over the guest's screen. + + :param defer_band_clear: retain the bar while a managed pager prepares its first + frame. The bridge clears it before commit; recovery and teardown clear it too. """ if self._depth == 0: return @@ -507,7 +517,19 @@ def release_region_for_handoff(self) -> None: if self._geometry is None: return geometry, self._geometry = self._geometry, None - self._terminal.release_region(geometry) + if defer_band_clear: + # A managed pager can prepare its full-screen frame while the old bar remains + # visible. Clear it only when that frame is ready to replace the main screen. + self._deferred_band = geometry + self._terminal.release_region() + else: + self._terminal.release_region(geometry) + + def clear_deferred_band(self) -> None: + """Clear a pager's retained main-screen bar before emission, recovery, or shutdown.""" + geometry, self._deferred_band = self._deferred_band, None + if geometry is not None: + self._terminal.release_region(geometry) def reacquire_region_after_handoff(self) -> None: """Re-establish the reservation after a program hands the terminal back. @@ -518,6 +540,7 @@ def reacquire_region_after_handoff(self) -> None: """ if not self._handoff_active or self._depth == 0: return + self.clear_deferred_band() self._handoff_active = False # Coming back is an ordinary reconfiguration, so it goes through the one path that # does the whole job: re-check backend capability, measure, install only if the diff --git a/cmd2/toolbar_mode.py b/cmd2/toolbar_mode.py index a83e896ec..3b39ce718 100644 --- a/cmd2/toolbar_mode.py +++ b/cmd2/toolbar_mode.py @@ -25,6 +25,7 @@ """ from enum import StrEnum +from importlib.metadata import PackageNotFoundError from importlib.metadata import version as _installed_version from typing import TYPE_CHECKING @@ -84,7 +85,15 @@ def _dependency_capability(version: str | None = None) -> tuple[bool, str]: :param version: the version to judge; the installed one by default :return: whether it is qualified, and a reason suitable for diagnostics """ - installed = version if version is not None else _installed_version("prompt_toolkit") + if version is not None: + installed = version + else: + try: + installed = _installed_version("prompt_toolkit") + except PackageNotFoundError: + # A frozen or vendored application may ship without package metadata. An unknown + # version is not a qualified one, and auto has to fall back rather than fail. + return False, "the installed prompt-toolkit version cannot be determined" if installed in QUALIFIED_PROMPT_TOOLKIT_VERSIONS: return True, "qualified prompt-toolkit" qualified = ", ".join(sorted(QUALIFIED_PROMPT_TOOLKIT_VERSIONS)) diff --git a/cmd2/toolbar_painter.py b/cmd2/toolbar_painter.py index 79310360c..063b7d416 100644 --- a/cmd2/toolbar_painter.py +++ b/cmd2/toolbar_painter.py @@ -355,7 +355,7 @@ def __init__( self._autowrap_after_paint = autowrap_after_paint self._last_frame: ToolbarFrame | None = None self._last_attrs: Mapping[str, Attrs] | None = None - self._last_band: tuple[int, int, int, object] | None = None + self._last_band: tuple[int, int, int, object, int] | None = None self._pending_error: BaseException | None = None @property @@ -438,7 +438,16 @@ def paint(self, prepared: PreparedFrame) -> bool: return False frame = prepared.frame - band = (geometry.physical_rows, geometry.columns, geometry.reserved_rows, geometry.buffer_id) + # Identical coordinates do not imply retained cells. A shrink below the + # reservation floor can erase the band, then growth can reacquire exactly + # the same dimensions and viewport. The new generation needs a full paint. + band = ( + geometry.physical_rows, + geometry.columns, + geometry.reserved_rows, + geometry.buffer_id, + geometry.generation, + ) previous = self._last_frame if band == self._last_band else None previous_attrs = self._last_attrs if previous is not None else None runs = _changed_runs(previous, previous_attrs, frame, prepared.attrs) diff --git a/examples/getting_started.py b/examples/getting_started.py index b5fe54312..aa88bdd11 100755 --- a/examples/getting_started.py +++ b/examples/getting_started.py @@ -60,6 +60,7 @@ def __init__(self) -> None: bottom_toolbar_mode=cmd2.ToolbarMode.AUTO, enable_rprompt=True, include_ipy=True, + include_py=True, persistent_history_file="cmd2_history.dat", refresh_interval=0.5, # refresh the UI twice a second to keep the bottom toolbar timestamp current shortcuts=shortcuts, diff --git a/examples/scripts/raise_exception.py b/examples/scripts/raise_exception.py new file mode 100644 index 000000000..857b3678d --- /dev/null +++ b/examples/scripts/raise_exception.py @@ -0,0 +1 @@ +raise ValueError("An example exception") diff --git a/examples/stage4_manual.py b/examples/stage4_manual.py new file mode 100644 index 000000000..a31b1ac14 --- /dev/null +++ b/examples/stage4_manual.py @@ -0,0 +1,40 @@ +## Example for testing nested input, secret input, and select input. Also demonstrates +## how to suspend the bottom toolbar and hand off the terminal to a guest. + +import time + +from getting_started import BasicApp + + +class ManualApp(BasicApp): + def do_nested(self, _arg): + value = self.read_input("Value> ", choices=["alpha", "beta"], history=["alpha", "beta"]) + self.poutput(repr(value)) + + def do_secret(self, _arg): + self.read_secret("Dummy secret> ") + self.poutput("Secret accepted; value not displayed") + + def do_choose(self, _arg): + self.poutput(self.select(["alpha", "beta"], "Choose> ")) + + def do_quiet(self, _arg): + time.sleep(10) + + def do_partial(self, _arg): + self.stdout.write("PARTIAL") + self.stdout.flush() + time.sleep(10) + self.stdout.write("END\n") + self.stdout.flush() + + def do_handoff(self, arg): + with self.suspend_bottom_toolbar(): + with self.suspend_bottom_toolbar(): + input("Guest owns terminal; resize, then Enter> ") + if arg.strip() == "fail": + raise RuntimeError("Intentional guest failure") + + +if __name__ == "__main__": + ManualApp().cmdloop() diff --git a/tests/test_command_toolbar.py b/tests/test_command_toolbar.py index a691d391c..29f96a784 100644 --- a/tests/test_command_toolbar.py +++ b/tests/test_command_toolbar.py @@ -1,6 +1,7 @@ """Command toolbar lifecycle and terminal integration tests.""" import contextlib +import subprocess import sys import threading import time @@ -119,6 +120,24 @@ def __getattr__(self, name): return getattr(self.file, name) +@pytest.mark.parametrize(("stdout_tty", "stderr_tty"), [(False, False), (True, False), (False, True), (True, True)]) +def test_pipeline_process_group_selection(toolbar_app, tmp_path, monkeypatch, running_pipe_process, stdout_tty, stderr_tty): + app, _, _ = toolbar_app + with (tmp_path / "stdout").open("w+") as out, (tmp_path / "stderr").open("w+") as err: + app.stdout = FileTerminal(out) if stdout_tty else out + monkeypatch.setattr(sys, "stderr", FileTerminal(err) if stderr_tty else err) + with mock.patch("subprocess.Popen", wraps=subprocess.Popen) as popen: + app.onecmd_plus_hooks(f'help | "{sys.executable}" -S -c "import sys; print(sys.stdin.read())"') + options = popen.call_args.kwargs + # A stream claiming isatty() is insufficient: these files do not refer to our + # controlling terminal, so no foreground handoff is safe. The pipeline is isolated + # the ordinary way instead: a new process group on Windows, a new session on POSIX. + isolation = "creationflags" if sys.platform == "win32" else "start_new_session" + expected = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else True + assert options[isolation] == expected + assert options.keys().isdisjoint({"process_group", "start_new_session", "creationflags"} - {isolation}) + + @pytest.mark.parametrize("builtin_pager", [False, True]) def test_command_toolbar_pipe_process_inherits_terminal(toolbar_app, tmp_path, builtin_pager, running_pipe_process) -> None: app, _, _ = toolbar_app @@ -223,6 +242,37 @@ def test_command_toolbar_ctrl_z(toolbar_app, supported, enabled) -> None: assert [key.key for key in keys] == ([] if supported and enabled else [Keys.ControlZ]) +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX job control") +@pytest.mark.parametrize("stopped", [True, False]) +def test_suspend_process_group_waits_for_the_stop_to_take_its_thread(stopped) -> None: + """The sender sleeps until the stop reaches it, or gives up once nothing has happened. + + A stop shows as time passing while asleep, so a clock that jumps across the first sleep + stands in for a job that was stopped and then resumed. A signal that was ignored, or + discarded for an orphaned group, lets the clock advance only by what was slept. + """ + import signal + + clock = [100.0] + sleeps = [] + + def sleep(seconds): + sleeps.append(seconds) + clock[0] += 5.0 if stopped and len(sleeps) == 1 else seconds + + with ( + mock.patch("cmd2.command_toolbar.time", SimpleNamespace(monotonic=lambda: clock[0], sleep=sleep)), + mock.patch("cmd2.command_toolbar.os.kill") as kill, + ): + command_toolbar.suspend_process_group() + kill.assert_called_once_with(0, signal.SIGTSTP) + if stopped: + assert sleeps == [0.01] + else: + assert len(sleeps) > 1 + assert clock[0] >= 100.25 - 1e-6 + + def test_command_toolbar_script_output_has_no_batching_delay(toolbar_app) -> None: app, _, output = toolbar_app sleep = mock.Mock() @@ -1138,6 +1188,164 @@ def test_the_refusal_lifts_when_the_display_finally_exits(toolbar_app, expire_st assert app.main_session.app.erase_when_done == prompt_erase +def test_cmdloop_waits_for_a_display_that_would_not_stop(toolbar_app, monkeypatch, capsys) -> None: + """A toolbar callback that will not return must not end the session. + + The loop reports the display it gave up on, then waits for the terminal back before + prompting on it again rather than failing out of cmdloop(). + """ + app, _, _ = toolbar_app + blocked = threading.Event() + monkeypatch.setattr(command_toolbar, "_SHUTDOWN_TIMEOUT", 0.01) + lines = iter(["wedge", "quit"]) + holders = [] + + def read_command_line(_prompt): + holders.append(app._display_holding_terminal) + return next(lines) + + def release_once_abandoned(): + deadline = time.monotonic() + 5 + while app._display_holding_terminal is None and time.monotonic() < deadline: + time.sleep(0.01) + blocked.set() + + def command(line, **kwargs): + if line == "wedge": + _block_the_display(app, blocked) + threading.Thread(target=release_once_abandoned, daemon=True).start() + return line == "quit" + + monkeypatch.setattr(app, "_read_command_line", read_command_line) + monkeypatch.setattr(app, "onecmd_plus_hooks", command) + try: + app._cmdloop() + finally: + blocked.set() + + assert holders == [None, None] + assert app.main_session.app.layout is app.main_session.layout + err = capsys.readouterr().err + assert "did not stop" in err + assert "Waiting for the bottom toolbar" in err + + +def test_cmdloop_gives_up_on_a_stuck_display_at_ctrl_c(toolbar_app, monkeypatch) -> None: + """A callback that never returns leaves Ctrl-C as the way out, which ends the loop cleanly.""" + app, _, _ = toolbar_app + + class Interrupted: + @property + def thread_is_alive(self) -> bool: + raise KeyboardInterrupt + + app._display_holding_terminal = Interrupted() + read = mock.Mock() + monkeypatch.setattr(app, "_read_command_line", read) + + app._cmdloop() + read.assert_not_called() + + +def test_ctrl_c_while_announcing_the_wait_ends_the_loop(toolbar_app, monkeypatch) -> None: + """The terminal is cooked before the notice, so Ctrl-C may land while it is still printing.""" + app, _, _ = toolbar_app + + class Stuck: + thread_is_alive = True + + def interrupted(*args, **kwargs): + raise KeyboardInterrupt + + app._display_holding_terminal = Stuck() + monkeypatch.setattr(app, "perror", interrupted) + read = mock.Mock() + monkeypatch.setattr(app, "_read_command_line", read) + + app._cmdloop() + read.assert_not_called() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pseudo-terminal") +def test_ctrl_c_reaches_the_wait_for_a_stuck_display(tmp_path) -> None: + """A real Ctrl-C, not a simulated KeyboardInterrupt, has to end the wait. + + The stuck display left the terminal in raw mode, where the driver does not turn Ctrl-C + into SIGINT. Only the display's own key binding does that, and its event loop is the one + blocked in the toolbar callback. + """ + import codecs + import os + import pty + import select + + application = tmp_path / "application.py" + application.write_text( + "import threading\n" + "from cmd2 import Cmd, ToolbarMode, command_toolbar\n" + "command_toolbar._SHUTDOWN_TIMEOUT = 0.2\n" + "wedged = threading.Event()\n" + "entered = threading.Event()\n" + "def toolbar():\n" + " if wedged.is_set():\n" + " entered.set()\n" + " threading.Event().wait()\n" + " return 'STATUS'\n" + "class App(Cmd):\n" + " def do_wedge(self, _):\n" + " wedged.set()\n" + " self._command_toolbar.app.invalidate()\n" + " assert entered.wait(5)\n" + "app = App(bottom_toolbar_mode=ToolbarMode.LEGACY)\n" + "app.prompt = 'TEST> '\n" + "app.main_session.bottom_toolbar = toolbar\n" + "app.cmdloop()\n" + "print('LOOP_ENDED', flush=True)\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + bootstrap = ( + "import os, fcntl, sys, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(sys.executable, [sys.executable, sys.argv[1]])" + ) + env = dict(os.environ, TERM="xterm-256color") + process = subprocess.Popen( + [sys.executable, "-c", bootstrap, str(application)], stdin=slave, stdout=slave, stderr=slave, env=env + ) + os.close(slave) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def wait_until(predicate): + nonlocal transcript + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + try: + data = decoder.decode(os.read(master, 65536)) + except OSError: + data = "" + transcript += data + if "\x1b[6n" in data: + # Answer prompt-toolkit's cursor-position request as a terminal would. + os.write(master, b"\x1b[1;1R") + if predicate(): + return + pytest.fail(f"terminal condition timed out:\n{transcript}") + + try: + wait_until(lambda: "TEST>" in transcript) + os.write(master, b"wedge\r") + wait_until(lambda: "Waiting for the bottom toolbar" in transcript) + os.write(master, b"\x03") + wait_until(lambda: "LOOP_ENDED" in transcript) + finally: + os.close(master) + process.kill() + process.wait(timeout=5) + + def test_a_surviving_display_stops_another_from_starting(toolbar_app, expire_startup) -> None: app, _, _ = toolbar_app blocked = expire_startup(app) diff --git a/tests/test_pager.py b/tests/test_pager.py index 8dfae5d08..0af1bbdd4 100644 --- a/tests/test_pager.py +++ b/tests/test_pager.py @@ -80,13 +80,13 @@ def observe(ui): def interact(): try: assert entered.wait(2) - pipe.send_text("\x1b[C" if chop else " ") + send_keys(app, pipe, "\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") + send_keys(app, pipe, "qnext\n") app.main_session.app.after_render += observe with ThreadPoolExecutor() as executor: @@ -99,10 +99,31 @@ def interact(): assert app._read_raw_input("Next: ", app.main_session) == "next" +def send_keys(app, pipe, text) -> None: + """Send text to the pipe input from the thread that runs the prompt-toolkit application. + + A Windows pipe input parses the text in the caller's thread, while the application + flushes that same parser from its event loop once ttimeoutlen has passed. The parser + is a generator, so the two overlapping raise "generator already executing", which a + free-threaded interpreter makes likely. A POSIX pipe input is read by the loop itself. + """ + loop = app.main_session.app.loop + if loop is not None: + try: + loop.call_soon_threadsafe(pipe.send_text, text) + except RuntimeError: + # The application finished and closed its loop in the meantime. + pass + else: + return + pipe.send_text(text) + + 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.app = app self.pipe = pipe self.pager = pager self.presses = 0 @@ -134,7 +155,7 @@ 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) + send_keys(self.app, self.pipe, keys) deadline = time.monotonic() + 5 index = 0 with self.updated: @@ -168,7 +189,7 @@ def interact(): assert entered.wait(5) script(PagerKeys(app, pipe, created[0])) finally: - pipe.send_text("q") + send_keys(app, pipe, "q") app.main_session.app.after_render += observe with mock.patch("cmd2.command_toolbar.Pager", side_effect=make_pager), ThreadPoolExecutor() as executor: @@ -310,11 +331,11 @@ def observe(ui) -> None: 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) + send_keys(app, pipe, "j") + send_keys(app, pipe, key) if not closed.wait(5): # Rescue the blocked main thread so this fails as an assertion, not a hang. - pipe.send_text("q") + send_keys(app, pipe, "q") raise AssertionError(f"{key!r} did not close the pager") app.main_session.app.after_render += observe diff --git a/tests/test_pipeline_job_control.py b/tests/test_pipeline_job_control.py index 7baf84d5b..73c2bba1a 100644 --- a/tests/test_pipeline_job_control.py +++ b/tests/test_pipeline_job_control.py @@ -62,7 +62,7 @@ def describe_processes(root: int, master: int) -> str: pytest.param("exit_sigint", True, True, "sh", "shell", "worker", id="wrapper-stop-and-interrupt-busy-shell-producer"), pytest.param("read_input", False, False, "direct", "command", "main", id="nested-prompt"), pytest.param("shell_input", False, True, "direct", "command", "main", id="shell-input"), - pytest.param("direct_input", False, False, "sh", "command", "main", id="wrapper-direct-input"), + pytest.param("direct_input", False, False, "sh", "command", "main", id="direct-input-with-toolbar-off"), pytest.param("direct_input", False, False, "exec", "command", "main", id="direct-input-in-orphaned-session"), pytest.param("interrupts", False, False, "exec", "command", "main", id="orphaned-job-control"), ], @@ -131,7 +131,7 @@ def test_pipeline_stops_with_cmd2_and_returns_terminal( interrupt_request = tmp_path / "interrupt.request" last_result = tmp_path / "last_result" application.write_text( - "from cmd2 import Cmd\n" + "from cmd2 import Cmd, ToolbarMode\n" "from cmd2.plugin import CommandFinalizationData\n" "import getpass, os, pathlib, signal, threading, time\n" "signal.signal(signal.SIGTSTP, signal.SIG_DFL)\n" @@ -162,10 +162,11 @@ def test_pipeline_stops_with_cmd2_and_returns_terminal( " def record_result(self, data: CommandFinalizationData) -> CommandFinalizationData:\n" f" pathlib.Path({str(last_result)!r}).write_text(repr(self.last_result))\n" " return data\n" - "app = App()\n" + f"app = App(bottom_toolbar_mode=ToolbarMode.{'OFF' if finish == 'direct_input' else 'RESERVED'})\n" "app.register_cmdfinalization_hook(app.record_result)\n" "app.prompt = 'TEST> '\n" "app.debug = True\n" + "app.main_session.bottom_toolbar = 'STATUS'\n" f"if {finish == 'interrupts'!r}:\n" " def interrupt_from_worker():\n" f" request = pathlib.Path({str(interrupt_request)!r})\n" @@ -480,8 +481,107 @@ def wait_until(predicate): process.wait(timeout=5) +@pytest.mark.parametrize("mode", ["reserved", "legacy"]) +def test_pager_resumes_with_raw_terminal_after_suspend(tmp_path, mode) -> None: + """Ctrl-Z in the built-in pager, then fg, must bring the pager back, not echoed cursor reports. + + The display thread sends the stop signal, which the kernel may hand to the main + thread. Unless the sending thread stops synchronously too, it restores raw mode + before the job has stopped, the shell puts its own modes back on fg, and every + cursor-position query afterwards is echoed as ^[[row;colR. + """ + import fcntl + import pty + import struct + import termios + + shell = shutil.which("bash") + if shell is None: + pytest.skip("requires an interactive bash shell") + application = tmp_path / "application.py" + application.write_text( + "from cmd2 import Cmd, ToolbarMode\n" + "class App(Cmd):\n" + " def do_page(self, statement):\n" + " self.ppaged('\\n'.join(f'line {n:04d}' for n in range(400)))\n" + # A refreshing toolbar, as in the examples: its periodic redraws are what turn a cooked terminal into a stream of echoed + # cursor reports. + f"app = App(bottom_toolbar_mode=ToolbarMode.{mode.upper()}, refresh_interval=0.5)\n" + "app.prompt = 'TEST> '\n" + "app.main_session.bottom_toolbar = 'STATUS'\n" + "app.cmdloop()\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) + bootstrap = ( + "import os, fcntl, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])" + ) + env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell) + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env) + os.close(slave) + screen = pyte.Screen(80, 24) + screen.write_process_input = lambda data: os.write(master, data.encode()) + stream = pyte.Stream(screen) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def send(data): + os.write(master, data.encode()) + + def pump(seconds, predicate=lambda: False): + nonlocal transcript + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + data = decoder.decode(os.read(master, 65536)) + transcript += data + stream.feed(data) + if predicate(): + return True + return predicate() + + def pager_shown(): + return any("q: quit" in line for line in screen.display) + + try: + assert pump(10, lambda: "OUTER> " in transcript) + send(f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}\n") + assert pump(10, lambda: "TEST>" in "\n".join(screen.display)), transcript + # Let the prompt finish its cursor-position handshake before typing at it. + pump(0.5) + send("page\n") + assert pump(10, pager_shown), transcript + start = len(transcript) + send("\x1a") + assert pump(10, lambda: "OUTER> " in transcript[start:]), transcript + start = len(transcript) + send("fg\n") + assert pump(10, pager_shown), transcript[start:] + # Give a broken resume time to start looping before judging the transcript. + pump(1.5) + resumed = transcript[start:] + echoed = re.findall(r"\^\[\[\d+;\d+R", resumed) + assert not echoed, f"cursor position reports were echoed after fg: {echoed[:5]}\n{resumed}" + assert pager_shown(), "\n".join(screen.display) + # A cooked terminal would hold the key back until Enter: the pager must quit on q alone. + start = len(transcript) + send("q") + assert pump(10, lambda: "TEST>" in transcript[start:]), f"pager did not quit on q after fg:\n{transcript[start:]}" + send("quit\n") + pump(5, lambda: os.tcgetpgrp(master) == process.pid) + finally: + os.close(master) + process.kill() + process.wait(timeout=5) + + +@pytest.mark.parametrize("mode", ["off", "reserved"]) @pytest.mark.parametrize("parent_delay", [False, True]) -def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path, parent_delay) -> None: +def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path, mode, parent_delay) -> None: """A pager such as less puts the terminal in raw mode as it starts, before reading its pipe. It has to own the terminal by then. A background tcsetattr() stops it with SIGTTOU, and @@ -521,7 +621,7 @@ def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path, parent_delay application = tmp_path / "application.py" application.write_text( "import pathlib, time\n" - "from cmd2 import Cmd, utils\n" + "from cmd2 import Cmd, ToolbarMode, utils\n" # The pipeline's first wait is cmd2's 0.2s startup check. Hold it open until the pager has reported, so the test does # not race that timer on a busy CI runner: what it checks is that the pager owns the terminal throughout the check, # however slowly it starts. @@ -541,8 +641,9 @@ def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path, parent_delay " time.sleep(0.5)\n" " return start_job_control(reader)\n" " utils.ProcReader._manage_terminal = late_job_control\n" - "app = Cmd()\n" + f"app = Cmd(bottom_toolbar_mode=ToolbarMode.{mode.upper()})\n" "app.prompt = 'TEST> '\n" + "app.main_session.bottom_toolbar = 'STATUS'\n" "app.cmdloop()\n", encoding="utf-8", ) diff --git a/tests/test_reserved_lifecycle.py b/tests/test_reserved_lifecycle.py index 3bf52fe46..cc19dec5a 100644 --- a/tests/test_reserved_lifecycle.py +++ b/tests/test_reserved_lifecycle.py @@ -603,8 +603,8 @@ def test_the_main_prompt_leaves_the_native_toolbar_hidden(self) -> None: finally: harness.close() - def test_another_session_still_gets_the_terminal_to_itself(self) -> None: - """A prompt cmd2 has not bound to the reservation renders outside it, for now.""" + def test_another_session_on_the_same_terminal_borrows_the_reservation(self) -> None: + """Managed nested input retains the main-screen band.""" harness = Harness(mode="reserved") try: with harness.app._reserved_toolbar_context(): @@ -616,7 +616,7 @@ def test_another_session_still_gets_the_terminal_to_itself(self) -> None: harness.app._read_raw_input("> ", other) - assert seen == [False] + assert seen == [True] assert toolbar.display.is_reserved is True finally: harness.close() diff --git a/tests/test_reserved_terminal.py b/tests/test_reserved_terminal.py index 611d5c97c..9cc600729 100644 --- a/tests/test_reserved_terminal.py +++ b/tests/test_reserved_terminal.py @@ -2,6 +2,7 @@ import asyncio import io +import signal import sys import threading import time @@ -14,6 +15,8 @@ import pytest from prompt_toolkit.application import run_in_terminal from prompt_toolkit.data_structures import Size +from prompt_toolkit.shortcuts import PromptSession +from prompt_toolkit.shortcuts.choice_input import ChoiceInput from cmd2 import command_toolbar from cmd2.reserved_toolbar import ReservedToolbar @@ -297,6 +300,76 @@ class TestResize: attached, so the poll on ``output.get_size()`` is the only way a resize arrives there. """ + @pytest.mark.parametrize("terminal_harness", [2, 3], indirect=True) + @pytest.mark.parametrize("text", ["", "typed"]) + def test_minimum_height_reacquisition_through_idle_polling(self, terminal_harness, monkeypatch, text) -> None: + """Fallback can erase a cached band before growth returns to exactly the same size.""" + harness, terminal = terminal_harness + ui = harness.app.main_session.app + ui.terminal_size_polling_interval = 0.01 + task = None + polls = 0 + errors = [] + + async def until(predicate): + async with asyncio.timeout(5): + while not predicate(): # noqa: ASYNC110 - observe upstream state without driving a redraw + await asyncio.sleep(0.01) + + async def transitions(): + try: + if text: + harness.pipe.send_text(text) + await until(lambda: ui.current_buffer.text == text) + for rows, columns in ((3, 80), (2, 80), (3, 80), (4, 40), (24, 100), (3, 80), (2, 80), (3, 80)): + baseline = polls + # Let upstream establish/settle its size baseline before changing it. + await until(lambda baseline=baseline: polls >= baseline + 2) + if harness.size != Size(rows=rows, columns=columns): + resize(harness, terminal, rows, columns) + toolbar = harness.app.reserved_toolbar + await until(lambda toolbar=toolbar, rows=rows: toolbar.is_active == (rows >= 3)) + if rows >= 3: + await until(lambda toolbar=toolbar, rows=rows: toolbar.display.geometry.physical_rows == rows) + await until(lambda: terminal.screen.display[-1].startswith("STATUS")) + await until(lambda: any(row.startswith("TEST> " + text) for row in terminal.screen.display)) + assert ui.current_buffer.text == text + assert ui.current_buffer.cursor_position == len(text) + except Exception as error: # noqa: BLE001 - report outside prompt-toolkit's event loop + errors.append(error) + finally: + harness.pipe.send_text("\n") + + def ready(app): + nonlocal task + if task is None: + task = ui.create_background_task(transitions()) + + ui.after_render += ready + try: + with harness.app._reserved_toolbar_context(): + get_size = ui.output.get_size + + def observe_size(): + nonlocal polls + try: + current = asyncio.current_task() + except RuntimeError: + current = None + if current is not None and current.get_coro().__name__ == "_poll_output_size": + polls += 1 + return get_size() + + monkeypatch.setattr(ui.output, "get_size", observe_size) + assert harness.app._read_raw_input("TEST> ", harness.app.main_session) == text + assert task is not None + task.result() + if errors: + raise errors[0] + finally: + ui.after_render -= ready + assert terminal.screen.margins is None + @pytest.mark.parametrize("terminal_harness", [2], indirect=True) def test_initially_short_prompt_grows_and_accepts_visible_input(self, terminal_harness) -> None: harness, terminal = terminal_harness @@ -327,6 +400,45 @@ def ready(app): watchdog.join() ui.after_render -= ready + @pytest.mark.parametrize("terminal_harness", [3], indirect=True) + @pytest.mark.parametrize("partial", ["", "PARTIAL"]) + def test_quiet_command_reacquires_the_same_band(self, terminal_harness, monkeypatch, partial) -> None: + """The command poll restores the whole band without disturbing unfinished output.""" + harness, terminal = terminal_harness + harness.app.main_session.app.terminal_size_polling_interval = 0.01 + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + toolbar = harness.app.reserved_toolbar + output = harness.app._command_toolbar.app.output + get_size = output.get_size + polls = 0 + + def observe_size(): + nonlocal polls + try: + current = asyncio.current_task() + except RuntimeError: + current = None + if current is not None and current.get_coro().__name__ == "_poll_output_size": + polls += 1 + return get_size() + + monkeypatch.setattr(output, "get_size", observe_size) + harness.app.stdout.write(partial) + harness.app.stdout.flush() + for rows in (2, 3, 4, 3, 2, 3): + baseline = polls + assert wait_for(lambda baseline=baseline: polls >= baseline + 2) + resize(harness, terminal, rows, 80) + assert wait_for(lambda rows=rows: toolbar.is_active == (rows >= 3)) + if rows >= 3: + assert wait_for(lambda rows=rows: toolbar.display.geometry.physical_rows == rows) + assert wait_for(lambda: terminal.screen.display[-1].startswith("STATUS")) + if partial: + assert any(row.startswith(partial) for row in terminal.screen.display) + harness.app.poutput(" COMPLETE") + assert any(row.startswith(partial + " COMPLETE") for row in terminal.screen.display) + assert terminal.screen.margins is None + @pytest.mark.parametrize("terminal_harness", [2], indirect=True) def test_initially_short_terminal_acquires_during_a_quiet_command(self, terminal_harness, monkeypatch) -> None: harness, terminal = terminal_harness @@ -609,7 +721,286 @@ def drive() -> None: return shown.is_set() +class TestNestedPrompts: + def test_abandoning_reservation_during_nested_input_restores_both_owners(self, terminal_harness) -> None: + harness, terminal = terminal_harness + session = PromptSession(input=harness.pipe, output=harness.backend, bottom_toolbar="STATUS") + original_render = session.app.renderer.render + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + reserved = harness.app.reserved_toolbar + + def abandon(): + reserved.stop() + harness.pipe.send_text("answer\n") + + assert harness.app._read_raw_input("Nested: ", session, pre_run=abandon) == "answer" + assert reserved.bridge is None + assert session.app.output is harness.backend + assert session.app.renderer.render == original_render + assert harness.app._command_toolbar._proxy is not None + assert terminal.screen.margins is None + + def test_nested_resize_and_failed_guest_restore_the_main_owner(self, terminal_harness) -> None: + harness, terminal = terminal_harness + session = PromptSession(input=harness.pipe, output=harness.backend, bottom_toolbar="STATUS") + original = session.app.renderer.render + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + main_bridge = harness.app.reserved_toolbar.bridge + + def fail(): + resize(harness, terminal, 12, 40) + session.app._on_resize() + raise ValueError("nested failure") + + with pytest.raises(ValueError, match="nested failure"): + harness.app._read_raw_input("Nested: ", session, pre_run=fail) + assert harness.app.reserved_toolbar.bridge is main_bridge + assert session.app.renderer.render == original + assert terminal.screen.margins == pyte.screens.Margins(0, 10) + assert terminal.screen.display[-1].startswith("STATUS") + read_prompt(harness, terminal) + + @pytest.mark.parametrize( + ("kind", "keys", "expected"), + [("input", "ans\t", "answer"), ("secret", "hidden-value\n", "hidden-value"), ("select", "\x1b[B\r", "two")], + ) + def test_managed_input_keeps_the_band_and_one_reader(self, terminal_harness, kind, keys, expected) -> None: + harness, terminal = terminal_harness + created = [] + seen = [] + accepted = False + selected = False + + def ready(app): + nonlocal accepted, selected + if not seen and app.renderer._min_available_height > 0: + seen.append((terminal.screen.margins, terminal.screen.display[-1])) + assert not harness.app._command_toolbar.thread_is_alive + harness.pipe.send_text(keys + ("" if kind == "input" else "trailing\n")) + elif kind == "input" and not selected and app.current_buffer.complete_state is not None: + selected = True + harness.pipe.send_text("\t") + elif kind == "input" and not accepted and app.current_buffer.text.rstrip() == "answer": + accepted = True + harness.pipe.send_text("\rtrailing\n") + + def make_session(*args, **kwargs): + session = PromptSession(*args, **kwargs) + created.append(session) + + session.app.after_render += ready + return session + + original_create = ChoiceInput._create_application + + def make_choice(choice): + app = original_create(choice) + app.after_render += ready + return app + + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + before = len(terminal.getvalue()) + watchdog = threading.Timer(5, harness.pipe.close) + watchdog.start() + try: + with ( + mock.patch("cmd2.cmd2.PromptSession", make_session), + mock.patch.object(ChoiceInput, "_create_application", make_choice), + ): + if kind == "input": + result = harness.app.read_input("Value: ", choices=["answer"]) + elif kind == "secret": + result = harness.app.read_secret("Secret: ") + else: + result = harness.app.select(["one", "two"]) + assert result.rstrip() == expected + assert seen[0][0] == pyte.screens.Margins(0, 22) + assert seen[0][1].startswith("STATUS") + assert "\x1b[r" not in terminal.getvalue()[before:] + assert harness.app._command_toolbar.thread_is_alive + assert harness.app._read_raw_input("Next: ", harness.app.main_session) == "trailing" + assert not harness.app.reserved_toolbar._nested_bridges + if kind == "secret": + assert "hidden-value" not in terminal.getvalue() + finally: + watchdog.cancel() + watchdog.join() + + @pytest.mark.parametrize("keys", ["\x03", "\x04"]) + def test_nested_input_interrupt_restores_bindings(self, terminal_harness, keys) -> None: + harness, terminal = terminal_harness + session = PromptSession(input=harness.pipe, output=harness.backend, bottom_toolbar="STATUS") + original_output = session.app.output + original_render = session.app.renderer.render + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + with pytest.raises((KeyboardInterrupt, EOFError)): + harness.app._read_raw_input("Nested: ", session, pre_run=lambda: harness.pipe.send_text(keys)) + assert session.app.output is original_output + assert session.app.renderer.render == original_render + assert harness.app._command_toolbar.thread_is_alive + assert terminal.screen.margins == pyte.screens.Margins(0, 22) + assert terminal.screen.display[-1].startswith("STATUS") + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX job control") +@pytest.mark.parametrize("owner", ["command", "prompt", "pager"]) +def test_ctrl_z_releases_before_signaling_and_reacquires_after_resume(terminal_harness, monkeypatch, owner) -> None: + harness, terminal = terminal_harness + harness.app.main_session.enable_suspend = True + stopped = threading.Event() + seen = [] + + def stop_process(pid, sig): + # This is the actual key-binding -> run_in_terminal -> signal path. The signal is + # replaced so the test runner never suspends itself or its parent process group. + seen.append((pid, sig, terminal.screen.margins, harness.app.main_session.app.output.get_size())) + assert harness.app.main_session.app._running_in_terminal + resize(harness, terminal, 12, 80) + stopped.set() + + monkeypatch.setattr("cmd2.command_toolbar.os.kill", stop_process) + with harness.app._reserved_toolbar_context(): + original_suspend = harness.app.main_session.app.suspend_to_background + if owner in ("command", "pager"): + with harness.app._command_toolbar_context(): + display = harness.app._command_toolbar + + def suspend() -> None: + harness.pipe.send_text("\x1a") + assert stopped.wait(5) + display._call_in_ui(lambda: None) + + if owner == "pager": + assert run_pager(harness, terminal, while_open=suspend) + else: + suspend() + assert terminal.screen.margins == pyte.screens.Margins(0, 10) + assert terminal.screen.display[-1].startswith("STATUS") + else: + ui = harness.app.main_session.app + accepted = False + + def ready(app): + nonlocal accepted + if stopped.is_set() and not accepted: + accepted = True + harness.pipe.send_text("\n") + + ui.after_render += ready + watchdog = threading.Timer(5, harness.pipe.close) + watchdog.start() + try: + assert ( + harness.app._read_raw_input( + "TEST> ", harness.app.main_session, pre_run=lambda: harness.pipe.send_text("\x1a") + ) + == "" + ) + assert stopped.is_set() + assert terminal.screen.margins == pyte.screens.Margins(0, 10) + finally: + watchdog.cancel() + watchdog.join() + ui.after_render -= ready + assert seen == [(0, signal.SIGTSTP, None, Size(24, 80))] + assert terminal.screen.margins is None + assert harness.app.main_session.app.suspend_to_background != original_suspend + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX termination handler") +def test_termination_unwinds_reserved_command_ownership(terminal_harness) -> None: + harness, terminal = terminal_harness + original_output = harness.app.main_session.app.output + displays = [] + + def terminate(): + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + displays.append(harness.app._command_toolbar) + harness.app.termination_signal_handler(signal.SIGTERM, None) + + with pytest.raises(SystemExit) as exit_info: + terminate() + assert exit_info.value.code == 128 + signal.SIGTERM + assert not displays[0].thread_is_alive + assert harness.app.main_session.app.output is original_output + assert terminal.screen.margins is None + + class TestPager: + @pytest.mark.parametrize("action", ["resume", "output", "external"]) + def test_retained_startup_bar_is_cleared_before_output_or_external_handoff(self, terminal_harness, action) -> None: + harness, terminal = terminal_harness + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + reserved = harness.app.reserved_toolbar + with reserved.suspended(defer_band_clear=True): + assert terminal.screen.display[-1].startswith("STATUS") + if action == "output": + # Blank lines can scroll the retained bar without overwriting its text. + harness.app.stdout.write("\n" * 50) + harness.app.stdout.flush() + elif action == "external": + with reserved.suspended(): + assert terminal.screen.display[-1].strip() == "" + else: + assert reserved.display._deferred_band is not None + assert reserved.display._deferred_band is None + # The command display keeps rendering on its own thread, so the bar it repaints + # after the handoff can land a frame later than the resume itself. + assert wait_for(lambda: terminal.screen.display[-1].startswith("STATUS")) + history = ["".join(line[x].data for x in sorted(line)) for line in terminal.screen.history.top] + assert not any("STATUS" in row for row in history) + + def test_toolbar_remains_visible_until_the_first_pager_frame_is_prepared(self, terminal_harness, monkeypatch) -> None: + harness, terminal = terminal_harness + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + reserved = harness.app.reserved_toolbar + display = harness.app._command_toolbar + original_prepare = reserved.bridge.prepare + seen = [] + + def prepare(*args, **kwargs): + frame = original_prepare(*args, **kwargs) + if display._paging and frame is not None and not seen: + # Observe after the expensive layout work, before commit. Checking only + # the completed pager would miss the blank interval while a file loads. + seen.append(terminal.screen.display[-1]) + assert display.app.output.get_size() == Size(24, 80) + return frame + + monkeypatch.setattr(reserved.bridge, "prepare", prepare) + assert run_pager(harness, terminal) + assert seen + assert seen[0].startswith("STATUS") + assert reserved.display._deferred_band is None + assert terminal.screen.display[-1].startswith("STATUS") + + def test_first_pager_frame_uses_full_geometry(self, terminal_harness) -> None: + harness, terminal = terminal_harness + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + display = harness.app._command_toolbar + sizes = [] + + def before_render(app): + if display._paging: + sizes.append(app.output.get_size()) + + display.app.before_render += before_render + try: + assert run_pager(harness, terminal) + assert sizes + assert all(size == Size(24, 80) for size in sizes) + finally: + display.app.before_render -= before_render + + def test_exact_usable_height_does_not_open_a_pager(self, terminal_harness) -> None: + harness, terminal = terminal_harness + with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context(): + text = "\n".join(f"line {i}" for i in range(23)) + with mock.patch.object(command_toolbar, "Pager", side_effect=AssertionError("unnecessary pager")): + harness.app._command_toolbar.page(text, chop=False) + assert any("line 22" in row for row in terminal.screen.display) + assert terminal.screen.display[-1].startswith("STATUS") + """The built-in pager renders a full screen of its own, so its frames must not be suppressed the way an ordinary command's empty frames are.""" @@ -771,6 +1162,7 @@ def drive() -> None: assert not forced_close.is_set(), "the quit callback failed to release page()" assert display.app.full_screen is False assert display.app.renderer.full_screen is False + assert display.app.renderer._in_alternate_screen is False assert display.app.layout is display._layout assert display.app.key_bindings is display._bindings or display.app.key_bindings is bindings assert display.app.editing_mode is editing_mode diff --git a/tests/test_reserved_toolbar.py b/tests/test_reserved_toolbar.py index 7f1cc7d03..a44289042 100644 --- a/tests/test_reserved_toolbar.py +++ b/tests/test_reserved_toolbar.py @@ -7,6 +7,7 @@ """ import io +import signal from typing import Any import pytest @@ -61,6 +62,34 @@ def app(self) -> Any: return self.session.app +@pytest.mark.parametrize("guest_fails", [False, True]) +def test_failed_reacquisition_releases_bindings_without_masking_the_guest(monkeypatch, guest_fails) -> None: + harness = Harness() + try: + with harness.toolbar: + display = harness.toolbar.display + + def fail(): + raise OSError("resume failed") + + monkeypatch.setattr(display, "reacquire_region_after_handoff", fail) + expected = ValueError if guest_fails else OSError + + def run_guest(): + with harness.toolbar.suspended(): + if guest_fails: + raise ValueError("guest failed") + + with pytest.raises(expected, match="guest failed" if guest_fails else "resume failed"): + run_guest() + assert harness.toolbar.bridge is None + assert harness.app.output is harness.backend + assert display.lease_depth == 0 + assert isinstance(harness.toolbar.take_pending_error(), OSError) + finally: + harness.close() + + class TestBinding: def test_starting_binds_the_application_and_its_renderer(self) -> None: harness = Harness() @@ -806,3 +835,86 @@ def test_suspending_a_toolbar_that_never_started_does_nothing(self) -> None: assert harness.written() == "" finally: harness.close() + + +class TestNestedPrompts: + def test_a_nested_session_without_a_toolbar_window_is_refused(self) -> None: + """Lending the reservation to a layout with no toolbar window would leave two toolbars.""" + harness = Harness() + try: + harness.toolbar.start() + main_bridge = harness.toolbar.bridge + nested: PromptSession[str] = PromptSession(input=harness.pipe, output=harness.backend) + nested.app.layout = Layout(Window()) + with pytest.raises(RuntimeError, match="nested session"), harness.toolbar.prompt_session(nested): + pass + assert nested.app.output is harness.backend + assert harness.toolbar.bridge is main_bridge + finally: + harness.close() + + def test_suspending_invalidates_a_nested_prompt_bridge_too(self) -> None: + """The guest program writes over the nested prompt's screen as much as the main one's.""" + harness = Harness() + try: + harness.toolbar.start() + nested: PromptSession[str] = PromptSession(input=harness.pipe, output=harness.backend, bottom_toolbar="NESTED") + with harness.toolbar.prompt_session(nested): + bridge = harness.toolbar.bridge + assert bridge is not None + assert bridge is not harness.toolbar._bridge + bridge.set_prompt_anchor(5) + with harness.toolbar.suspended(): + assert bridge.prompt_anchor is None + assert bridge.needs_resynchronization is True + assert "handed to another program" in (bridge.resynchronization_reason or "") + assert "came back from another program" in (bridge.resynchronization_reason or "") + finally: + harness.close() + + +class TestJobControl: + def test_stopping_leaves_the_application_as_it_found_it(self) -> None: + """Upstream's method is not put back as an instance attribute when the toolbar stops. + + The command display installs its thread-safe Ctrl-Z handler only where the application + has none of its own, so a restored bound method would keep it out for the rest of the + session after a fallback from reserved rendering. + """ + harness = Harness() + try: + with harness.toolbar: + assert "suspend_to_background" in vars(harness.app) + assert "suspend_to_background" not in vars(harness.app) + + def installed() -> None: + pass + + harness.app.suspend_to_background = installed + with harness.toolbar: + assert harness.app.suspend_to_background is not installed + assert harness.app.suspend_to_background is installed + finally: + harness.close() + + @pytest.mark.parametrize("missing", ["support", "signal"]) + def test_suspending_to_background_does_nothing_where_it_is_unsupported( + self, monkeypatch: pytest.MonkeyPatch, missing: str + ) -> None: + """Without a stop signal or a supporting platform there is no process to stop.""" + harness = Harness() + try: + harness.toolbar.start() + calls: list[Any] = [] + monkeypatch.setattr("cmd2.reserved_toolbar.run_in_terminal", calls.append) + if missing == "support": + monkeypatch.setattr("cmd2.reserved_toolbar.suspend_to_background_supported", lambda: False) + else: + monkeypatch.delattr(signal, "SIGTSTP", raising=False) + harness.clear() + harness.app.suspend_to_background() + assert calls == [] + assert harness.written() == "" + assert harness.toolbar.is_active is True + finally: + harness.close() diff --git a/tests/test_terminal_display.py b/tests/test_terminal_display.py index 0508c002c..c2ed5eb7c 100644 --- a/tests/test_terminal_display.py +++ b/tests/test_terminal_display.py @@ -52,6 +52,17 @@ def make_resizable_output(rows: int = 24, columns: int = 80) -> tuple[Vt100_Outp class TestGeometry: + def test_shutdown_clears_a_bar_retained_for_an_unfinished_pager(self) -> None: + output, stream = make_output() + display = TerminalDisplay(output) + display.acquire() + display.release_region_for_handoff(defer_band_clear=True) + stream.seek(0) + stream.truncate() + display.release() + assert "\x1b[24;1H\x1b[0m\x1b[J" in stream.getvalue() + assert display._deferred_band is None + def test_usable_rows_subtracts_the_reservation_exactly_once(self) -> None: """The double-subtraction mutation: at 24 rows with 1 reserved, 22 is the wrong answer.""" geometry = Geometry(generation=1, physical_rows=24, columns=80, reserved_rows=1) diff --git a/tests/test_toolbar_mode.py b/tests/test_toolbar_mode.py index 6911e7366..9726cc4bf 100644 --- a/tests/test_toolbar_mode.py +++ b/tests/test_toolbar_mode.py @@ -7,6 +7,7 @@ """ import io +from importlib.metadata import PackageNotFoundError import pytest from prompt_toolkit.data_structures import Size @@ -14,6 +15,7 @@ from prompt_toolkit.output.vt100 import Vt100_Output import cmd2 +from cmd2 import toolbar_mode from cmd2.toolbar_mode import ( QUALIFIED_PROMPT_TOOLKIT_VERSIONS, ToolbarMode, @@ -65,6 +67,22 @@ def test_an_unqualified_version_is_reported_with_its_number(self) -> None: assert supported is False assert "3.0.99" in reason + def test_missing_package_metadata_is_unqualified(self, monkeypatch) -> None: + """A frozen or vendored application may have no metadata to read the version from.""" + + def missing(name: str) -> str: + raise PackageNotFoundError(name) + + monkeypatch.setattr(toolbar_mode, "_installed_version", missing) + supported, reason = _dependency_capability() + assert supported is False + assert "cannot be determined" in reason + + # auto falls back rather than failing the command loop. + mode, reason = _select_toolbar_mode("auto", qualified_output(), toolbar_enabled=True, interactive=True) + assert mode is ToolbarMode.LEGACY + assert "cannot be determined" in reason + class TestAutomaticSelection: def test_a_qualified_terminal_selects_reserved(self) -> None: diff --git a/tests/test_toolbar_painter.py b/tests/test_toolbar_painter.py index 603327c81..c1d9095f7 100644 --- a/tests/test_toolbar_painter.py +++ b/tests/test_toolbar_painter.py @@ -470,6 +470,23 @@ def test_a_geometry_change_forces_a_full_repaint(self) -> None: assert harness.paint("hi") is True assert "\x1b[12;1Hhi " in harness.visible() + @pytest.mark.parametrize("prepare_while_inactive", [False, True]) + def test_reacquiring_the_same_band_repaints_all_cells(self, prepare_while_inactive: bool) -> None: + """Returning from fallback must not reuse cells painted before the reservation ended.""" + harness = Harness(rows=3) + assert harness.paint("hi") is True + harness.display.resize(2) + assert not harness.display.is_reserved + if prepare_while_inactive: + assert harness.painter.prepare(lambda: "hi") is None + harness.display.resize(3) + harness.clear() + assert harness.paint("hi") is True + assert "\x1b[3;1Hhi " in harness.visible() + harness.clear() + assert harness.paint("hi") is False + assert harness.written() == "" + def test_empty_content_is_painted_rather_than_skipped(self) -> None: """An empty toolbar is an intentional visibility change and must reach the band.""" harness = Harness()