mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 15:37:55 +08:00
cli/notify: notify a waiting prompt through the desktop or the terminal
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`.
This commit is contained in:
@@ -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
|
||||
@@ -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 == []
|
||||
@@ -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: ...
|
||||
Reference in New Issue
Block a user