diff --git a/tests/test_cli/test_limits.py b/tests/test_cli/test_limits.py index e20eacc..e007e75 100644 --- a/tests/test_cli/test_limits.py +++ b/tests/test_cli/test_limits.py @@ -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 diff --git a/tests/test_cli/test_notify.py b/tests/test_cli/test_notify.py index 6689828..996e54b 100644 --- a/tests/test_cli/test_notify.py +++ b/tests/test_cli/test_notify.py @@ -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) diff --git a/tests/test_prompting.py b/tests/test_prompting.py index b674f11..b53e07a 100644 --- a/tests/test_prompting.py +++ b/tests/test_prompting.py @@ -267,6 +267,25 @@ def test_non_interactive_confirm_uses_default() -> None: assert confirm("ok?", default=True) is True +def test_confirm_takes_a_host_answer() -> None: + """``allow`` / ``deny`` from a notification answer a yes/no like ``y`` / ``n``.""" + with temporary_backend(ScriptedBackend([])): + assert confirm("Go?", source=_AnswerSource("allow")) is True + assert confirm("Go?", source=_AnswerSource("deny")) is False + assert confirm("Go?", source=_AnswerSource("n")) is False + # Nothing arrived: the prompt falls back to its default. + assert confirm("Go?", default=True, source=_AnswerSource(None)) is True + + +def test_confirm_takes_an_answer_the_hook_offers() -> None: + backend = ScriptedBackend([], confirms=[False]) + set_prompt_wait_hook(lambda _wait: contextlib.nullcontext(_AnswerSource("allow"))) + with temporary_backend(backend): + assert confirm("Go?") is True + # The answer came from the surface, so the backend was never asked. + assert backend.confirms == [False] + + def test_prompt_output_hook_runs_before_chrome(capsys: pytest.CaptureFixture[str]) -> None: """A registered hook runs before echo/secho write; clearing it removes it.