Jean-Pierre MartineauandClaude Opus 5
deea9ebdd2
Four screens carry a find-as-you-type filter and Esc meant two things across them: the map and the mesh walk cleared the query and stayed, while the select list and the path composer left the screen, filter and all. Typing "yul" on the map then pressing Esc showed you the map; typing "yul" on a contact list then pressing Esc dropped you at the main menu. Same affordance, same glyphs, opposite outcome — and the select list is by far the most-used of the four. They peel now, all of them: the typed query is the most recent thing the reader entered, so the first Esc takes it back and the second leaves. The footer says which, swapping its trailing Esc verb for "Esc clear" while a filter stands — the word the map and the walk have always used, via a _esc_verb helper beside the two atom-splicers that already bend that line. The composer's Esc now layers exactly as its own backspace does: the entry first, then the screen. This was finding D in docs/survey-navigation.md, open since the survey was written and the last thing left telling a reader what Esc does. The rule is now stated in CLAUDE.md under Navigation. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
670 lines
27 KiB
Python
670 lines
27 KiB
Python
"""Tests for the navigation stack: strict push/pop, kept state, and the two global chords.
|
|
|
|
The app's realized path is a stack — the main menu at the bottom, one frame per screen or
|
|
dialog entered, one pop per Esc — and the machinery that guarantees it lives in
|
|
:mod:`meshterm.ui.tui.session` (``stay``/``Visit``/``run_screen``) and
|
|
:mod:`meshterm.ui.tui.screen` (``POP_ALL``/``PopToMenu``). These exercise the guarantees:
|
|
that a visit pushes and pops exactly once however many rounds it runs, that the screen
|
|
object outlives its sub-screens so its cursor and filter are still there on the way back,
|
|
and that ^W unwinds the whole stack while ^Q leaves the app — from anywhere, and never
|
|
through a dialog that is still asking something.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from prompt_toolkit.input.defaults import create_pipe_input
|
|
from prompt_toolkit.output import DummyOutput
|
|
|
|
from meshterm.ui.tui.progress import ProgressScreen
|
|
from meshterm.ui.tui.prompt import ButtonDialog
|
|
from meshterm.ui.tui.screen import CANCEL, POP_ALL, BusyScreen, PopToMenu, Screen, ScrollScreen
|
|
from meshterm.ui.tui.select import Choice, SelectScreen
|
|
from meshterm.ui.tui.session import _CTRL_LETTER_CHORDS, _KEY_ACTIONS, TuiSession
|
|
|
|
#: What the terminal actually sends for the two global chords (verified against the
|
|
#: prompt_toolkit key tables, and on the Canadian Multilingual layout with ``ToUnicodeEx``).
|
|
POP_ALL_KEY = "\x17" # ^W
|
|
QUIT_KEY = "\x11" # ^Q
|
|
|
|
#: The two keys the picker/live-screen flow is driven by, spelled out so no raw
|
|
#: control byte has to sit in the source.
|
|
ESC = "\x1b"
|
|
ENTER = "\r"
|
|
|
|
|
|
def _async_list(items):
|
|
"""A zero-argument async stub returning a fresh copy of ``items`` (a devstate read)."""
|
|
|
|
async def read():
|
|
return list(items)
|
|
|
|
return read
|
|
|
|
|
|
def _session(inp) -> TuiSession:
|
|
"""A headless session driven by a pipe, rendering nowhere."""
|
|
return TuiSession(input=inp, output=DummyOutput())
|
|
|
|
|
|
# --- the stack discipline ----------------------------------------------------
|
|
|
|
|
|
async def test_a_visit_pushes_once_and_pops_once_however_many_rounds_it_runs() -> None:
|
|
"""``stay`` is one push and one pop, not one per round.
|
|
|
|
This is the difference from ``run_screen`` in a loop, which pops on every resolve and
|
|
leaves the screen off the stack while whatever it committed to runs.
|
|
"""
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
screen = SelectScreen("hub", [Choice("a", 1), Choice("b", 2)])
|
|
depths: list[int] = []
|
|
|
|
async def main() -> None:
|
|
async with session.stay(screen) as visit:
|
|
for _ in range(3):
|
|
depths.append(len(session._stack))
|
|
screen.resolve(1)
|
|
assert await visit.result() == 1
|
|
# A sub-screen opened from inside the loop nests *above* the hub.
|
|
child = ScrollScreen("child")
|
|
session.push(child)
|
|
depths.append(len(session._stack))
|
|
session.pop(child)
|
|
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
assert depths == [1, 2, 1, 2, 1, 2]
|
|
assert session._stack == [], "the visit must pop its screen on the way out"
|
|
|
|
|
|
async def test_a_visit_keeps_the_screens_cursor_and_filter_across_a_sub_screen() -> None:
|
|
"""Coming back from a sub-screen lands on the row you left, filter and all.
|
|
|
|
Reusing the screen *object* is what buys this. A ``default=`` restore could only ever
|
|
recover the cursor, and only when the row it names still exists.
|
|
"""
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
screen = SelectScreen(
|
|
"hub", [Choice("alpha", 1), Choice("beta", 2), Choice("gamma", 3)]
|
|
)
|
|
|
|
async def main() -> None:
|
|
async with session.stay(screen) as visit:
|
|
screen.handle("text", "a") # filter to the rows carrying an "a"
|
|
screen.handle("down")
|
|
before = (screen._filter, screen._index)
|
|
screen.resolve("go")
|
|
await visit.result()
|
|
await session.run_screen(ScrollScreen("a sub-screen"))
|
|
assert (screen._filter, screen._index) == before
|
|
|
|
session_task = session.run(main())
|
|
# The sub-screen is dismissed with Esc, which is the only key this flow needs.
|
|
inp.send_text("\x1b")
|
|
await asyncio.wait_for(session_task, timeout=5)
|
|
|
|
|
|
async def test_a_visited_screen_is_armed_from_the_moment_it_is_pushed() -> None:
|
|
"""A key that lands before the caller loops back is kept, not dropped.
|
|
|
|
The screen is on the stack for the whole visit, so it can be resolved at any moment —
|
|
including while the caller is still finishing the previous round. Arming only inside
|
|
:meth:`~meshterm.ui.tui.session.Visit.result` would leave a gap in which the press has
|
|
nowhere to go.
|
|
"""
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
screen = ScrollScreen("hub")
|
|
|
|
async def main() -> None:
|
|
async with session.stay(screen) as visit:
|
|
assert screen.future is not None, "armed by the push, before any round"
|
|
screen.resolve("early") # resolved before the caller asks for it
|
|
assert await visit.result() == "early"
|
|
# And the next round is live immediately, without another await first.
|
|
assert screen.future is not None and not screen.future.done()
|
|
screen.resolve("next")
|
|
assert await visit.result() == "next"
|
|
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
|
|
|
|
async def test_the_trace_target_picker_stays_pushed_under_the_live_screen(monkeypatch) -> None:
|
|
"""Esc out of a trace is one pop, back onto the picker — not a drop to the menu.
|
|
|
|
The picker used to be a one-shot prompt gathered in ``prompt_params``: it resolved and
|
|
popped *before* the live screen opened, so the trace screen sat straight on the menu and
|
|
a single Esc skipped the list it was launched from. Now the tool keeps the picker for the
|
|
whole visit and the live screen nests above it.
|
|
"""
|
|
from meshterm.core.models import Contact
|
|
from meshterm.tools.trace import TraceTool
|
|
from meshterm.ui.surface import TuiUi
|
|
|
|
contacts = [
|
|
Contact(name="YUL-Poly", public_key="a1" + "0" * 62, key_prefix="a1"),
|
|
Contact(name="YUL-Cartierville", public_key="3d" + "0" * 62, key_prefix="3d"),
|
|
]
|
|
|
|
class _Repo:
|
|
def target_last_traced(self):
|
|
return {}
|
|
|
|
def heard_nodes(self):
|
|
return []
|
|
|
|
class _Devstate:
|
|
async def contacts(self):
|
|
return list(contacts)
|
|
|
|
class _Ctx:
|
|
repo = _Repo()
|
|
devstate = _Devstate()
|
|
is_connected = False # keeps the best-effort path-hash read off the radio
|
|
settings = type("S", (), {"connect_on_start": False})()
|
|
|
|
ctx = _Ctx()
|
|
depths: list[int] = []
|
|
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
ctx.ui = TuiUi(session)
|
|
|
|
async def fake_open_trace(_ctx, target: str, *, initial_spec: str = "") -> int:
|
|
"""Stand in for the live screen: note the stack under it, then Esc out of it."""
|
|
depths.append(len(session._stack))
|
|
inp.send_text(ESC) # the Esc that used to land on the main menu
|
|
return 2
|
|
|
|
monkeypatch.setattr("meshterm.ui.trace_screen.open_trace", fake_open_trace)
|
|
|
|
async def main() -> None:
|
|
inp.send_text(ENTER) # Enter on the leading row commits it as the target
|
|
result = await TraceTool()._run_live(ctx)
|
|
assert result.summary == {"sessions": 1, "traces": 2}
|
|
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
|
|
assert depths == [1], "the picker is still on the stack while the trace screen runs"
|
|
assert session._stack == [], "and the visit pops it on the way out"
|
|
|
|
|
|
async def _screen_at(session: TuiSession, depth: int, timeout: float = 2.0):
|
|
"""Spin the loop until the stack is exactly ``depth`` deep, then return its top screen.
|
|
|
|
Lets a test drive a nested flow by resolving each screen in turn, without depending on
|
|
when a piped keystroke happens to be read.
|
|
"""
|
|
loop = asyncio.get_running_loop()
|
|
deadline = loop.time() + timeout
|
|
while len(session._stack) != depth:
|
|
if loop.time() > deadline: # pragma: no cover - only on a regression
|
|
raise AssertionError(f"stack never reached depth {depth}: {session._stack}")
|
|
await asyncio.sleep(0)
|
|
return session._stack[-1]
|
|
|
|
|
|
async def test_the_tx_optimize_pickers_stay_pushed_under_the_sweep(monkeypatch) -> None:
|
|
"""The two-picker entry flow is a stack: sweep over target list over node list.
|
|
|
|
Both pickers used to be gathered in ``prompt_params`` and popped before the sweep
|
|
opened, so Esc from anywhere in the flow landed on the main menu — and picking a second
|
|
target meant reopening the tool and re-picking the node to tune.
|
|
"""
|
|
from meshterm.core.models import NODE_TYPE_REPEATER, Contact
|
|
from meshterm.tools.tx_optimize import TxOptimizeTool
|
|
from meshterm.ui.surface import TuiUi
|
|
|
|
contacts = [
|
|
Contact(
|
|
name="YUL-Cartierville", public_key="3d" + "0" * 62, key_prefix="3d",
|
|
node_type=NODE_TYPE_REPEATER,
|
|
),
|
|
Contact(name="YUL-Poly", public_key="a1" + "0" * 62, key_prefix="a1"),
|
|
]
|
|
|
|
class _Ctx:
|
|
admin_store = type("A", (), {"get": staticmethod(lambda _c: None)})()
|
|
devstate = type("D", (), {"contacts": staticmethod(_async_list(contacts))})()
|
|
|
|
ctx = _Ctx()
|
|
swept: list[tuple[str, str, int]] = []
|
|
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
ctx.ui = TuiUi(session)
|
|
|
|
async def fake_sweep(_ctx, *, admin_node, target_label, target_hash):
|
|
"""Stand in for the sweep screen: note who it tunes and what is under it."""
|
|
swept.append((admin_node.name, target_label, len(session._stack)))
|
|
return {"best": 20}
|
|
|
|
monkeypatch.setattr("meshterm.ui.tx_screen.open_tx_optimize", fake_sweep)
|
|
|
|
async def main() -> None:
|
|
run = asyncio.ensure_future(TxOptimizeTool()._run_live(ctx))
|
|
node_list = await _screen_at(session, 1)
|
|
node_list.resolve("YUL-Cartierville")
|
|
target_list = await _screen_at(session, 2)
|
|
target_list.resolve("YUL-Poly")
|
|
while not swept:
|
|
await asyncio.sleep(0)
|
|
# Esc from the sweep leaves the target list up; Esc there, the node list.
|
|
target_list.resolve(CANCEL)
|
|
node_list.resolve(CANCEL)
|
|
result = await asyncio.wait_for(run, timeout=2)
|
|
assert result.summary == {"best": 20}, "the last sweep's summary comes back"
|
|
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
|
|
assert swept == [("YUL-Cartierville", "YUL-Poly", 2)], "both pickers under the sweep"
|
|
assert session._stack == [], "and both visits pop on the way out"
|
|
|
|
|
|
# --- refreshing a visited list's rows ----------------------------------------
|
|
|
|
|
|
def test_replace_items_keeps_the_filter_and_follows_the_highlighted_row() -> None:
|
|
"""A list whose content is data can change under a reader without moving them.
|
|
|
|
The editors' rows carry live values and their titles count what is staged, so they have
|
|
to be rebuilt after every action. Rebuilding them as a whole new *screen* threw away the
|
|
typed filter and left the cursor to a ``default=`` restore; swapping the rows in place
|
|
keeps both, and follows the highlighted row by value wherever it moved to.
|
|
"""
|
|
screen = SelectScreen(
|
|
"Editor · nothing staged",
|
|
[Choice("alpha", "a"), Choice("beta", "b"), Choice("gamma", "g")],
|
|
)
|
|
screen.handle("text", "a") # filters to alpha / beta / gamma — all carry an "a"
|
|
screen.handle("down") # …onto beta
|
|
assert screen._current_choice().value == "b"
|
|
|
|
# beta moves to the front and its label gains a staged value; the title recounts.
|
|
screen.replace_items(
|
|
[Choice("beta → 12", "b"), Choice("alpha", "a"), Choice("gamma", "g")],
|
|
title="Editor · 1 staged",
|
|
)
|
|
assert screen.title == "Editor · 1 staged"
|
|
assert screen._filter == "a", "the typed filter survives the swap"
|
|
assert screen._current_choice().value == "b", "the highlight followed its row"
|
|
|
|
|
|
def test_replace_items_clamps_to_the_position_when_the_row_is_gone() -> None:
|
|
"""Deleting the row you were on leaves you where it was, not back at the top."""
|
|
screen = SelectScreen("Queue", [Choice(name, name) for name in ("a", "b", "c", "d")])
|
|
for _ in range(2):
|
|
screen.handle("down")
|
|
assert screen._current_choice().value == "c"
|
|
|
|
screen.replace_items([Choice(name, name) for name in ("a", "b", "d")])
|
|
assert screen._current_choice().value == "d", "the row that slid into the gap"
|
|
|
|
# And a swap that empties the list entirely must not leave a stale index behind.
|
|
screen.replace_items([])
|
|
assert screen._current_choice() is None
|
|
|
|
|
|
def test_update_rows_resorts_the_lanes_and_keeps_the_highlight_on_its_contact() -> None:
|
|
"""The Trace picker's TRACED lane ages under the reader without moving them.
|
|
|
|
The list is sorted by the very column a returning trace changes, so the node just
|
|
walked jumps to the top — and the cursor has to ride it there, or the reader would be
|
|
put back on whatever row inherited their position.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from meshterm.ui.contactlist import (
|
|
TRACE_SORT_COLUMNS,
|
|
TRACE_SORT_OPENS_ASCENDING,
|
|
ContactListScreen,
|
|
ContactRow,
|
|
)
|
|
from meshterm.ui.widgets import ContactsSort
|
|
|
|
now = datetime.now(timezone.utc)
|
|
rows = [
|
|
ContactRow(value="alpha", name="alpha", last_traced=now - timedelta(hours=2)),
|
|
ContactRow(value="beta", name="beta", last_traced=None), # never traced
|
|
]
|
|
screen = ContactListScreen(
|
|
"Trace target — pick a target",
|
|
rows=rows,
|
|
prefix_bytes=1,
|
|
sort=ContactsSort.from_name(
|
|
"traced", TRACE_SORT_COLUMNS, TRACE_SORT_OPENS_ASCENDING
|
|
),
|
|
show_traced=True,
|
|
)
|
|
screen.handle("text", "bet") # find-as-you-type down to the never-traced node
|
|
assert screen._current_choice().value == "beta"
|
|
|
|
# beta is traced: it now leads the TRACED sort, where alpha was.
|
|
screen.update_rows(
|
|
[
|
|
ContactRow(value="alpha", name="alpha", last_traced=now - timedelta(hours=2)),
|
|
ContactRow(value="beta", name="beta", last_traced=now),
|
|
]
|
|
)
|
|
assert screen._filter == "bet", "the typed filter survives the swap"
|
|
assert screen._current_choice().value == "beta", "the highlight followed its contact"
|
|
for _ in range(3): # ⌫ back out of the filter: the whole, re-sorted list is there
|
|
screen.handle("backspace")
|
|
assert [c.value for c in screen._choices()] == ["beta", "alpha"]
|
|
|
|
|
|
# --- Esc peels the filter before it leaves -----------------------------------
|
|
|
|
|
|
async def test_a_typed_filter_is_peeled_before_the_list_is_left() -> None:
|
|
"""Esc clears a standing filter and stays; the next Esc leaves.
|
|
|
|
The map and the mesh walk always did this. The select list and the path composer did
|
|
not — the same keys on the same affordance, opposite outcomes — so typing three letters
|
|
on a contact list and pressing Esc dropped you at the main menu.
|
|
"""
|
|
screen = SelectScreen("Contacts", [Choice(n, n) for n in ("alpha", "beta", "gamma")])
|
|
screen.future = asyncio.get_running_loop().create_future()
|
|
screen.handle("text", "b")
|
|
assert [c.value for c in screen._choices()] == ["beta"]
|
|
|
|
screen.handle("escape")
|
|
assert screen._filter == "", "the filter is what the first Esc peels"
|
|
assert not screen.future.done(), "…and the screen is still here"
|
|
assert [c.value for c in screen._choices()] == ["alpha", "beta", "gamma"]
|
|
|
|
screen.handle("escape")
|
|
assert screen.future.result() is CANCEL, "the second Esc leaves"
|
|
|
|
|
|
def test_the_footer_says_clear_while_a_filter_is_standing() -> None:
|
|
"""Esc's verb follows what the press will actually do — the map and walk's wording."""
|
|
screen = SelectScreen(
|
|
"Contacts",
|
|
[Choice("alpha", "a")],
|
|
footer_hint="↑↓ move · Enter open · type to filter · Esc back",
|
|
)
|
|
assert screen.footer_hint.endswith("Esc back")
|
|
screen.handle("text", "a")
|
|
assert screen.footer_hint.endswith("Esc clear")
|
|
screen.handle("backspace")
|
|
assert screen.footer_hint.endswith("Esc back")
|
|
|
|
|
|
async def test_the_path_composer_peels_its_typed_entry_first() -> None:
|
|
"""Esc unwinds the composer the way ⌫ already did: the entry, then the screen."""
|
|
from meshterm.services.topology import build_topology
|
|
from meshterm.ui.path_composer import PathComposerScreen
|
|
|
|
self_id = "aa" * 6 + "0" * 52
|
|
screen = PathComposerScreen(
|
|
device_label="Homestead",
|
|
device_hash=self_id,
|
|
topology=build_topology(
|
|
self_id=self_id, contacts=[], trace_paths=[], packet_paths=[],
|
|
neighbour_links=[],
|
|
),
|
|
width_bytes=1,
|
|
)
|
|
screen.future = asyncio.get_running_loop().create_future()
|
|
screen.handle("text", "a")
|
|
screen.handle("text", "1")
|
|
assert screen.footer_hint.endswith("Esc clear")
|
|
|
|
screen.handle("escape")
|
|
assert not screen.future.done(), "the entry is peeled, the composer stays"
|
|
assert screen.footer_hint.endswith("Esc cancel")
|
|
|
|
screen.handle("escape")
|
|
assert screen.future.result() is CANCEL
|
|
|
|
|
|
# --- ^W, the pop-all ---------------------------------------------------------
|
|
|
|
|
|
async def test_pop_all_unwinds_every_frame_at_once() -> None:
|
|
"""^W from three screens deep pops all three and lands the caller past the whole flow."""
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
popped: list[str] = []
|
|
landed: list[str] = []
|
|
|
|
async def main() -> None:
|
|
hub = ScrollScreen("hub")
|
|
try:
|
|
async with session.stay(hub) as visit:
|
|
try:
|
|
await session.run_screen(ScrollScreen("middle"))
|
|
finally:
|
|
popped.append("middle")
|
|
await visit.result()
|
|
except PopToMenu:
|
|
landed.append("menu")
|
|
finally:
|
|
popped.append("hub")
|
|
|
|
inp.send_text(POP_ALL_KEY)
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
assert landed == ["menu"], "the unwind must reach the frame that catches it"
|
|
assert popped == ["middle", "hub"], "every frame pops itself on the way out"
|
|
assert session._stack == []
|
|
|
|
|
|
async def test_pop_all_passes_straight_through_an_except_exception_guard() -> None:
|
|
"""An unwind is not a tool failure.
|
|
|
|
The app wraps tool runs in ``except Exception`` so a broken tool cannot take the menu
|
|
down with it (:func:`meshterm.ui.menu._run_selection` is the widest such guard). An
|
|
unwind has to cross those without being caught and reported as an error, which is why
|
|
:class:`PopToMenu` derives from ``BaseException``.
|
|
"""
|
|
assert issubclass(PopToMenu, BaseException)
|
|
assert not issubclass(PopToMenu, Exception)
|
|
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
swallowed: list[str] = []
|
|
landed: list[str] = []
|
|
|
|
async def a_tool() -> None:
|
|
try:
|
|
await session.run_screen(ScrollScreen("a tool's screen"))
|
|
except Exception: # noqa: BLE001 - the guard under test
|
|
swallowed.append("caught")
|
|
|
|
async def main() -> None:
|
|
try:
|
|
await a_tool()
|
|
except PopToMenu:
|
|
landed.append("menu")
|
|
|
|
inp.send_text(POP_ALL_KEY)
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
assert swallowed == []
|
|
assert landed == ["menu"]
|
|
|
|
|
|
async def test_pop_all_armed_while_nothing_is_pushed_fires_at_the_next_screen() -> None:
|
|
"""^W during a device read still unwinds — at the next screen the flow tries to open.
|
|
|
|
There is no future to carry the sentinel while the app sits under the busy overlay with
|
|
an empty stack, so the flag is what remembers the press.
|
|
"""
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
opened: list[str] = []
|
|
|
|
async def main() -> None:
|
|
inp.send_text(POP_ALL_KEY)
|
|
await asyncio.sleep(0.05) # let the key reach the dispatcher
|
|
assert session.top is None, "nothing is pushed while the read is in flight"
|
|
with pytest.raises(PopToMenu):
|
|
await session.run_screen(ScrollScreen("the screen the read was for"))
|
|
opened.append("refused")
|
|
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
assert opened == ["refused"]
|
|
|
|
|
|
def test_pop_all_is_refused_over_anything_modal() -> None:
|
|
"""A dialog is an unanswered question and progress is work in flight; ^W waits."""
|
|
session = TuiSession(output=DummyOutput())
|
|
for layer in (
|
|
ButtonDialog("Delete it?", [("Cancel", False), ("Delete", True)]),
|
|
ProgressScreen("Working"),
|
|
BusyScreen("checking…"),
|
|
):
|
|
session.push(layer)
|
|
assert session.request_pop_all() is False, f"{type(layer).__name__} must block ^W"
|
|
assert session._unwinding is False
|
|
session.pop(layer)
|
|
|
|
|
|
def test_pop_all_is_a_no_op_at_the_navigation_root() -> None:
|
|
"""You cannot go back to where you already are."""
|
|
session = TuiSession(output=DummyOutput())
|
|
menu = SelectScreen("What would you like to do?", [Choice("a tool", "a")])
|
|
session.set_root(menu)
|
|
session.push(menu)
|
|
assert session.request_pop_all() is False
|
|
assert session._unwinding is False
|
|
# One screen deeper, the same key fires.
|
|
deeper = ScrollScreen("a tool")
|
|
session.push(deeper)
|
|
assert session.request_pop_all() is True
|
|
assert deeper.future is None or True # resolving a screen with no future is harmless
|
|
|
|
|
|
def test_pop_all_resolves_the_top_screen_with_the_sentinel() -> None:
|
|
"""The sentinel travels through the future; no caller in the app ever sees it."""
|
|
session = TuiSession(output=DummyOutput())
|
|
screen = ScrollScreen("deep")
|
|
|
|
class _Fut:
|
|
def __init__(self) -> None:
|
|
self.value = None
|
|
|
|
def done(self) -> bool:
|
|
return False
|
|
|
|
def set_result(self, value: object) -> None:
|
|
self.value = value
|
|
|
|
screen.future = _Fut() # type: ignore[assignment]
|
|
session.push(screen)
|
|
assert session.request_pop_all() is True
|
|
assert screen.future.value is POP_ALL # type: ignore[union-attr]
|
|
|
|
|
|
def test_a_stack_reset_disarms_a_pending_unwind() -> None:
|
|
"""A disconnect abandons the flow itself, so an armed ^W has nothing left to unwind."""
|
|
session = TuiSession(output=DummyOutput())
|
|
session.push(ScrollScreen("deep"))
|
|
session.request_pop_all()
|
|
assert session._unwinding is True
|
|
session.reset()
|
|
assert session._unwinding is False
|
|
|
|
|
|
# --- ^Q, the quit ------------------------------------------------------------
|
|
|
|
|
|
async def test_quit_from_a_nested_screen_still_runs_the_apps_teardown() -> None:
|
|
"""^Q leaves from anywhere — and the flow's ``finally`` blocks still run.
|
|
|
|
The chord exits the prompt_toolkit application straight from the key handler, so the
|
|
coroutine driving the app is parked on a screen when it returns. Left abandoned there,
|
|
the teardown that closes the history and chat runs, stops the services and arms the exit
|
|
watchdog would never happen.
|
|
"""
|
|
with create_pipe_input() as inp:
|
|
session = _session(inp)
|
|
torn_down: list[str] = []
|
|
|
|
async def main() -> None:
|
|
try:
|
|
await session.run_screen(ScrollScreen("two screens deep"))
|
|
finally:
|
|
torn_down.append("services closed")
|
|
|
|
inp.send_text(QUIT_KEY)
|
|
await asyncio.wait_for(session.run(main()), timeout=5)
|
|
assert torn_down == ["services closed"]
|
|
|
|
|
|
# --- the bindings themselves -------------------------------------------------
|
|
|
|
|
|
def test_both_global_chords_are_bound_on_both_ctrl_keys() -> None:
|
|
"""^W and ^Q reach their action through the plain binding and the right-Ctrl rescue.
|
|
|
|
The single chord table is what guarantees the second half: a chord added there can never
|
|
be bound without its rescue, which is the only way a layout that claims the right Ctrl as
|
|
a character-group modifier reaches it at all.
|
|
"""
|
|
assert _CTRL_LETTER_CHORDS["w"] == "to_menu"
|
|
assert _CTRL_LETTER_CHORDS["q"] == "quit"
|
|
from prompt_toolkit.keys import Keys
|
|
|
|
assert _KEY_ACTIONS[Keys.ControlW] == "to_menu"
|
|
assert _KEY_ACTIONS[Keys.ControlQ] == "quit"
|
|
|
|
|
|
def test_neither_global_chord_is_advertised_anywhere() -> None:
|
|
"""Both are deliberately undiscoverable in the UI — the About page states them instead.
|
|
|
|
A global verb has no screen to belong to, so a footer atom or an F-key chip would have to
|
|
ride on *every* screen; the lane only has three free slots per screen to begin with.
|
|
"""
|
|
import ast
|
|
import re
|
|
from pathlib import Path
|
|
|
|
root = Path(__file__).resolve().parents[1] / "meshterm"
|
|
chord = re.compile(r"\^[WQ]\b")
|
|
holders = (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
|
|
offenders: list[str] = []
|
|
for path in sorted(root.rglob("*.py")):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
# Docstrings explain the chords at length and are not what a reader of the UI sees;
|
|
# what matters is every string the module *states*.
|
|
docstrings = {
|
|
id(node.body[0].value)
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, holders)
|
|
and node.body
|
|
and isinstance(node.body[0], ast.Expr)
|
|
and isinstance(node.body[0].value, ast.Constant)
|
|
and isinstance(node.body[0].value.value, str)
|
|
}
|
|
for node in ast.walk(tree):
|
|
if (
|
|
isinstance(node, ast.Constant)
|
|
and isinstance(node.value, str)
|
|
and id(node) not in docstrings
|
|
and chord.search(node.value)
|
|
):
|
|
offenders.append(f"{path.name}:{node.lineno}: {node.value.strip()[:70]}")
|
|
assert not offenders, "the global chords must not appear in any hint or chip:\n" + "\n".join(
|
|
offenders
|
|
)
|
|
|
|
|
|
def test_only_navigational_layers_are_non_modal() -> None:
|
|
"""``modal`` is about owning the keyboard, ``floating`` only about being drawn as a box.
|
|
|
|
They are easy to conflate — most dialogs are both — so the two screens that break the
|
|
correlation are asserted outright: a plain select list floats and is not modal, and the
|
|
busy splash is modal without floating at all.
|
|
"""
|
|
assert SelectScreen("list", [Choice("a", 1)]).floating is True
|
|
assert SelectScreen("list", [Choice("a", 1)]).modal is False
|
|
assert BusyScreen("checking…").floating is False
|
|
assert BusyScreen("checking…").modal is True
|
|
assert Screen().modal is False
|
|
assert CANCEL is not POP_ALL
|