diff --git a/cmd2/output_recorder.py b/cmd2/output_recorder.py new file mode 100644 index 000000000..f68a68080 --- /dev/null +++ b/cmd2/output_recorder.py @@ -0,0 +1,415 @@ +"""Record a renderer's output operations instead of performing them. + +prompt-toolkit's ``Renderer.render()`` evaluates the application's layout, filters and styles +*while* it emits. Holding the terminal lock around that whole call would run application +callbacks inside the transaction, which the wait contract forbids and which is how a toolbar +callback ends up deadlocking against the thread that is painting it. + +The way out is to separate preparation from emission. The renderer runs against a +:class:`RecordingOutput`, which answers its questions from facts captured beforehand and +writes its operations to a list. The resulting :class:`OperationBatch` is replayed onto the +real backend later, inside one terminal transaction, once the bridge has revalidated that the +frame is still current. + +Two properties make the recording safe to build off-lock: + +**It never touches a backend.** The recorder holds no output object at all -- only an +immutable :class:`PreflightFacts`. Using the real backend as a recording sink would advance +its buffered text, attribute and cursor caches and console modes, so a discarded frame would +leave the backend believing things about the terminal that were never emitted. + +**Reads are answered, not deferred.** The renderer asks for the size and for the rows below +the cursor mid-render and branches on the answers. Those come from the preflight snapshot, so +every operation in a batch was decided against one consistent view of the terminal -- the same +view the bridge revalidates before replaying it. + +A batch is a delay, not a translation: replaying it produces exactly the bytes the backend +would have produced had the renderer written to it directly. +""" + +import io +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from prompt_toolkit.data_structures import Size +from prompt_toolkit.output import ColorDepth, Output + +if TYPE_CHECKING: # pragma: no cover + from prompt_toolkit.cursor_shapes import CursorShape + from prompt_toolkit.styles import Attrs + + +class UnrecordableOperationError(RuntimeError): + """Raised for an operation that cannot be deferred to commit time. + + Screen-buffer transitions and viewport moves change what every later coordinate means. + They are ownership boundaries, taken explicitly before a frame is prepared for the new + screen, so meeting one during preparation means a frame is being prepared against a + terminal that is no longer the one it was measured on. + """ + + +@dataclass(frozen=True) +class PreflightFacts: + """The terminal as one preparation saw it, captured before recording begins. + + Everything the renderer can *ask* is here, so recording needs no backend. Capturing these + is a short serialized read with no application callbacks in it. + """ + + #: The size the application is rendering against, already reservation-adjusted. + size: Size + + #: Native rows below the cursor, or ``None`` where the backend has no such notion and the + #: renderer must fall back to a cursor-position report. + rows_below_cursor: int | None + + #: The backend's encoding. + encoding: str + + #: The backend's default color depth. + default_color_depth: ColorDepth + + #: Whether the backend answers cursor-position reports. + responds_to_cpr: bool + + #: The backend's file descriptor, or ``None`` where it has none. + fileno: int | None + + @classmethod + def capture(cls, output: Output) -> "PreflightFacts": + """Read every fact a recorded render may need, without writing anything. + + :param output: the backend to read from + :return: the captured facts + """ + try: + rows_below: int | None = output.get_rows_below_cursor_position() + except NotImplementedError: + rows_below = None + try: + descriptor: int | None = output.fileno() + except (io.UnsupportedOperation, NotImplementedError, AttributeError, OSError): + # Backends disagree on how to say "no descriptor": Vt100 lets StringIO raise + # UnsupportedOperation, DummyOutput raises NotImplementedError outright. + descriptor = None + return cls( + size=output.get_size(), + rows_below_cursor=rows_below, + encoding=output.encoding(), + default_color_depth=output.get_default_color_depth(), + responds_to_cpr=output.responds_to_cpr, + fileno=descriptor, + ) + + +@dataclass(frozen=True) +class Operation: + """One recorded output call, ready to be replayed by name onto a real backend.""" + + #: The :class:`~prompt_toolkit.output.Output` method that was called. + name: str + + #: Its positional arguments. Output's interface takes no keyword-only arguments. + args: tuple[Any, ...] = () + + def replay(self, output: Output) -> None: + """Perform this operation on a real backend. + + :param output: the backend to emit through + """ + getattr(output, self.name)(*self.args) + + +@dataclass(frozen=True) +class OperationBatch: + """An immutable, ordered recording of one prepared frame. + + Immutability is what makes commit-time validation meaningful: a batch that could still + grow after it was validated would be replayed as something other than what was checked. + """ + + #: The operations to replay, in the order the renderer made them. + operations: tuple[Operation, ...] + + #: The terminal facts the operations were decided against. + facts: PreflightFacts + + def replay(self, output: Output) -> None: + """Replay every operation onto a real backend, in order. + + Replay deliberately does not catch anything. A failure part-way through means some + bytes have already reached the terminal and others have not, which is a physical state + the caller has to recover from -- swallowing the exception here would hide exactly the + case that needs handling. + + :param output: the backend to emit through + :raises Exception: whatever the backend raises, after the earlier operations have run + """ + for operation in self.operations: + operation.replay(output) + + +class RecordingOutput(Output): + """An :class:`~prompt_toolkit.output.Output` that records rather than emits. + + Every abstract method is written out. Answering by ``__getattr__`` would leave the class + abstract, and would silently record whatever prompt-toolkit adds next without anyone + deciding whether it is safe to defer. + """ + + #: Marks this as a recorder so the physical layer refuses to wrap it. + is_reserved_adapter = True + + def __init__(self, facts: PreflightFacts) -> None: + """Record a frame against a fixed view of the terminal. + + :param facts: the preflight snapshot to answer reads from + """ + self._facts = facts + self._operations: list[Operation] = [] + # Output declares stdout as writable. There is no stream to offer here, and offering + # the real one would invite a caller to write around the recording. + self.stdout = None + + @property + def facts(self) -> PreflightFacts: + """The preflight snapshot this frame was recorded against.""" + return self._facts + + @property + def operations(self) -> tuple[Operation, ...]: + """The operations recorded so far, in order.""" + return tuple(self._operations) + + def batch(self) -> OperationBatch: + """Freeze what has been recorded so far into a replayable batch. + + :return: the immutable batch + """ + return OperationBatch(operations=self.operations, facts=self._facts) + + def _record(self, name: str, *args: Any) -> None: + """Append one operation to the recording. + + :param name: the output method that was called + :param args: its positional arguments + """ + self._operations.append(Operation(name, args)) + + # -- reads, answered from the preflight snapshot --------------------------------------- + + def get_size(self) -> Size: + """Report the size this frame is being prepared against.""" + return self._facts.size + + def get_rows_below_cursor_position(self) -> int: + """Report the native row count, or raise as a backend without one would. + + :return: rows below the cursor within the usable region + :raises NotImplementedError: if the backend has no native answer + """ + if self._facts.rows_below_cursor is None: + raise NotImplementedError + return self._facts.rows_below_cursor + + def encoding(self) -> str: + """Report the backend's encoding.""" + return self._facts.encoding + + def get_default_color_depth(self) -> ColorDepth: + """Report the backend's default color depth.""" + return self._facts.default_color_depth + + @property + def responds_to_cpr(self) -> bool: + """Whether the backend answers cursor-position reports.""" + return self._facts.responds_to_cpr + + def fileno(self) -> int: + """Report the backend's file descriptor. + + :return: the descriptor + :raises io.UnsupportedOperation: if the backend has none + """ + if self._facts.fileno is None: + raise io.UnsupportedOperation("the recorded backend has no file descriptor") + return self._facts.fileno + + # -- ownership boundaries, which cannot be deferred ------------------------------------ + + def enter_alternate_screen(self) -> None: + """Refuse to record a switch to the alternate screen. + + :raises UnrecordableOperationError: always + """ + raise UnrecordableOperationError( + "entering the alternate screen is an ownership transition and cannot be recorded in a frame" + ) + + def quit_alternate_screen(self) -> None: + """Refuse to record a return to the main screen. + + :raises UnrecordableOperationError: always + """ + raise UnrecordableOperationError( + "leaving the alternate screen is an ownership transition and cannot be recorded in a frame" + ) + + def scroll_buffer_to_prompt(self) -> None: + """Refuse to record a viewport move. + + :raises UnrecordableOperationError: always + """ + raise UnrecordableOperationError("moving the viewport invalidates the geometry this frame was prepared against") + + # -- recorded operations --------------------------------------------------------------- + + def write(self, data: str) -> None: + """Record a text write. + + :param data: the text + """ + self._record("write", data) + + def write_raw(self, data: str) -> None: + """Record a raw write. + + :param data: the raw data + """ + self._record("write_raw", data) + + def flush(self) -> None: + """Record a flush boundary.""" + self._record("flush") + + def set_title(self, title: str) -> None: + """Record a title change. + + :param title: the title + """ + self._record("set_title", title) + + def clear_title(self) -> None: + """Record clearing the title.""" + self._record("clear_title") + + def erase_screen(self) -> None: + """Record a screen erase.""" + self._record("erase_screen") + + def erase_down(self) -> None: + """Record an erase from the cursor to the bottom.""" + self._record("erase_down") + + def erase_end_of_line(self) -> None: + """Record an erase to the end of the line.""" + self._record("erase_end_of_line") + + def set_attributes(self, attrs: "Attrs", color_depth: ColorDepth) -> None: + """Record an attribute change. + + :param attrs: the attributes + :param color_depth: the color depth to render them at + """ + self._record("set_attributes", attrs, color_depth) + + def reset_attributes(self) -> None: + """Record an attribute reset.""" + self._record("reset_attributes") + + def disable_autowrap(self) -> None: + """Record turning automatic wrapping off.""" + self._record("disable_autowrap") + + def enable_autowrap(self) -> None: + """Record turning automatic wrapping on.""" + self._record("enable_autowrap") + + def cursor_goto(self, row: int = 0, column: int = 0) -> None: + """Record a cursor move. + + :param row: zero-based row + :param column: zero-based column + """ + self._record("cursor_goto", row, column) + + def cursor_up(self, amount: int) -> None: + """Record moving the cursor up. + + :param amount: rows to move + """ + self._record("cursor_up", amount) + + def cursor_down(self, amount: int) -> None: + """Record moving the cursor down. + + :param amount: rows to move + """ + self._record("cursor_down", amount) + + def cursor_forward(self, amount: int) -> None: + """Record moving the cursor right. + + :param amount: columns to move + """ + self._record("cursor_forward", amount) + + def cursor_backward(self, amount: int) -> None: + """Record moving the cursor left. + + :param amount: columns to move + """ + self._record("cursor_backward", amount) + + def hide_cursor(self) -> None: + """Record hiding the cursor.""" + self._record("hide_cursor") + + def show_cursor(self) -> None: + """Record showing the cursor.""" + self._record("show_cursor") + + def set_cursor_shape(self, cursor_shape: "CursorShape") -> None: + """Record a cursor-shape change. + + :param cursor_shape: the shape + """ + self._record("set_cursor_shape", cursor_shape) + + def reset_cursor_shape(self) -> None: + """Record restoring the default cursor shape.""" + self._record("reset_cursor_shape") + + def enable_mouse_support(self) -> None: + """Record turning mouse reporting on.""" + self._record("enable_mouse_support") + + def disable_mouse_support(self) -> None: + """Record turning mouse reporting off.""" + self._record("disable_mouse_support") + + def enable_bracketed_paste(self) -> None: + """Record turning bracketed paste on.""" + self._record("enable_bracketed_paste") + + def disable_bracketed_paste(self) -> None: + """Record turning bracketed paste off.""" + self._record("disable_bracketed_paste") + + def reset_cursor_key_mode(self) -> None: + """Record restoring the default cursor-key mode.""" + self._record("reset_cursor_key_mode") + + def ask_for_cpr(self) -> None: + """Record a cursor-position request. + + The request is emitted when the batch is replayed, in its recorded position, so the + bridge that owns pending-request correlation registers it at commit time rather than + while a frame is still provisional. + """ + self._record("ask_for_cpr") + + def bell(self) -> None: + """Record ringing the bell.""" + self._record("bell") diff --git a/cmd2/prompt_toolkit_bridge.py b/cmd2/prompt_toolkit_bridge.py new file mode 100644 index 000000000..9ee14d58b --- /dev/null +++ b/cmd2/prompt_toolkit_bridge.py @@ -0,0 +1,643 @@ +"""Prepare, commit and recover prompt-toolkit renderer frames. + +Upstream's renderer emits while it thinks. It evaluates the layout, resolves styles, decides a +height, writes the difference against its own last screen, and updates a dozen fields as it +goes -- all inside one call. Two consequences shape this module. + +**Emission has to be separated from preparation.** The render runs against a recorder, off the +terminal lock, and produces an ordered batch. The bridge revalidates that batch against the +current geometry, output, owner and terminal generations and only then replays it, inside one +transaction. A command write, a resize, a handoff or an owner change between those two moments +retires the batch without emitting a byte of it. + +**Discarding the output is only half of discarding the frame.** The renderer advanced its own +state while preparing: ``_last_screen`` became a baseline for a frame the terminal never +received, and the mode flags -- bracketed paste, mouse, cursor-key mode, cursor shape -- latch +next to their emission. A latched flag never re-emits its sequence, so a full repaint does not +repair it. Recovery therefore does two things: it drops the diff baseline, and it re-establishes +a small, explicitly enumerated terminal-state contract, physically, and then tells the renderer +what is now true. + +This is deliberately not a snapshot-and-restore of upstream's fields, and deliberately not a +call to upstream's ``reset()``: that emits operations of its own and rewrites available-height +bookkeeping. It is a narrow initialization contract tied to a qualified prompt-toolkit version, +and :mod:`tests.test_prompt_toolkit_bridge` holds it to that version by name. +""" + +from collections import deque +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from prompt_toolkit.data_structures import Point +from prompt_toolkit.layout.mouse_handlers import MouseHandlers + +from .output_recorder import OperationBatch, PreflightFacts, RecordingOutput +from .terminal_transaction import TerminalLock, assert_no_terminal_transaction + +if TYPE_CHECKING: # pragma: no cover + from collections.abc import Callable + + from prompt_toolkit.application import Application + from prompt_toolkit.renderer import Renderer + + from .terminal_display import TerminalDisplay + + +class ReservedModeFailureError(RuntimeError): + """Raised when reserved rendering cannot continue safely. + + The caller's answer to this is to release the reservation and fall back to compatibility + rendering -- after the release, never before it. Continuing to emit into a terminal whose + state cannot be established is how a prompt ends up drawn over committed output. + """ + + +@dataclass(frozen=True) +class Generations: + """What a prepared frame was prepared against. + + Every field is something that changes where or how the frame's operations would land. + Content is *not* here: a toolbar content change must never authorize an otherwise stale + batch, so it is tracked separately. + """ + + #: The geometry snapshot's generation. + geometry: int + + #: Identity of the output object the frame was recorded against. + output: int + + #: Which UI owner prepared it. + owner: int + + #: Bumped by every managed write, clear or handoff that reaches the terminal. + terminal: int + + +@dataclass(frozen=True) +class PreparedRender: + """One recorded frame and the generations it must still match to be emitted.""" + + #: The operations to replay. + batch: OperationBatch + + #: What the frame was prepared against. + generations: Generations + + +@dataclass(frozen=True) +class TerminalModePolicy: + """The modes the current owner wants established, evaluated on the UI thread. + + Resolving these runs application filters, so it happens before the transaction rather than + between two writes to the terminal. + """ + + #: Whether mouse reporting should be on, per the renderer's filter. + mouse_support: bool + + +class PromptToolkitBridge: + """Binds one renderer to the reserved terminal, and owns what is known about it.""" + + def __init__(self, renderer: "Renderer", display: "TerminalDisplay", lock: TerminalLock) -> None: + """Bind to a renderer and the display that owns the reservation. + + :param renderer: the application's renderer + :param display: the owner of the reservation and its geometry + :param lock: the terminal transaction lock shared by all cmd2-controlled output + """ + self._renderer = renderer + self._display = display + self._lock = lock + self._owner_generation = 0 + self._terminal_generation = 0 + self._content_generation = 0 + self._in_flight: PreparedRender | None = None + self._preparing = False + self._committed: Generations | None = None + self._needs_resynchronization = False + self._reserved_emission_stopped = False + self._redraw_pending = False + self._redraw_scheduler: Callable[[], None] | None = None + self._pending_error: BaseException | None = None + self._prompt_anchor: int | None = None + self._resynchronization_reason: str | None = None + self._pending_cpr: deque[Generations] = deque() + + # -- what is known --------------------------------------------------------------------- + + @property + def needs_resynchronization(self) -> bool: + """Whether recovery is owed before another frame may be prepared.""" + return self._needs_resynchronization + + @property + def reserved_emission_stopped(self) -> bool: + """Whether reserved rendering has been abandoned after unrecoverable failure.""" + return self._reserved_emission_stopped + + @property + def can_dispatch_input(self) -> bool: + """Whether input and after-render notifications may consume the renderer's state. + + False from the moment a preparation starts: the renderer's mouse handlers, visible + windows and cursor position describe a screen the terminal has not been shown, and + they are provisional from the first layout callback -- not merely once the render + call returns. + """ + return not ( + self._needs_resynchronization or self._reserved_emission_stopped or self._preparing or self._in_flight is not None + ) + + @property + def resynchronization_reason(self) -> str | None: + """Why recovery is owed, for diagnostics, or ``None`` when none is.""" + return self._resynchronization_reason + + @property + def in_flight(self) -> PreparedRender | None: + """The frame prepared but not yet committed, if there is one.""" + return self._in_flight + + @property + def redraw_pending(self) -> bool: + """Whether a redraw has been requested and not yet served.""" + return self._redraw_pending + + @property + def content_generation(self) -> int: + """How many times the toolbar's content has been invalidated.""" + return self._content_generation + + @property + def prompt_anchor(self) -> int | None: + """The physical row the prompt is known to start on, or ``None``.""" + return self._prompt_anchor + + def generations(self) -> Generations: + """Snapshot what a frame prepared now would have to match at commit. + + :return: the current generations + """ + geometry = self._display.geometry + return Generations( + geometry=geometry.generation if geometry is not None else 0, + output=id(self._display.output), + owner=self._owner_generation, + terminal=self._terminal_generation, + ) + + def take_pending_error(self) -> BaseException | None: + """Take the failure waiting to be reported, if there is one. + + :return: the error to report once, or ``None`` + """ + error, self._pending_error = self._pending_error, None + return error + + # -- invalidation ---------------------------------------------------------------------- + + def set_redraw_scheduler(self, scheduler: "Callable[[], None]") -> None: + """Install the callback that asks the UI owner for another frame. + + :param scheduler: called once per coalesced redraw request + """ + self._redraw_scheduler = scheduler + + def note_managed_write(self, prompt_anchor: int | None = None) -> None: + """Record that managed output reached the terminal. + + This invalidates the committed baseline rather than only bumping a generation. + Generation comparison catches a write that lands *during* a preparation, but a write + between two frames leaves the renderer believing its last screen is still displayed + and its cursor still where that screen ended -- and the output just emitted moved the + cursor, and may have scrolled everything above it. The next frame would be diffed + against a screen the terminal no longer shows, from an origin it no longer has. + + :param prompt_anchor: the physical row the prompt now starts on, where the layer that + emitted the output knows it. Passing nothing *forgets* the origin rather than + keeping the old one: the write moved the cursor and may have scrolled the screen, + so the remembered row is exactly what is no longer true, and recovery asks the + terminal instead. + """ + self._terminal_generation += 1 + self._prompt_anchor = prompt_anchor + self.require_resynchronization("managed output reached the terminal") + self._request_redraw() + + def note_owner_change(self) -> None: + """Record that a different UI owner now holds the terminal.""" + self._owner_generation += 1 + self._request_redraw() + + def note_geometry_change(self) -> None: + """Record that the terminal's geometry changed under us.""" + self.require_resynchronization("the geometry changed") + self._request_redraw() + + def note_content_change(self) -> None: + """Record that the toolbar's content changed. + + This deliberately does not touch the generations a renderer batch is validated + against: a content change is not a reason to emit a frame prepared against a terminal + that has since moved on. + """ + self._content_generation += 1 + + def require_resynchronization(self, reason: str) -> None: + """Mark that recovery is owed before anything else may be rendered. + + :param reason: why, for diagnostics + """ + self._needs_resynchronization = True + self._resynchronization_reason = reason + self._retire() + + def stop_reserved_emission(self, error: BaseException) -> None: + """Abandon reserved rendering after a failure that could not be cleaned up. + + :param error: what went wrong, to be reported once + """ + self._reserved_emission_stopped = True + self._pending_error = error + self._retire() + + def set_prompt_anchor(self, physical_row: int) -> None: + """Record the physical row the prompt starts on. + + :param physical_row: the one-based row + """ + self._prompt_anchor = physical_row + + def forget_prompt_anchor(self) -> None: + """Record that the prompt's origin is no longer known.""" + self._prompt_anchor = None + + def _request_redraw(self) -> None: + """Ask the owner for another frame, coalescing repeated requests into one.""" + if self._redraw_pending: + return + self._redraw_pending = True + if self._redraw_scheduler is not None: + self._redraw_scheduler() + + def _retire(self) -> None: + """Drop the in-flight frame and the baseline it would have become.""" + self._in_flight = None + self._renderer._last_screen = None + + # -- prepare and commit ---------------------------------------------------------------- + + def prepare(self, app: "Application[Any]") -> PreparedRender | None: + """Record a full renderer frame without emitting anything. + + :param app: the application to render + :return: the prepared frame, or ``None`` if one cannot be prepared right now + """ + assert_no_terminal_transaction("preparing a renderer frame") + if self._reserved_emission_stopped or self._needs_resynchronization or self._preparing or self._in_flight is not None: + return None + + # One transaction, so a managed write or resize cannot land between reading the + # terminal and recording which generation was read: facts and generations have to + # describe the same terminal, or the batch is validated against one snapshot and + # rendered from another. + with self._lock.transaction("preflight"): + facts = PreflightFacts.capture(self._display.output) + generations = self.generations() + + recorder = RecordingOutput(facts) + original = self._renderer.output + self._renderer.output = recorder + # Marked before the renderer is invoked, not after it returns. Layout, filter and + # style callbacks run inside that call, and from the first of them the renderer's + # state is provisional -- a second render or an input dispatch started from one of + # them would consume a screen that does not exist. + self._preparing = True + try: + self._renderer.render(app, app.layout) + except Exception as error: # noqa: BLE001 - a layout callback must not end a command + self._pending_error = error + self.require_resynchronization("preparing the frame raised") + return None + finally: + self._renderer.output = original + self._preparing = False + + with self._lock.transaction("publish"): + # Checking and publishing are one step. Apart, they are two, and a writer that + # abandons the reservation between them has its retirement overwritten by the + # publication that follows -- leaving a frame in flight that nothing invalidated + # and everything downstream believes is current. + if self.needs_resynchronization or self.reserved_emission_stopped: + # Something invalidated the terminal while the frame was being prepared -- a + # managed write from inside a layout callback, say, or a failure that + # abandoned the reservation outright. Either way the operations are recorded + # against a terminal that has moved on. + self._retire() + return None + + prepared = PreparedRender(batch=recorder.batch(), generations=generations) + self._in_flight = prepared + return prepared + + def commit(self, prepared: PreparedRender) -> bool: + """Revalidate a prepared frame and, if it is still current, emit it. + + :param prepared: the frame to commit + :return: whether the frame was emitted in full + """ + assert_no_terminal_transaction("committing a prepared frame") + with self._lock.transaction("commit", generation=prepared.generations.geometry): + if prepared is not self._in_flight: + # Already retired, or from a previous attempt. Replaying it would emit a frame + # nothing has validated, and possibly emit it twice. + # + # Read here, after the terminal has been acquired: read before the wait, it + # answers a question about a terminal somebody else still held, since a writer + # can retire the batch while this call queues, changing neither the + # generations nor the size. + return False + if self._needs_resynchronization or self._reserved_emission_stopped: + # Not covered by the identity test above. Retirement clears the frame in + # flight, but a preparation completing concurrently can publish a new one over + # that retirement, and the frame it publishes matches on identity and on every + # generation. What makes it uncommittable is the state of the terminal, so + # that is what is asked. + return False + if self.generations() != prepared.generations: + self.require_resynchronization("the terminal changed between preparing and committing") + return False + if prepared.batch.facts.size != self._display.output.get_size(): + # The generations can agree while the frame was laid out for a different + # terminal -- the geometry is read once per snapshot, and the batch carries + # what the renderer actually branched on. The frame's own facts are the last + # word on whether it still fits. + self.require_resynchronization("the frame was laid out for a different size") + return False + try: + prepared.batch.replay(self._display.output) + self._display.output.flush() + except Exception as error: # noqa: BLE001 - the terminal's state is now unknown + # Some of the batch reached the terminal and some did not, and nothing here + # knows where the boundary was. The frame is never replayed: a retry would + # duplicate whatever already landed. + self._pending_error = error + self.require_resynchronization("a frame was only partly emitted") + self._attempt_cleanup() + return False + + self._in_flight = None + self._committed = prepared.generations + self._redraw_pending = False + return True + + def _attempt_cleanup(self) -> bool: + """Bring the terminal back to a known state after a partial commit. + + :return: whether a known state was re-established + """ + try: + output = self._display.output + output.reset_attributes() + output.enable_autowrap() + output.flush() + self._display.reconfigure() + except Exception as error: # noqa: BLE001 - cleanup failing is itself the answer + self._reserved_emission_stopped = True + self._pending_error = error + return False + return True + + # -- recovery -------------------------------------------------------------------------- + + def resynchronize(self) -> None: + """Re-establish a known terminal state and a known prompt origin. + + Call this on the UI owner's thread: the mode policy is resolved from application + filters, which need the application's context and must not run under the lock. + + :raises ReservedModeFailureError: if reserved rendering has stopped, or no prompt + origin can be established at all + """ + assert_no_terminal_transaction("resynchronizing the terminal") + # Resolved before the terminal is taken: this runs application filters, which the + # wait contract keeps off the lock. + policy = self._desired_policy() + + with self._lock.transaction("resynchronize"): + if self._reserved_emission_stopped: + # Checked here rather than before the wait. Rendering can be abandoned while + # this call queues for the terminal, and recovery would then write cursor and + # mode sequences into a terminal nothing is allowed to emit to any more. + raise ReservedModeFailureError("reserved emission has stopped; release before rendering again") + # The origin is read *here*, not before the wait. Recovery can queue behind + # another writer for as long as that writer holds the terminal, and what it does + # in the meantime -- emitting output, moving the prompt, resizing -- is exactly + # what changes where the prompt now starts. An origin read beforehand describes a + # terminal somebody else still owned. + origin = self._usable_prompt_anchor() + if origin is None: + can_report = self._display.output.responds_to_cpr + else: + self._establish(policy, origin) + return + + if not can_report: + raise ReservedModeFailureError("the prompt's origin is unknown and the terminal does not report its cursor") + # The reply establishes the origin. Recovery stays owed until it arrives; guessing + # would repaint the prompt over committed output. + self.request_cursor_position() + + def _establish(self, policy: TerminalModePolicy, origin: int) -> None: + """Put the terminal into the known state, from inside the transaction. + + Recovery is marked complete here rather than after the lock is given back: whoever + takes the terminal next must not find a recovery still owed against work that has + already been done. + + :param policy: the mode policy to establish + :param origin: the physical row to place the cursor on + """ + output = self._display.output + output.write_raw(f"\x1b[{origin};1H") + # Upstream enables bracketed paste on every render and latches a flag beside the + # emission, so the policy here is not conditional: it is on, and the flag is made + # to agree with an enable that actually reached the terminal. + output.enable_bracketed_paste() + if policy.mouse_support: + output.enable_mouse_support() + else: + output.disable_mouse_support() + output.reset_cursor_key_mode() + output.reset_attributes() + output.enable_autowrap() + output.reset_cursor_shape() + output.show_cursor() + output.flush() + self._needs_resynchronization = False + self._resynchronization_reason = None + self._in_flight = None + self._initialize_renderer(policy, origin) + + def _usable_rows(self) -> int: + """How many rows the application may use right now. + + :return: the usable height + """ + geometry = self._display.geometry + if geometry is not None: + return geometry.usable_rows + return int(self._display.output.get_size().rows) + + def _usable_prompt_anchor(self) -> int | None: + """Return the remembered prompt origin, if it is still inside the usable region. + + A remembered row survives a resize that the row does not: after the terminal shrinks, + row 20 may be inside the reserved band or off the screen entirely. Rendering from + there would put the prompt in the toolbar's rows, so an anchor that no longer fits is + forgotten and re-established rather than trusted. + + :return: the anchor, or ``None`` if there is none or it is out of range + """ + anchor = self._prompt_anchor + if anchor is None: + return None + if not 1 <= anchor <= self._usable_rows(): + self._prompt_anchor = None + return None + return anchor + + def _desired_policy(self) -> TerminalModePolicy: + """Evaluate the current owner's mode policy, off the terminal lock. + + :return: the policy to establish + """ + return TerminalModePolicy(mouse_support=bool(self._renderer.mouse_support())) + + def _initialize_renderer(self, policy: TerminalModePolicy, origin: int) -> None: + """Tell the renderer what the terminal now is. + + This is the version-specific half of recovery. Each assignment answers a field that + upstream advances during rendering and never re-checks: the diff baseline and its size + and style, the latched mode flags, the believed cursor position that a full repaint + moves *from*, the mouse handlers a discarded frame published, and the available-height + bookkeeping that a visible toolbar says nothing about. + + :param policy: the policy just established physically + :param origin: the physical row the cursor was just placed on + """ + renderer = self._renderer + renderer._bracketed_paste_enabled = True + renderer._mouse_support_enabled = policy.mouse_support + renderer._cursor_key_mode_reset = True + # Cleared rather than set: the shape was reset physically, and clearing the cache is + # what makes the next frame establish the application's own shape again. + renderer._last_cursor_shape = None + renderer._cursor_pos = Point(x=0, y=0) + renderer._last_screen = None + renderer._last_size = None + renderer._last_style = None + renderer.mouse_handlers = MouseHandlers() + # Not zero. The cursor was just placed on a known row inside the usable region, so the + # height below it is known by the same arithmetic a cursor-position reply would give: + # zeroing it would leave the prompt's height unknown while input was allowed to + # resume, which is the invalid state recovery exists to leave behind. + renderer._min_available_height = self._usable_rows() - origin + 1 + + # -- cursor position reports ----------------------------------------------------------- + + def request_cursor_position(self) -> bool: + """Ask the terminal where the cursor is, in its own transaction. + + The request is never emitted inside another transaction. A paint temporarily occupies + the band and restores the cursor afterwards; a request made in the middle of one would + be answered with the painter's cursor, not the prompt's. + + :return: whether a request was emitted + """ + assert_no_terminal_transaction("requesting a cursor position report") + output = self._display.output + if not output.responds_to_cpr: + return False + with self._lock.transaction("cursor position request"): + if self._reserved_emission_stopped: + # Recovery gives the terminal back before asking for it again to send this + # request, so emission can be abandoned in between. Checked here for the same + # reason recovery checks it here: before the wait, the answer describes a + # terminal somebody else still held. + return False + # Recorded here, not before the wait. A managed write can land while this call + # queues for the terminal, and a request stamped with the generations from before + # that write would have its own reply rejected as stale. + # + # The whole generation tuple, not just the geometry: the terminal samples the + # cursor when it processes the request, so output written afterwards moves the + # very thing the reply describes. + generations = self.generations() + output.ask_for_cpr() + output.flush() + self._pending_cpr.append(generations) + return True + + def report_cursor_row(self, row: int) -> bool: + """Take a cursor-position reply, in physical coordinates. + + Replies carry no generation on the wire, so they are correlated by order against the + requests this bridge made. A reply from before a geometry change describes a screen + that no longer exists and must not satisfy the request made after it. + + A reply is stale when anything about the terminal has changed since the request went + out -- a resize, an owner change, or managed output that moved the cursor the terminal + was about to sample. + + A row inside the reserved band is the failure named in the design: upstream would + compute ``U - r + 1``, which is zero at the first reserved row and negative below it, + and would leave the prompt's height silently invalid rather than raising. + + Validating the reply and publishing the origin it establishes happen in one terminal + transaction. Split, they are two steps a managed write can land between: the reply + passes as current, the write moves the prompt, and the anchor it just recorded is then + overwritten by a row that is no longer where the prompt is. Managed output reaches the + terminal inside this same lock, so a reply validated here cannot be overtaken by one. + + Completing the renderer's own pending report only schedules its callbacks on the event + loop, which is not a wait and dispatches no application code, so it belongs inside the + transaction with the decision it settles. + + :param row: the one-based physical row the terminal reported + :return: whether the reply was accepted and used + """ + with self._lock.transaction("cursor position report"): + if not self._pending_cpr: + # Nothing outstanding: a late reply from a stream that was already drained. It + # must not be allowed to answer a request that was never made. + self._settle_renderer_cpr() + return False + # Popped whatever the outcome: replies correlate by order, so dropping one without + # taking it off the queue would answer every later request with its predecessor. + generations = self._pending_cpr.popleft() + if generations != self.generations(): + self._settle_renderer_cpr() + return False + + usable = self._usable_rows() + if not 1 <= row <= usable: + self._settle_renderer_cpr() + self.require_resynchronization(f"cursor position report row {row} is inside the reserved band") + return False + + self._prompt_anchor = row + self._renderer.report_absolute_cursor_row(row) + return True + + def _settle_renderer_cpr(self) -> None: + """Resolve one of the renderer's own pending reports, if it has any. + + A rejected reply still has to settle the bookkeeping it would have settled. Left + pending, the renderer waits for a report that is never coming. + """ + futures = self._renderer._waiting_for_cpr_futures + if futures: + futures.popleft().set_result(None) diff --git a/cmd2/terminal_transaction.py b/cmd2/terminal_transaction.py new file mode 100644 index 000000000..babc0ed69 --- /dev/null +++ b/cmd2/terminal_transaction.py @@ -0,0 +1,244 @@ +"""The terminal transaction lock and the wait contract that keeps it deadlock-free. + +Every byte cmd2 sends to the terminal -- a renderer frame replayed by the bridge, a toolbar +paint, a managed write, a margin change -- is emitted inside one *terminal transaction*. +Serializing individual output methods is not enough: a paint that lands between a renderer's +cursor move and its text write puts the text somewhere other than where the renderer meant +it, and both calls were individually locked. + +:class:`TerminalLock` is ``L_terminal``: the last lock in the output path. The rules it +enforces are the ones that make "last" true. + +**Nothing higher-level may be held while taking it.** Stream routing locks, ownership and +lifecycle locks, queue locks -- all are released first. :class:`HigherLevelLock` records that +a thread holds one, so taking the terminal lock underneath it is refused *before* a real +blocking acquire rather than discovered as a deadlock. + +**Nothing that can wait may run while holding it.** No application callback, no proxy drain, +no future or event or queue wait, no join, no sleep. :func:`guarded_call` refuses those +synchronously, before the primitive is entered; a violation that fires after the wait has +begun protects nothing. + +The unavoidable exception is leaf I/O. ``write()``, ``flush()`` and native console calls can +block on the operating system, and they belong inside the boundary precisely because they are +the emission. Isolating them behind the physical backend keeps them from calling back into +cmd2; it does not make them non-blocking, and this module claims no such thing. + +The guard is per-thread and always on. Its cost is a thread-local attribute read, which is +cheaper than the class of bug it catches is to diagnose from a hung terminal. +""" + +import threading +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from types import TracebackType +from typing import Any, Protocol, Self, TypeVar + +_T = TypeVar("_T") + + +class TerminalTransactionViolationError(RuntimeError): + """Raised when the lock or wait contract would be broken. + + This is deliberately an error rather than a warning. The operations it guards deadlock or + corrupt the display when they are allowed through, and both failures are far harder to + attribute after the fact than an exception at the call site. + + The design calls this ``TerminalTransactionViolation``; the ``Error`` suffix is the + repository's naming rule for exceptions. + """ + + +class _Lock(Protocol): + """The subset of a lock this module uses, so tests can supply a non-blocking double.""" + + def acquire(self, *args: Any, **kwargs: Any) -> bool: + """Take the lock.""" + ... # pragma: no cover + + def release(self) -> None: + """Give the lock back.""" + ... # pragma: no cover + + +@dataclass(frozen=True) +class TransactionState: + """What the debug guard records about the transaction a thread is inside.""" + + #: What the transaction is for, for diagnostics: ``"paint"``, ``"commit"``, and so on. + kind: str + + #: How many nested emission helpers are sharing this transaction. + depth: int + + #: The thread that owns it. A transaction is never visible to another thread. + thread_id: int + + #: The geometry generation the transaction validated against, where it has one. + generation: int | None = None + + +class _GuardState(threading.local): + """Per-thread record of the transaction and the higher-level locks this thread holds.""" + + def __init__(self) -> None: + self.transaction: TransactionState | None = None + self.held_locks: list[str] = [] + + +_state = _GuardState() + + +def current_transaction() -> TransactionState | None: + """Report the terminal transaction this thread is inside, if any. + + :return: the active transaction state, or ``None`` + """ + return _state.transaction + + +def held_higher_level_locks() -> tuple[str, ...]: + """Report the higher-level locks this thread holds, outermost first. + + :return: the names of the held locks + """ + return tuple(_state.held_locks) + + +def assert_no_terminal_transaction(operation: str) -> None: + """Refuse an operation that must not run inside a terminal transaction. + + Call this *before* every blocking helper, callback dispatch, proxy drain or close, and + join that cmd2 owns -- not after, and not inside the primitive. + + :param operation: what was about to happen, named for the error message + :raises TerminalTransactionViolationError: if this thread is inside a transaction + """ + active = _state.transaction + if active is not None: + raise TerminalTransactionViolationError( + f"{operation} is not allowed inside the {active.kind} terminal transaction " + f"(depth {active.depth}); release the terminal lock first" + ) + + +def guarded_call(operation: str, func: Callable[..., _T], *args: Any, **kwargs: Any) -> _T: + """Run a call that may block, refusing it inside a terminal transaction. + + :param operation: what the call is, named for the error message + :param func: the callable to run + :param args: positional arguments for ``func`` + :param kwargs: keyword arguments for ``func`` + :return: whatever ``func`` returns + :raises TerminalTransactionViolationError: if this thread is inside a transaction + """ + assert_no_terminal_transaction(operation) + return func(*args, **kwargs) + + +class HigherLevelLock: + """A lock that ranks above ``L_terminal`` and must be released before it is taken. + + Stream routing, ownership and lifecycle, application state and work queues all live here. + Holding one while acquiring the terminal lock is the deadlock: the thread holding the + terminal lock cannot finish emitting until a worker gets the routing lock back, and the + worker cannot until the emitter gives it up. + """ + + def __init__(self, name: str, lock: _Lock | None = None) -> None: + """Wrap a lock under a name that appears in violation messages. + + :param name: what this lock protects, for diagnostics + :param lock: the lock to wrap; a fresh :class:`threading.RLock` by default + """ + self._name = name + self._lock: _Lock = lock if lock is not None else threading.RLock() + + @property + def name(self) -> str: + """What this lock protects.""" + return self._name + + def __enter__(self) -> Self: + """Take the lock, refusing to do so from inside a terminal transaction. + + :return: this lock + :raises TerminalTransactionViolationError: if this thread is inside a transaction + """ + assert_no_terminal_transaction(f"acquiring the {self._name} lock") + self._lock.acquire() + _state.held_locks.append(self._name) + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Give the lock back, including when the body raised.""" + _state.held_locks.pop() + self._lock.release() + + +class TerminalLock: + """``L_terminal``: the final lock in the output path. + + Re-entrant on one thread so that an emission helper may call another, but a nested + transaction joins the outer one rather than starting its own -- the outer transaction's + kind and generation are what a violation message should name, because the outer one is + what validated against the terminal. + """ + + def __init__(self, lock: _Lock | None = None) -> None: + """Build a terminal lock. + + :param lock: the lock to serialize on; a fresh :class:`threading.RLock` by default + """ + self._lock: _Lock = lock if lock is not None else threading.RLock() + + @property + def active(self) -> bool: + """Whether this thread is currently inside a terminal transaction.""" + return _state.transaction is not None + + @contextmanager + def transaction(self, kind: str, generation: int | None = None) -> Iterator[TransactionState]: + """Hold the terminal for the duration of one transaction. + + :param kind: what the transaction is for, for diagnostics + :param generation: the geometry generation it was validated against, if any + :return: a context manager yielding the transaction state + :raises TerminalTransactionViolationError: if this thread holds a higher-level lock + """ + active = _state.transaction + if active is not None: + nested = TransactionState( + kind=active.kind, + depth=active.depth + 1, + thread_id=active.thread_id, + generation=active.generation, + ) + _state.transaction = nested + try: + yield nested + finally: + _state.transaction = active + return + + if _state.held_locks: + held = ", ".join(_state.held_locks) + raise TerminalTransactionViolationError( + f"cannot start the {kind} terminal transaction while holding {held}; " + f"higher-level locks are released before the terminal lock is taken" + ) + + self._lock.acquire() + state = TransactionState(kind=kind, depth=1, thread_id=threading.get_ident(), generation=generation) + _state.transaction = state + try: + yield state + finally: + _state.transaction = None + self._lock.release() diff --git a/cmd2/toolbar_painter.py b/cmd2/toolbar_painter.py new file mode 100644 index 000000000..51a0f40ec --- /dev/null +++ b/cmd2/toolbar_painter.py @@ -0,0 +1,477 @@ +"""Lay toolbar content out as display cells and paint only what changed. + +The toolbar in reserved mode is not a prompt-toolkit window. It is painted independently into +rows the application cannot reach, which is what keeps it out of the renderer's diff and out +of its erases -- and it is why the layout work upstream would have done has to be done here. + +A :class:`ToolbarFrame` is a grid of what the terminal will *show*: one :class:`Cell` per +display column, per row of the band. Comparing two frames therefore answers "will the user see +a difference", not "did the Python string change", which is the comparison that decides +whether anything is emitted at all. A wide character owns two cells, a combining character +owns none of its own, and a wide character is never split across the right edge -- half a +character at the edge is what makes a terminal wrap a row into the one below it. + +Content is laid out rather than passed through: + +**Zero-width escape fragments are dropped.** ``[ZeroWidthEscape]`` fragments carry raw +terminal control, and arbitrary control inside a paint moves the physical cursor out of the +band and into the application's rows. + +**Mouse handlers are dropped, text and style are kept.** The band is outside prompt-toolkit's +mouse map, so a handler here would never be called; rendering the visible part is the honest +subset rather than advertising support that does not exist. + +**Carriage returns are dropped and tabs are expanded.** Both are cursor motion in a context +where the painter owns the cursor. +""" + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from prompt_toolkit.formatted_text import to_formatted_text +from prompt_toolkit.output import ColorDepth +from prompt_toolkit.utils import get_cwidth + +from .scroll_region import cursor_restore_sequence, cursor_save_sequence +from .terminal_transaction import TerminalLock, assert_no_terminal_transaction + +if TYPE_CHECKING: # pragma: no cover + from collections.abc import Callable, Mapping + + from prompt_toolkit.formatted_text import AnyFormattedText + from prompt_toolkit.styles import Attrs, BaseStyle + + from .terminal_display import TerminalDisplay + +#: Columns between tab stops. Tabs are expanded during layout because the painter positions +#: the cursor itself; letting a tab reach the terminal would move it by an amount the frame +#: does not model. +TAB_WIDTH = 8 + +#: Fragments whose style contains this carry raw terminal control rather than text. +_ZERO_WIDTH_ESCAPE = "[ZeroWidthEscape]" + + +@dataclass(frozen=True) +class Cell: + """One display column of the toolbar band.""" + + #: What is drawn here: a character, a character plus its combining marks, or ``""`` for + #: the second half of a wide character and for a blank continuation cell. + char: str + + #: The style string that applies to this column. + style: str + + #: Whether this cell is the right half of a wide character in the cell before it. + is_continuation: bool = False + + @property + def width(self) -> int: + """How many columns this cell's content occupies: two for a wide character, else one.""" + if self.is_continuation: + return 0 + return get_cwidth(self.char) + + +@dataclass(frozen=True) +class ToolbarFrame: + """An immutable grid of cells: exactly ``height`` rows of exactly ``width`` cells.""" + + #: The rows of the band, top first. + rows: tuple[tuple[Cell, ...], ...] + + @property + def height(self) -> int: + """How many rows the frame occupies.""" + return len(self.rows) + + @property + def width(self) -> int: + """How many columns each row occupies.""" + return len(self.rows[0]) if self.rows else 0 + + @classmethod + def build( + cls, + content: "AnyFormattedText", + width: int, + height: int, + default_style: str = "", + ) -> "ToolbarFrame": + """Lay content out into a frame of exactly ``height`` by ``width`` cells. + + Content that does not fill the frame is padded with default-styled spaces: the pad is + what overwrites a longer previous frame, so it is content rather than absence of it. + Content taller than the band is truncated here -- growing the toolbar is a geometry + transition, and writing the extra rows would put them outside the reservation. + + :param content: the formatted text to lay out + :param width: the terminal width in columns + :param height: the height of the reserved band in rows + :param default_style: the style for padding cells + :return: the frame + :raises ValueError: if ``width`` or ``height`` is not positive + """ + if width < 1: + raise ValueError(f"a frame needs a positive width, got {width}") + if height < 1: + raise ValueError(f"a frame needs a positive height, got {height}") + + rows = _layout(content, width, default_style) + blank = tuple(Cell(" ", default_style) for _ in range(width)) + while len(rows) < height: + rows.append(list(blank)) + return cls(rows=tuple(tuple(row) for row in rows[:height])) + + +def measure_toolbar_height(content: "AnyFormattedText", width: int) -> int: + """Measure how many rows content needs at a given width. + + This is what sizes the reservation, so it counts wrapping and explicit newlines the same + way :meth:`ToolbarFrame.build` lays them out. Empty content still measures one row: an + empty toolbar is an intentional visibility change, not a request for no reservation. + + :param content: the formatted text to measure + :param width: the terminal width in columns + :return: the number of rows required, at least one + :raises ValueError: if ``width`` is not positive + """ + if width < 1: + raise ValueError(f"measuring needs a positive width, got {width}") + return max(1, len(_layout(content, width, ""))) + + +def _layout(content: "AnyFormattedText", width: int, default_style: str) -> list[list[Cell]]: + """Lay content out into as many full-width rows as it needs. + + :param content: the formatted text to lay out + :param width: the terminal width in columns + :param default_style: the style for padding cells + :return: the rows, each padded to ``width`` cells + """ + rows: list[list[Cell]] = [] + row: list[Cell] = [] + + def finish_row() -> None: + """Pad the row in progress and start a new one.""" + row.extend(Cell(" ", default_style) for _ in range(width - len(row))) + rows.append(list(row)) + row.clear() + + for fragment in to_formatted_text(content): + style, text = fragment[0], fragment[1] + if _ZERO_WIDTH_ESCAPE in style: + continue + for char in text: + if char == "\n": + finish_row() + continue + if char == "\r": + continue + if char == "\t": + spaces = TAB_WIDTH - (len(row) % TAB_WIDTH) + for _ in range(spaces): + if len(row) == width: + finish_row() + row.append(Cell(" ", style)) + continue + + char_width = get_cwidth(char) + if char_width == 0 and row: + # A combining mark belongs to the character it follows; it occupies no column + # of its own, so it joins that cell rather than becoming one. After a wide + # character the cell to the left is that character's right half, and the mark + # belongs to the half that carries the text -- attaching it to the + # continuation cell would give the mark a column of its own and shift every + # later cell one place right of where it is on the screen. + base = len(row) - 1 + if row[base].is_continuation: + base -= 1 + previous = row[base] + row[base] = Cell(previous.char + char, previous.style, previous.is_continuation) + continue + columns = max(1, char_width) + if len(row) + columns > width: + # Padding rather than splitting: half a wide character at the right edge is + # what makes the terminal wrap the row itself, which would put toolbar cells + # in a row the frame does not own. + finish_row() + row.append(Cell(char, style)) + if columns == 2: + row.append(Cell("", style, is_continuation=True)) + + if row: + finish_row() + return rows + + +@dataclass(frozen=True) +class PreparedFrame: + """A frame with every style already resolved, ready to emit. + + Style resolution runs application-supplied style rules, so it happens here -- during + preparation, off the terminal lock -- rather than between two writes to the band. + """ + + #: The cells to paint. + frame: ToolbarFrame + + #: Resolved attributes for every style string the frame uses. + attrs: "Mapping[str, Attrs]" + + #: The color depth those attributes are rendered at. + color_depth: ColorDepth + + #: The geometry generation the frame was laid out for. The band's physical rows come + #: from this snapshot, so emitting the frame against any other one writes into rows the + #: application now owns. + generation: int + + +class ToolbarPainter: + """Paints the reserved band, writing only the cells that changed. + + The painter is independent of the renderer: it writes to physical rows the application's + geometry excludes, so its output is never part of a renderer diff and never erased by one. + That independence is the whole mechanism, and it is also why the painter has to restore + everything it touches -- the renderer's next frame assumes the cursor, attributes and wrap + mode are where it left them. + + Nothing is erased before writing. An erase followed by a write is two visible states, and + the flicker of that pair is what this design exists to remove; a changed run is overwritten + in place and a shortened frame's tail is padded, which is one state. + """ + + def __init__( + self, + display: "TerminalDisplay", + lock: TerminalLock, + style: "BaseStyle", + color_depth: ColorDepth, + default_style: str = "", + autowrap_after_paint: bool = True, + ) -> None: + """Bind a painter to the display that owns the reservation. + + The painter takes the display rather than an output and a geometry. The band's rows + are physical, so a frame laid out for one geometry addresses the wrong rows under any + other -- and a caller that passes both an output and a snapshot can pass a stale one. + Reading the geometry from its owner, inside the transaction, removes that possibility. + + Painting goes to the *original* backend: the band is outside the application's + geometry, so the reserved adapter is a view that excludes it. + + :param display: the owner of the reservation and its geometry + :param lock: the terminal transaction lock shared by all cmd2-controlled output + :param style: the style rules used to resolve fragment styles + :param color_depth: the color depth to render attributes at + :param default_style: the style for padding cells + :param autowrap_after_paint: the committed autowrap policy to restore afterwards. + Upstream's renderer leaves autowrap enabled between frames, which is the default + here; a bridge that has committed a different policy passes it instead. + """ + self._display = display + self._output = display.terminal.output + self._lock = lock + self._style = style + self._color_depth = color_depth + self._default_style = default_style + 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._pending_error: BaseException | None = None + + @property + def last_frame(self) -> ToolbarFrame | None: + """The frame currently believed to be on the screen, or ``None`` after invalidation.""" + return self._last_frame + + def take_pending_error(self) -> BaseException | None: + """Take the unreported content-callback error, if there is one. + + Taking it is what re-arms evaluation: until the user has been told, retrying the same + failing callback on every refresh would report the same error forever. + + :return: the error to report once, or ``None`` + """ + error, self._pending_error = self._pending_error, None + return error + + def invalidate(self) -> None: + """Forget what is on the screen, so the next paint writes the whole band. + + Recovery, a geometry change and a handoff all leave the band's contents unknown. A + diff against a frame that may no longer be displayed would write nothing at all. + """ + self._last_frame = None + self._last_attrs = None + self._last_band = None + + def prepare(self, content: "Callable[[], AnyFormattedText]") -> PreparedFrame | None: + """Evaluate the toolbar's content once and lay it out, off the terminal lock. + + :param content: the callback returning the toolbar's formatted text + :return: the prepared frame, or ``None`` if there is no reservation to paint, or + evaluation failed, or a previous failure is still waiting to be reported + """ + assert_no_terminal_transaction("evaluating the toolbar's content") + if self._pending_error is not None: + return None + geometry = self._display.geometry + if geometry is None: + return None + try: + text = content() + except Exception as error: # noqa: BLE001 - a toolbar callback must not end a command + # The last good frame stays on screen. A toolbar that blanks itself because a + # callback raised is a worse failure than a stale one, and the command that was + # running is not this callback's to interrupt. + self._pending_error = error + return None + frame = ToolbarFrame.build( + text, + width=geometry.columns, + height=geometry.reserved_rows, + default_style=self._default_style, + ) + styles = {cell.style for row in frame.rows for cell in row} + return PreparedFrame( + frame=frame, + attrs={style: self._style.get_attrs_for_style_str(style) for style in styles}, + color_depth=self._color_depth, + generation=geometry.generation, + ) + + def paint(self, prepared: PreparedFrame) -> bool: + """Write the changed cells of the band, inside one terminal transaction. + + The geometry is read from its owner *inside* the transaction and checked against the + one the frame was laid out for. Between preparing and painting the terminal can be + resized, released, or handed to another program, and each of those makes the band's + physical rows rows the application owns instead. A refused frame publishes no + baseline: what the band is showing is then unknown, so the next paint must be full. + + :param prepared: the frame to paint + :return: whether anything was written + """ + with self._lock.transaction("paint", generation=prepared.generation): + geometry = self._display.geometry + if geometry is None or geometry.generation != prepared.generation: + self.invalidate() + return False + + frame = prepared.frame + band = (geometry.physical_rows, geometry.columns, geometry.reserved_rows, geometry.buffer_id) + 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) + if not runs: + self._last_frame = frame + self._last_attrs = prepared.attrs + self._last_band = band + return False + + top_row = geometry.physical_rows - geometry.reserved_rows + 1 + # Anything another writer left buffered goes out first, so the band is painted + # after the output it was meant to follow rather than in the middle of it. + self._output.flush() + # DECSC saves the cursor *and* the current attributes, and DECRC restores both, so + # the renderer's next write lands where and how it expects. + self._output.write_raw(cursor_save_sequence()) + self._output.disable_autowrap() + for row_index, column, cells in runs: + self._output.write_raw(_cursor_position_sequence(top_row + row_index, column + 1)) + style: str | None = None + for cell in cells: + if cell.is_continuation: + continue + if cell.style != style: + self._output.set_attributes(prepared.attrs[cell.style], prepared.color_depth) + style = cell.style + self._output.write(cell.char) + if self._autowrap_after_paint: + self._output.enable_autowrap() + self._output.write_raw(cursor_restore_sequence()) + self._output.flush() + + self._last_frame = frame + self._last_attrs = prepared.attrs + self._last_band = band + return True + + +def _cursor_position_sequence(row: int, column: int) -> str: + """Build a one-based absolute cursor move. + + The band is addressed in physical coordinates, which the application's view does not + contain, so this deliberately does not go through ``cursor_goto``. + + :param row: one-based physical row + :param column: one-based column + :return: the CUP escape sequence + """ + return f"\x1b[{row};{column}H" + + +def _same_cell( + old: Cell, + old_attrs: "Mapping[str, Attrs]", + new: Cell, + new_attrs: "Mapping[str, Attrs]", +) -> bool: + """Decide whether two cells would look identical on the terminal. + + Style *strings* are not enough. A style rule can be changed under a class name -- a theme + switch, a dynamic style -- leaving ``class:status`` naming a different colour than it did + last frame. Comparing the resolved attributes is what makes the comparison a question + about the screen rather than about the text of the style. + + :param old: the cell believed to be displayed + :param old_attrs: resolved attributes as they were when it was painted + :param new: the cell to display + :param new_attrs: resolved attributes for the new frame + :return: whether the terminal would show the same thing + """ + if old.char != new.char or old.is_continuation != new.is_continuation: + return False + return old_attrs.get(old.style) == new_attrs.get(new.style) + + +def _changed_runs( + previous: ToolbarFrame | None, + previous_attrs: "Mapping[str, Attrs] | None", + frame: ToolbarFrame, + attrs: "Mapping[str, Attrs]", +) -> list[tuple[int, int, tuple[Cell, ...]]]: + """Find the spans of cells that differ from what is believed to be on screen. + + A run always begins on a whole character: a wide character's two cells carry the same + style and change together, so a difference can never start on the right half of one. + + :param previous: the frame believed to be displayed, or ``None`` for a full repaint + :param previous_attrs: the attributes that frame was painted with + :param frame: the frame to display + :param attrs: resolved attributes for the frame to display + :return: ``(row index, first column, cells)`` for each run, in order + """ + runs: list[tuple[int, int, tuple[Cell, ...]]] = [] + old_attrs: Mapping[str, Attrs] = previous_attrs if previous_attrs is not None else {} + for row_index, row in enumerate(frame.rows): + previous_row = previous.rows[row_index] if previous is not None else None + column = 0 + while column < len(row): + if previous_row is not None and _same_cell(previous_row[column], old_attrs, row[column], attrs): + column += 1 + continue + start = column + while column < len(row) and ( + previous_row is None or not _same_cell(previous_row[column], old_attrs, row[column], attrs) + ): + column += 1 + # A wide character whose halves straddle the end of the run comes along whole. + while column < len(row) and row[column].is_continuation: + column += 1 + runs.append((row_index, start, tuple(row[start:column]))) + return runs diff --git a/tests/test_output_recorder.py b/tests/test_output_recorder.py new file mode 100644 index 000000000..a24d37900 --- /dev/null +++ b/tests/test_output_recorder.py @@ -0,0 +1,300 @@ +"""Tests for recording a renderer's output operations instead of performing them. + +The property every test here is really about is that preparation leaves the real backend +untouched: not its stream, not its buffered text, not its attribute or cursor caches, not its +console modes. A recorder that quietly used the backend as its sink would pass a test that +only compared the replayed operation list. +""" + +import io +from typing import Any + +import pytest +from prompt_toolkit.cursor_shapes import CursorShape +from prompt_toolkit.data_structures import Size +from prompt_toolkit.output import ColorDepth, DummyOutput +from prompt_toolkit.output.vt100 import Vt100_Output +from prompt_toolkit.styles import Attrs + +from cmd2.output_recorder import ( + Operation, + OperationBatch, + PreflightFacts, + RecordingOutput, + UnrecordableOperationError, +) + +ATTRS = Attrs( + color="ansired", + bgcolor=None, + bold=True, + underline=False, + strike=False, + italic=False, + blink=False, + reverse=False, + hidden=False, + dim=False, +) + + +def make_output(rows: int = 24, cols: int = 80) -> tuple[Vt100_Output, io.StringIO]: + """Build a real Vt100_Output over a string buffer.""" + stream = io.StringIO() + size = Size(rows=rows, columns=cols) + return Vt100_Output(stream, lambda: size), stream + + +def make_recorder(rows: int = 23, cols: int = 80) -> RecordingOutput: + """Build a recorder over fixed preflight facts.""" + facts = PreflightFacts( + size=Size(rows=rows, columns=cols), + rows_below_cursor=None, + encoding="utf-8", + default_color_depth=ColorDepth.DEPTH_8_BIT, + responds_to_cpr=True, + fileno=7, + ) + return RecordingOutput(facts) + + +class TestPreflightFacts: + def test_facts_are_captured_from_the_backend(self) -> None: + output, _stream = make_output() + facts = PreflightFacts.capture(output) + assert facts.size == Size(rows=24, columns=80) + assert facts.encoding == output.encoding() + assert facts.responds_to_cpr == output.responds_to_cpr + assert facts.default_color_depth == output.get_default_color_depth() + + def test_a_backend_without_a_native_row_count_records_none(self) -> None: + """POSIX backends raise here and fall back to CPR; that is a fact, not a failure.""" + output, _stream = make_output() + with pytest.raises(NotImplementedError): + output.get_rows_below_cursor_position() + assert PreflightFacts.capture(output).rows_below_cursor is None + + def test_a_backend_without_a_file_descriptor_records_none(self) -> None: + output = Vt100_Output(io.StringIO(), lambda: Size(rows=24, columns=80)) + assert PreflightFacts.capture(output).fileno is None + + def test_capturing_facts_writes_nothing(self) -> None: + output, stream = make_output() + PreflightFacts.capture(output) + output.flush() + assert stream.getvalue() == "" + + +class TestRecording: + def test_operations_are_recorded_in_order(self) -> None: + recorder = make_recorder() + recorder.cursor_goto(3, 5) + recorder.write("hello") + recorder.erase_down() + assert recorder.operations == ( + Operation("cursor_goto", (3, 5)), + Operation("write", ("hello",)), + Operation("erase_down", ()), + ) + + def test_a_flush_is_recorded_as_a_boundary(self) -> None: + """Flush boundaries are operations, not a signal to touch a real stream.""" + recorder = make_recorder() + recorder.write("hello") + recorder.flush() + assert recorder.operations[-1] == Operation("flush", ()) + + def test_reads_are_answered_from_the_preflight_facts(self) -> None: + recorder = make_recorder(rows=23) + assert recorder.get_size() == Size(rows=23, columns=80) + assert recorder.encoding() == "utf-8" + assert recorder.responds_to_cpr is True + assert recorder.get_default_color_depth() == ColorDepth.DEPTH_8_BIT + assert recorder.fileno() == 7 + + def test_reads_are_not_recorded_as_operations(self) -> None: + """A read is not something to replay; replaying one would emit nothing anyway.""" + recorder = make_recorder() + recorder.get_size() + recorder.encoding() + assert recorder.operations == () + + def test_a_missing_native_row_count_raises_as_the_backend_would(self) -> None: + recorder = make_recorder() + with pytest.raises(NotImplementedError): + recorder.get_rows_below_cursor_position() + + def test_a_missing_file_descriptor_raises_as_the_backend_would(self) -> None: + facts = PreflightFacts( + size=Size(rows=23, columns=80), + rows_below_cursor=None, + encoding="utf-8", + default_color_depth=ColorDepth.DEPTH_8_BIT, + responds_to_cpr=True, + fileno=None, + ) + with pytest.raises(io.UnsupportedOperation): + RecordingOutput(facts).fileno() + + def test_the_recorder_holds_no_reference_to_a_backend(self) -> None: + """The strongest form of 'preparation has no physical side effects'.""" + recorder = make_recorder() + assert not any(isinstance(value, Vt100_Output) for value in vars(recorder).values()) + + def test_a_screen_buffer_transition_cannot_be_recorded(self) -> None: + """Buffer transitions are ownership boundaries, taken before a frame is prepared.""" + recorder = make_recorder() + for transition in (recorder.enter_alternate_screen, recorder.quit_alternate_screen): + with pytest.raises(UnrecordableOperationError): + transition() + + def test_a_viewport_move_cannot_be_recorded(self) -> None: + recorder = make_recorder() + with pytest.raises(UnrecordableOperationError): + recorder.scroll_buffer_to_prompt() + + def test_a_rejected_operation_is_not_left_in_the_batch(self) -> None: + recorder = make_recorder() + recorder.write("hello") + with pytest.raises(UnrecordableOperationError): + recorder.enter_alternate_screen() + assert recorder.operations == (Operation("write", ("hello",)),) + + +class TestReplay: + def test_a_batch_replays_its_operations_onto_a_real_backend(self) -> None: + recorder = make_recorder() + recorder.cursor_goto(2, 0) + recorder.write("hello") + recorder.flush() + + output, stream = make_output() + recorder.batch().replay(output) + # Upstream passes cursor_goto's arguments straight into CUP; the recorder is a delay, + # not a correction, so replay reproduces that byte for byte. + assert stream.getvalue() == "\x1b[2;0Hhello" + + def test_recording_and_replay_produce_what_the_backend_would_have(self) -> None: + """The recorder is a delay, not a translation: the bytes must be identical.""" + direct, direct_stream = make_output() + direct.set_attributes(ATTRS, ColorDepth.DEPTH_8_BIT) + direct.write("hello") + direct.erase_end_of_line() + direct.reset_attributes() + direct.flush() + + recorder = make_recorder() + recorder.set_attributes(ATTRS, ColorDepth.DEPTH_8_BIT) + recorder.write("hello") + recorder.erase_end_of_line() + recorder.reset_attributes() + recorder.flush() + + replayed, replayed_stream = make_output() + recorder.batch().replay(replayed) + assert replayed_stream.getvalue() == direct_stream.getvalue() + + def test_a_batch_is_immutable_once_taken(self) -> None: + """A batch that kept growing after preparation could not be validated at commit.""" + recorder = make_recorder() + recorder.write("first") + batch = recorder.batch() + recorder.write("second") + assert batch.operations == (Operation("write", ("first",)),) + + def test_replay_stops_at_the_operation_that_failed(self) -> None: + """Partial replay is a real state; the batch must not paper over it.""" + + class FailingOutput(Vt100_Output): + def erase_down(self) -> None: + raise OSError("terminal went away") + + recorder = make_recorder() + recorder.write("hello") + recorder.erase_down() + recorder.write("never") + + stream = io.StringIO() + output = FailingOutput(stream, lambda: Size(rows=24, columns=80)) + with pytest.raises(OSError, match="terminal went away"): + recorder.batch().replay(output) + output.flush() + assert stream.getvalue() == "hello" + + def test_an_empty_batch_replays_nothing(self) -> None: + output, stream = make_output() + OperationBatch(operations=(), facts=make_recorder().facts).replay(output) + output.flush() + assert stream.getvalue() == "" + + def test_the_batch_carries_the_facts_it_was_prepared_against(self) -> None: + recorder = make_recorder(rows=23) + assert recorder.batch().facts.size == Size(rows=23, columns=80) + + +#: Every operation the recorder defers, with arguments where it takes them. Replaying each one +#: must produce exactly what calling it on the backend produces, so this table is what keeps a +#: mistyped delegation -- a wrong method name, a dropped argument -- from reaching the terminal. +RECORDED_OPERATIONS: list[tuple[str, tuple[Any, ...]]] = [ + ("write", ("hello",)), + ("write_raw", ("\x1b[7m",)), + ("set_title", ("cmd2",)), + ("clear_title", ()), + ("erase_screen", ()), + ("erase_down", ()), + ("erase_end_of_line", ()), + ("set_attributes", (ATTRS, ColorDepth.DEPTH_8_BIT)), + ("reset_attributes", ()), + ("disable_autowrap", ()), + ("enable_autowrap", ()), + ("cursor_goto", (4, 2)), + ("cursor_up", (3,)), + ("cursor_down", (3,)), + ("cursor_forward", (3,)), + ("cursor_backward", (3,)), + ("hide_cursor", ()), + ("show_cursor", ()), + ("set_cursor_shape", (CursorShape.BLOCK,)), + ("reset_cursor_shape", ()), + ("enable_mouse_support", ()), + ("disable_mouse_support", ()), + ("enable_bracketed_paste", ()), + ("disable_bracketed_paste", ()), + ("reset_cursor_key_mode", ()), + ("ask_for_cpr", ()), + ("bell", ()), + ("flush", ()), +] + + +class TestEveryRecordedOperation: + @pytest.mark.parametrize(("name", "args"), RECORDED_OPERATIONS, ids=[name for name, _ in RECORDED_OPERATIONS]) + def test_replay_matches_a_direct_call_on_the_backend(self, name: str, args: tuple[Any, ...]) -> None: + direct, direct_stream = make_output() + getattr(direct, name)(*args) + direct.flush() + + recorder = make_recorder() + getattr(recorder, name)(*args) + assert recorder.operations[0].name == name + + replayed, replayed_stream = make_output() + recorder.batch().replay(replayed) + replayed.flush() + assert replayed_stream.getvalue() == direct_stream.getvalue() + + def test_a_native_row_count_is_reported_when_the_backend_has_one(self) -> None: + """Windows answers this natively; the recorded render must see the same number.""" + facts = PreflightFacts( + size=Size(rows=23, columns=80), + rows_below_cursor=9, + encoding="utf-8", + default_color_depth=ColorDepth.DEPTH_8_BIT, + responds_to_cpr=True, + fileno=7, + ) + assert RecordingOutput(facts).get_rows_below_cursor_position() == 9 + + def test_a_backend_that_refuses_a_file_descriptor_records_none(self) -> None: + """DummyOutput raises NotImplementedError rather than UnsupportedOperation.""" + assert PreflightFacts.capture(DummyOutput()).fileno is None diff --git a/tests/test_prompt_toolkit_bridge.py b/tests/test_prompt_toolkit_bridge.py new file mode 100644 index 000000000..74bcdc65a --- /dev/null +++ b/tests/test_prompt_toolkit_bridge.py @@ -0,0 +1,891 @@ +"""Tests for preparing, committing and recovering renderer frames. + +Several of these are the named regressions from design section 13.2. The property underneath +all of them is that a frame the terminal never received must never become the baseline the +next frame is diffed against: upstream advances its own state while it renders, so discarding +the *output* is only half of discarding the frame. + +The renderer here is a real prompt-toolkit renderer driving a real application, so the +recorded operations and the flags left behind are the ones production would see. +""" + +import io +import threading +from collections import deque +from concurrent.futures import Future +from typing import Any + +import pytest +from prompt_toolkit.application import Application +from prompt_toolkit.application.current import set_app +from prompt_toolkit.data_structures import Point, Size +from prompt_toolkit.input import DummyInput +from prompt_toolkit.layout import Layout, Window +from prompt_toolkit.layout.controls import FormattedTextControl +from prompt_toolkit.output.vt100 import Vt100_Output + +from cmd2.output_recorder import PreflightFacts +from cmd2.prompt_toolkit_bridge import ( + PromptToolkitBridge, + ReservedModeFailureError, +) +from cmd2.terminal_display import TerminalDisplay +from cmd2.terminal_transaction import TerminalLock, current_transaction + + +class TtyStringIO(io.StringIO): + """A stream that claims to be a terminal, so the backend will use cursor reports.""" + + def isatty(self) -> bool: + return True + + +class Harness: + """A real application over a reserved terminal, with the stream it writes to.""" + + def __init__( + self, + rows: int = 24, + columns: int = 40, + reserved_rows: int = 1, + content: Any = "hello", + lock: TerminalLock | None = None, + ) -> None: + self.stream = TtyStringIO() + self.size = Size(rows=rows, columns=columns) + self.backend = Vt100_Output(self.stream, lambda: self.size) + self.display = TerminalDisplay(self.backend, reserved_rows=reserved_rows) + assert self.display.acquire() is True + self.lock = lock or TerminalLock() + self.app: Application[Any] = Application( + layout=Layout(Window(FormattedTextControl(content))), + output=self.display.output, + input=DummyInput(), + ) + self.bridge = PromptToolkitBridge(renderer=self.app.renderer, display=self.display, lock=self.lock) + self.bridge.set_prompt_anchor(1) + self.clear() + + @property + def renderer(self) -> Any: + """The application's renderer.""" + return self.app.renderer + + def clear(self) -> str: + """Take everything written so far, leaving the stream empty.""" + written = self.stream.getvalue() + self.stream.truncate(0) + self.stream.seek(0) + return written + + def written(self) -> str: + """What has been written since the last clear.""" + return self.stream.getvalue() + + def prepare(self) -> Any: + """Prepare one frame, as the bridge's caller would.""" + with set_app(self.app): + return self.bridge.prepare(self.app) + + def render(self) -> bool: + """Prepare and commit one frame.""" + prepared = self.prepare() + assert prepared is not None + return self.bridge.commit(prepared) + + def resize(self, rows: int, columns: int = 40) -> None: + """Resize the terminal and re-establish the reservation for the new geometry.""" + self.size = Size(rows=rows, columns=columns) + self.display.reconfigure() + + def resynchronize(self) -> None: + """Run recovery in the application's context.""" + with set_app(self.app): + self.bridge.resynchronize() + + +class TestPreparation: + def test_preparation_writes_nothing_to_the_terminal(self) -> None: + harness = Harness() + assert harness.prepare() is not None + assert harness.written() == "" + + def test_render_callbacks_run_before_terminal_commit(self) -> None: + """Named test 13.2: layout and style callbacks must not run inside the transaction.""" + seen: list[object] = [] + + def content() -> str: + seen.append(current_transaction()) + return "hello" + + harness = Harness(content=content) + assert harness.prepare() is not None + assert seen + assert all(state is None for state in seen) + + def test_the_prepared_frame_is_what_gets_emitted(self) -> None: + harness = Harness() + assert harness.render() is True + assert "hello" in harness.written() + + def test_preparation_does_not_advance_the_backend(self) -> None: + """A discarded frame must leave no trace in the backend's own buffering.""" + harness = Harness() + harness.prepare() + harness.backend.flush() + assert harness.written() == "" + + def test_the_application_renders_against_the_usable_height(self) -> None: + """The reservation is subtracted once: 24 physical rows, one reserved, 23 usable.""" + harness = Harness(rows=24, reserved_rows=1) + prepared = harness.prepare() + assert prepared is not None + assert prepared.batch.facts.size == Size(rows=23, columns=40) + + def test_a_failing_preparation_reports_and_asks_for_resynchronization(self) -> None: + def boom() -> str: + raise RuntimeError("layout failed") + + harness = Harness(content=boom) + assert harness.prepare() is None + assert harness.bridge.needs_resynchronization is True + assert isinstance(harness.bridge.take_pending_error(), RuntimeError) + assert harness.written() == "" + + def test_no_frame_is_prepared_while_resynchronization_is_owed(self) -> None: + harness = Harness() + harness.bridge.require_resynchronization("test") + assert harness.prepare() is None + + +class TestCommitValidation: + def test_stale_prepared_frame_never_becomes_diff_baseline(self) -> None: + """Named test 13.2: intervening output retires the batch without emitting it.""" + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.clear() + + harness.bridge.note_managed_write() + assert harness.bridge.commit(prepared) is False + + assert harness.written() == "" + assert harness.renderer._last_screen is None + assert harness.bridge.needs_resynchronization is True + + def test_a_resize_between_prepare_and_commit_retires_the_batch(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.size = Size(rows=12, columns=40) + harness.display.reconfigure() + harness.clear() + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" + + def test_an_owner_change_retires_the_batch(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.note_owner_change() + assert harness.bridge.commit(prepared) is False + + def test_a_content_invalidation_does_not_authorize_a_stale_batch(self) -> None: + """Content generation is tracked separately so it cannot mask a real invalidation.""" + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.note_managed_write() + harness.bridge.note_content_change() + assert harness.bridge.commit(prepared) is False + + def test_a_content_invalidation_alone_does_not_retire_a_batch(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.note_content_change() + assert harness.bridge.commit(prepared) is True + + def test_the_committed_frame_is_the_next_diff_baseline(self) -> None: + harness = Harness() + harness.render() + assert harness.renderer._last_screen is not None + assert harness.bridge.needs_resynchronization is False + + def test_uncommitted_frame_metadata_is_not_dispatched(self) -> None: + """Named test 13.2: provisional handlers and windows stay invisible until commit.""" + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.note_managed_write(prompt_anchor=1) + assert harness.bridge.can_dispatch_input is False + assert harness.bridge.commit(prepared) is False + assert harness.bridge.can_dispatch_input is False + harness.resynchronize() + assert harness.bridge.can_dispatch_input is True + + +class TestPartialCommitFailure: + def test_partial_commit_failure_does_not_replay_frame(self) -> None: + """Named test 13.2: some bytes are already out; replaying would duplicate them.""" + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.clear() + + failures = {"count": 0} + real_write = harness.backend.write_raw + + def failing_write_raw(data: str) -> None: + failures["count"] += 1 + if failures["count"] == 3: + raise OSError("terminal went away") + real_write(data) + + harness.backend.write_raw = failing_write_raw # type: ignore[method-assign] + assert harness.bridge.commit(prepared) is False + harness.backend.write_raw = real_write # type: ignore[method-assign] + + assert isinstance(harness.bridge.take_pending_error(), OSError) + assert harness.bridge.needs_resynchronization is True + assert harness.renderer._last_screen is None + emitted = harness.clear() + + # The batch is not replayed: a retry would duplicate whatever already reached the + # terminal, and nothing here knows how much that was. + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" + assert emitted != "" + + def test_failed_cleanup_stops_reserved_emission(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + + def always_fails(data: str) -> None: + raise OSError("terminal went away") + + harness.backend.write_raw = always_fails # type: ignore[method-assign] + assert harness.bridge.commit(prepared) is False + assert harness.bridge.reserved_emission_stopped is True + + def test_nothing_is_prepared_once_reserved_emission_has_stopped(self) -> None: + harness = Harness() + harness.bridge.stop_reserved_emission(OSError("terminal went away")) + assert harness.prepare() is None + + +class TestRecovery: + def test_discarded_frame_resynchronizes_terminal_modes(self) -> None: + """Named test 13.2: a latched flag never re-emits its sequence on its own.""" + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + # Preparation advanced the flag even though the terminal saw nothing. + assert harness.renderer._bracketed_paste_enabled is True + harness.bridge.note_managed_write(prompt_anchor=1) + harness.bridge.commit(prepared) + harness.clear() + + harness.resynchronize() + written = harness.written() + assert "\x1b[?2004h" in written # bracketed paste, re-established physically + assert harness.renderer._bracketed_paste_enabled is True + assert harness.renderer._last_screen is None + assert harness.renderer._last_size is None + assert harness.renderer._last_style is None + assert harness.renderer._last_cursor_shape is None + + def test_recovery_restores_the_baseline_only_after_a_full_frame(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.note_managed_write(prompt_anchor=1) + harness.bridge.commit(prepared) + harness.resynchronize() + harness.clear() + assert harness.render() is True + assert "hello" in harness.written() + + def test_discarded_frame_recovers_current_prompt_origin(self) -> None: + """Named test 13.2: intervening output scrolled the prompt; the stale cursor is wrong.""" + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.note_managed_write() + harness.bridge.set_prompt_anchor(9) + harness.bridge.commit(prepared) + harness.clear() + + harness.resynchronize() + assert "\x1b[9;1H" in harness.written() + assert harness.renderer._cursor_pos == Point(x=0, y=0) + + def test_recovery_without_a_known_origin_fails_explicitly(self) -> None: + """Guessing an origin would paint the prompt over committed output.""" + harness = Harness() + harness.bridge.forget_prompt_anchor() + harness.backend.enable_cpr = False + with pytest.raises(ReservedModeFailureError), set_app(harness.app): + harness.bridge.resynchronize() + + def test_recovery_invalidates_provisional_mouse_metadata(self) -> None: + harness = Harness() + harness.render() + handlers = harness.renderer.mouse_handlers + harness.resynchronize() + assert harness.renderer.mouse_handlers is not handlers + + def test_recovery_reestablishes_available_height_from_the_origin(self) -> None: + """A visible marker does not prove prompt geometry; height has to be re-established. + + Zeroing it would leave ``height_is_known`` false while input was allowed to resume, + which is the same invalid state recovery exists to leave behind. + """ + harness = Harness(rows=24, reserved_rows=1) + harness.bridge.set_prompt_anchor(6) + harness.renderer._min_available_height = 17 + harness.resynchronize() + assert harness.renderer._min_available_height == 23 - 6 + 1 + assert harness.renderer.height_is_known is True + + def test_recovery_does_not_use_the_upstream_reset(self) -> None: + """Upstream reset() emits operations and rewrites available-height bookkeeping.""" + harness = Harness() + calls: list[int] = [] + harness.renderer.reset = lambda *args, **kwargs: calls.append(1) # type: ignore[method-assign] + harness.resynchronize() + assert calls == [] + + def test_the_narrow_reset_names_fields_this_prompt_toolkit_has(self) -> None: + """A version contract: a renamed upstream field must fail here, not silently do nothing.""" + harness = Harness() + for name in ( + "_last_screen", + "_last_size", + "_last_style", + "_last_cursor_shape", + "_cursor_pos", + "_min_available_height", + "_bracketed_paste_enabled", + "_mouse_support_enabled", + "_cursor_key_mode_reset", + "mouse_handlers", + ): + assert hasattr(harness.renderer, name), name + + +class TestInvalidationCoalescing: + def test_repeated_invalidation_yields_to_managed_output(self) -> None: + """Named test 13.2: contention coalesces into one redraw and does not spin.""" + harness = Harness() + scheduled: list[int] = [] + harness.bridge.set_redraw_scheduler(lambda: scheduled.append(1)) + + for _ in range(5): + harness.bridge.note_managed_write(prompt_anchor=1) + + assert len(scheduled) == 1 + assert harness.bridge.redraw_pending is True + + harness.resynchronize() + assert harness.render() is True + assert harness.bridge.redraw_pending is False + + def test_a_redraw_is_requested_again_after_it_is_served(self) -> None: + harness = Harness() + scheduled: list[int] = [] + harness.bridge.set_redraw_scheduler(lambda: scheduled.append(1)) + harness.bridge.note_managed_write(prompt_anchor=1) + harness.resynchronize() + harness.render() + harness.bridge.note_managed_write(prompt_anchor=1) + assert len(scheduled) == 2 + + +class TestCursorPositionReports: + def test_cpr_uses_row_one_coordinate_contract(self) -> None: + """A valid row yields exactly U - r + 1 because the region is anchored at row one.""" + harness = Harness(rows=24, reserved_rows=1) + assert harness.bridge.request_cursor_position() is True + assert harness.bridge.report_cursor_row(4) is True + assert harness.renderer._min_available_height == 23 - 4 + 1 + + def test_cpr_in_reserved_band_is_rejected(self) -> None: + """Named test 13.1: the upstream formula gives zero at U+1 and negative below it.""" + harness = Harness(rows=24, reserved_rows=1) + harness.bridge.request_cursor_position() + assert 23 - 24 + 1 == 0 + assert harness.bridge.report_cursor_row(24) is False + assert harness.renderer._min_available_height == 0 + assert harness.bridge.needs_resynchronization is True + + deeper = Harness(rows=24, reserved_rows=2) + deeper.bridge.request_cursor_position() + assert 22 - 24 + 1 == -1 + assert deeper.bridge.report_cursor_row(24) is False + assert deeper.renderer._min_available_height == 0 + + def test_a_rejected_reply_settles_the_pending_request(self) -> None: + """A stuck future would leave the renderer waiting for a report that never comes.""" + harness = Harness() + harness.bridge.request_cursor_position() + harness.bridge.report_cursor_row(24) + assert harness.renderer.waiting_for_cpr is False + + def test_an_uncorrelated_late_reply_is_dropped(self) -> None: + harness = Harness() + assert harness.bridge.report_cursor_row(4) is False + assert harness.renderer._min_available_height == 0 + + def test_a_stale_generation_reply_cannot_satisfy_a_newer_request(self) -> None: + """Named test 13.2: replies correlate by order, and the wire carries no generation.""" + harness = Harness() + harness.bridge.request_cursor_position() + harness.size = Size(rows=12, columns=40) + harness.display.reconfigure() + harness.bridge.note_geometry_change() + harness.bridge.request_cursor_position() + + # The first reply belongs to the request made before the resize. + assert harness.bridge.report_cursor_row(4) is False + assert harness.renderer._min_available_height == 0 + # The second one is the current generation's and is accepted. + assert harness.bridge.report_cursor_row(4) is True + assert harness.renderer._min_available_height == 11 - 4 + 1 + + def test_cpr_request_cannot_interleave_with_paint(self) -> None: + """The request is emitted in its own transaction, never inside another one.""" + harness = Harness() + seen: list[object] = [] + real_ask = harness.backend.ask_for_cpr + + def watched_ask() -> None: + seen.append(current_transaction()) + real_ask() + + harness.backend.ask_for_cpr = watched_ask # type: ignore[method-assign] + harness.bridge.request_cursor_position() + assert len(seen) == 1 + state = seen[0] + assert state is not None + assert state.kind == "cursor position request" + + def test_no_request_is_made_when_the_backend_does_not_answer(self) -> None: + harness = Harness() + harness.backend.enable_cpr = False + assert harness.bridge.request_cursor_position() is False + + +class TestRecoveryEdges: + def test_recovery_after_reserved_emission_stopped_is_refused(self) -> None: + harness = Harness() + harness.bridge.stop_reserved_emission(OSError("terminal went away")) + with pytest.raises(ReservedModeFailureError), set_app(harness.app): + harness.bridge.resynchronize() + + def test_an_unknown_origin_asks_the_terminal_and_stays_owed(self) -> None: + """The reply establishes the origin; recovery is not finished until it arrives.""" + harness = Harness() + harness.bridge.require_resynchronization("test") + harness.bridge.forget_prompt_anchor() + harness.clear() + harness.resynchronize() + assert "\x1b[6n" in harness.written() + assert harness.bridge.needs_resynchronization is True + + def test_a_cursor_report_establishes_the_prompt_origin(self) -> None: + harness = Harness() + harness.bridge.forget_prompt_anchor() + harness.bridge.request_cursor_position() + assert harness.bridge.report_cursor_row(6) is True + assert harness.bridge.prompt_anchor == 6 + + def test_mouse_support_is_established_when_the_application_wants_it(self) -> None: + harness = Harness() + harness.renderer.mouse_support = lambda: True + harness.clear() + harness.resynchronize() + assert "\x1b[?1000h" in harness.written() + assert harness.renderer._mouse_support_enabled is True + + def test_a_rejected_reply_resolves_a_renderer_future(self) -> None: + """Whoever asked is waiting; a rejected reply still has to settle that bookkeeping.""" + harness = Harness() + pending: Future[None] = Future() + harness.renderer._waiting_for_cpr_futures.append(pending) + harness.bridge.request_cursor_position() + assert harness.bridge.report_cursor_row(24) is False + assert pending.done() is True + + +class TestBookkeeping: + def test_content_invalidations_are_counted(self) -> None: + harness = Harness() + assert harness.bridge.content_generation == 0 + harness.bridge.note_content_change() + assert harness.bridge.content_generation == 1 + + def test_the_prompt_anchor_is_reported(self) -> None: + harness = Harness() + assert harness.bridge.prompt_anchor == 1 + harness.bridge.forget_prompt_anchor() + assert harness.bridge.prompt_anchor is None + + def test_only_one_frame_is_in_flight_at_a_time(self) -> None: + """Two provisional frames would mean two claims on the renderer's state.""" + harness = Harness() + assert harness.prepare() is not None + assert harness.prepare() is None + + +class TestReviewRegressions: + def test_a_batch_prepared_against_a_stale_size_is_not_committed(self, monkeypatch: Any) -> None: + """Review finding 1: facts and generations must describe the same terminal.""" + harness = Harness(rows=24) + capture = PreflightFacts.capture + + def capture_then_resize(output: Any) -> PreflightFacts: + facts = capture(output) + harness.resize(12) + return facts + + monkeypatch.setattr(PreflightFacts, "capture", staticmethod(capture_then_resize)) + prepared = harness.prepare() + monkeypatch.undo() + assert prepared is not None + assert prepared.batch.facts.size == Size(rows=23, columns=40) + harness.clear() + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" + + def test_a_frame_is_not_committed_while_recovery_is_owed(self) -> None: + """Review finding 1: an invalidated preparation must not become committable. + + Owing a recovery retires the frame in flight, and that retirement is the single + mechanism commit checks -- so this asserts it happened, not only that the commit was + refused, since a refusal for some other reason would prove nothing. + """ + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.require_resynchronization("something invalidated the terminal") + assert harness.bridge.in_flight is None + harness.clear() + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" + + def test_abandoning_reserved_emission_retires_the_frame_in_flight(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.stop_reserved_emission(OSError("terminal went away")) + assert harness.bridge.in_flight is None + assert harness.bridge.commit(prepared) is False + + def test_input_cannot_be_dispatched_while_a_frame_is_being_prepared(self) -> None: + """Review finding 2: the renderer's state is provisional from the first callback on.""" + seen: list[bool] = [] + harness = Harness(content=lambda: seen.append(harness.bridge.can_dispatch_input) or "hello") + harness.prepare() + assert seen + assert not any(seen) + + def test_a_recursive_preparation_is_refused(self) -> None: + """Review finding 2: two provisional frames would both claim the renderer's state.""" + recursive: list[Any] = [] + + def content() -> str: + with set_app(harness.app): + recursive.append(harness.bridge.prepare(harness.app)) + return "hello" + + harness = Harness(content=content) + assert harness.prepare() is not None + assert recursive == [None] + + def test_managed_output_between_frames_invalidates_the_baseline(self) -> None: + """Review finding 3: the committed cursor relationship does not survive a write.""" + harness = Harness() + assert harness.render() is True + harness.bridge.note_managed_write() + assert harness.bridge.needs_resynchronization is True + assert harness.prepare() is None + + def test_a_managed_write_can_supply_the_new_prompt_origin(self) -> None: + """The layer that emitted the output is the one that knows where it ended.""" + harness = Harness() + harness.render() + harness.bridge.note_managed_write(prompt_anchor=7) + assert harness.bridge.prompt_anchor == 7 + harness.clear() + harness.resynchronize() + assert "\x1b[7;1H" in harness.written() + + def test_recovery_refuses_an_anchor_outside_the_usable_region(self) -> None: + """Review finding 5: a shrunken terminal makes a remembered row point into the band.""" + harness = Harness(rows=24) + harness.bridge.set_prompt_anchor(20) + harness.resize(12) + harness.bridge.note_geometry_change() + harness.clear() + harness.resynchronize() + assert "\x1b[20;1H" not in harness.written() + assert harness.bridge.prompt_anchor is None + assert harness.bridge.needs_resynchronization is True + + def test_an_out_of_range_anchor_falls_back_when_the_terminal_cannot_report(self) -> None: + harness = Harness(rows=24) + harness.bridge.set_prompt_anchor(20) + harness.resize(12) + harness.backend.enable_cpr = False + with pytest.raises(ReservedModeFailureError), set_app(harness.app): + harness.bridge.resynchronize() + + def test_the_resynchronization_reason_is_reported(self) -> None: + harness = Harness() + assert harness.bridge.resynchronization_reason is None + harness.bridge.require_resynchronization("a command wrote to the terminal") + assert harness.bridge.resynchronization_reason == "a command wrote to the terminal" + harness.resynchronize() + assert harness.bridge.resynchronization_reason is None + + def test_a_write_from_inside_a_layout_callback_retires_the_frame(self) -> None: + """The operations were recorded against a terminal that moved on mid-render.""" + + def content() -> str: + harness.bridge.note_managed_write() + return "hello" + + harness = Harness(content=content) + assert harness.prepare() is None + assert harness.bridge.needs_resynchronization is True + assert harness.bridge.can_dispatch_input is False + + def test_a_cursor_report_is_validated_against_the_screen_when_released(self) -> None: + """With no reservation the whole screen is usable, and the band no longer exists.""" + harness = Harness(rows=24) + harness.display.release() + harness.bridge.request_cursor_position() + assert harness.bridge.report_cursor_row(24) is True + assert harness.renderer._min_available_height == 24 - 24 + 1 + + +class RetiringLock: + """A lock that runs scheduled callbacks at the moments it is handed over. + + This stands in for another writer changing the terminal while a caller waits for it. + Driving that with two real threads cannot say *where* the waiting thread had got to before + the lock was released -- the interleaving these tests are about is the one where it is + already past its own checks -- so the handover itself is the seam to inject at. + + Callbacks are scheduled per acquisition, in order, so an operation that takes the terminal + more than once can be interrupted at the handover that matters. ``None`` skips one. + """ + + def __init__(self) -> None: + self._lock = threading.RLock() + self._schedule: deque[Any] = deque() + + def schedule(self, *callbacks: Any) -> None: + """Queue one callback per upcoming acquisition.""" + self._schedule.extend(callbacks) + + def acquire(self, *args: Any, **kwargs: Any) -> bool: + acquired = self._lock.acquire(*args, **kwargs) + if self._schedule: + callback = self._schedule.popleft() + if callback is not None: + callback() + return acquired + + def release(self) -> None: + self._lock.release() + + +class TestReviewRegressionsRoundTwo: + def test_a_managed_write_without_an_origin_forgets_the_old_one(self) -> None: + """Review finding 1: the output moved the cursor, so the remembered row is stale.""" + harness = Harness() + harness.render() + assert harness.bridge.prompt_anchor == 1 + harness.bridge.note_managed_write() + assert harness.bridge.prompt_anchor is None + + harness.clear() + harness.resynchronize() + written = harness.written() + assert "\x1b[1;1H" not in written + assert "\x1b[6n" in written + assert harness.bridge.needs_resynchronization is True + + def test_a_frame_prepared_after_emission_stopped_is_not_published(self) -> None: + """Review finding 2: stopping is not the same state as owing a recovery.""" + + def content() -> str: + harness.bridge.stop_reserved_emission(OSError("terminal went away")) + return "hello" + + harness = Harness(content=content) + assert harness.prepare() is None + assert harness.bridge.in_flight is None + assert harness.bridge.reserved_emission_stopped is True + + def test_a_frame_retired_while_the_commit_waits_is_not_emitted(self) -> None: + """Review finding 3: retirement only replaces an explicit guard if read under the lock.""" + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + prepared = harness.prepare() + assert prepared is not None + harness.clear() + + handover.schedule(lambda: harness.bridge.require_resynchronization("another writer")) + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" + assert harness.bridge.needs_resynchronization is True + + def test_recovery_uses_the_origin_it_finds_after_taking_the_terminal(self) -> None: + """Review finding: an origin read before the wait describes a terminal someone held.""" + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + harness.bridge.set_prompt_anchor(1) + harness.clear() + + handover.schedule(lambda: harness.bridge.note_managed_write(prompt_anchor=7)) + harness.resynchronize() + + written = harness.written() + assert "\x1b[7;1H" in written + assert "\x1b[1;1H" not in written + assert harness.renderer._min_available_height == 23 - 7 + 1 + assert harness.bridge.needs_resynchronization is False + + def test_recovery_completes_before_the_terminal_is_released(self) -> None: + """Whoever takes the terminal next must not find a recovery still owed.""" + seen: list[bool] = [] + harness = Harness() + original = harness.bridge._initialize_renderer + + def watched(policy: Any, origin: int) -> None: + original(policy, origin) + seen.append(harness.bridge.needs_resynchronization) + + harness.bridge._initialize_renderer = watched # type: ignore[method-assign] + harness.bridge.require_resynchronization("test") + harness.resynchronize() + assert seen == [False] + + def test_a_cursor_report_invalidated_by_managed_output_is_rejected(self) -> None: + """Review finding: the write moved the cursor the terminal was sampling.""" + harness = Harness() + harness.bridge.request_cursor_position() + harness.bridge.note_managed_write(prompt_anchor=7) + assert harness.bridge.report_cursor_row(4) is False + assert harness.bridge.prompt_anchor == 7 + assert harness.renderer._min_available_height == 0 + + def test_replies_still_correlate_by_order_after_one_is_invalidated(self) -> None: + """Rejecting a reply must not desynchronize the queue behind it.""" + harness = Harness() + harness.bridge.request_cursor_position() + harness.bridge.note_managed_write(prompt_anchor=7) + harness.bridge.request_cursor_position() + + assert harness.bridge.report_cursor_row(4) is False + assert harness.bridge.report_cursor_row(5) is True + assert harness.bridge.prompt_anchor == 5 + assert harness.renderer._min_available_height == 23 - 5 + 1 + + def test_a_reply_cannot_overtake_a_write_that_lands_while_it_waits(self) -> None: + """Review finding: validating a reply and publishing its origin must be one step.""" + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + harness.bridge.request_cursor_position() + + handover.schedule(lambda: harness.bridge.note_managed_write(prompt_anchor=7)) + assert harness.bridge.report_cursor_row(4) is False + assert harness.bridge.prompt_anchor == 7 + + def test_a_request_records_the_terminal_it_was_actually_sent_to(self) -> None: + """A write during acquisition must not make the reply to a later request look stale.""" + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + + handover.schedule(lambda: harness.bridge.note_managed_write(prompt_anchor=7)) + assert harness.bridge.request_cursor_position() is True + # The request went out after that write, so its reply describes the current terminal. + assert harness.bridge.report_cursor_row(4) is True + assert harness.bridge.prompt_anchor == 4 + + def test_recovery_that_finds_emission_stopped_writes_nothing(self) -> None: + """Review finding: rendering can be abandoned while recovery queues for the terminal.""" + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + harness.bridge.require_resynchronization("test") + harness.clear() + + handover.schedule(lambda: harness.bridge.stop_reserved_emission(OSError("terminal went away"))) + with pytest.raises(ReservedModeFailureError): + harness.resynchronize() + assert harness.written() == "" + assert harness.bridge.needs_resynchronization is True + + def test_an_unknown_origin_request_that_finds_emission_stopped_writes_nothing(self) -> None: + """Review finding: recovery's second transaction is a second chance to be abandoned. + + Recovery without an anchor releases the terminal and asks for it again to send the + cursor request. Emission can be given up in between, and the request would otherwise + write into a terminal nothing may emit to any more. + """ + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + harness.bridge.forget_prompt_anchor() + harness.bridge.require_resynchronization("test") + harness.clear() + + # Skip recovery's own transaction; abandon emission as the request takes the terminal. + handover.schedule(None, lambda: harness.bridge.stop_reserved_emission(OSError("terminal went away"))) + harness.resynchronize() + + assert harness.bridge.reserved_emission_stopped is True + assert harness.written() == "" + # Nothing was queued either: a reply now would be answering a request never made. + assert harness.bridge.report_cursor_row(4) is False + + def test_a_frame_is_not_published_over_an_abandonment(self) -> None: + """Review finding: checking and publishing must be one step, or one overwrites the other.""" + handover = RetiringLock() + harness = Harness(lock=TerminalLock(lock=handover)) + # Skip the preflight acquisition; abandon emission as the publication takes the lock. + handover.schedule(None, lambda: harness.bridge.stop_reserved_emission(OSError("terminal went away"))) + + assert harness.prepare() is None + assert harness.bridge.in_flight is None + + def test_a_frame_in_flight_while_emission_is_abandoned_is_not_committed(self) -> None: + """Retirement alone cannot carry this: a publication can overwrite a retirement. + + The state is built directly because that is the point -- commit must reject a frame + that is in flight while emission is abandoned, whatever sequence produced that pair, + rather than trusting that nothing can produce it. + """ + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.stop_reserved_emission(OSError("terminal went away")) + harness.bridge._in_flight = prepared + harness.clear() + + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" + + def test_a_frame_in_flight_while_recovery_is_owed_is_not_committed(self) -> None: + harness = Harness() + prepared = harness.prepare() + assert prepared is not None + harness.bridge.require_resynchronization("another writer") + harness.bridge._in_flight = prepared + harness.clear() + + assert harness.bridge.commit(prepared) is False + assert harness.written() == "" diff --git a/tests/test_prompt_toolkit_contracts.py b/tests/test_prompt_toolkit_contracts.py index 3ed7400d6..d81001b81 100644 --- a/tests/test_prompt_toolkit_contracts.py +++ b/tests/test_prompt_toolkit_contracts.py @@ -9,12 +9,21 @@ import inspect import sys +from typing import Any +import prompt_toolkit.renderer import pytest +from prompt_toolkit.application import Application +from prompt_toolkit.application.current import set_app +from prompt_toolkit.input import DummyInput +from prompt_toolkit.layout import Layout, Window +from prompt_toolkit.layout.controls import FormattedTextControl from prompt_toolkit.output import DummyOutput, Output from prompt_toolkit.renderer import Renderer from prompt_toolkit.styles import default_ui_style +from cmd2.output_recorder import Operation, PreflightFacts, RecordingOutput + WINDOWS_ONLY = pytest.mark.skipif(sys.platform != "win32", reason="Windows backend is importable only on Windows") @@ -22,6 +31,15 @@ def make_renderer(output: Output) -> Renderer: return Renderer(default_ui_style(), output) +def make_app() -> "Application[Any]": + """Build a minimal application to render.""" + return Application( + layout=Layout(Window(FormattedTextControl("hello"))), + output=DummyOutput(), + input=DummyInput(), + ) + + class TestCursorPositionArithmetic: """Protects the geometry model: physical CPR rows against a virtual total.""" @@ -152,3 +170,54 @@ def test_legacy_win32_erase_down_is_a_separate_implementation(self) -> None: from prompt_toolkit.output.win32 import Win32Output assert Win32Output.erase_down is not Vt100_Output.erase_down + + +class TestDiscardedFrameRecoveryContract: + """Protects the section 7.2.1 recovery contract against a dependency upgrade. + + Each test proves one of the reasons recovery is what it is: why clearing the diff baseline + is not sufficient, and why upstream's own ``reset()`` is not an acceptable substitute for + the narrow initialization the bridge performs. + """ + + def test_render_latches_mode_flags_beside_their_emission(self) -> None: + """A discarded frame leaves the flag ahead of the terminal, and nothing re-tests it.""" + facts = PreflightFacts.capture(DummyOutput()) + renderer = make_renderer(DummyOutput()) + app = make_app() + first = RecordingOutput(facts) + renderer.output = first + with set_app(app): + renderer.render(app, app.layout) + assert Operation("enable_bracketed_paste") in first.operations + assert renderer._bracketed_paste_enabled is True + + # Discard that frame's output. A second render emits nothing to re-enable it. + second = RecordingOutput(facts) + renderer.output = second + renderer._last_screen = None + with set_app(app): + renderer.render(app, app.layout) + assert Operation("enable_bracketed_paste") not in second.operations + + def test_the_upstream_reset_emits_operations(self) -> None: + """Which is why recovery cannot simply call it: it writes to a terminal we are mid-fix.""" + renderer = make_renderer(DummyOutput()) + recorder = RecordingOutput(PreflightFacts.capture(DummyOutput())) + renderer.output = recorder + renderer.reset() + assert recorder.operations != () + + def test_the_upstream_reset_rewrites_available_height_bookkeeping(self) -> None: + renderer = make_renderer(DummyOutput()) + renderer.report_absolute_cursor_row(5) + assert renderer._min_available_height > 0 + renderer.reset() + assert renderer._min_available_height == 0 + + def test_a_full_repaint_still_moves_from_the_believed_cursor_position(self) -> None: + """So an uncommitted cursor position sends the repaint to the wrong origin.""" + source = inspect.getsource(prompt_toolkit.renderer._output_screen_diff) + assert "current_pos" in source + assert '"\\r\\n" * (new.y - current_y)' in source or "\\r\\n" in source + assert "_cursor_pos" in inspect.getsource(Renderer.render) diff --git a/tests/test_terminal_transaction.py b/tests/test_terminal_transaction.py new file mode 100644 index 000000000..cc17ef496 --- /dev/null +++ b/tests/test_terminal_transaction.py @@ -0,0 +1,324 @@ +"""Tests for the terminal transaction lock and its wait contract. + +Two of these are the named regressions from design section 13.2 -- +``test_paint_transaction_rejects_blocking_wait`` and +``test_terminal_lock_order_violation_is_detected``. Both are written against sentinels rather +than real blocking primitives, so a guard that fails to fire makes the test *fail* instead of +hanging the suite, and both assert that the sentinel was never entered: raising after the +wait has already begun would be no protection at all. +""" + +import queue +import threading +import time +from concurrent.futures import Future +from typing import Any, Self + +import pytest + +from cmd2.terminal_transaction import ( + HigherLevelLock, + TerminalLock, + TerminalTransactionViolationError, + assert_no_terminal_transaction, + current_transaction, + guarded_call, + held_higher_level_locks, +) + + +class Sentinel: + """A stand-in for a blocking primitive that records being entered instead of blocking.""" + + def __init__(self) -> None: + self.entered = False + + def __call__(self, *args: Any, **kwargs: Any) -> str: + self.entered = True + return "finished" + + +class RecordingLock: + """A lock that records acquisition rather than ever blocking.""" + + def __init__(self) -> None: + self.acquired = 0 + + def acquire(self, *args: Any, **kwargs: Any) -> bool: + self.acquired += 1 + return True + + def release(self) -> None: + pass + + def __enter__(self) -> Self: + self.acquire() + return self + + def __exit__(self, *args: object) -> None: + self.release() + + +def blocking_primitives() -> list[tuple[str, Any]]: + """Build one sentinel-backed call per prohibited blocking primitive. + + Each entry is the operation name and a zero-argument callable that would enter the + primitive if the guard let it through. + """ + event = threading.Event() + event.set() + condition = threading.Condition() + pending: Future[str] = Future() + pending.set_result("done") + work: queue.Queue[str] = queue.Queue() + work.put("item") + thread = threading.Thread(target=lambda: None) + thread.start() + proxy = Sentinel() + + def condition_wait() -> Any: + with condition: + return guarded_call("condition wait", condition.wait, 0.01) + + return [ + ("ui future wait", lambda: guarded_call("ui future wait", pending.result, 0.01)), + ("event wait", lambda: guarded_call("event wait", event.wait, 0.01)), + ("condition wait", condition_wait), + ("queue wait", lambda: guarded_call("queue wait", work.get, True, 0.01)), + ("thread join", lambda: guarded_call("thread join", thread.join, 0.01)), + ("sleep", lambda: guarded_call("sleep", time.sleep, 0.01)), + ("proxy drain", lambda: guarded_call("proxy drain", proxy)), + ("proxy close", lambda: guarded_call("proxy close", proxy)), + ] + + +class TestTransactionState: + def test_no_transaction_is_active_by_default(self) -> None: + assert current_transaction() is None + + def test_a_transaction_reports_its_kind_generation_and_thread(self) -> None: + lock = TerminalLock() + with lock.transaction("paint", generation=7): + state = current_transaction() + assert state is not None + assert state.kind == "paint" + assert state.generation == 7 + assert state.depth == 1 + assert state.thread_id == threading.get_ident() + + def test_the_transaction_is_gone_after_the_block(self) -> None: + lock = TerminalLock() + with lock.transaction("paint"): + pass + assert current_transaction() is None + + def test_the_transaction_is_released_when_the_body_raises(self) -> None: + lock = TerminalLock() + with pytest.raises(ZeroDivisionError), lock.transaction("paint"): + raise ZeroDivisionError + assert current_transaction() is None + + def test_same_thread_nesting_reuses_the_transaction(self) -> None: + """An emission helper called from inside another one shares one transaction.""" + lock = TerminalLock() + with lock.transaction("commit", generation=3): + with lock.transaction("write"): + state = current_transaction() + assert state is not None + assert state.depth == 2 + # The outer transaction keeps its identity; a nested helper does not + # relabel the transaction it joined. + assert state.kind == "commit" + assert state.generation == 3 + outer = current_transaction() + assert outer is not None + assert outer.depth == 1 + + def test_another_thread_sees_no_transaction_of_its_own(self) -> None: + """The guard is per-thread: it must not report one thread's transaction to another.""" + lock = TerminalLock() + seen: list[Any] = [] + with lock.transaction("paint"): + worker = threading.Thread(target=lambda: seen.append(current_transaction())) + worker.start() + worker.join() + assert seen == [None] + + +class TestWaitContract: + @pytest.mark.parametrize("index", range(len(blocking_primitives()))) + def test_paint_transaction_rejects_blocking_wait(self, index: int) -> None: + """Named test 13.2: every prohibited wait is refused before the primitive is entered.""" + name, invoke = blocking_primitives()[index] + lock = TerminalLock() + with lock.transaction("paint"), pytest.raises(TerminalTransactionViolationError, match=name): + invoke() + + @pytest.mark.parametrize("index", range(len(blocking_primitives()))) + def test_the_primitive_is_never_entered(self, index: int) -> None: + """Raising after the wait has started would be no protection: prove it never starts.""" + _name, invoke = blocking_primitives()[index] + entered: list[str] = [] + lock = TerminalLock() + + def watched() -> Any: + entered.append("yes") + return invoke() + + with lock.transaction("paint"), pytest.raises(TerminalTransactionViolationError): + watched() + # ``watched`` itself ran; what must not have happened is the guarded primitive. + assert entered == ["yes"] + + def test_the_same_calls_are_allowed_outside_a_transaction(self) -> None: + """The guard rejects a context, not the operations themselves.""" + for _name, invoke in blocking_primitives(): + invoke() + + def test_bypassing_the_helper_reaches_the_primitive(self) -> None: + """The guard is what detects this; a sentinel proves the test would fail without it.""" + proxy = Sentinel() + lock = TerminalLock() + with lock.transaction("paint"): + proxy() + assert proxy.entered is True + + def test_assert_no_terminal_transaction_names_the_operation(self) -> None: + lock = TerminalLock() + with lock.transaction("paint"), pytest.raises(TerminalTransactionViolationError, match="draining the stdout proxy"): + assert_no_terminal_transaction("draining the stdout proxy") + + def test_assert_no_terminal_transaction_passes_when_released(self) -> None: + assert_no_terminal_transaction("draining the stdout proxy") + + +class TestLockOrder: + def test_terminal_lock_order_violation_is_detected(self) -> None: + """Named test 13.2: both nesting directions are rejected, the correct order is not.""" + underlying = RecordingLock() + terminal = TerminalLock(lock=underlying) + + # Higher-level lock held, then the terminal lock: refused before the blocking acquire. + with ( + HigherLevelLock("routing", lock=RecordingLock()), + pytest.raises(TerminalTransactionViolationError, match="routing"), + terminal.transaction("paint"), + ): + pass + assert underlying.acquired == 0 + + # Terminal lock held, then the higher-level lock: refused the same way. + routing = RecordingLock() + with ( + terminal.transaction("paint"), + pytest.raises(TerminalTransactionViolationError, match="routing"), + HigherLevelLock("routing", lock=routing), + ): + pass + assert routing.acquired == 0 + + # Release then acquire is the supported order and is not obstructed. + with HigherLevelLock("routing", lock=routing): + pass + with terminal.transaction("paint"): + pass + assert underlying.acquired == 2 + assert routing.acquired == 1 + + def test_a_higher_level_lock_is_released_when_the_body_raises(self) -> None: + routing = HigherLevelLock("routing", lock=RecordingLock()) + with pytest.raises(ZeroDivisionError), routing: + raise ZeroDivisionError + terminal = TerminalLock(lock=RecordingLock()) + with terminal.transaction("paint"): + pass + + def test_nested_higher_level_locks_are_all_reported(self) -> None: + terminal = TerminalLock(lock=RecordingLock()) + with ( + HigherLevelLock("lifecycle", lock=RecordingLock()), + HigherLevelLock("routing", lock=RecordingLock()), + pytest.raises(TerminalTransactionViolationError) as info, + terminal.transaction("paint"), + ): + pass + assert "lifecycle" in str(info.value) + assert "routing" in str(info.value) + + def test_a_higher_level_lock_may_be_taken_after_the_transaction_ends(self) -> None: + terminal = TerminalLock(lock=RecordingLock()) + routing = RecordingLock() + with terminal.transaction("paint"): + pass + with HigherLevelLock("routing", lock=routing): + pass + assert routing.acquired == 1 + + def test_another_thread_may_hold_a_higher_level_lock(self) -> None: + """Lock order is a per-thread rule; another thread's routing lock is not our problem.""" + terminal = TerminalLock(lock=RecordingLock()) + started = threading.Event() + finished = threading.Event() + + def worker() -> None: + with HigherLevelLock("routing", lock=RecordingLock()): + started.set() + finished.wait(timeout=5) + + thread = threading.Thread(target=worker) + thread.start() + started.wait(timeout=5) + try: + with terminal.transaction("paint"): + pass + finally: + finished.set() + thread.join(timeout=5) + + +class TestSerialization: + def test_two_threads_never_hold_the_terminal_at_once(self) -> None: + """The lock is what serializes emission; without it the two bodies meet. + + The overlap is detected with a barrier rather than a sleep. A sleep would only make + an overlap *likely* to be observed; a barrier that both threads must reach inside the + transaction can only be satisfied if they are genuinely inside it together. + """ + terminal = TerminalLock() + both_inside = threading.Barrier(2, timeout=0.2) + start = threading.Barrier(2, timeout=5) + overlaps: list[int] = [] + + def emit() -> None: + start.wait() + with terminal.transaction("paint"): + try: + both_inside.wait() + except threading.BrokenBarrierError: + return + overlaps.append(1) + + threads = [threading.Thread(target=emit) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + assert overlaps == [] + + +class TestDiagnostics: + def test_the_guard_reports_held_locks_outermost_first(self) -> None: + assert held_higher_level_locks() == () + with HigherLevelLock("lifecycle", lock=RecordingLock()), HigherLevelLock("routing", lock=RecordingLock()): + assert held_higher_level_locks() == ("lifecycle", "routing") + assert held_higher_level_locks() == () + + def test_a_higher_level_lock_reports_what_it_protects(self) -> None: + assert HigherLevelLock("routing", lock=RecordingLock()).name == "routing" + + def test_the_terminal_lock_reports_whether_this_thread_holds_it(self) -> None: + terminal = TerminalLock(lock=RecordingLock()) + assert terminal.active is False + with terminal.transaction("paint"): + assert terminal.active is True + assert terminal.active is False diff --git a/tests/test_toolbar_painter.py b/tests/test_toolbar_painter.py new file mode 100644 index 000000000..8a7894163 --- /dev/null +++ b/tests/test_toolbar_painter.py @@ -0,0 +1,619 @@ +"""Tests for laying toolbar content out as display cells. + +Cells, not string length: a frame is a grid of what the terminal will show, so that comparing +two frames answers "will the user see a difference" rather than "did the Python string +change". Wide characters occupy two cells, combining characters occupy none of their own, and +a wide character is never split across the right edge. +""" + +import io +import re +import threading + +import pytest +from prompt_toolkit.data_structures import Size +from prompt_toolkit.formatted_text import AnyFormattedText +from prompt_toolkit.output import ColorDepth +from prompt_toolkit.output.vt100 import Vt100_Output +from prompt_toolkit.styles import BaseStyle, DummyStyle, DynamicStyle, Style + +from cmd2.terminal_display import TerminalDisplay +from cmd2.terminal_transaction import TerminalLock, current_transaction, held_higher_level_locks +from cmd2.toolbar_painter import Cell, ToolbarFrame, ToolbarPainter, measure_toolbar_height + + +def text_of(frame: ToolbarFrame, row: int = 0) -> str: + """Join one row's cells into the text a terminal would show.""" + return "".join(cell.char for cell in frame.rows[row]) + + +def styles_of(frame: ToolbarFrame, row: int = 0) -> list[str]: + """List one row's per-cell styles.""" + return [cell.style for cell in frame.rows[row]] + + +class TestShape: + def test_a_frame_is_always_exactly_its_declared_size(self) -> None: + frame = ToolbarFrame.build("hi", width=10, height=2) + assert len(frame.rows) == 2 + assert all(len(row) == 10 for row in frame.rows) + + def test_short_content_is_padded_with_default_style_spaces(self) -> None: + """The pad is what overwrites a longer previous frame; it is part of the content.""" + frame = ToolbarFrame.build([("class:toolbar", "hi")], width=5, height=1, default_style="class:toolbar") + assert text_of(frame) == "hi " + assert styles_of(frame) == ["class:toolbar"] * 5 + + def test_content_wider_than_the_terminal_wraps(self) -> None: + frame = ToolbarFrame.build("abcdef", width=3, height=2) + assert text_of(frame, 0) == "abc" + assert text_of(frame, 1) == "def" + + def test_content_taller_than_the_band_is_truncated(self) -> None: + """Growing past the band is a geometry transition, never a write outside it.""" + frame = ToolbarFrame.build("one\ntwo\nthree", width=10, height=2) + assert text_of(frame, 0) == "one " + assert text_of(frame, 1) == "two " + + def test_an_explicit_newline_starts_a_row(self) -> None: + frame = ToolbarFrame.build("a\nb", width=3, height=2) + assert text_of(frame, 0) == "a " + assert text_of(frame, 1) == "b " + + def test_empty_content_is_a_blank_frame(self) -> None: + frame = ToolbarFrame.build("", width=4, height=1) + assert text_of(frame) == " " + + def test_frames_with_the_same_cells_are_equal(self) -> None: + assert ToolbarFrame.build("hi", width=4, height=1) == ToolbarFrame.build("hi", width=4, height=1) + + def test_a_style_only_change_is_not_equal(self) -> None: + """Same text, different attributes, is a visible change and must not compare equal.""" + plain = ToolbarFrame.build([("", "hi")], width=4, height=1) + bold = ToolbarFrame.build([("bold", "hi")], width=4, height=1) + assert plain != bold + + +class TestStyles: + def test_each_fragment_styles_its_own_cells(self) -> None: + frame = ToolbarFrame.build([("bold", "ab"), ("italic", "c")], width=4, height=1, default_style="base") + assert styles_of(frame) == ["bold", "bold", "italic", "base"] + + def test_a_zero_width_escape_fragment_is_dropped(self) -> None: + """Passing arbitrary cursor control through the painter would move the real cursor.""" + content = [("", "a"), ("[ZeroWidthEscape]", "\x1b[6n"), ("", "b")] + frame = ToolbarFrame.build(content, width=4, height=1) + assert text_of(frame) == "ab " + + def test_a_mouse_handler_fragment_keeps_its_text_and_style(self) -> None: + """Handlers are not supported in the band; the visible part still renders.""" + content = [("bold", "click", lambda event: None)] + frame = ToolbarFrame.build(content, width=6, height=1) + assert text_of(frame) == "click " + assert styles_of(frame)[0] == "bold" + + +class TestCellWidths: + def test_a_wide_character_occupies_two_cells(self) -> None: + frame = ToolbarFrame.build("广", width=4, height=1) + assert frame.rows[0][0].char == "广" + assert frame.rows[0][1].char == "" + assert frame.rows[0][1].is_continuation is True + assert text_of(frame) == "广 " + + def test_a_wide_character_is_never_split_at_the_right_edge(self) -> None: + """Half a wide character at the edge is what wraps a row into the one below it.""" + frame = ToolbarFrame.build("a广", width=2, height=2) + assert text_of(frame, 0) == "a " + assert frame.rows[1][0].char == "广" + + def test_the_pad_before_a_wrapped_wide_character_uses_the_default_style(self) -> None: + frame = ToolbarFrame.build([("bold", "a广")], width=2, height=2, default_style="base") + assert styles_of(frame, 0) == ["bold", "base"] + + def test_a_combining_character_joins_the_cell_before_it(self) -> None: + frame = ToolbarFrame.build("e\u0301x", width=4, height=1) + assert frame.rows[0][0].char == "e\u0301" + assert frame.rows[0][1].char == "x" + + def test_a_leading_combining_character_gets_its_own_cell(self) -> None: + """There is nothing to combine with; dropping it would silently lose content.""" + frame = ToolbarFrame.build("\u0301a", width=4, height=1) + assert frame.rows[0][0].char == "\u0301" + assert frame.rows[0][1].char == "a" + + def test_a_tab_advances_to_the_next_tab_stop(self) -> None: + frame = ToolbarFrame.build("a\tb", width=12, height=1) + assert text_of(frame) == "a b " + + def test_a_carriage_return_does_not_reach_the_terminal(self) -> None: + """A stray CR would move the cursor within the band rather than print.""" + frame = ToolbarFrame.build("a\rb", width=4, height=1) + assert text_of(frame) == "ab " + + +class TestMeasurement: + def test_a_short_toolbar_is_one_row(self) -> None: + assert measure_toolbar_height("hi", width=10) == 1 + + def test_wrapping_is_counted(self) -> None: + assert measure_toolbar_height("abcdef", width=3) == 2 + + def test_newlines_are_counted(self) -> None: + assert measure_toolbar_height("a\nb\nc", width=10) == 3 + + def test_empty_content_still_measures_one_row(self) -> None: + """An empty toolbar is an intentional visibility change, not a zero-row reservation.""" + assert measure_toolbar_height("", width=10) == 1 + + def test_measurement_matches_the_frame_it_would_build(self) -> None: + content = [("bold", "wide 广 content that wraps around")] + height = measure_toolbar_height(content, width=12) + frame = ToolbarFrame.build(content, width=12, height=height) + # Nothing was truncated: the last row is where the content ended. + assert measure_toolbar_height(content, width=12) == len(frame.rows) + assert text_of(frame, height - 1).strip() != "" + + +class TestValidation: + def test_a_frame_needs_a_positive_width(self) -> None: + with pytest.raises(ValueError, match="width"): + ToolbarFrame.build("hi", width=0, height=1) + + def test_a_frame_needs_a_positive_height(self) -> None: + with pytest.raises(ValueError, match="height"): + ToolbarFrame.build("hi", width=4, height=0) + + def test_a_cell_reports_its_display_width(self) -> None: + assert Cell("a", "").width == 1 + assert Cell("广", "").width == 2 + assert Cell("", "", is_continuation=True).width == 0 + + def test_measuring_needs_a_positive_width(self) -> None: + with pytest.raises(ValueError, match="width"): + measure_toolbar_height("hi", width=0) + + def test_a_frame_reports_its_own_size(self) -> None: + frame = ToolbarFrame.build("hi", width=6, height=2) + assert (frame.height, frame.width) == (2, 6) + + def test_an_empty_frame_reports_zero_width(self) -> None: + assert ToolbarFrame(rows=()).width == 0 + + def test_a_tab_wraps_when_it_reaches_the_edge(self) -> None: + """The expansion is cells, so it wraps like any other run of them.""" + frame = ToolbarFrame.build("a\tb", width=4, height=3) + assert text_of(frame, 0) == "a " + assert text_of(frame, 1) == " " + assert text_of(frame, 2) == "b " + + def test_a_combining_mark_after_a_wide_character_joins_that_character(self) -> None: + """Attaching it to the continuation cell instead shifts every later column.""" + frame = ToolbarFrame.build("广́x", width=6, height=1) + assert frame.rows[0][0].char == "广́" + assert frame.rows[0][1].is_continuation is True + assert frame.rows[0][2].char == "x" + + +class ResizableDisplay(TerminalDisplay): + """A display whose terminal can be resized between preparing and painting.""" + + def __init__(self, output: Vt100_Output, screen: dict[str, int], reserved_rows: int = 1) -> None: + super().__init__(output, reserved_rows=reserved_rows) + self._screen = screen + + def resize(self, rows: int) -> None: + """Change the terminal's height and re-establish the reservation.""" + self._screen["rows"] = rows + self.reconfigure() + + +class RecordingStream(io.StringIO): + """The terminal end of the backend, recording what reached it and when. + + Deliberately a stream rather than an ``Output`` subclass: backend capability is decided by + exact class identity, so a subclassed backend would never be granted a reservation, and + every test here would pass for the wrong reason. + """ + + def __init__(self) -> None: + super().__init__() + self.flushes = 0 + self.transaction_during_write: list[object] = [] + + def write(self, text: str) -> int: + self.transaction_during_write.append(current_transaction()) + return super().write(text) + + def flush(self) -> None: + self.flushes += 1 + super().flush() + + +class Harness: + """A painter over a real reservation, and the stream the terminal receives.""" + + def __init__( + self, + rows: int = 24, + columns: int = 5, + reserved: int = 1, + style: BaseStyle | None = None, + ) -> None: + self.stream = RecordingStream() + self.screen = {"rows": rows, "columns": columns} + self.output = Vt100_Output(self.stream, lambda: Size(rows=self.screen["rows"], columns=self.screen["columns"])) + self.display = ResizableDisplay(self.output, self.screen, reserved_rows=reserved) + assert self.display.acquire() is True + self.painter = ToolbarPainter( + display=self.display, + lock=TerminalLock(), + style=style or DummyStyle(), + color_depth=ColorDepth.DEPTH_8_BIT, + ) + self.clear() + + def clear(self) -> None: + """Discard everything written so far.""" + self.stream.truncate(0) + self.stream.seek(0) + + def written(self) -> str: + """Everything written since the last clear.""" + return self.stream.getvalue() + + def visible(self) -> str: + """What was written, with attribute changes stripped out.""" + return re.sub(r"\x1b\[[0-9;]*m", "", self.stream.getvalue()) + + def paint(self, content: AnyFormattedText) -> bool: + """Prepare and paint content in one step, as a refresh would.""" + prepared = self.painter.prepare(lambda: content) + assert prepared is not None + return self.painter.paint(prepared) + + +class TestPainting: + def test_the_first_paint_writes_the_whole_band_at_its_physical_row(self) -> None: + harness = Harness() + assert harness.paint("hi") is True + assert "\x1b[24;1H" in harness.visible() + assert "hi " in harness.visible() + + def test_a_multirow_band_writes_each_row_at_its_own_physical_row(self) -> None: + harness = Harness(columns=2, reserved=2) + harness.paint("ab\ncd") + written = harness.visible() + assert "\x1b[23;1Hab" in written + assert "\x1b[24;1Hcd" in written + + def test_an_unchanged_frame_emits_nothing(self) -> None: + """Same cells and attributes: the toolbar produces no output at all.""" + harness = Harness() + harness.paint("hi") + harness.clear() + flushes = harness.stream.flushes + assert harness.paint("hi") is False + assert harness.written() == "" + assert harness.stream.flushes == flushes + + def test_only_the_changed_run_is_rewritten(self) -> None: + harness = Harness() + harness.paint("abcd") + harness.clear() + harness.paint("abXd") + written = harness.visible() + assert "\x1b[24;3HX" in written + assert "abX" not in written + + def test_nothing_is_cleared_before_painting(self) -> None: + """An erase before the write is exactly the flicker this design exists to remove.""" + harness = Harness() + harness.paint("abcd") + harness.paint("z") + written = harness.written() + for erase in ("\x1b[K", "\x1b[0K", "\x1b[2K", "\x1b[J", "\x1b[M"): + assert erase not in written + + def test_a_shorter_frame_pads_its_tail_rather_than_erasing_it(self) -> None: + harness = Harness() + harness.paint("abcd") + harness.clear() + harness.paint("z") + # The final column was already blank in the previous frame, so it is not rewritten: + # the run stops where the difference does. + assert "\x1b[24;1Hz " in harness.visible() + + def test_a_style_only_change_repaints_those_cells(self) -> None: + # A real style, not DummyStyle: under DummyStyle "bold" and "" resolve to the same + # attributes, so the terminal would show the same thing and not painting is correct. + harness = Harness(style=Style.from_dict({})) + harness.paint([("", "hi")]) + harness.clear() + assert harness.paint([("bold", "hi")]) is True + assert "hi" in harness.visible() + + def test_a_wide_character_is_replaced_as_a_whole(self) -> None: + """Both of its cells change together, so a run never begins on the right half.""" + harness = Harness() + harness.paint("a广b") + harness.clear() + harness.paint("aXYb") + assert "\x1b[24;2HXY" in harness.visible() + + def test_replacing_one_wide_character_with_another_repaints_both_cells(self) -> None: + """The two halves compare equal, so the run has to be extended over the second one.""" + harness = Harness() + harness.paint("a广b") + harness.clear() + assert harness.paint("a国b") is True + assert "\x1b[24;2H国" in harness.visible() + + def test_the_cursor_is_saved_and_restored_around_the_paint(self) -> None: + harness = Harness() + harness.paint("hi") + written = harness.written() + assert written.startswith("\x1b7") + assert written.endswith("\x1b8") + + def test_autowrap_is_disabled_during_the_paint_and_restored(self) -> None: + """Writing the last column with autowrap on would push the band into another row.""" + harness = Harness() + harness.paint("hi") + written = harness.written() + assert written.index("\x1b[?7l") < written.index("\x1b[24;1H") + assert written.index("\x1b[?7h") > written.index("\x1b[24;1H") + + def test_the_paint_is_flushed(self) -> None: + harness = Harness() + harness.paint("hi") + assert harness.stream.flushes >= 1 + + def test_every_write_happens_inside_a_terminal_transaction(self) -> None: + harness = Harness() + harness.stream.transaction_during_write.clear() + harness.paint("hi") + assert harness.stream.transaction_during_write + assert all(state is not None for state in harness.stream.transaction_during_write) + + def test_invalidating_forces_a_full_repaint(self) -> None: + """After recovery the terminal's contents are unknown, so the diff baseline is gone.""" + harness = Harness() + harness.paint("hi") + harness.painter.invalidate() + harness.clear() + assert harness.paint("hi") is True + assert "\x1b[24;1Hhi " in harness.visible() + + def test_a_geometry_change_forces_a_full_repaint(self) -> None: + """The band moved; cells matching the old frame are not on the screen any more.""" + harness = Harness() + harness.paint("hi") + harness.display.resize(12) + harness.clear() + assert harness.paint("hi") is True + assert "\x1b[12;1Hhi " in harness.visible() + + 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() + harness.paint("hi") + harness.clear() + assert harness.paint("") is True + # Only the two cells that held text are rewritten; the rest of the band was already + # blank. Blanking by writing spaces is a paint, not an erase. + assert "\x1b[24;1H " in harness.visible() + + +class TestOwnershipValidation: + def test_a_paint_is_refused_when_the_terminal_changed_between_prepare_and_paint(self) -> None: + """The band prepared for row 24 is in the command area once the terminal grows.""" + harness = Harness(rows=24) + prepared = harness.painter.prepare(lambda: "hi") + assert prepared is not None + harness.display.resize(40) + harness.clear() + assert harness.painter.paint(prepared) is False + assert harness.written() == "" + + def test_a_refused_paint_does_not_become_the_baseline(self) -> None: + """Publishing it would make the next diff skip changes the terminal never received.""" + harness = Harness(rows=24) + prepared = harness.painter.prepare(lambda: "hi") + assert prepared is not None + harness.display.resize(40) + harness.painter.paint(prepared) + assert harness.painter.last_frame is None + + def test_a_paint_is_refused_while_the_reservation_is_released(self) -> None: + """With no reservation there is no band to own, and no rows to write into.""" + harness = Harness() + prepared = harness.painter.prepare(lambda: "hi") + assert prepared is not None + harness.display.release() + harness.clear() + assert harness.painter.paint(prepared) is False + assert harness.written() == "" + + def test_nothing_is_prepared_while_the_reservation_is_released(self) -> None: + harness = Harness() + harness.display.release() + assert harness.painter.prepare(lambda: "hi") is None + + +class TestContentEvaluation: + def test_the_callback_runs_outside_the_terminal_transaction(self) -> None: + """Named rule 13.2: a content callback must never run while the terminal is held.""" + harness = Harness() + seen: list[object] = [] + harness.painter.prepare(lambda: seen.append(current_transaction()) or "hi") + assert seen == [None] + + def test_the_callback_runs_once_per_requested_refresh(self) -> None: + harness = Harness() + calls = 0 + + def content() -> str: + nonlocal calls + calls += 1 + return "hi" + + harness.painter.prepare(content) + assert calls == 1 + + def test_a_failing_callback_keeps_the_last_good_frame(self) -> None: + harness = Harness(columns=6) + harness.paint("good") + good = harness.painter.last_frame + + def boom() -> str: + raise RuntimeError("callback failed") + + assert harness.painter.prepare(boom) is None + assert harness.painter.last_frame == good + assert "good" in harness.visible() + + def test_a_failing_callback_is_not_called_again(self) -> None: + """Repeated failing updates would report the same error on every refresh.""" + harness = Harness() + calls = 0 + + def boom() -> str: + nonlocal calls + calls += 1 + raise RuntimeError("callback failed") + + harness.painter.prepare(boom) + harness.painter.prepare(boom) + assert calls == 1 + + def test_the_error_is_reported_once(self) -> None: + harness = Harness() + + def boom() -> str: + raise RuntimeError("callback failed") + + harness.painter.prepare(boom) + assert isinstance(harness.painter.take_pending_error(), RuntimeError) + assert harness.painter.take_pending_error() is None + + def test_taking_the_error_lets_content_be_evaluated_again(self) -> None: + """Reporting is what re-arms it: the user has been told, so a retry is not a loop.""" + harness = Harness(columns=12) + failures = [True] + + def content() -> str: + if failures[0]: + raise RuntimeError("callback failed") + return "recovered" + + harness.painter.prepare(content) + harness.painter.take_pending_error() + failures[0] = False + prepared = harness.painter.prepare(content) + assert prepared is not None + assert "recovered" in "".join(cell.char for cell in prepared.frame.rows[0]) + + +class TestResolvedStyles: + def test_a_resolved_style_change_repaints_the_cells(self) -> None: + """The class string is unchanged, but what the terminal shows is not.""" + rules = {"status": "fg:ansired"} + style = DynamicStyle(lambda: Style.from_dict(dict(rules))) + harness = Harness(style=style) + + content = [("class:status", "hi")] + assert harness.paint(content) is True + harness.clear() + + rules["status"] = "fg:ansiblue" + assert harness.paint(content) is True + assert "hi" in harness.visible() + + def test_an_unchanged_resolved_style_still_emits_nothing(self) -> None: + style = DynamicStyle(lambda: Style.from_dict({"status": "fg:ansired"})) + harness = Harness(style=style) + content = [("class:status", "hi")] + harness.paint(content) + harness.clear() + assert harness.paint(content) is False + assert harness.written() == "" + + +class BlockingStream(io.StringIO): + """A stream whose first write blocks until the test releases it.""" + + def __init__(self) -> None: + super().__init__() + self.armed = False + self.blocked = threading.Event() + self.entered = threading.Event() + self.locks_held_while_blocked: tuple[str, ...] | None = None + + def write(self, text: str) -> int: + if self.armed and not self.entered.is_set(): + self.entered.set() + self.locks_held_while_blocked = held_higher_level_locks() + self.blocked.wait(timeout=5) + return super().write(text) + + +class TestBackpressure: + def test_paint_preserves_transaction_order_with_blocked_sink(self) -> None: + """Named test 13.2: a blocked writer holds the terminal, and nothing slips past it.""" + stream = BlockingStream() + screen = {"rows": 24, "columns": 5} + output = Vt100_Output(stream, lambda: Size(rows=screen["rows"], columns=screen["columns"])) + display = ResizableDisplay(output, screen) + assert display.acquire() is True + lock = TerminalLock() + painter = ToolbarPainter( + display=display, + lock=lock, + style=DummyStyle(), + color_depth=ColorDepth.DEPTH_8_BIT, + ) + order: list[str] = [] + # Arm only now: the reservation's own margin write happens during setup, and blocking + # that would stall the harness rather than the case under test. + stream.truncate(0) + stream.seek(0) + stream.armed = True + + def command_output() -> None: + with lock.transaction("managed write"): + order.append("write start") + output.write("output from a command\n") + output.flush() + order.append("write end") + + ready = threading.Event() + + def toolbar_paint() -> None: + prepared = painter.prepare(lambda: "hi") + assert prepared is not None + ready.set() + painter.paint(prepared) + order.append("paint end") + + writer = threading.Thread(target=command_output) + writer.start() + assert stream.entered.wait(timeout=5) + + painter_thread = threading.Thread(target=toolbar_paint) + painter_thread.start() + assert ready.wait(timeout=5) + # The painter has its frame and is asking for the terminal, but the blocked writer + # holds it, so not one byte of the band can have reached the stream. + assert "\x1b7" not in stream.getvalue() + + stream.blocked.set() + writer.join(timeout=5) + painter_thread.join(timeout=5) + + assert order == ["write start", "write end", "paint end"] + written = stream.getvalue() + assert written.index("output from a command") < written.index("\x1b7") + # The writer blocked inside leaf I/O, holding no higher-level lock -- which is what + # keeps the rest of cmd2 able to make progress while the terminal is backed up. + assert stream.locks_held_while_blocked == ()