From 036e0b40ad143c33e9c0a6ff47c42018c90acfde Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 21 Oct 2026 19:57:17 +0800 Subject: [PATCH] cli/notify: notify a waiting prompt through the desktop or the terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A prompt that has waited `grace_seconds` is now announced, and where the platform allows it the notification carries the answer: allow / deny / review for a confirmation, a reply field for an `ask_user_*` question. Picking "review" withdraws the notification and keeps waiting at the terminal, and a notification is dismissed as soon as the prompt returns — answered or not. Two channels. The desktop one is the optional `notify` extra, run on its own event loop in a daemon thread because a notification's callbacks only fire while the loop that owns it is running (the library's synchronous wrapper does not provide that). The terminal one needs nothing: it writes OSC 777 / OSC 9 to `/dev/tty` — so a redirect cannot swallow it — picking the protocol from the environment and wrapping it for tmux / screen, falling back to a plain bell. kitty speaks OSC 99 only, which is not implemented, so it keeps the bell. Nothing here can break a prompt: a missing extra, a desktop that refuses, a failed write all fall through to the next channel and never raise. Titles are flattened and stripped of control characters, and the reason line (which can name a command or a URL) only travels with `include_reason`. --- src/plyngent/cli/notify.py | 501 ++++++++++++++++++++++++++ tests/test_cli/test_notify.py | 383 ++++++++++++++++++++ typings/desktop_notifier/__init__.pyi | 66 ++++ 3 files changed, 950 insertions(+) create mode 100644 src/plyngent/cli/notify.py create mode 100644 tests/test_cli/test_notify.py create mode 100644 typings/desktop_notifier/__init__.pyi diff --git a/src/plyngent/cli/notify.py b/src/plyngent/cli/notify.py new file mode 100644 index 0000000..260f36c --- /dev/null +++ b/src/plyngent/cli/notify.py @@ -0,0 +1,501 @@ +"""Notify the human when a prompt is waiting on them. + +A prompt blocks the turn, so someone who switched to another window learns about +it only when they come back. The notifier announces a prompt once it has waited +``grace_seconds`` — answering at the terminal stays silent — and, where the +platform allows it, offers the answer itself: allow / deny / review buttons for +a confirmation, a reply field for a question. + +Two channels. The desktop one is the optional ``desktop-notifier`` extra +(``pip install plyngent[notify]``); the terminal one writes an OSC escape +sequence to the controlling terminal, which every modern terminal turns into a +native notification without any dependency at all. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import importlib.util +import os +import sys +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from queue import Empty, SimpleQueue +from typing import TYPE_CHECKING, Any, Protocol, Self, TypeVar, runtime_checkable + +from plyngent.prompting import PromptWait, get_prompt_backend, set_prompt_wait_hook + +if TYPE_CHECKING: + from collections.abc import Callable, Coroutine, Mapping + from types import TracebackType + + from plyngent.config.models import NotifyConfig + +ALLOW = "allow" +DENY = "deny" +REVIEW = "review" + +#: Kinds whose notification carries answer buttons (``ask`` gets a reply field). +_ANSWER_KINDS = frozenset({"confirm", "batch", "policy", "access", "skills"}) + +_BUTTONS: tuple[tuple[str, str], ...] = ( + (ALLOW, "Allow once"), + (DENY, "Deny"), + (REVIEW, "Go to terminal"), +) + +_MAX_TITLE = 80 +_MAX_MESSAGE = 200 + + +def _clean(text: str, limit: int) -> str: + """One line, no control characters, truncated to *limit* characters.""" + flat = " ".join(str(text).split()) + stripped = flat.replace("\x1b", "").replace("\x07", "") + return stripped if len(stripped) <= limit else stripped[: limit - 1] + "…" + + +# --------------------------------------------------------------------------- # +# Desktop channel (optional `desktop-notifier` extra) +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True, slots=True) +class DesktopRequest: + """One notification to show on the desktop, with the answers it offers.""" + + title: str + message: str + buttons: tuple[tuple[str, str], ...] = () + reply_field: bool = False + on_answer: Callable[[str], None] | None = None + on_reply: Callable[[str], None] | None = None + + +@runtime_checkable +class DesktopChannel(Protocol): + """A desktop notification surface (the extra, or a test double).""" + + def capabilities(self) -> frozenset[str]: + """``buttons`` / ``reply`` when answers can be offered.""" + ... + + def send(self, request: DesktopRequest) -> str | None: + """Show *request*; the returned id dismisses it again.""" + ... + + def clear(self, identifier: str | None) -> None: + """Dismiss a notification (a no-op for an unknown id).""" + ... + + +class DesktopNotifierChannel: + """``desktop-notifier`` on its own loop in a daemon thread. + + A notification's callbacks only fire while the event loop that owns it is + running, which the library's synchronous wrapper does not provide — the + async API therefore runs here on a dedicated loop, and the prompt thread + waits on the results. + """ + + def __init__(self) -> None: + self._loop: asyncio.AbstractEventLoop = asyncio.new_event_loop() + self._started: threading.Event = threading.Event() + self._notifier: object = None + self._capabilities: frozenset[str] = frozenset() + self._thread: threading.Thread = threading.Thread(target=self._serve, name="plyngent-notify", daemon=True) + self._thread.start() + _ = self._started.wait(5.0) + + def _serve(self) -> None: + asyncio.set_event_loop(self._loop) + try: + self._loop.run_until_complete(self._init()) + except Exception: # noqa: BLE001 - absent extra, or a platform that refuses + self._notifier = None + self._capabilities = frozenset() + finally: + self._started.set() + self._loop.run_forever() + + async def _init(self) -> None: + from desktop_notifier import ( + Capability, + DesktopNotifier, + ) + + notifier = DesktopNotifier(app_name="plyngent") + capabilities = await notifier.get_capabilities() + names: set[str] = set() + if Capability.BUTTONS in capabilities: + names.add("buttons") + if Capability.REPLY_FIELD in capabilities: + names.add("reply") + self._notifier = notifier + self._capabilities = frozenset(names) + + def _call(self, coro: Coroutine[Any, Any, _T]) -> _T: + """Run *coro* on the channel's loop and wait for its result.""" + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + return future.result(5.0) + + def capabilities(self) -> frozenset[str]: + return self._capabilities + + def send(self, request: DesktopRequest) -> str | None: + if self._notifier is None: + msg = "no desktop notifier" + raise RuntimeError(msg) + from desktop_notifier import Button, ReplyField + + buttons = [ + Button(title=label, on_pressed=_button_handler(request.on_answer, key)) for key, label in request.buttons + ] + reply = ReplyField(title="Answer", on_replied=_reply_handler(request.on_reply)) if request.reply_field else None + send = getattr(self._notifier, "send") # noqa: B009 - typed at runtime + return str( + self._call( + send( + title=request.title, + message=request.message, + buttons=buttons, + reply_field=reply, + ) + ) + ) + + def clear(self, identifier: str | None) -> None: + if identifier is None or self._notifier is None: + return + with contextlib.suppress(Exception): + clear = getattr(self._notifier, "clear") # noqa: B009 - typed at runtime + _ = self._call(clear(identifier)) + + +def _button_handler(callback: Callable[[str], None] | None, key: str) -> Callable[[], None]: + """A button callback carrying the answer *key*.""" + + def _run() -> None: + if callback is not None: + callback(key) + + return _run + + +def _reply_handler(callback: Callable[[str], None] | None) -> Callable[[str], None]: + """A reply-field callback carrying the typed text.""" + + def _run(text: str) -> None: + if callback is not None: + callback(text) + + return _run + + +def load_desktop_channel() -> DesktopChannel | None: + """The desktop channel, or ``None`` when the extra is not installed.""" + if importlib.util.find_spec("desktop_notifier") is None: + return None + channel = DesktopNotifierChannel() + return channel if channel.capabilities() else None + + +# --------------------------------------------------------------------------- # +# Terminal channel (no dependency) +# --------------------------------------------------------------------------- # + +_OSC9 = "\x1b]9;{title}: {message}\x07" +_OSC777 = "\x1b]777;notify;{title};{message}\x07" +_BELL = "\a" +#: VTE (GNOME Terminal, Tilix, Terminator, …) speaks OSC 777 from this version on. +_VTE_OSC777 = 4600 + + +def _vte_version(env: Mapping[str, str]) -> int: + try: + return int(env.get("VTE_VERSION", "0")) + except ValueError: + return 0 + + +def detect_protocol(env: Mapping[str, str], override: str = "") -> str: + """Pick the terminal escape protocol: ``osc777``, ``osc9`` or ``bell``. + + Detection is environment-only — no capability query — so it never blocks and + never touches the input stream. kitty speaks OSC 99 only, which this module + does not implement, so it keeps the plain bell. + """ + forced = override.strip().lower() + if forced in {"osc777", "osc9", "bell"}: + return forced + program = env.get("TERM_PROGRAM", "") + if env.get("KITTY_WINDOW_ID"): + return "bell" + if program in {"iTerm.app", "Apple_Terminal"}: + return "osc9" + if program in {"WezTerm", "ghostty"} or env.get("WT_SESSION"): + return "osc777" + if _vte_version(env) >= _VTE_OSC777: + return "osc777" + term = env.get("TERM", "") + return "osc777" if any(hint in term for hint in ("wezterm", "ghostty", "rxvt")) else "bell" + + +def detect_multiplexer(env: Mapping[str, str]) -> str: + """``tmux`` / ``screen`` when the sequence must be wrapped, else ``""``.""" + if env.get("TMUX"): + return "tmux" + return "screen" if env.get("STY") else "" + + +def terminal_sequence(protocol: str, title: str, message: str, *, multiplexer: str = "") -> str: + """The escape sequence for one notification, wrapped for a multiplexer.""" + if protocol == "bell": + return _BELL + safe_title = _clean(title, _MAX_TITLE) + safe_message = _clean(message, _MAX_MESSAGE) + if protocol == "osc9": + seq = _OSC9.format(title=safe_title, message=safe_message) + else: + seq = _OSC777.format(title=safe_title, message=safe_message) + if multiplexer == "tmux": + # tmux needs `set -g allow-passthrough on` and an ESC-prefixed payload. + return "\x1bPtmux;\x1b" + seq.replace("\x1b", "\x1b\x1b") + "\x1b\\" + if multiplexer == "screen": + return "\x1bP" + seq + "\x1b\\" + return seq + + +def write_terminal(data: str) -> None: + """Write *data* to the controlling terminal; never raises. + + ``/dev/tty`` rather than stdout/stderr so a redirect cannot swallow the + notification or mix it into the program's output. + """ + payload = data.encode("utf-8", "replace") + if sys.platform != "win32": + try: + with Path("/dev/tty").open("wb") as tty: + _ = tty.write(payload) + tty.flush() + except OSError: + pass + else: + return + buffer = getattr(sys.stderr, "buffer", None) + if buffer is None: + return + with contextlib.suppress(OSError, ValueError): + _ = buffer.write(payload) + _ = sys.stderr.flush() + + +# --------------------------------------------------------------------------- # +# The notifier +# --------------------------------------------------------------------------- # + + +class _Resolve: + """Sentinel: resolve the desktop channel lazily from the extra.""" + + +_RESOLVE = _Resolve() +_T = TypeVar("_T") + + +class WaitHandle: + """One waiting prompt: its grace timer, its answers, its notification.""" + + notifier: WaitNotifier + wait: PromptWait + identifier: str | None + #: Whether the notification offered answers, so a read may poll for them. + supports_actions: bool + + def __init__(self, notifier: WaitNotifier, wait: PromptWait) -> None: + self.notifier = notifier + self.wait = wait + self.identifier = None + self.supports_actions = False + self._answers: SimpleQueue[str] = SimpleQueue() + self._timer: threading.Timer | None = None + self._closed: bool = False + + def take(self) -> str | None: + """The next answer offered by the notification, without blocking.""" + try: + return self._answers.get_nowait() + except Empty: + return None + + def answer(self, key: str) -> None: + """Record one answer from the notification (a button, or a reply).""" + if self._closed: + return + if key == REVIEW and self.wait.kind != "ask": + # "review" means "I will answer in the terminal": withdraw and wait. + self.notifier.dismiss(self) + return + self._answers.put(key) + + def __enter__(self) -> Self: + timer = threading.Timer(self.notifier.grace_seconds, self._notify) + timer.daemon = True + self._timer = timer + timer.start() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + del exc_type, exc, tb + self._closed = True + if self._timer is not None: + self._timer.cancel() + self.notifier.dismiss(self) + + def _notify(self) -> None: + if self._closed or not self.notifier.may_notify(): + return + if not self.notifier.notify_desktop(self): + self.notifier.notify_terminal(self.wait) + + +class WaitNotifier: + """Announce a waiting prompt once it has waited long enough. + + Instances are the ``prompt_wait`` hook: calling one returns the + :class:`WaitHandle` context manager the prompt read consults. + """ + + _config: NotifyConfig + _resolve_desktop: bool + _desktop: DesktopChannel | None + _write: Callable[[str], None] + _env: Mapping[str, str] + _clock: Callable[[], float] + _last_sent: float + + def __init__( + self, + config: NotifyConfig, + *, + desktop: DesktopChannel | _Resolve | None = _RESOLVE, + write: Callable[[str], None] | None = None, + env: Mapping[str, str] | None = None, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._config = config + if isinstance(desktop, _Resolve): + self._resolve_desktop = True + self._desktop = None + else: + self._resolve_desktop = False + self._desktop = desktop + self._write = write or write_terminal + self._env = os.environ if env is None else env + self._clock = clock + self._last_sent = float("-inf") + + # -- hook ------------------------------------------------------------- # + + def __call__(self, wait: PromptWait) -> WaitHandle: + return WaitHandle(self, wait) + + @property + def grace_seconds(self) -> float: + return self._config.grace_seconds + + def _wants(self, channel: str) -> bool: + channels = self._config.channels + return channel in channels or "auto" in channels + + def may_notify(self) -> bool: + """Whether a notification is worth sending right now.""" + if not self._config.enabled or not get_prompt_backend().is_interactive(): + return False + now = self._clock() + if now - self._last_sent < self._config.min_interval_seconds: + return False + self._last_sent = now + return True + + def _message(self, wait: PromptWait) -> str: + title = _clean(wait.title, _MAX_TITLE) + if self._config.include_reason and wait.detail: + return f"{title}\n{_clean(wait.detail, _MAX_MESSAGE)}" + return title + + def desktop_channel(self) -> DesktopChannel | None: + """The desktop surface, resolved once (importing the extra on demand).""" + if self._resolve_desktop: + self._desktop = load_desktop_channel() + self._resolve_desktop = False + return self._desktop + + def notify_desktop(self, handle: WaitHandle) -> bool: + """Show the desktop notification; False when it could not be shown.""" + if not self._wants("desktop"): + return False + channel = self.desktop_channel() + if channel is None: + return False + wait = handle.wait + capabilities = channel.capabilities() + buttons: tuple[tuple[str, str], ...] = () + if self._config.actions and wait.kind in _ANSWER_KINDS and "buttons" in capabilities: + buttons = _BUTTONS + reply = self._config.actions and wait.kind == "ask" and "reply" in capabilities + try: + handle.identifier = channel.send( + DesktopRequest( + title=_clean(wait.title, _MAX_TITLE), + message=self._message(wait), + buttons=buttons, + reply_field=reply, + on_answer=handle.answer, + on_reply=handle.answer, + ) + ) + except Exception: # noqa: BLE001 - a notification never breaks a prompt + return False + handle.supports_actions = bool(buttons) or reply + return True + + def notify_terminal(self, wait: PromptWait) -> None: + """Write one terminal notification (silent on any failure).""" + if not self._wants("terminal"): + return + sequence = terminal_sequence( + detect_protocol(self._env, self._config.protocol), + wait.title, + self._message(wait), + multiplexer=detect_multiplexer(self._env), + ) + self._write(sequence) + + def dismiss(self, handle: WaitHandle) -> None: + """Dismiss the handle's desktop notification, if one is showing.""" + identifier = handle.identifier + if identifier is None: + return + handle.identifier = None + channel = self._desktop + if channel is not None: + channel.clear(identifier) + + +def install_notifier(config: NotifyConfig) -> WaitNotifier | None: + """Install the prompt-wait hook from ``[notify]`` (None clears it).""" + if not config.enabled: + set_prompt_wait_hook(None) + return None + notifier = WaitNotifier(config) + set_prompt_wait_hook(notifier) + return notifier diff --git a/tests/test_cli/test_notify.py b/tests/test_cli/test_notify.py new file mode 100644 index 0000000..6689828 --- /dev/null +++ b/tests/test_cli/test_notify.py @@ -0,0 +1,383 @@ +from __future__ import annotations + +import time +from typing import TYPE_CHECKING + +import pytest + +from plyngent.cli.notify import ( + ALLOW, + DENY, + REVIEW, + DesktopRequest, + WaitNotifier, + detect_multiplexer, + detect_protocol, + install_notifier, + terminal_sequence, +) +from plyngent.config.models import NotifyConfig +from plyngent.prompting import ( + PromptWait, + configure_prompting, + get_prompt_wait_hook, + reset_prompting, + temporary_backend, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + +class _InteractiveBackend: + """An interactive backend that never actually reads (the prompts do not run).""" + + def is_interactive(self) -> bool: + return True + + def read_line(self, prompt: str, *, default: str | None = None, completions: object = None) -> str: + del prompt, completions + return default or "" + + def read_secret_line(self, prompt: str) -> str: + del prompt + return "" + + def confirm(self, prompt: str, *, default: bool = False) -> bool: + del prompt + return default + + def echo(self, message: str = "", *, err: bool = False) -> None: + del message, err + + def secho(self, message: str, *, fg: str | None = None, err: bool = False) -> None: + del message, fg, err + + +class _FakeDesktop: + """A desktop channel that records what it was asked to show.""" + + def __init__( + self, + capabilities: frozenset[str] | None = None, + *, + fail: bool = False, + ) -> None: + self.caps = frozenset({"buttons", "reply"}) if capabilities is None else capabilities + self.fail = fail + self.requests: list[DesktopRequest] = [] + self.cleared: list[str | None] = [] + + def capabilities(self) -> frozenset[str]: + return self.caps + + def send(self, request: DesktopRequest) -> str | None: + if self.fail: + msg = "no notification server" + raise RuntimeError(msg) + self.requests.append(request) + return "id-1" + + def clear(self, identifier: str | None) -> None: + self.cleared.append(identifier) + + +@pytest.fixture(autouse=True) +def _prompting_state() -> Iterator[None]: + reset_prompting() + configure_prompting(backend=_InteractiveBackend()) + yield + reset_prompting() + + +def _notifier( + *, + desktop: _FakeDesktop | None = None, + grace_seconds: float = 60.0, + min_interval_seconds: float = 0.0, + **kwargs: object, +) -> tuple[WaitNotifier, list[str]]: + """A notifier with a captured terminal writer and a fake desktop channel.""" + written: list[str] = [] + config = NotifyConfig( + grace_seconds=grace_seconds, + min_interval_seconds=min_interval_seconds, + **kwargs, # type: ignore[arg-type] + ) + notifier = WaitNotifier( + config, + desktop=desktop, + write=written.append, + env={"TERM_PROGRAM": "WezTerm"}, + ) + return notifier, written + + +# --------------------------------------------------------------------------- # +# Terminal channel +# --------------------------------------------------------------------------- # + + +def test_detect_protocol_reads_the_environment() -> None: + assert detect_protocol({"TERM_PROGRAM": "iTerm.app"}) == "osc9" + assert detect_protocol({"TERM_PROGRAM": "Apple_Terminal"}) == "osc9" + assert detect_protocol({"TERM_PROGRAM": "WezTerm"}) == "osc777" + assert detect_protocol({"TERM_PROGRAM": "ghostty"}) == "osc777" + assert detect_protocol({"WT_SESSION": "1"}) == "osc777" + assert detect_protocol({"VTE_VERSION": "5000"}) == "osc777" + assert detect_protocol({"TERM": "rxvt-unicode"}) == "osc777" + # kitty speaks OSC 99 only, which this module does not implement. + assert detect_protocol({"KITTY_WINDOW_ID": "1"}) == "bell" + assert detect_protocol({"TERM": "dumb"}) == "bell" + + +def test_detect_protocol_honours_the_override() -> None: + assert detect_protocol({"TERM_PROGRAM": "WezTerm"}, "osc9") == "osc9" + assert detect_protocol({"TERM_PROGRAM": "iTerm.app"}, "bell") == "bell" + # An unknown override falls back to detection. + assert detect_protocol({"TERM_PROGRAM": "WezTerm"}, "osc99") == "osc777" + + +def test_detect_multiplexer() -> None: + assert detect_multiplexer({"TMUX": "/tmp/tmux-1000/default,1,0"}) == "tmux" + assert detect_multiplexer({"STY": "1234.pts-0.host"}) == "screen" + assert detect_multiplexer({}) == "" + + +def test_terminal_sequence_osc777_and_osc9() -> None: + assert terminal_sequence("osc777", "T", "M") == "\x1b]777;notify;T;M\x07" + assert terminal_sequence("osc9", "T", "M") == "\x1b]9;T: M\x07" + assert terminal_sequence("bell", "T", "M") == "\a" + + +def test_terminal_sequence_is_wrapped_for_tmux() -> None: + seq = terminal_sequence("osc777", "T", "M", multiplexer="tmux") + assert seq.startswith("\x1bPtmux;\x1b") + assert seq.endswith("\x1b\\") + # The inner ESCs are doubled so tmux forwards them verbatim. + assert "\x1b\x1b]777" in seq + + +def test_terminal_sequence_strips_control_characters() -> None: + seq = terminal_sequence("osc777", "a\x07b\nc", "d\te") + assert "\x07" not in seq.removesuffix("\x07") + assert "ab c" in seq + assert "\n" not in seq + + +# --------------------------------------------------------------------------- # +# Grace, throttle, channels +# --------------------------------------------------------------------------- # + + +def test_nothing_is_shown_before_the_grace_period() -> None: + notifier, written = _notifier(grace_seconds=0.05) + with notifier(PromptWait("confirm", "Allow?")): + pass + time.sleep(0.1) + assert written == [] + + +def test_a_waited_prompt_is_announced() -> None: + notifier, written = _notifier(grace_seconds=0.01) + with notifier(PromptWait("confirm", "Allow?")) as handle: + deadline = time.monotonic() + 1.0 + while not written and time.monotonic() < deadline: + time.sleep(0.01) + assert handle.supports_actions is False # no desktop channel here + assert written + assert "777" in written[0] + assert "Allow?" in written[0] + + +def test_terminal_channel_can_be_pinned() -> None: + desktop = _FakeDesktop() + notifier, written = _notifier(desktop=desktop, channels=["terminal"]) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert desktop.requests == [] + assert written + + +def test_desktop_channel_can_be_pinned() -> None: + desktop = _FakeDesktop() + notifier, written = _notifier(desktop=desktop, channels=["desktop"]) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert len(desktop.requests) == 1 + assert written == [] + + +def test_terminal_fallback_when_the_desktop_refuses() -> None: + desktop = _FakeDesktop(fail=True) + notifier, written = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert written + + +def test_a_burst_is_throttled() -> None: + notifier, written = _notifier(grace_seconds=0.0, min_interval_seconds=60.0) + for _ in range(2): + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert len(written) == 1 + + +def test_nothing_is_shown_on_a_non_interactive_prompt() -> None: + from plyngent.prompting import NonInteractiveBackend + + notifier, written = _notifier() + configure_prompting(backend=NonInteractiveBackend()) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert written == [] + + +def test_disabled_notifications_send_nothing() -> None: + notifier, written = _notifier(grace_seconds=0.0, enabled=False) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert written == [] + + +# --------------------------------------------------------------------------- # +# Answers +# --------------------------------------------------------------------------- # + + +def test_a_confirmation_offers_three_buttons() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert handle.supports_actions is True + request = desktop.requests[0] + assert [key for key, _ in request.buttons] == [ALLOW, DENY, REVIEW] + assert request.reply_field is False + + +def test_an_ask_offers_a_reply_field_instead_of_buttons() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("ask", "Which branch?")) as handle: + handle._notify() + request = desktop.requests[0] + assert request.reply_field is True + assert request.buttons == () + + +def test_a_limit_prompt_carries_no_answers() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("limit", "Round limit reached")) as handle: + handle._notify() + assert handle.supports_actions is False + assert desktop.requests[0].buttons == () + assert desktop.requests[0].reply_field is False + + +def test_actions_can_be_switched_off() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop, actions=False) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert handle.supports_actions is False + assert desktop.requests[0].buttons == () + + +def test_platform_without_buttons_gets_a_text_notification() -> None: + desktop = _FakeDesktop(capabilities=frozenset()) + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert handle.supports_actions is False + assert desktop.requests[0].buttons == () + + +def test_a_button_answer_reaches_the_prompt() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + request = desktop.requests[0] + assert request.on_answer is not None + request.on_answer(ALLOW) + assert handle.take() == ALLOW + assert handle.take() is None + + +def test_a_reply_reaches_the_prompt() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("ask", "Which branch?")) as handle: + handle._notify() + request = desktop.requests[0] + assert request.on_reply is not None + request.on_reply("feature/x") + assert handle.take() == "feature/x" + + +def test_review_keeps_waiting_and_dismisses() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + request = desktop.requests[0] + assert request.on_answer is not None + request.on_answer(REVIEW) + assert handle.take() is None + assert desktop.cleared == ["id-1"] + + +def test_an_answer_after_the_prompt_is_ignored() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + request = desktop.requests[0] + assert request.on_answer is not None + request.on_answer(ALLOW) + assert handle.take() is None + + +def test_the_notification_is_dismissed_when_the_prompt_returns() -> None: + desktop = _FakeDesktop() + notifier, _ = _notifier(desktop=desktop) + with notifier(PromptWait("confirm", "Allow?")) as handle: + handle._notify() + assert handle.identifier == "id-1" + assert handle.identifier is None + assert desktop.cleared == ["id-1"] + + +# --------------------------------------------------------------------------- # +# Installation +# --------------------------------------------------------------------------- # + + +def test_install_notifier_sets_the_hook() -> None: + notifier = install_notifier(NotifyConfig()) + assert notifier is not None + assert get_prompt_wait_hook() is not None + + +def test_install_notifier_can_be_disabled() -> None: + assert install_notifier(NotifyConfig(enabled=False)) is None + assert get_prompt_wait_hook() is None + + +def test_reason_stays_out_unless_asked_for() -> None: + notifier, _ = _notifier() + wait = PromptWait("confirm", "Allow?", "argv: rm -rf /") + assert notifier._message(wait) == "Allow?" + notifier, _ = _notifier(grace_seconds=0.0, include_reason=True) + assert "rm -rf" in notifier._message(wait) + + +def test_temporary_backend_is_transparent_to_the_notifier() -> None: + """The notifier is only a hook: without installation nothing is shown.""" + notifier, written = _notifier() + with temporary_backend(_InteractiveBackend()), notifier(PromptWait("confirm", "Allow?")): + pass + assert written == [] diff --git a/typings/desktop_notifier/__init__.pyi b/typings/desktop_notifier/__init__.pyi new file mode 100644 index 0000000..10e2cd3 --- /dev/null +++ b/typings/desktop_notifier/__init__.pyi @@ -0,0 +1,66 @@ +# Minimal stubs for the optional ``desktop-notifier`` extra. +# Only the surface ``plyngent.cli.notify`` uses is declared; the real package is +# installed with ``pip install plyngent[notify]`` and is absent by default. + +from collections.abc import Sequence +from enum import Enum +from typing import Any + +class Capability(Enum): + APP_NAME = ... + TITLE = ... + MESSAGE = ... + URGENCY = ... + ICON = ... + BUTTONS = ... + REPLY_FIELD = ... + ATTACHMENT = ... + ON_DISPATCHED = ... + ON_CLICKED = ... + ON_DISMISSED = ... + SOUND = ... + THREAD = ... + TIMEOUT = ... + +class Button: + def __init__( + self, + title: str, + on_pressed: Any = None, + identifier: str | None = None, + ) -> None: ... + +class ReplyField: + def __init__( + self, + title: str = ..., + button_title: str = ..., + on_replied: Any = None, + ) -> None: ... + +class DesktopNotifier: + def __init__( + self, + app_name: str = ..., + app_icon: Any = ..., + notification_limit: int | None = None, + ) -> None: ... + async def get_capabilities(self) -> frozenset[Capability]: ... + async def send( + self, + title: str, + message: str, + urgency: Any = ..., + icon: Any = ..., + buttons: Sequence[Button] = ..., + reply_field: ReplyField | None = None, + on_dispatched: Any = None, + on_clicked: Any = None, + on_dismissed: Any = None, + attachment: Any = ..., + sound: Any = ..., + thread: str | None = None, + timeout: int = -1, # noqa: ASYNC109 - the library's own parameter name + ) -> str: ... + async def clear(self, identifier: str) -> None: ... + async def clear_all(self) -> None: ...