test: cover answers from the notification

Three seams: the offer is known when the wait starts (a confirm and a question,
with and without the platform capability), a button pressed after the
notification appears answers the tool confirm it belongs to — a pty stands in
for the terminal so the read has to consult both surfaces — and `confirm()`
takes an answer passed in and one offered by the wait hook, without asking the
backend.
This commit is contained in:
2026-10-23 10:24:56 +08:00
parent 80ad34f8fe
commit 16c5e01e44
3 changed files with 102 additions and 0 deletions
+55
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import contextlib
import sys
from typing import TYPE_CHECKING
import pytest
@@ -253,6 +254,60 @@ def test_tool_confirm_denied_by_the_notification() -> None:
assert prompt_confirm_tool("run_argv", {}, "runs a command") is False
@pytest.mark.skipif(sys.platform == "win32", reason="a pty is POSIX-only")
def test_a_button_replaces_the_terminal_read(monkeypatch: pytest.MonkeyPatch) -> None:
"""A button pressed in the notification answers the waiting prompt.
The read consults stdin *and* the notification while it blocks: the button
only exists once the notification is shown, which is after the read began.
"""
import os
import pty
import threading
from plyngent.cli.notify import ALLOW, DesktopRequest, WaitNotifier
from plyngent.config.models import NotifyConfig
from plyngent.prompting import set_prompt_wait_hook
class PressesTheButton:
"""A desktop surface that answers once the notification is up."""
def __init__(self) -> None:
self.requests: list[DesktopRequest] = []
def capabilities(self) -> frozenset[str]:
return frozenset({"buttons", "reply"})
def send(self, request: DesktopRequest) -> str | None:
self.requests.append(request)
answer = request.on_answer
if answer is not None:
threading.Timer(0.2, answer, args=(ALLOW,)).start()
return "id-1"
def clear(self, identifier: str | None) -> None:
del identifier
desktop = PressesTheButton()
notifier = WaitNotifier(
NotifyConfig(grace_seconds=0.0, min_interval_seconds=0.0),
desktop=desktop,
write=lambda _sequence: None,
)
master, slave = pty.openpty()
stream = os.fdopen(slave)
set_prompt_wait_hook(notifier)
monkeypatch.setattr(sys, "stdin", stream)
try:
with temporary_backend(ScriptedBackend([])):
assert prompt_confirm_tool("run_argv", {}, "runs a command") is True
finally:
set_prompt_wait_hook(None)
stream.close()
os.close(master)
assert desktop.requests
def test_batch_confirm_takes_a_notification_answer() -> None:
from plyngent.agent import ToolConfirmRequest
from plyngent.cli.limits import _prompt_confirm_batch_sync
+28
View File
@@ -257,6 +257,34 @@ def test_a_confirmation_offers_three_buttons() -> None:
assert request.reply_field is False
def test_a_waiting_prompt_is_answerable_before_it_is_announced() -> None:
"""The offer is decided when the wait starts, not when the notification shows.
A notification only appears after the grace period, so a prompt that learned
about its answers from ``notify_desktop`` would already be blocked on stdin.
"""
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop, grace_seconds=60.0)
with notifier(PromptWait("confirm", "Allow?")) as handle:
assert handle.supports_actions is True
assert desktop.requests == [] # not announced yet
def test_a_question_is_answerable_before_it_is_announced() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop, grace_seconds=60.0)
with notifier(PromptWait("ask", "Which branch?")) as handle:
assert handle.supports_actions is True
assert desktop.requests == []
def test_no_answer_is_expected_without_the_capability() -> None:
desktop = _FakeDesktop(capabilities=frozenset({"buttons"}))
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("ask", "Which branch?")) as handle:
assert handle.supports_actions is False
def test_an_ask_offers_a_reply_field_instead_of_buttons() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)