forked from vhaudiquet/weekly-activity
refactor(wizard): replace hand-rolled arrow-key menu with questionary
The reviewer asked for a known-good interactive-menu library instead of the custom termios/select key reader; arrow keys had failed in their terminal. The TTY path now delegates to questionary.select (pointer ">", bold highlight via pointer/selected styles), keeping choice values mapped to list indices. Cancellable menus graft an eager Escape binding onto the prompt's own key-binding registry so Esc backs out while Ctrl-C still aborts fixed menus and backs out of cancellable ones. The non-TTY numbered fallback is untouched, and hand-rolled internals (_arrow_menu/_read_keypress/_option_line/_draw_menu_rows) are gone.
This commit is contained in:
+208
-107
@@ -2,20 +2,21 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import pty
|
||||
import re
|
||||
import select
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
from unittest import mock
|
||||
|
||||
from weekly_activity.wizard import _menu, _option_line
|
||||
from weekly_activity.wizard import _CANCELLED, _menu
|
||||
|
||||
|
||||
class NumberedMenuFallbackTest(unittest.TestCase):
|
||||
@@ -52,58 +53,211 @@ class NumberedMenuFallbackTest(unittest.TestCase):
|
||||
self.assertEqual(output.count("Please enter a number between 1 and 3."), 2)
|
||||
|
||||
|
||||
class OptionLineRenderingTest(unittest.TestCase):
|
||||
"""Selected rows carry the '>' marker and bold codes; others stay plain."""
|
||||
|
||||
def test_selected_line_is_bold_and_marked(self) -> None:
|
||||
self.assertEqual(_option_line("Exit", True), "\x1b[1m> Exit\x1b[0m")
|
||||
|
||||
def test_unselected_line_is_indented_and_plain(self) -> None:
|
||||
self.assertEqual(_option_line("Remove a source", False), " Remove a source")
|
||||
# --- Interactive (TTY) path -----------------------------------------------
|
||||
|
||||
|
||||
_CHILD_SCRIPT = """
|
||||
class _Tty(io.StringIO):
|
||||
"""A stream claiming to be a terminal so _menu routes to questionary."""
|
||||
|
||||
def isatty(self) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
class _FakeKeyBindings:
|
||||
"""Registry mimicking prompt_toolkit's enough to capture added handlers."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.added: list[tuple[tuple[Any, ...], Any]] = []
|
||||
|
||||
def add(self, *keys: Any, eager: bool = False):
|
||||
def register(handler):
|
||||
self.added.append((keys, handler))
|
||||
return handler
|
||||
|
||||
return register
|
||||
|
||||
|
||||
class _FakeApplication:
|
||||
def __init__(self) -> None:
|
||||
self.key_bindings = _FakeKeyBindings()
|
||||
self.exit_calls: list[dict[str, Any]] = []
|
||||
|
||||
def exit(self, result: Any = None, exception: BaseException | None = None, style=None) -> None:
|
||||
self.exit_calls.append({"result": result, "exception": exception})
|
||||
|
||||
|
||||
class _FakeQuestion:
|
||||
def __init__(self, outcome: Any) -> None:
|
||||
self.application = _FakeApplication()
|
||||
self.outcome = outcome
|
||||
|
||||
def unsafe_ask(self) -> Any:
|
||||
if isinstance(self.outcome, BaseException):
|
||||
raise self.outcome
|
||||
return self.outcome
|
||||
|
||||
|
||||
class _FakeQuestionary:
|
||||
"""Stand-in exposing exactly the surface wizard._tty_menu touches."""
|
||||
|
||||
def __init__(self, outcome: Any) -> None:
|
||||
self.outcome = outcome
|
||||
self.select_outcome = _FakeQuestion(outcome)
|
||||
self.seen: dict[str, Any] = {}
|
||||
|
||||
def Choice(self, **kwargs: Any) -> Any:
|
||||
return types.SimpleNamespace(**kwargs)
|
||||
|
||||
@staticmethod
|
||||
def Style(styles: list[tuple[str, str]]) -> list[tuple[str, str]]:
|
||||
return styles
|
||||
|
||||
def select(self, message: str, choices: list[Any], **kwargs: Any) -> _FakeQuestion:
|
||||
self.seen.update(message=message, choices=list(choices), **kwargs)
|
||||
return self.select_outcome
|
||||
|
||||
|
||||
class QuestionaryDelegationTest(unittest.TestCase):
|
||||
"""On terminals _menu must hand selection to questionary, not re-implement it."""
|
||||
|
||||
def ask_tty(
|
||||
self,
|
||||
options: list[str],
|
||||
outcome: Any,
|
||||
*,
|
||||
cancellable: bool,
|
||||
) -> tuple[int | None, _FakeQuestion, dict[str, Any]]:
|
||||
"""Run the TTY branch against an injected fake questionary module."""
|
||||
fake = _FakeQuestionary(outcome)
|
||||
|
||||
with (
|
||||
mock.patch.object(sys, "stdin", _Tty()),
|
||||
mock.patch.object(sys, "stdout", _Tty()),
|
||||
mock.patch.dict(sys.modules, {"questionary": cast("Any", fake)}),
|
||||
):
|
||||
picked = _menu("Pick one", options, cancellable=cancellable)
|
||||
return picked, fake.select_outcome, fake.seen
|
||||
|
||||
def test_non_cancellable_menu_delegates_and_maps_choice_value_to_index(self) -> None:
|
||||
picked, question, seen = self.ask_tty(
|
||||
["Alpha", "Beta", "Gamma"],
|
||||
outcome=2,
|
||||
cancellable=False,
|
||||
)
|
||||
self.assertEqual(picked, 2)
|
||||
self.assertEqual(seen["message"], "Pick one")
|
||||
self.assertEqual(seen["pointer"], ">")
|
||||
self.assertEqual(
|
||||
[(choice.title, choice.value) for choice in seen["choices"]],
|
||||
[("Alpha", 0), ("Beta", 1), ("Gamma", 2)],
|
||||
)
|
||||
# A fixed menu grafts nothing onto questionary's own bindings.
|
||||
self.assertEqual(question.application.key_bindings.added, [])
|
||||
|
||||
def test_cancellable_menu_keeps_fixed_menus_behaviour_on_ctrl_c(self) -> None:
|
||||
with self.assertRaises(KeyboardInterrupt):
|
||||
self.ask_tty(["Alpha", "Beta"], outcome=KeyboardInterrupt(), cancellable=False)
|
||||
|
||||
def test_cancellable_menu_maps_ctrl_c_to_none(self) -> None:
|
||||
picked, _, _ = self.ask_tty(
|
||||
["Alpha", "Beta"],
|
||||
outcome=KeyboardInterrupt(),
|
||||
cancellable=True,
|
||||
)
|
||||
self.assertIsNone(picked)
|
||||
|
||||
def test_cancellable_menu_grafts_escape_onto_the_prompt_bindings(self) -> None:
|
||||
picked, question, _ = self.ask_tty(
|
||||
["Alpha", "Beta"],
|
||||
outcome=_CANCELLED,
|
||||
cancellable=True,
|
||||
)
|
||||
self.assertIsNone(picked) # a non-int answer means "backed out"
|
||||
keys, handler = question.application.key_bindings.added[-1]
|
||||
self.assertIn("escape", keys)
|
||||
handler(types.SimpleNamespace(app=question.application))
|
||||
self.assertIs(question.application.exit_calls[0]["result"], _CANCELLED)
|
||||
|
||||
|
||||
# --- Real-terminal proof over a pty ----------------------------------------
|
||||
|
||||
|
||||
_MENU_CHILD = """
|
||||
import os
|
||||
import sys
|
||||
import termios
|
||||
import tty
|
||||
|
||||
sys.path.insert(0, os.environ["WIZARD_SRC"])
|
||||
from weekly_activity.wizard import _read_keypress # noqa: E402
|
||||
from weekly_activity.wizard import _tty_menu # noqa: E402
|
||||
|
||||
fd = sys.stdin.fileno()
|
||||
saved = termios.tcgetattr(fd)
|
||||
tty.setcbreak(fd)
|
||||
try:
|
||||
for _ in range(int(sys.argv[1])):
|
||||
print("READY", flush=True)
|
||||
print(f"KEY={_read_keypress()}", flush=True)
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, saved)
|
||||
picked = _tty_menu("Pick one", ["Alpha", "Beta", "Gamma"], cancellable=True)
|
||||
print(f"PICKED={picked}", flush=True)
|
||||
"""
|
||||
|
||||
|
||||
_KEY_LINE = re.compile(rb"KEY=([a-z]+)\r\n")
|
||||
_PICKED_LINE = re.compile(rb"PICKED=(\d+|None)")
|
||||
|
||||
|
||||
class ArrowKeyKeypressTest(unittest.TestCase):
|
||||
"""CSI sequences must classify over a real pty instead of surfacing as bare Esc.
|
||||
class ArrowMenuPtyTest(unittest.TestCase):
|
||||
"""Drive the genuine questionary menu across a real pty with raw keypresses.
|
||||
|
||||
Drives the genuine ``_read_keypress()`` on a cbreak-mode terminal — the path the
|
||||
arrow-key menu uses, unreachable through scripted ``io.StringIO`` stdin.
|
||||
This is the regression case the reviewer hit: arrow presses must move the
|
||||
highlight of a live menu, which scripted ``io.StringIO`` stdin cannot reach.
|
||||
"""
|
||||
|
||||
TIMEOUT_S = 20.0
|
||||
TIMEOUT_S = 30.0
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.child_env = {
|
||||
**os.environ,
|
||||
"WIZARD_SRC": str(Path(__file__).resolve().parents[1] / "src"),
|
||||
}
|
||||
|
||||
def drive_menu(self, sends: list[bytes]) -> tuple[str, int]:
|
||||
"""Render the menu on a pty, send keystrokes, return screen + exit code."""
|
||||
master, slave = pty.openpty()
|
||||
child = subprocess.Popen(
|
||||
[sys.executable, "-c", _MENU_CHILD],
|
||||
stdin=slave,
|
||||
stdout=slave,
|
||||
stderr=subprocess.PIPE,
|
||||
env=self.child_env,
|
||||
)
|
||||
os.close(slave)
|
||||
screen = bytearray()
|
||||
|
||||
try:
|
||||
self._await(screen, master, child, lambda: b"Gamma" in screen, "menu render")
|
||||
for payload in sends:
|
||||
os.write(master, payload)
|
||||
time.sleep(0.2) # let the vt100 parser see keystrokes separately
|
||||
self._await(
|
||||
screen,
|
||||
master,
|
||||
child,
|
||||
lambda: _PICKED_LINE.search(bytes(screen)) is not None,
|
||||
"final PICKED line",
|
||||
)
|
||||
child.wait(timeout=self.TIMEOUT_S)
|
||||
err = child.stderr.read().decode("utf-8", "replace") if child.stderr else ""
|
||||
self.assertEqual(child.returncode, 0, f"menu child failed: {err}")
|
||||
return bytes(screen).decode("utf-8", "replace"), child.returncode
|
||||
finally:
|
||||
if child.poll() is None:
|
||||
child.kill()
|
||||
child.wait()
|
||||
if child.stderr is not None:
|
||||
child.stderr.close()
|
||||
os.close(master)
|
||||
|
||||
def _await(
|
||||
self,
|
||||
screen: bytearray,
|
||||
master: int,
|
||||
child: subprocess.Popen[bytes],
|
||||
box: dict[str, bytes],
|
||||
condition: Callable[[], bool],
|
||||
condition,
|
||||
stage: str,
|
||||
) -> None:
|
||||
"""Block until ``condition`` holds on the pty stream, else fail loudly."""
|
||||
"""Read the pty until ``condition`` holds on accumulated output."""
|
||||
deadline = time.monotonic() + self.TIMEOUT_S
|
||||
while not condition():
|
||||
remaining = deadline - time.monotonic()
|
||||
@@ -116,90 +270,37 @@ class ArrowKeyKeypressTest(unittest.TestCase):
|
||||
except OSError: # slave closed: the child exited early
|
||||
break
|
||||
if chunk:
|
||||
box["stream"] += chunk
|
||||
screen.extend(chunk)
|
||||
else:
|
||||
return
|
||||
if child.poll() is None:
|
||||
child.kill()
|
||||
child.wait(timeout=5)
|
||||
err = child.stderr.read().decode("utf-8", "replace") if child.stderr else ""
|
||||
self.fail(f"{stage}: stalled; tail={box['stream'][-300:]!r} stderr={err!r}")
|
||||
self.fail(f"{stage}: stalled; tail={bytes(screen)[-300:]!r} stderr={err!r}")
|
||||
|
||||
def _child_tokens(self, sends: list[bytes]) -> list[str]:
|
||||
"""Send each payload after its READY prompt; return printed key tokens."""
|
||||
src_dir = str(Path(__file__).resolve().parents[1] / "src")
|
||||
master, slave = pty.openpty()
|
||||
child = subprocess.Popen(
|
||||
[sys.executable, "-c", _CHILD_SCRIPT, str(len(sends))],
|
||||
stdin=slave,
|
||||
stdout=slave,
|
||||
stderr=subprocess.PIPE,
|
||||
env={**os.environ, "WIZARD_SRC": src_dir},
|
||||
)
|
||||
os.close(slave)
|
||||
box: dict[str, bytes] = {"stream": b""}
|
||||
def _picked(self, screen: str) -> str:
|
||||
found = _PICKED_LINE.search(screen.encode())
|
||||
if found is None:
|
||||
self.fail(f"no PICKED line in {screen[-300:]!r}")
|
||||
return found.group(1).decode()
|
||||
|
||||
try:
|
||||
tokens: list[str] = []
|
||||
for index, payload in enumerate(sends):
|
||||
self._await(
|
||||
master,
|
||||
child,
|
||||
box,
|
||||
lambda i=index: box["stream"].count(b"READY") > i,
|
||||
f"READY prompt {index} before {payload!r}",
|
||||
)
|
||||
seen = len(_KEY_LINE.findall(box["stream"]))
|
||||
os.write(master, payload)
|
||||
self._await(
|
||||
master,
|
||||
child,
|
||||
box,
|
||||
lambda n=seen: len(_KEY_LINE.findall(box["stream"])) > n,
|
||||
f"token line after {payload!r}",
|
||||
)
|
||||
tokens.append(_KEY_LINE.findall(box["stream"])[-1].decode())
|
||||
def test_down_arrow_moves_highlight_and_enter_accepts(self) -> None:
|
||||
# Start on "Alpha"; one Down must land on "Beta" — the reported bug.
|
||||
screen, _ = self.drive_menu([b"\x1b[B", b"\r"])
|
||||
self.assertEqual(self._picked(screen), "1")
|
||||
|
||||
self.assertEqual(
|
||||
box["stream"].count(b"READY"),
|
||||
len(sends),
|
||||
f"unexpected output tail {box['stream'][-300:]!r}",
|
||||
)
|
||||
self.assertEqual(len(_KEY_LINE.findall(box["stream"])), len(sends))
|
||||
try:
|
||||
child.wait(timeout=self.TIMEOUT_S)
|
||||
except subprocess.TimeoutExpired:
|
||||
child.kill()
|
||||
child.wait()
|
||||
self.fail("child ignored termination after completing every keystroke")
|
||||
err = child.stderr.read().decode("utf-8", "replace") if child.stderr else ""
|
||||
self.assertEqual(child.returncode, 0, err)
|
||||
return tokens
|
||||
finally:
|
||||
if child.poll() is None:
|
||||
child.kill()
|
||||
child.wait()
|
||||
if child.stderr is not None:
|
||||
child.stderr.close()
|
||||
os.close(master)
|
||||
def test_repeated_down_arrows_reach_third_option(self) -> None:
|
||||
screen, _ = self.drive_menu([b"\x1b[B", b"\x1b[B", b"\r"])
|
||||
self.assertEqual(self._picked(screen), "2")
|
||||
|
||||
def test_arrow_home_end_sequences_map_to_selection_keys(self) -> None:
|
||||
tokens = self._child_tokens([b"\x1b[A", b"\x1b[B", b"\x1b[H", b"\x1b[F"])
|
||||
self.assertEqual(tokens, ["up", "down", "home", "end"])
|
||||
def test_enter_without_movement_accepts_first_option(self) -> None:
|
||||
screen, _ = self.drive_menu([b"\r"])
|
||||
self.assertEqual(self._picked(screen), "0")
|
||||
|
||||
def test_single_byte_keystrokes_still_classify(self) -> None:
|
||||
# The reader switched to raw fd bytes; letters/Enter must classify as before.
|
||||
tokens = self._child_tokens([b"\r", b"k", b"j", b"x"])
|
||||
self.assertEqual(tokens, ["enter", "up", "down", "other"])
|
||||
|
||||
def test_bare_escape_is_classified_as_escape(self) -> None:
|
||||
tokens = self._child_tokens([b"\x1b"])
|
||||
self.assertEqual(tokens, ["escape"])
|
||||
|
||||
def test_escape_sequence_tail_does_not_leak_into_next_keypress(self) -> None:
|
||||
# The original report: an arrow press poisoned the following reads.
|
||||
tokens = self._child_tokens([b"\x1b[B", b"j", b"\r"])
|
||||
self.assertEqual(tokens, ["down", "down", "enter"])
|
||||
def test_escape_backs_out_of_a_cancellable_menu(self) -> None:
|
||||
screen, _ = self.drive_menu([b"\x1b"])
|
||||
self.assertEqual(self._picked(screen), "None")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user