mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
feat(cli): add terminal OSC 9;4 progress & escape helper (#3347)
Closes #3119 --- Adds a small `terminal_escape` module that centralizes the "fire-and-forget" pattern the CLI uses for cosmetic terminal control sequences: - `write_terminal_escape(sequence)` — prefers `/dev/tty` so output reaches the terminal even when stdout/stderr are redirected; falls back to `sys.__stderr__` only when it is a TTY; swallows `OSError` so cosmetic writes can never crash the app. - `write_osc(command, payload, *, st=False)` — composes an OSC sequence with BEL or ST termination. - `TerminalProgressState` + `set_terminal_progress` / `clear_terminal_progress` — `OSC 9;4` taskbar-progress helpers with validation (clamps to `[0, 100]` for determinate states, normalizes `INDETERMINATE` and `CLEAR` to `0`). - Opt-out via `DEEPAGENTS_CLI_NO_TERMINAL_ESCAPE=1` for terminals where users prefer to suppress it. Wires indeterminate progress into the existing agent-turn lifecycle in `DeepAgentsApp._set_spinner`: the indicator turns on whenever a spinner is shown, clears when the spinner is hidden. `atexit` cleanup is registered lazily — only after a non-clear emit succeeds — so the helper stays a true no-op on test imports, redirected runs, or quick exits. Unsupported terminals silently ignore unknown OSC sequences (the design constraint the issue calls out), so callers can fire unconditionally without runtime probing. Existing OSC 52 clipboard and iTerm cursor-guide writers are left as-is — the issue lists migrating them as a "consider," not a must — and remain easy follow-ups. _Opened collaboratively by Mason Daugherty and open-swe._ --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> Co-authored-by: Mason Daugherty <61371264+mdrxy@users.noreply.github.com> Co-authored-by: Mason Daugherty <mason@langchain.dev> Co-authored-by: Mason Daugherty <github@mdrxy.com>
This commit is contained in:
open-swe[bot]
@@ -110,6 +110,9 @@ KITTY_KEYBOARD = "DEEPAGENTS_CLI_KITTY_KEYBOARD"
|
||||
LANGSMITH_PROJECT = "DEEPAGENTS_CLI_LANGSMITH_PROJECT"
|
||||
"""Override LangSmith project name for agent traces."""
|
||||
|
||||
NO_TERMINAL_ESCAPE = "DEEPAGENTS_CLI_NO_TERMINAL_ESCAPE"
|
||||
"""Disable all terminal escape/control sequence output when enabled."""
|
||||
|
||||
NO_UPDATE_CHECK = "DEEPAGENTS_CLI_NO_UPDATE_CHECK"
|
||||
"""Disable automatic update checking when set."""
|
||||
|
||||
|
||||
@@ -3293,16 +3293,37 @@ class DeepAgentsApp(App):
|
||||
async def _set_spinner(self, status: SpinnerStatus) -> None:
|
||||
"""Show, update, or hide the loading spinner.
|
||||
|
||||
Also drives the terminal's `OSC 9;4` progress indicator, when
|
||||
supported, so taskbar / dock / tab badges reflect agent activity while
|
||||
the user is in another window.
|
||||
|
||||
Args:
|
||||
status: The spinner status to display, or `None` to hide.
|
||||
"""
|
||||
from deepagents_cli.terminal_escape import (
|
||||
TerminalProgressState,
|
||||
clear_terminal_progress,
|
||||
set_terminal_progress,
|
||||
)
|
||||
|
||||
if status is None:
|
||||
# Hide
|
||||
if self._loading_widget:
|
||||
await self._loading_widget.remove()
|
||||
self._loading_widget = None
|
||||
try:
|
||||
clear_terminal_progress()
|
||||
except Exception:
|
||||
# Cosmetic only — must never break spinner lifecycle.
|
||||
logger.exception("clear_terminal_progress raised unexpectedly")
|
||||
return
|
||||
|
||||
try:
|
||||
set_terminal_progress(state=TerminalProgressState.INDETERMINATE)
|
||||
except Exception:
|
||||
# Cosmetic only — must never break spinner lifecycle.
|
||||
logger.exception("set_terminal_progress raised unexpectedly")
|
||||
|
||||
try:
|
||||
messages = self.query_one("#messages", Container)
|
||||
except NoMatches:
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
"""Best-effort writer for terminal escape/control sequences.
|
||||
|
||||
Centralizes the "fire and forget" pattern the CLI uses for cosmetic terminal
|
||||
control (OSC 9;4 taskbar progress today; eventually OSC 52 clipboard and the
|
||||
iTerm2 cursor guide). Writes prefer `/dev/tty` so output reaches the terminal
|
||||
even when stdout/stderr are redirected, fall back to `sys.__stderr__`, and
|
||||
never raise — cosmetic control output must not crash the app.
|
||||
|
||||
Set `DEEPAGENTS_CLI_NO_TERMINAL_ESCAPE=1` to disable all output (useful for
|
||||
unsupported terminals or noisy logs).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import atexit
|
||||
import logging
|
||||
import pathlib
|
||||
import sys
|
||||
import threading
|
||||
from enum import StrEnum
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from deepagents_cli._env_vars import NO_TERMINAL_ESCAPE, is_env_truthy
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from typing import TextIO
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_PROGRESS_MIN = 0
|
||||
"""Lower clamp bound for determinate `OSC 9;4` progress percentages."""
|
||||
|
||||
_PROGRESS_MAX = 100
|
||||
"""Upper clamp bound for determinate `OSC 9;4` progress percentages."""
|
||||
|
||||
|
||||
class TerminalProgressState(StrEnum):
|
||||
"""`OSC 9;4` progress states.
|
||||
|
||||
See https://learn.microsoft.com/en-us/windows/terminal/tutorials/progress-bar-sequences.
|
||||
"""
|
||||
|
||||
CLEAR = "0"
|
||||
"""Remove any progress indicator. Percentage is ignored."""
|
||||
|
||||
NORMAL = "1"
|
||||
"""Determinate progress shown with the default (success) color."""
|
||||
|
||||
ERROR = "2"
|
||||
"""Determinate progress shown with the error/red color."""
|
||||
|
||||
INDETERMINATE = "3"
|
||||
"""Activity in progress with no known percentage; renders as a pulse."""
|
||||
|
||||
WARNING = "4"
|
||||
"""Determinate progress shown with the warning/yellow color."""
|
||||
|
||||
|
||||
def _is_disabled() -> bool:
|
||||
"""Return whether terminal-escape output is opt-out disabled."""
|
||||
return is_env_truthy(NO_TERMINAL_ESCAPE)
|
||||
|
||||
|
||||
def _open_tty() -> TextIO | None:
|
||||
"""Return an open `/dev/tty` handle, or `None` if unavailable."""
|
||||
try:
|
||||
return pathlib.Path("/dev/tty").open("w", encoding="utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def _is_stream_tty(stream: TextIO | None) -> bool:
|
||||
"""Return whether `stream` is a real TTY."""
|
||||
if stream is None:
|
||||
return False
|
||||
try:
|
||||
return bool(stream.isatty())
|
||||
except (ValueError, OSError):
|
||||
return False
|
||||
|
||||
|
||||
def write_terminal_escape(sequence: str) -> bool:
|
||||
r"""Best-effort write of a terminal control sequence.
|
||||
|
||||
Prefers `/dev/tty` so the sequence reaches the terminal even when stdout
|
||||
or stderr are redirected. Falls back to `sys.__stderr__` only if it is a
|
||||
TTY.
|
||||
|
||||
Returns `False` (no-op) when output is disabled or no TTY is reachable.
|
||||
|
||||
Args:
|
||||
sequence: Raw escape sequence to write, including leading `\x1b`/`ESC`
|
||||
and terminator.
|
||||
|
||||
Returns:
|
||||
`True` if the sequence was written and flushed without error.
|
||||
"""
|
||||
if _is_disabled() or not sequence:
|
||||
return False
|
||||
tty = _open_tty()
|
||||
if tty is not None:
|
||||
try:
|
||||
with tty:
|
||||
tty.write(sequence)
|
||||
tty.flush()
|
||||
except (OSError, UnicodeError) as exc:
|
||||
logger.debug("terminal_escape /dev/tty write failed: %s", exc)
|
||||
else:
|
||||
return True
|
||||
stderr = sys.__stderr__
|
||||
if stderr is not None and _is_stream_tty(stderr):
|
||||
try:
|
||||
stderr.write(sequence)
|
||||
stderr.flush()
|
||||
except (OSError, ValueError) as exc:
|
||||
logger.debug("terminal_escape stderr write failed: %s", exc)
|
||||
return False
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def write_osc(command: str, payload: str = "", *, st: bool = False) -> bool:
|
||||
r"""Write an `OSC <command>;<payload>` sequence.
|
||||
|
||||
Args:
|
||||
command: The numeric OSC command (e.g. ``"9;4"`` for taskbar progress).
|
||||
payload: Optional semicolon-joined payload appended after the command.
|
||||
st: When `True`, terminate with String Terminator (`ESC \`) instead of
|
||||
the default BEL (`\a`).
|
||||
|
||||
BEL matches the Windows Terminal docs and works on most terminals;
|
||||
VTE-derived terminals may prefer ST.
|
||||
|
||||
Returns:
|
||||
`True` if the sequence was written.
|
||||
"""
|
||||
body = f"{command};{payload}" if payload else command
|
||||
terminator = "\x1b\\" if st else "\a"
|
||||
return write_terminal_escape(f"\x1b]{body}{terminator}")
|
||||
|
||||
|
||||
_progress_active = False
|
||||
_atexit_registered = False
|
||||
_atexit_lock = threading.Lock()
|
||||
|
||||
|
||||
def _validate_progress(progress: int | None, state: TerminalProgressState) -> int:
|
||||
"""Clamp/normalize `progress` for a given `state`.
|
||||
|
||||
Determinate states (`NORMAL`, `ERROR`, `WARNING`) clamp to `[0, 100]`;
|
||||
`INDETERMINATE` and `CLEAR` always emit `0`. A non-`None` `progress`
|
||||
supplied with `CLEAR`/`INDETERMINATE` is dropped with a debug log so
|
||||
misuse stays observable without raising on a cosmetic write path. A
|
||||
`progress` that can't be coerced to `int` is treated the same way.
|
||||
|
||||
Args:
|
||||
progress: Raw percentage, or `None`.
|
||||
state: The OSC 9;4 progress state.
|
||||
|
||||
Returns:
|
||||
The normalized progress integer to emit.
|
||||
"""
|
||||
if state in {TerminalProgressState.CLEAR, TerminalProgressState.INDETERMINATE}:
|
||||
if progress is not None and progress != 0:
|
||||
logger.debug(
|
||||
"terminal_progress: ignoring progress=%r for state=%s",
|
||||
progress,
|
||||
state.name,
|
||||
)
|
||||
return 0
|
||||
if progress is None:
|
||||
return 0
|
||||
try:
|
||||
coerced = int(progress)
|
||||
except (TypeError, ValueError) as exc:
|
||||
logger.debug(
|
||||
"terminal_progress: non-numeric progress=%r ignored (%s)", progress, exc
|
||||
)
|
||||
return 0
|
||||
return max(_PROGRESS_MIN, min(_PROGRESS_MAX, coerced))
|
||||
|
||||
|
||||
def set_terminal_progress(
|
||||
progress: int | None = None,
|
||||
*,
|
||||
state: TerminalProgressState = TerminalProgressState.NORMAL,
|
||||
) -> bool:
|
||||
"""Set the terminal's `OSC 9;4` progress indicator.
|
||||
|
||||
Fires unconditionally — terminals that don't recognize `OSC 9;4` silently
|
||||
ignore the sequence. Set `DEEPAGENTS_CLI_NO_TERMINAL_ESCAPE=1` to opt out
|
||||
entirely.
|
||||
|
||||
Args:
|
||||
progress: Percentage `0-100` for determinate states. Ignored for
|
||||
`INDETERMINATE` and `CLEAR`.
|
||||
state: One of `TerminalProgressState`.
|
||||
|
||||
Returns:
|
||||
`True` if the sequence was written.
|
||||
"""
|
||||
global _atexit_registered, _progress_active # noqa: PLW0603
|
||||
|
||||
value = _validate_progress(progress, state)
|
||||
payload = f"{state.value};{value}"
|
||||
written = write_osc("9;4", payload)
|
||||
if written and state is not TerminalProgressState.CLEAR:
|
||||
with _atexit_lock:
|
||||
if not _atexit_registered:
|
||||
atexit.register(_atexit_clear)
|
||||
_atexit_registered = True
|
||||
_progress_active = True
|
||||
elif state is TerminalProgressState.CLEAR:
|
||||
_progress_active = False
|
||||
return written
|
||||
|
||||
|
||||
def clear_terminal_progress() -> bool:
|
||||
"""Clear the terminal's progress indicator.
|
||||
|
||||
Emits `OSC 9;4;0;0`.
|
||||
|
||||
Returns:
|
||||
`True` if the sequence was written.
|
||||
"""
|
||||
return set_terminal_progress(state=TerminalProgressState.CLEAR)
|
||||
|
||||
|
||||
def _atexit_clear() -> None:
|
||||
"""`atexit` hook that clears any leftover progress indicator."""
|
||||
if _progress_active:
|
||||
clear_terminal_progress()
|
||||
@@ -7946,3 +7946,128 @@ class TestExternalBypassFieldHonored:
|
||||
|
||||
# Local import for BypassTier in TestExternalBypassFieldHonored.
|
||||
from deepagents_cli.command_registry import BypassTier # noqa: E402
|
||||
|
||||
|
||||
class TestSetSpinnerTerminalProgress:
|
||||
"""`_set_spinner` should drive the `OSC 9;4` terminal progress indicator."""
|
||||
|
||||
async def test_status_triggers_indeterminate_progress(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""A non-None spinner status should set indeterminate progress."""
|
||||
from deepagents_cli import terminal_escape
|
||||
|
||||
calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
|
||||
|
||||
def _record_set(*args: object, **kwargs: object) -> bool:
|
||||
calls.append(("set", args, dict(kwargs)))
|
||||
return True
|
||||
|
||||
def _record_clear() -> bool:
|
||||
calls.append(("clear", (), {}))
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(terminal_escape, "set_terminal_progress", _record_set)
|
||||
monkeypatch.setattr(terminal_escape, "clear_terminal_progress", _record_clear)
|
||||
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-osc")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
calls.clear()
|
||||
await app._set_spinner("Thinking")
|
||||
await pilot.pause()
|
||||
|
||||
assert any(
|
||||
entry[0] == "set"
|
||||
and entry[2].get("state")
|
||||
is terminal_escape.TerminalProgressState.INDETERMINATE
|
||||
for entry in calls
|
||||
)
|
||||
|
||||
async def test_none_status_clears_progress(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Hiding the spinner should clear terminal progress."""
|
||||
from deepagents_cli import terminal_escape
|
||||
|
||||
calls: list[str] = []
|
||||
|
||||
def _record_set(*_args: object, **_kwargs: object) -> bool:
|
||||
calls.append("set")
|
||||
return True
|
||||
|
||||
def _record_clear() -> bool:
|
||||
calls.append("clear")
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(terminal_escape, "set_terminal_progress", _record_set)
|
||||
monkeypatch.setattr(terminal_escape, "clear_terminal_progress", _record_clear)
|
||||
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-osc")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
await app._set_spinner("Thinking")
|
||||
await pilot.pause()
|
||||
calls.clear()
|
||||
await app._set_spinner(None)
|
||||
await pilot.pause()
|
||||
|
||||
assert "clear" in calls
|
||||
|
||||
async def test_consecutive_set_spinner_calls_keep_emitting(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Re-showing after a clear should re-emit indeterminate progress."""
|
||||
from deepagents_cli import terminal_escape
|
||||
|
||||
calls: list[str] = []
|
||||
|
||||
def _record_set(*_args: object, **_kwargs: object) -> bool:
|
||||
calls.append("set")
|
||||
return True
|
||||
|
||||
def _record_clear() -> bool:
|
||||
calls.append("clear")
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(terminal_escape, "set_terminal_progress", _record_set)
|
||||
monkeypatch.setattr(terminal_escape, "clear_terminal_progress", _record_clear)
|
||||
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-osc-rep")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
calls.clear()
|
||||
await app._set_spinner("Thinking")
|
||||
await app._set_spinner("Thinking")
|
||||
await app._set_spinner(None)
|
||||
await app._set_spinner("Thinking")
|
||||
await pilot.pause()
|
||||
|
||||
assert calls.count("set") >= 3
|
||||
assert "clear" in calls
|
||||
|
||||
async def test_set_spinner_swallows_terminal_escape_errors(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Unexpected exceptions from `terminal_escape` must not break the UI."""
|
||||
from deepagents_cli import terminal_escape
|
||||
|
||||
def _boom_set(*_args: object, **_kwargs: object) -> bool:
|
||||
msg = "boom"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
def _boom_clear() -> bool:
|
||||
msg = "boom"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
monkeypatch.setattr(terminal_escape, "set_terminal_progress", _boom_set)
|
||||
monkeypatch.setattr(terminal_escape, "clear_terminal_progress", _boom_clear)
|
||||
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-osc-boom")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
# Must not raise even though both calls explode.
|
||||
await app._set_spinner("Thinking")
|
||||
await pilot.pause()
|
||||
await app._set_spinner(None)
|
||||
await pilot.pause()
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
"""Tests for `deepagents_cli.terminal_escape`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import logging
|
||||
import pathlib
|
||||
|
||||
import pytest
|
||||
|
||||
from deepagents_cli import terminal_escape
|
||||
from deepagents_cli.terminal_escape import (
|
||||
TerminalProgressState,
|
||||
_validate_progress,
|
||||
clear_terminal_progress,
|
||||
set_terminal_progress,
|
||||
write_osc,
|
||||
write_terminal_escape,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_active_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Reset the module-level progress sentinel between tests."""
|
||||
monkeypatch.setattr(terminal_escape, "_progress_active", False)
|
||||
monkeypatch.setattr(terminal_escape, "_atexit_registered", False)
|
||||
monkeypatch.delenv(terminal_escape.NO_TERMINAL_ESCAPE, raising=False)
|
||||
|
||||
|
||||
class _FakeTTY(io.StringIO):
|
||||
"""`StringIO` with a context-manager that doesn't truncate on close."""
|
||||
|
||||
def __enter__(self) -> _FakeTTY: # noqa: PYI034 # _FakeTTY is a test helper
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc: object) -> None:
|
||||
pass
|
||||
|
||||
|
||||
class TestWriteTerminalEscape:
|
||||
"""Tests for `write_terminal_escape`."""
|
||||
|
||||
def test_writes_to_tty_when_available(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert write_terminal_escape("\x1b[?25l") is True
|
||||
assert fake.getvalue() == "\x1b[?25l"
|
||||
|
||||
def test_falls_back_to_stderr_when_tty_missing(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: None)
|
||||
monkeypatch.setattr(terminal_escape, "_is_stream_tty", lambda _stream: True)
|
||||
buf = io.StringIO()
|
||||
monkeypatch.setattr("sys.__stderr__", buf)
|
||||
assert write_terminal_escape("\x1b]9;4;0;0\a") is True
|
||||
assert buf.getvalue() == "\x1b]9;4;0;0\a"
|
||||
|
||||
def test_no_op_when_no_tty_and_stderr_redirected(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: None)
|
||||
monkeypatch.setattr(terminal_escape, "_is_stream_tty", lambda _stream: False)
|
||||
assert write_terminal_escape("\x1b]9;4;0;0\a") is False
|
||||
|
||||
def test_no_op_when_disabled_by_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv(terminal_escape.NO_TERMINAL_ESCAPE, "1")
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert write_terminal_escape("\x1b]9;4;0;0\a") is False
|
||||
assert fake.getvalue() == ""
|
||||
|
||||
def test_no_op_for_empty_sequence(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert write_terminal_escape("") is False
|
||||
assert fake.getvalue() == ""
|
||||
|
||||
def test_oserror_during_write_returns_false(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
class _RaisingTTY(_FakeTTY):
|
||||
def write(self, _data: str) -> int:
|
||||
msg = "disconnected"
|
||||
raise OSError(msg)
|
||||
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: _RaisingTTY())
|
||||
monkeypatch.setattr(terminal_escape, "_is_stream_tty", lambda _stream: False)
|
||||
assert write_terminal_escape("\x1b]9;4;0;0\a") is False
|
||||
|
||||
def test_real_open_tty_oserror_falls_through(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Exercise the real `_open_tty` body, not a stub."""
|
||||
|
||||
def _raising_open(*_args: object, **_kwargs: object) -> None:
|
||||
msg = "no tty"
|
||||
raise OSError(msg)
|
||||
|
||||
monkeypatch.setattr(pathlib.Path, "open", _raising_open)
|
||||
monkeypatch.setattr(terminal_escape, "_is_stream_tty", lambda _stream: False)
|
||||
assert write_terminal_escape("\x1b]9;4;0;0\a") is False
|
||||
|
||||
def test_stderr_fallback_write_failure_returns_false(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""When `/dev/tty` is missing and stderr write raises, return `False`."""
|
||||
|
||||
class _RaisingStderr:
|
||||
def write(self, _data: str) -> int:
|
||||
msg = "stderr closed"
|
||||
raise OSError(msg)
|
||||
|
||||
def flush(self) -> None: ...
|
||||
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: None)
|
||||
monkeypatch.setattr(terminal_escape, "_is_stream_tty", lambda _stream: True)
|
||||
monkeypatch.setattr("sys.__stderr__", _RaisingStderr())
|
||||
assert write_terminal_escape("\x1b]9;4;0;0\a") is False
|
||||
|
||||
def test_unicode_payload_round_trips(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""OSC payloads with non-ASCII bytes are written verbatim."""
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert write_osc("9;4", "テスト") is True
|
||||
assert fake.getvalue() == "\x1b]9;4;テスト\a"
|
||||
|
||||
|
||||
class TestWriteOsc:
|
||||
"""Tests for `write_osc`."""
|
||||
|
||||
def test_default_terminator_is_bel(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
write_osc("9;4", "3;0")
|
||||
assert fake.getvalue() == "\x1b]9;4;3;0\a"
|
||||
|
||||
def test_st_terminator(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
write_osc("9;4", "0;0", st=True)
|
||||
assert fake.getvalue() == "\x1b]9;4;0;0\x1b\\"
|
||||
|
||||
def test_empty_payload_omits_separator(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
write_osc("0")
|
||||
assert fake.getvalue() == "\x1b]0\a"
|
||||
|
||||
|
||||
class TestValidateProgress:
|
||||
"""Tests for `_validate_progress`."""
|
||||
|
||||
def test_clear_state_normalizes_to_zero(self) -> None:
|
||||
assert _validate_progress(50, TerminalProgressState.CLEAR) == 0
|
||||
|
||||
def test_indeterminate_normalizes_to_zero(self) -> None:
|
||||
assert _validate_progress(50, TerminalProgressState.INDETERMINATE) == 0
|
||||
|
||||
def test_determinate_passthrough(self) -> None:
|
||||
assert _validate_progress(42, TerminalProgressState.NORMAL) == 42
|
||||
|
||||
def test_determinate_clamps_low(self) -> None:
|
||||
assert _validate_progress(-10, TerminalProgressState.NORMAL) == 0
|
||||
|
||||
def test_determinate_clamps_high(self) -> None:
|
||||
assert _validate_progress(250, TerminalProgressState.ERROR) == 100
|
||||
|
||||
def test_none_progress_for_determinate_becomes_zero(self) -> None:
|
||||
assert _validate_progress(None, TerminalProgressState.NORMAL) == 0
|
||||
|
||||
def test_non_numeric_progress_coerces_to_zero(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Bad types are logged + treated as zero, never raised."""
|
||||
with caplog.at_level(logging.DEBUG, logger=terminal_escape.__name__):
|
||||
assert _validate_progress("nope", TerminalProgressState.NORMAL) == 0 # type: ignore[arg-type]
|
||||
assert any("non-numeric" in record.message for record in caplog.records)
|
||||
|
||||
def test_clear_with_nonzero_progress_is_logged(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Supplying progress to CLEAR/INDETERMINATE is observable misuse."""
|
||||
with caplog.at_level(logging.DEBUG, logger=terminal_escape.__name__):
|
||||
assert _validate_progress(42, TerminalProgressState.CLEAR) == 0
|
||||
assert any("ignoring progress" in record.message for record in caplog.records)
|
||||
|
||||
|
||||
class TestSetTerminalProgress:
|
||||
"""Tests for `set_terminal_progress` / `clear_terminal_progress`."""
|
||||
|
||||
def test_normal_progress_writes_state_and_percent(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert set_terminal_progress(75, state=TerminalProgressState.NORMAL) is True
|
||||
assert fake.getvalue() == "\x1b]9;4;1;75\a"
|
||||
|
||||
def test_indeterminate_emits_zero_progress(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
set_terminal_progress(state=TerminalProgressState.INDETERMINATE)
|
||||
assert fake.getvalue() == "\x1b]9;4;3;0\a"
|
||||
|
||||
def test_clear_emits_clear_state(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert clear_terminal_progress() is True
|
||||
assert fake.getvalue() == "\x1b]9;4;0;0\a"
|
||||
|
||||
def test_fires_unconditionally_regardless_of_terminal(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""No terminal allowlist — unsupported terminals ignore the sequence."""
|
||||
monkeypatch.setenv("TERM_PROGRAM", "iTerm.app")
|
||||
monkeypatch.delenv("WT_SESSION", raising=False)
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
assert set_terminal_progress(state=TerminalProgressState.INDETERMINATE) is True
|
||||
assert fake.getvalue() == "\x1b]9;4;3;0\a"
|
||||
|
||||
def test_active_sentinel_set_and_cleared(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
registered: list[object] = []
|
||||
monkeypatch.setattr("atexit.register", lambda fn: registered.append(fn) or fn)
|
||||
set_terminal_progress(state=TerminalProgressState.INDETERMINATE)
|
||||
assert terminal_escape._progress_active is True
|
||||
assert registered == [terminal_escape._atexit_clear]
|
||||
clear_terminal_progress()
|
||||
assert terminal_escape._progress_active is False
|
||||
|
||||
def test_atexit_registered_only_once(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
registered: list[object] = []
|
||||
monkeypatch.setattr("atexit.register", lambda fn: registered.append(fn) or fn)
|
||||
set_terminal_progress(state=TerminalProgressState.INDETERMINATE)
|
||||
set_terminal_progress(50, state=TerminalProgressState.NORMAL)
|
||||
clear_terminal_progress()
|
||||
set_terminal_progress(state=TerminalProgressState.INDETERMINATE)
|
||||
assert registered == [terminal_escape._atexit_clear]
|
||||
|
||||
def test_failed_write_does_not_register_atexit(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: None)
|
||||
monkeypatch.setattr(terminal_escape, "_is_stream_tty", lambda _stream: False)
|
||||
registered: list[object] = []
|
||||
monkeypatch.setattr("atexit.register", lambda fn: registered.append(fn) or fn)
|
||||
assert set_terminal_progress(state=TerminalProgressState.INDETERMINATE) is False
|
||||
assert registered == []
|
||||
assert terminal_escape._progress_active is False
|
||||
|
||||
|
||||
class TestAtexitClear:
|
||||
"""`_atexit_clear` should only emit a clear when progress was set."""
|
||||
|
||||
def test_emits_clear_when_active(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fake = _FakeTTY()
|
||||
monkeypatch.setattr(terminal_escape, "_open_tty", lambda: fake)
|
||||
monkeypatch.setattr(terminal_escape, "_progress_active", True)
|
||||
terminal_escape._atexit_clear()
|
||||
assert fake.getvalue() == "\x1b]9;4;0;0\a"
|
||||
|
||||
def test_skips_when_not_active(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
called: list[bool] = []
|
||||
monkeypatch.setattr(
|
||||
terminal_escape,
|
||||
"clear_terminal_progress",
|
||||
lambda: called.append(True) or False,
|
||||
)
|
||||
monkeypatch.setattr(terminal_escape, "_progress_active", False)
|
||||
terminal_escape._atexit_clear()
|
||||
assert called == []
|
||||
Reference in New Issue
Block a user