# SPDX-License-Identifier: Apache-2.0 """Tests for the device-configuration registry, backup and restore, and the safety checks. All the tests run against :class:`MockDevice`, so no hardware is necessary. """ from __future__ import annotations from pathlib import Path import pytest import typer from meshterm.core.config_io import backup_config, plan_restore, read_backup from meshterm.core.connection import MockDevice from meshterm.core.device_config import ( RADIO_PRESETS, DeviceConfigError, build_snapshot, current_preset, effective_maximum, format_value, get_spec, parse_value, ) from meshterm.tools.config import _require_yes async def _connected_mock() -> MockDevice: """A MockDevice that is connected.""" device = MockDevice() await device.connect() return device # -- parsing and formatting ---------------------------------------------------- def test_parse_value_types() -> None: """Each value type converts correctly from a string.""" assert parse_value(get_spec("name"), "Node-7") == "Node-7" assert parse_value(get_spec("radio_sf"), "9") == 9 assert parse_value(get_spec("radio_freq"), "868.0") == 868.0 assert parse_value(get_spec("manual_add_contacts"), "yes") is True assert parse_value(get_spec("manual_add_contacts"), "off") is False def test_parse_value_range_and_enum_errors() -> None: """A value out of range, and an enum value that is not valid, raise DeviceConfigError.""" with pytest.raises(DeviceConfigError, match="<= 12"): parse_value(get_spec("radio_sf"), "99") with pytest.raises(DeviceConfigError, match=">= 5"): parse_value(get_spec("radio_cr"), "1") with pytest.raises(DeviceConfigError, match="one of"): parse_value(get_spec("telemetry_mode_base"), "7") with pytest.raises(DeviceConfigError, match="boolean"): parse_value(get_spec("manual_add_contacts"), "maybe") def test_format_value() -> None: """The format shows booleans, enums, and unknown values in a form that a person can read.""" assert format_value(get_spec("manual_add_contacts"), True) == "true" assert format_value(get_spec("telemetry_mode_base"), 1) == "1 (by contact)" assert format_value(get_spec("name"), None) == "?" def test_path_hash_mode_is_strict_enum() -> None: """path_hash_mode takes the three modes that the firmware accepts. It refuses mode 3.""" assert parse_value(get_spec("path_hash_mode"), "2") == 2 assert format_value(get_spec("path_hash_mode"), 0) == "0 (1-byte hashes (default))" with pytest.raises(DeviceConfigError, match="one of"): parse_value(get_spec("path_hash_mode"), "3") def test_device_pin_is_zero_or_six_digits() -> None: """The firmware refuses each other PIN, so MeshTerm refuses the value before it sends it.""" spec = get_spec("device_pin") assert parse_value(spec, "0") == 0 assert parse_value(spec, "123456") == 123456 with pytest.raises(DeviceConfigError, match="six digits"): parse_value(spec, "1234") def test_telemetry_modes_stop_at_the_firmware_s_three() -> None: """The modes are deny, by contact, and allow all. There is no fourth mode to offer.""" with pytest.raises(DeviceConfigError, match="one of"): parse_value(get_spec("telemetry_mode_env"), "3") def test_highlighted_hash_highlights_path_hash_prefix() -> None: """A key lane lights only its path hash prefix. The lane shows the full key. It lights those bytes in the palette hue that comes from the hash of the node. The name of the node has the same hue. """ from meshterm.ui.theme import node_style from meshterm.ui.widgets import highlighted_hash pub = "aabbccddee" + "00" * 27 # Mode 2 means hashes of 3 bytes, so the first 6 hex characters are the prefix that # MeshTerm can address. text = highlighted_hash(pub, prefix_bytes=3) assert text.plain == pub # the full key is shown highlighted = [s for s in text.spans if s.style == node_style(pub)] assert len(highlighted) == 1 assert text.plain[highlighted[0].start : highlighted[0].end] == "aabbcc" def test_highlighted_hash_truncates_on_a_byte_boundary() -> None: """A key that is too long for its width is truncated on a byte boundary. It keeps whole leading bytes (an even number of digits) and an ellipsis. It has padding to the exact width, so a digit of half a byte never shows. """ from meshterm.ui.widgets import highlighted_hash pub = "ab" * 32 # 64 hex digits # An even width: width - 1 is odd, so the function removes the last digit (half a byte). # The cell that this frees becomes padding. The lane shows 14 digits (7 bytes), not 15. text = highlighted_hash(pub, prefix_bytes=1, width=16) assert len(text.plain) == 16 # the lane still has the full width visible = text.plain.rstrip().rstrip("…") assert visible == "ab" * 7 and len(visible) % 2 == 0 assert "…" in text.plain # An odd width is already on a boundary: width - 1 = 14 digits, with no padding. assert highlighted_hash(pub, prefix_bytes=1, width=15).plain == "ab" * 7 + "…" def _contacts_names(counts, sort: str, prefix_bytes: int = 3): """Render a contacts table for ``sort``. Return it with the plain text of the name column.""" from meshterm.core.models import Contact from meshterm.ui.widgets import ContactsSort, contacts_table contacts = [ Contact(name="Bob", public_key="9f1a2b" + "00" * 29, key_prefix="9f1a2b"), Contact(name="Alice", public_key="3d63c6" + "00" * 29, key_prefix="3d63c6"), ] group = contacts_table( "Homestead", "aabbcc" + "00" * 29, contacts, prefix_bytes, counts, ContactsSort.from_name(sort), ) table = group.renderables[0] # (table, blank line, legend) names = [c.plain if hasattr(c, "plain") else str(c) for c in table.columns[1].cells] return table, names def test_contacts_table_lists_us_first_with_full_keys() -> None: """The contacts table pins our node first, sorts by name, and shows full lit keys.""" counts = {"3d63c6000000": 14} # Alice was heard. Bob was not heard. table, names = _contacts_names(counts, sort="name") # Column 1 is the name. Our node is first, then the contacts in alphabetical order # (Alice before Bob). assert names[0].startswith("Homestead") assert names[1] == "Alice" assert names[2] == "Bob" # Packet counts (column 3): Alice has her count, and Bob has the em dash. pkts = [c.plain if hasattr(c, "plain") else str(c) for c in table.columns[3].cells] assert pkts[1] == "14" assert pkts[2] == "—" # Keys (column 4) are the full key with the path hash prefix lit. The model does not # truncate. Rich truncates with an ellipsis only at render time, when the terminal is # too narrow. from meshterm.ui.theme import node_style keys = list(table.columns[4].cells) assert keys[1].plain == "3d63c6" + "00" * 29 # The prefix is lit in the hue that comes from the hash of the key. assert any(s.style == node_style("3d63c6") for s in keys[1].spans) def test_contacts_table_sorts_by_heard_and_packets() -> None: """A sort that is not the default changes the order of the contacts. Our node stays first.""" from datetime import timedelta from meshterm.core.models import Contact, utcnow # MeshTerm heard Bob more recently than Alice, but MeshTerm heard more packets from Alice. contacts = [ Contact( name="Bob", public_key="9f1a2b" + "00" * 29, key_prefix="9f1a2b", last_seen=utcnow() - timedelta(minutes=5), ), Contact( name="Alice", public_key="3d63c6" + "00" * 29, key_prefix="3d63c6", last_seen=utcnow() - timedelta(days=2), ), ] counts = {"3d63c6000000": 14, "9f1a2b000000": 3} from meshterm.ui.widgets import ContactsSort, contacts_table def names(sort: ContactsSort): group = contacts_table("Us", "aabbcc" + "00" * 29, contacts, 3, counts, sort) cells = group.renderables[0].columns[1].cells return [c.plain if hasattr(c, "plain") else str(c) for c in cells] # Most recent first: Bob (5m) before Alice (2d) in the default heard sort (ascending age). assert names(ContactsSort.from_name("heard"))[1:] == ["Bob", "Alice"] # Most packets first: Alice (14) before Bob (3) in the default packets sort (descending). assert names(ContactsSort.from_name("packets"))[1:] == ["Alice", "Bob"] # Descending changes the order: the contact heard longest ago is first, so Alice (2d) # is above Bob (5m). assert names(ContactsSort(column="heard", ascending=False))[1:] == ["Alice", "Bob"] def test_contacts_table_breaks_metric_ties_by_name_ascending() -> None: """Two contacts with the same metric keep the A→Z order in both sort directions.""" from datetime import timedelta from meshterm.core.models import Contact, utcnow from meshterm.ui.widgets import ContactsSort, ordered_contacts same = utcnow() - timedelta(minutes=5) contacts = [ Contact(name="Charlie", public_key="aa" * 16, last_seen=same), Contact(name="Alice", public_key="bb" * 16, last_seen=same), Contact(name="Bob", public_key="cc" * 16, last_seen=utcnow() - timedelta(hours=3)), ] def order(ascending: bool) -> list[str]: sort = ContactsSort(column="heard", ascending=ascending) return [c.name for c in ordered_contacts(contacts, {}, sort)] # Most recent first: the two contacts of 5 minutes are first, in the order Alice→Charlie, # then Bob (3h). assert order(ascending=True) == ["Alice", "Charlie", "Bob"] # Oldest first: Bob is first, but the tie still breaks Alice→Charlie, not Charlie→Alice. assert order(ascending=False) == ["Bob", "Alice", "Charlie"] async def test_contacts_tool_opens_sorted_by_heard(monkeypatch: pytest.MonkeyPatch) -> None: """The Contacts list opens with the most recently heard contact first. The user sees who is on the mesh now. The list is not a roll call. """ from meshterm.core.models import Contact from meshterm.tools.contacts import ContactsTool from meshterm.ui import widgets captured: dict = {} real_order = widgets.ordered_contacts def fake_order(contacts, counts, sort): # noqa: ANN001, ANN202 captured["sort"] = sort return real_order(contacts, counts, sort) monkeypatch.setattr(widgets, "ordered_contacts", fake_order) class _Devstate: async def self_info(self) -> dict: return {"name": "Us", "public_key": "aa" * 32} async def contacts(self) -> list: return [Contact(name="Alice", public_key="d4" * 32)] async def path_hash_mode(self) -> int: return 0 class _Ctx: devstate = _Devstate() repo = type("R", (), {"heard_nodes": staticmethod(lambda: [])})() ui = type("U", (), {"show": staticmethod(lambda *r: None)})() await ContactsTool().run(_Ctx(), {}) # type: ignore[arg-type] assert (captured["sort"].column, captured["sort"].ascending) == ("heard", True) def _contacts_sort(name: str = "name"): """The sort of the interactive Contacts screen: the ring of four columns of the shared list.""" from meshterm.ui.contactlist import SORT_COLUMNS, SORT_OPENS_ASCENDING from meshterm.ui.widgets import ContactsSort return ContactsSort.from_name(name, SORT_COLUMNS, SORT_OPENS_ASCENDING) def test_contacts_screen_ctrl_arrows_steer_the_sort() -> None: """The Ctrl+arrow keys control the Contacts sort. Ctrl+←/→ move through the shared ring of four columns, and the ring wraps. Ctrl+↑/↓ set the direction. These are the keys of the Time Machine picker, and the Contacts screen now uses them too. """ from meshterm.ui.contacts_screen import ContactsScreen screen = ContactsScreen("Us", "aabbcc" + "00" * 29, [], 3, {}, _contacts_sort()) assert (screen._sort.column, screen._sort.ascending) == ("name", True) screen.handle("ctrl_right") # name -> heard, which opens in its natural (ascending) direction assert (screen._sort.column, screen._sort.ascending) == ("heard", True) screen.handle("ctrl_right") # heard -> packets, which opens descending assert (screen._sort.column, screen._sort.ascending) == ("packets", False) screen.handle("ctrl_up") # set ascending assert screen._sort.ascending is True screen.handle("ctrl_down") # set descending assert screen._sort.ascending is False screen.handle("ctrl_right") # packets -> hash, the fourth column of the ring assert (screen._sort.column, screen._sort.ascending) == ("hash", True) screen.handle("ctrl_right") # hash -> wraps to name assert screen._sort.column == "name" screen.handle("ctrl_left") # name -> wraps to hash assert screen._sort.column == "hash" def test_contacts_screen_lists_contacts_in_the_shared_lanes() -> None: """The screen lists contacts in the shared lanes, with our node pinned first. The lanes are NAME, HEARD, PKTS, and KEY. The arrow keys with no modifier only move the highlight, and Enter resolves the highlighted node. """ import re from meshterm.core.models import Contact, utcnow from meshterm.ui.contacts_screen import _ARCHIVE, YOU, ContactsScreen from meshterm.ui.tui.screen import CANCEL contacts = [ Contact(name="Alice", public_key="aa" * 32, last_seen=utcnow()), Contact(name="Bob", public_key="bb" * 32), ] counts = {"aa" * 6: 7} # MeshTerm heard Alice. It never heard Bob. screen = ContactsScreen("Us", "cc" * 32, contacts, 1, counts, _contacts_sort()) body = "\n".join(re.sub(r"\x1b\[[0-9;]*m", "", ln) for ln in screen.render_body(72)) # The shared column header (it is stored as sticky, so it pins when the page scrolls # past it)... assert "NAME" in body and "HEARD" in body and "PKTS" in body and "KEY" in body assert screen._sticky_headers # ...our node is first in the lanes, then the contacts with their counts. choices = screen._choices() assert choices[0].value == YOU assert "(you)" in choices[0].label.plain assert "Alice" in body and "Bob" in body assert f"{7:>5}" in body # the packets heard from Alice, right-aligned in the lane assert "never" in body # Bob has no last_seen # Each contact row has the Contact itself, so Enter gives the whole record to the caller. # The list ends with the archive action (there is no exit row, because Esc leaves). assert choices[-1].value == _ARCHIVE assert all(isinstance(c.value, Contact) for c in choices[1:-1]) # The arrow keys with no modifier move the highlight and do not change the sort. before = (screen._sort.column, screen._sort.ascending) screen.handle("down") assert (screen._sort.column, screen._sort.ascending) == before highlighted = screen._current_choice().value assert highlighted != YOU and isinstance(highlighted, Contact) # Enter resolves the highlighted contact (open_contacts opens its Node detail). # Esc still leaves with CANCEL. resolved: list = [] screen.resolve = lambda value: resolved.append(value) # type: ignore[method-assign] screen.handle("enter") assert resolved == [highlighted] screen.handle("escape") assert resolved == [highlighted, CANCEL] def test_archive_age_rungs_split_stale_from_never_heard() -> None: """An age rung takes only contacts that were heard and are now quiet. Never-heard has its own rung. The second section of the ladder is a plain operation with a predictable result, which a ranking cannot express. One point needs care: a contact with no advert time must not be in it. A contact that the user added a short time ago has not yet had time to send an advert, so it looks infinitely old. If an age rung swept it, the sweep deleted nodes because they are new. """ from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import Contact from meshterm.ui.sweep_screen import _DAY, _NEVER, victims_for def scored(name: str, days, score: float) -> ScoredContact: # noqa: ANN001 contact = Contact(name=name, public_key=f"{ord(name[0]):02x}" * 32) return ScoredContact( contact=contact, signals=ContactSignals(node="0" * 12, heard_age_days=days), score=score, percentile=int(score), ) # Strongest first, in the order that `rank_contacts` returns. ranked = [ scored("Fresh", 0.04, 90.0), scored("Quiet", 3.0, 70.0), scored("Old", 45.0, 50.0), scored("Ancient", 400.0, 30.0), scored("Unheard", None, 10.0), ] names = lambda picked: [v.contact.name for v in victims_for(ranked, ranked, picked)] # noqa: E731 # A week takes the two old contacts. It does not take the fresh contact or the # never-heard contact. assert set(names(("age", 7 * _DAY))) == {"Old", "Ancient"} # A year takes only the very old contact. assert names(("age", 365 * _DAY)) == ["Ancient"] # The never-heard group is only the contact with no advert time. assert names(("age", _NEVER)) == ["Unheard"] # Weakest first on the age route also, so a user who looks quickly at the first screen # of the preview sees the contacts that the user is least likely to want back. assert names(("age", 7 * _DAY)) == ["Ancient", "Old"] async def test_archive_sweeps_under_a_progress_bar_and_reports_in_a_dialog() -> None: """The whole sweep, from start to end: ladder → preview → amber confirm → bar → outcome dialog. The busy overlay could not do two things (JP, 2026-08-10). First, it only said "working", and only in the gaps between screens. Over the contacts list that was pushed, it drew nothing. Thus a long sweep looked like a screen that did not answer, and the arrow keys still moved a highlight where nothing was in progress. Second, the count of the removed contacts went to a note that the user saw only when the user left the screen. Thus the sweep now uses the progress dialog of the app (it swallows each key while it exists) while it runs. When the sweep is done, a dialog that the user can close shows the count. """ import asyncio import logging from datetime import timedelta from types import SimpleNamespace from meshterm.core.contact_score import ContactSignals from meshterm.core.models import Contact, utcnow from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import archive_contacts from meshterm.ui.tui.progress import ProgressScreen from meshterm.ui.tui.session import TuiSession old = utcnow() - timedelta(days=400) victims = [ Contact( name=f"Old{i}", public_key=f"{i:02x}" * 32, key_prefix=f"{i:02x}" * 6, last_seen=old ) for i in range(3) ] removed_from_device: list[str] = [] archived: list[str] = [] class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 await asyncio.sleep(0) # a command of a real companion takes one turn of the loop removed_from_device.append(contact.name) class _Repo: @staticmethod def contact_signals(nodes): # noqa: ANN001 # Silent for a long time, no messages, heard a few times. This is the shape of # the contacts that the sweep exists to remove. return { node: ContactSignals(node=node, heard_age_days=400.0, packets=2, known_days=420.0) for node in nodes } @staticmethod def channel_post_counts(): return {} session = TuiSession() ui = TuiUi(session) asked: list[dict] = [] async def _dialog(prompt, buttons, **kw): # noqa: ANN001 asked.append({"prompt": prompt, "buttons": buttons, **kw}) return True ui.dialog = _dialog # type: ignore[method-assign] ctx = SimpleNamespace( ui=ui, log=logging.getLogger("test.archive"), repo=_Repo(), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, archive=lambda dev, contact, when: archived.append(contact.name), ), watch_store=None, admin_store=None, devstate=SimpleNamespace( contacts=lambda: _contacts(victims), self_info=lambda: _self_info({}), invalidate_contacts=lambda: None, ), device=lambda: _device(_Device()), ) task = asyncio.ensure_future(archive_contacts(ctx, "cc" * 32)) # Move past the five standing rungs into the second section and select "not heard in 1 # year". It takes all three contacts. The standing rungs keep a share, so none of them # takes a whole table. ladder = await _step_until_screen(session, lambda s: s.title.startswith("Archive contacts")) body = _screen_text(ladder) # A rung has columns for its counts. What it keeps is the number that must fit the # device, and it is before what the rung archives. assert body.index("KEEPS") < body.index("ARCHIVES") for _ in range(9): ladder.handle("down") ladder.handle("enter") # The preview lists exactly the contacts that will go, and it commits on its Apply row # (the default). preview = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) body = _screen_text(preview) # The evidence has one lane for each kind. It is never a score, and it is not a # percentile now: the row shows what MeshTerm measured, and the order of the list is the # ranking. lanes = ["NAME", "HEARD", "PKTS", "MSGS", "HOPS"] assert [body.index(word) for word in lanes] == sorted(body.index(word) for word in lanes) assert all(v.name in body for v in victims) preview.handle("enter") # The confirm is amber and has only buttons. The action is reversible, so it does not use # the reserved red or a typed word. bar = await _step_until_screen(session, lambda s: isinstance(s, ProgressScreen)) assert asked and asked[0]["danger"] is True assert "destructive" not in asked[0] assert [label for label, _v in asked[0]["buttons"]] == ["Cancel", "Archive"] assert "restored" in asked[0]["prompt"] # The sweep runs under the progress dialog, which is the top screen for its whole # lifetime. Thus the user cannot move in the list under it while the sweep works. assert bar.title == "Archive contacts" assert bar.render_body(60) # a real bar, with a real total to count down bar.handle("down") # the dialog swallows each key, so nothing under it moves # …and the count is in a dialog, not in a note that the user reads when leaving. done = await _step_until_screen( session, lambda s: not isinstance(s, ProgressScreen) and "archived" in _screen_text(s) ) assert "archived 3 contacts" in _screen_text(done) done.handle("escape") assert await task == 3 assert sorted(removed_from_device) == ["Old0", "Old1", "Old2"] # The contacts are off the device, but MeshTerm keeps them: each victim is in the store as # archived. assert sorted(archived) == ["Old0", "Old1", "Old2"] def test_a_preview_row_leads_with_the_contacts_type_mark() -> None: """Each contact that the sweep takes starts with its type mark, as in the Contacts list. The mark is a lane of its own before the name. It has the colour of the type, not the colour of the name. The header moves with it: ``NAME`` is over the name, and each evidence word is still over its values. ``←→`` keeps the mark pinned beside its name. """ from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import NODE_TYPE_REPEATER, Contact from meshterm.ui.sweep_screen import ( _MARK_W, _NAME_W, _lane_widths, _preview_header, _preview_items, _victim_row, ) from meshterm.ui.theme import name_style def scored(name: str, key: str, node_type: int | None) -> ScoredContact: return ScoredContact( contact=Contact(name=name, public_key=key, key_prefix=key[:12], node_type=node_type), signals=ContactSignals(node=key[:12], packets=3), score=1.0, percentile=1, ) victims = [scored("Hilltop", "3d" * 32, NODE_TYPE_REPEATER), scored("Walker", "5e" * 32, None)] widths = _lane_widths(victims) hilltop, walker = (_victim_row(v, widths) for v in victims) assert hilltop.plain.startswith("▲ Hilltop") assert walker.plain.startswith("● Walker") # it never sent a type in an advert: the plain node def styles_at(row, index: int) -> list[str]: # noqa: ANN001 return [str(span.style) for span in row.spans if span.start <= index < span.end] assert styles_at(hilltop, 0) == ["type.repeater"] assert styles_at(hilltop, hilltop.plain.index("Hilltop")) == [name_style("Hilltop", "3d" * 32)] # The header starts at the edge of the pointer column, and a row starts just after it. header = _preview_header(widths, 72) assert header.index("NAME") == 2 + hilltop.plain.index("Hilltop") assert header.index("PKTS") + len("PKTS") - 1 == 2 + hilltop.plain.index("3") items = _preview_items(victims) rows = [c for c in items if isinstance(getattr(c, "value", None), ScoredContact)] assert len(rows) == 2 and all(c.hscroll_from == _MARK_W + _NAME_W for c in rows) async def test_sparing_a_row_shrinks_the_sweep_without_leaving_the_preview() -> None: """Delete on a preview row removes that contact from the sweep, in place. Before, the user had to go back and select a shallower rung. This spared the one contact that the user knew, but it also spared forty contacts that the user did not care about. The list refreshes through ``replace_items`` and is not rebuilt. Thus the user keeps the place in the list, and the list does not go back to the top when the user was halfway down. The contact that the user removes is also removed from the contacts that the sweep archives. """ import asyncio from types import SimpleNamespace from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import Contact from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import _preview from meshterm.ui.tui.session import TuiSession def scored(name: str, pct: int) -> ScoredContact: key = f"{ord(name[-1]):02x}" * 32 return ScoredContact( contact=Contact(name=name, public_key=key, key_prefix=key[:12]), signals=ContactSignals(node=key[:12]), score=float(pct), percentile=pct, ) victims = [scored("Weak", 3), scored("Middling", 9), scored("Keeper", 14)] session = TuiSession() ctx = SimpleNamespace(ui=TuiUi(session)) task = asyncio.ensure_future(_preview(ctx, victims)) screen = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) assert "3 contacts to archive" in screen.title # The user can spare each contact row. The user cannot spare the Apply and Back pair, # because a decision has nothing to remove. The percentile lane is pinned, so the ←→ # scroll does not move it. It is the place of the user in a list that the percentile # orders. rows = [c for c in screen._choices() if isinstance(c.value, ScoredContact)] assert len(rows) == 3 and all(c.deletable and c.hscroll_from > 0 for c in rows) assert not any(c.deletable for c in screen._choices() if not isinstance(c.value, ScoredContact)) # The highlight starts on Apply (the row that commits is the default). Move to the top of # the contacts and down to "Keeper", then spare it. The screen stays open, because this # is an edit and not an exit. screen.handle("home") screen.handle("down") screen.handle("down") assert screen._current_choice().value.contact.name == "Keeper" screen.handle("delete") # Wait for the loop to take the delete. Then make sure that the refresh went to the same # screen object and not to a rebuilt one, because this keeps the place of the user. same = await _step_until_screen( session, lambda s: "2 contacts to archive" in getattr(s, "title", "") ) assert same is screen, "the preview refreshed in place rather than being rebuilt" body = _screen_text(screen) assert "Keeper" not in body assert "Weak" in body and "Middling" in body assert "Archive 2 contacts" in body # the Apply row counts again what is left # If the user commits now, the sweep archives exactly the two contacts that stayed after # the edit. End goes to Back, and one step up is Apply. screen.handle("end") screen.handle("up") assert screen._current_choice().value == ("apply",) screen.handle("enter") assert await task is True assert [v.contact.name for v in victims] == ["Weak", "Middling"] async def test_sparing_every_row_leaves_the_sweep_with_nothing_to_do() -> None: """If the user empties the preview, it goes back. It does not confirm an empty archive.""" import asyncio from types import SimpleNamespace from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import Contact from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import _preview from meshterm.ui.tui.session import TuiSession key = "aa" * 32 only = ScoredContact( contact=Contact(name="Solo", public_key=key, key_prefix=key[:12]), signals=ContactSignals(node=key[:12]), score=1.0, percentile=1, ) victims = [only] session = TuiSession() task = asyncio.ensure_future(_preview(SimpleNamespace(ui=TuiUi(session)), victims)) screen = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) screen.handle("home") # away from the default Apply row and onto the only contact screen.handle("delete") assert await task is False assert victims == [] def _victim(name: str, pct: int = 1): # noqa: ANN202 """A scored contact for the preview, with a full key made from its last letter.""" from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import Contact key = f"{ord(name[-1]):02x}" * 32 return ScoredContact( contact=Contact(name=name, public_key=key, key_prefix=key[:12]), signals=ContactSignals(node=key[:12]), score=float(pct), percentile=pct, ) async def test_shift_delete_stages_a_lock_and_apply_returns_it() -> None: """Shift+Delete removes the row, as Delete does, and it also stages a lock on the contact. The lock is staged, not written (JP, 2026-10-19: "on apply"). The preview adds the contact to the list of locks of the caller, and the caller writes the locks on Apply. The Apply row counts the locks, and the Back row tells that Back discards them too. """ import asyncio from types import SimpleNamespace from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import _preview from meshterm.ui.tui.session import TuiSession victims = [_victim("Weak", 3), _victim("Middling", 9), _victim("Keeper", 14)] locks: list = [] session = TuiSession() task = asyncio.ensure_future(_preview(SimpleNamespace(ui=TuiUi(session)), victims, locks)) screen = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) screen.handle("home") screen.handle("down") screen.handle("down") assert screen._current_choice().value.contact.name == "Keeper" screen.handle("shift_delete") same = await _step_until_screen( session, lambda s: "2 contacts to archive" in getattr(s, "title", "") ) assert same is screen, "the preview refreshed in place rather than being rebuilt" assert [v.contact.name for v in victims] == ["Weak", "Middling"] assert [v.contact.name for v in locks] == ["Keeper"] body = _screen_text(screen) assert "Keeper" not in body assert "Archive 2 contacts and lock 1" in body assert "Back — change nothing" in body # Delete on a row still spares it only: it stages no lock. screen.handle("home") screen.handle("delete") await _step_until_screen(session, lambda s: "1 contact to archive" in getattr(s, "title", "")) assert [v.contact.name for v in locks] == ["Keeper"] screen.handle("end") screen.handle("up") assert screen._current_choice().value == ("apply",) screen.handle("enter") assert await task is True assert [v.contact.name for v in victims] == ["Middling"] async def test_locking_every_row_leaves_a_preview_that_only_locks() -> None: """When the user locks the last victim, the preview stays, because Apply still has work. A preview with no victims and no locks goes back (refer to the test above). A preview with no victims but with staged locks has a real choice left: it shows an empty state in place of the rows, and its Apply row writes only the locks. """ import asyncio from types import SimpleNamespace from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import _preview from meshterm.ui.tui.session import TuiSession victims = [_victim("Solo")] locks: list = [] session = TuiSession() task = asyncio.ensure_future(_preview(SimpleNamespace(ui=TuiUi(session)), victims, locks)) screen = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) screen.handle("home") screen.handle("shift_delete") await _step_until_screen(session, lambda s: "no contacts to archive" in getattr(s, "title", "")) body = _screen_text(screen) assert "no contacts left to archive" in body assert "Lock 1 contact" in body screen.handle("end") screen.handle("up") screen.handle("enter") assert await task is True assert victims == [] and [v.contact.name for v in locks] == ["Solo"] async def test_esc_with_staged_locks_asks_before_it_discards_them() -> None: """Esc with staged locks asks first, as an editor with staged values does. "Keep editing" stays on the same list. "Discard" leaves, and the caller writes nothing. Esc with only spared rows does not ask, because a spared row is not a staged change. """ import asyncio from types import SimpleNamespace from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import _preview from meshterm.ui.tui.session import TuiSession victims = [_victim("Weak"), _victim("Keeper")] locks: list = [] session = TuiSession() ui = TuiUi(session) answers = iter(["keep", "discard"]) asked: list[str] = [] async def _dialog(prompt, buttons, **kw): # noqa: ANN001 asked.append(prompt) return next(answers) ui.dialog = _dialog # type: ignore[method-assign] task = asyncio.ensure_future(_preview(SimpleNamespace(ui=ui), victims, locks)) screen = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) screen.handle("home") screen.handle("shift_delete") await _step_until_screen(session, lambda s: "1 contact to archive" in getattr(s, "title", "")) screen.handle("escape") await _step_until_screen(session, lambda s: len(asked) == 1 and s is screen) assert "Discard 1 staged change" in asked[0] assert not task.done(), "Keep editing stays on the preview" screen.handle("escape") assert await task is False assert len(asked) == 2 def test_the_preview_names_the_lock_key_only_where_the_hint_has_room() -> None: """The ``⇧Del lock`` atom shows on a contact row only when the hint stays in 72 cells. JP, 2026-10-19: "If footer too long, don't put it in the footer. If there's room, put it." No other atom leaves the hint to make room for it. A contact row also says ``Enter open``, because Enter opens the node page there. """ from rich.cells import cell_len from meshterm.ui.sweep_screen import _preview_screen victims = [_victim("Weak"), _victim("Keeper")] screen = _preview_screen(victims, []) screen.render_body(72) assert "⇧Del lock" not in screen.footer_hint # the highlight starts on Apply assert "Enter select" in screen.footer_hint screen.handle("home") hint = screen.footer_hint assert "Enter open" in hint assert hint.index("Del keep") < hint.index("⇧Del lock") < hint.index("Esc back") assert cell_len(hint) <= 72 # A typed filter changes the Esc verb, and the hint still has room. screen.handle("text", "w") assert "⇧Del lock" in screen.footer_hint and "Esc clear" in screen.footer_hint # At a width where the row overflows, ←→ scroll takes the room, and the lock atom goes. screen.handle("escape") screen.render_body(30) hint = screen.footer_hint assert "←→ scroll" in hint and "Del keep" in hint assert "⇧Del lock" not in hint # A caller that cannot store a lock gives no lock key at all. plain = _preview_screen(victims) plain.handle("home") assert "⇧Del lock" not in plain.footer_hint resolved: list = [] plain.resolve = resolved.append # type: ignore[method-assign] plain.handle("shift_delete") assert resolved == [] def test_the_preview_lane_puts_lock_behind_the_delete_chip() -> None: """On a handheld, Shift+F3 is ``Lock``, behind the ``Delete`` chip on F3. The chip is dim where the key does nothing (the Apply and Back rows). The border of the floating preview does not repeat ``Del keep`` or ``⇧Del lock``, because the lane already shows both keys. """ from meshterm.ui.sweep_screen import _preview_screen from meshterm.ui.tui.fkeys import strip_lane_atoms screen = _preview_screen([_victim("Weak"), _victim("Keeper")], []) slot = screen.picocalc_lyra_lane[2] assert (slot.label, slot.opp_label, slot.opp_action) == ("Delete", "Lock", "shift_delete") assert slot.opp_enabled is False # the highlight starts on Apply screen.handle("home") lane = screen.picocalc_lyra_lane assert lane[2].opp_enabled is True border = strip_lane_atoms(screen.footer_hint, lane) assert "Del keep" not in border and "⇧Del lock" not in border # Without a lock store, the slot has no Shift half. assert _preview_screen([_victim("Weak")]).picocalc_lyra_lane[2].opp_label == "" def test_shift_delete_reaches_the_screens_from_each_front_end() -> None: """The terminal, the Windows console, and the emulator all deliver ``shift_delete``. xterm spells Shift+Delete as ``CSI 3;2 ~``, and the emulator types the same bytes. prompt_toolkit reads them as ``ShiftDelete``, and the session maps it to the action. """ from prompt_toolkit.input.vt100_parser import Vt100Parser from prompt_toolkit.keys import Keys from meshterm.emulator.keys import Key, encode from meshterm.ui.tui.session import _KEY_ACTIONS spelled = encode(Key(name="delete", shift=True)) assert spelled == "\x1b[3;2~" parsed: list = [] parser = Vt100Parser(parsed.append) parser.feed_and_flush(spelled) assert [press.key for press in parsed] == [Keys.ShiftDelete] assert _KEY_ACTIONS[Keys.ShiftDelete] == "shift_delete" async def test_the_sweep_writes_staged_locks_after_the_confirm() -> None: """Apply writes each staged lock, and the confirm and the outcome both say so. With victims, the amber confirm also names the locks, and Cancel writes nothing. With only locks, there is no confirm and no device work: a lock archives nothing. """ from contextlib import contextmanager from types import SimpleNamespace from meshterm.ui.sweep_screen import _sweep locked: list[str] = [] archived: list[str] = [] removed: list[str] = [] outcomes: list[str] = [] answers: list[bool] = [] prompts: list[str] = [] class _Progress: def add_task(self, *args, **kwargs) -> int: # noqa: ANN002, ANN003 return 0 def advance(self, *args) -> None: # noqa: ANN002 pass @contextmanager def _progress(title): # noqa: ANN001, ANN202 yield _Progress() async def _dialog(prompt, buttons, **kw): # noqa: ANN001 prompts.append(prompt) return answers.pop(0) async def _message(outcome, **kw): # noqa: ANN001 outcomes.append(outcome.plain) class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 removed.append(contact.name) ctx = SimpleNamespace( ui=SimpleNamespace( dialog=_dialog, progress=_progress, session=SimpleNamespace(message_dialog=_message), ), log=SimpleNamespace(debug=lambda *a: None), contact_store=SimpleNamespace( set_locked=lambda dev, contact, flag: locked.append(contact.name), archive=lambda dev, contact, when: archived.append(contact.name), ), devstate=SimpleNamespace(invalidate_contacts=lambda: None), device=lambda: _device(_Device()), ) # Cancel on the confirm writes nothing, the locks included. answers.append(False) assert await _sweep(ctx, "cc" * 32, [_victim("Weak")], [_victim("Keeper")]) == 0 assert "also locks 1 contact" in prompts[-1] assert locked == [] and removed == [] # Archive writes the lock and archives the victim. answers.append(True) assert await _sweep(ctx, "cc" * 32, [_victim("Weak")], [_victim("Keeper")]) == 1 assert locked == ["Keeper"] and removed == ["Weak"] and archived == ["Weak"] assert outcomes[-1] == "✓ archived 1 contact and locked 1" # Only locks: no confirm, no device. asked = len(prompts) assert await _sweep(ctx, "cc" * 32, [], [_victim("Solo")]) == 0 assert len(prompts) == asked assert locked == ["Keeper", "Solo"] and removed == ["Weak"] assert outcomes[-1] == "✓ locked 1 contact" @pytest.mark.parametrize("platform_name", ["regular", "picocalc-lyra"]) def test_the_archive_preview_keeps_its_cursor_in_its_box_walking_both_ways( platform_name: str, ) -> None: """The highlight of the preview stays in its box, for each distance that the user moves. A long sweep scrolls. It is a list with a prompt over a pinned column header. Two faults once let the highlight leave the window on the way back up. First, MeshTerm counted the prompt two times in the line that the frame keeps visible, so the real row was above the top edge of the box. Second, at the bottom, the slide of the header removed the row that the user had just moved to. The test goes through the real dialog compositor on each platform, at the readable width of that platform. """ from rich.text import Text from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import Contact from meshterm.platforms import PICOCALC_LYRA, REGULAR, set_platform from meshterm.ui.sweep_screen import _preview_screen from meshterm.ui.tui.frame import compose_dialog platform = {"regular": REGULAR, "picocalc-lyra": PICOCALC_LYRA}[platform_name] set_platform(platform) def scored(index: int) -> ScoredContact: key = f"{index:02x}" * 32 return ScoredContact( contact=Contact(name=f"Faraway {index}", public_key=key, key_prefix=key[:12]), signals=ContactSignals(node=key[:12], heard_age_days=40.0 + index, packets=index), score=float(index), percentile=index, ) screen = _preview_screen([scored(i) for i in range(40)]) cols, rows = platform.readable_cols, 24 # The highlight starts on Apply, at the bottom. Move to the first contact, down to Back, # and up again. Each step is a paint that the user sees. walk = [None] + ["up"] * 45 + ["down"] * 45 + ["up"] * 45 for step, key in enumerate(walk): if key is not None: screen.handle(key) box = Text.from_ansi(compose_dialog(screen, cols, rows)).plain assert "❯" in box, f"step {step} ({key}): the highlight left the box\n{box}" async def test_deleting_one_contact_drops_it_from_the_device_and_the_store() -> None: """The delete of one contact: a red confirm, then MeshTerm forgets both parts of the union. The list on a screen is the union of the contact table of the device and the contacts that MeshTerm remembers for that device. A removal that reached only the radio merges back at the next read, and the screen looks as if it did nothing. The delete also removes the cached contacts, so the list that the user returns to reads them again. """ import logging from types import SimpleNamespace from meshterm.core.models import Contact from meshterm.ui.node_detail_screen import _remove_contact from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession hub = Contact(name="Hub", public_key="3d" * 32, key_prefix="3d" * 6) removed: list[str] = [] forgotten: list[tuple[str, str]] = [] invalidated: list[bool] = [] class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 removed.append(contact.name) ui = TuiUi(TuiSession()) asked: list[dict] = [] async def _dialog(prompt, buttons, **kw): # noqa: ANN001 asked.append({"prompt": prompt, "buttons": buttons, **kw}) return True # the button that commits ui.dialog = _dialog # type: ignore[method-assign] ctx = SimpleNamespace( ui=ui, log=logging.getLogger("test.remove"), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, forget=lambda dev, key: forgotten.append((dev, key)), ), devstate=SimpleNamespace(invalidate_contacts=lambda: invalidated.append(True)), device=lambda: _device(_Device()), ) assert await _remove_contact(ctx, hub, "cc" * 32, "Hub") is True # The confirm is the delete of one record in the app: red, with Cancel on the left and # the verb that commits on the right, as the default. It says clearly that this delete is # final. It also points to the archive, which is the reversible way to free the same slot # on the device. assert asked[0]["destructive"] is True assert [label for label, _v in asked[0]["buttons"]] == ["Cancel", "Delete"] assert asked[0]["default"] == 1 prompt = asked[0]["prompt"] assert "Hub" in prompt and "for good" in prompt and "restored" in prompt assert "archive it instead" in prompt # Both parts of the union, then the cache that the list reads again. assert removed == ["Hub"] assert forgotten == [("cc" * 32, "3d" * 32)] assert invalidated == [True] async def test_cancelling_the_remove_confirm_touches_nothing() -> None: """Esc or Cancel on the confirm leaves the contact on the device and in the caller's list.""" import logging from types import SimpleNamespace from meshterm.core.models import Contact from meshterm.ui.node_detail_screen import _remove_contact from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession touched: list[str] = [] class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 touched.append(contact.name) ui = TuiUi(TuiSession()) async def _declined(*a, **k): # noqa: ANN001, ANN002, ANN003 return None # the result that Esc gives for a button dialog ui.dialog = _declined # type: ignore[method-assign] ctx = SimpleNamespace( ui=ui, log=logging.getLogger("test.remove"), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, forget=lambda *a: touched.append("forgot"), ), devstate=SimpleNamespace(invalidate_contacts=lambda: touched.append("invalidated")), device=lambda: _device(_Device()), ) assert ( await _remove_contact(ctx, Contact(name="Hub", public_key="3d" * 32), "cc" * 32, "Hub") is False ) assert touched == [] async def test_a_refused_removal_is_shown_and_the_contact_stays() -> None: """A dialog reports a device that refuses. The row must not go away because of a false result. Only the device can remove a contact from its table. If the device refuses and MeshTerm forgets its own part of the union, the list hides a contact that the radio still has. Thus MeshTerm forgets nothing, the dialog shows the failure to the user, and the page stays open on the node. """ import asyncio import logging from types import SimpleNamespace from meshterm.core.models import Contact from meshterm.ui.node_detail_screen import _remove_contact from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession forgotten: list = [] class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 await asyncio.sleep(0) raise RuntimeError("contact table is busy") session = TuiSession() ui = TuiUi(session) async def _accepted(*a, **k): # noqa: ANN001, ANN002, ANN003 return True ui.dialog = _accepted # type: ignore[method-assign] ctx = SimpleNamespace( ui=ui, log=logging.getLogger("test.remove"), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, forget=lambda *a: forgotten.append(a), ), devstate=SimpleNamespace(invalidate_contacts=lambda: forgotten.append("cache")), device=lambda: _device(_Device()), ) task = asyncio.ensure_future( _remove_contact(ctx, Contact(name="Hub", public_key="3d" * 32), "cc" * 32, "Hub") ) failed = await _step_until_screen(session, lambda s: "couldn't delete" in _screen_text(s)) assert "contact table is busy" in _screen_text(failed) failed.handle("escape") assert await task is False assert forgotten == [] # the device still has it, so MeshTerm does not forget it async def test_a_contact_the_device_never_held_is_still_removed_here() -> None: """“Not on the device” is not a refusal. The removal finishes on our side, and says so. The list that a screen removes from is the union of the table of the device and the contacts that MeshTerm remembers for it. The entry can exist only on our side, or the firmware can remove it after MeshTerm read the table. When the removal failed in this case, the user could not remove the row. Now the user sees that the device had no part in the removal, MeshTerm forgets our part, and the visit ends on the contact like each other removal. """ import asyncio import logging from types import SimpleNamespace from meshterm.core.connection import ContactNotOnDeviceError from meshterm.core.models import Contact from meshterm.ui.node_detail_screen import _remove_contact from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession hub = Contact(name="Hub", public_key="3d" * 32, key_prefix="3d" * 6) forgotten: list[tuple[str, str]] = [] invalidated: list[bool] = [] class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 await asyncio.sleep(0) raise ContactNotOnDeviceError(contact) session = TuiSession() ui = TuiUi(session) async def _accepted(*a, **k): # noqa: ANN001, ANN002, ANN003 return True ui.dialog = _accepted # type: ignore[method-assign] ctx = SimpleNamespace( ui=ui, log=logging.getLogger("test.remove"), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, forget=lambda dev, key: forgotten.append((dev, key)), ), devstate=SimpleNamespace(invalidate_contacts=lambda: invalidated.append(True)), device=lambda: _device(_Device()), ) task = asyncio.ensure_future(_remove_contact(ctx, hub, "cc" * 32, "Hub")) told = await _step_until_screen(session, lambda s: "wasn't in this device" in _screen_text(s)) text = _screen_text(told) assert "Hub" in text and "⚠" in text # a warning, not the ✗ for a command that failed told.handle("escape") assert await task is True assert forgotten == [("cc" * 32, "3d" * 32)] assert invalidated == [True] async def _self_info(info): # noqa: ANN001, ANN201 """Wait and return the given self-info dictionary (the device read that the ranking makes).""" return info async def _true() -> bool: return True async def _contacts(rows): # noqa: ANN001 return list(rows) async def _device(device): # noqa: ANN001 return device def _screen_text(screen) -> str: # noqa: ANN001 """All the text of a screen that is pushed, as one plain string.""" import re body = screen.render_body(60) lines = body if isinstance(body, list) else list(body) drawn = [line() if callable(line) else line for line in lines] return re.sub(r"\[[0-9;]*m", "", " ".join(drawn) + " " + str(screen.title)) async def _step_until_screen(session, predicate, *, limit: int = 500): # noqa: ANN001 """Yield to the loop until the top screen satisfies ``predicate``. Then return that screen.""" import asyncio for _ in range(limit): top = session.top if top is not None and predicate(top): return top await asyncio.sleep(0) raise AssertionError("the expected screen never reached the top of the stack") async def test_the_list_rebuilds_when_the_detail_page_deletes_the_contact( monkeypatch: pytest.MonkeyPatch, ) -> None: """A contact that the user removes on its own page is not in the list that the user returns to. The detail page is where the user deletes one contact, so the list must know. It gives a contact to the page and gets back whether that contact stayed. If it did not stay, the list reads the device again and rebuilds. Without the rebuild, the user returns to a row for a contact that the radio no longer has. """ import asyncio import logging from types import SimpleNamespace from meshterm.core.models import Contact from meshterm.ui import node_detail_screen from meshterm.ui.contacts_screen import ContactsScreen, open_contacts from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession alice = Contact(name="Alice", public_key="aa" * 32) bob = Contact(name="Bob", public_key="bb" * 32) remaining = [alice, bob] detailed: list = [] async def _fake_detail(ctx, contact): # noqa: ANN001 """Replace the page: report that it deleted the contact that it received.""" detailed.append(contact) remaining.remove(contact) return True monkeypatch.setattr(node_detail_screen, "open_node_detail", _fake_detail) session = TuiSession() ctx = SimpleNamespace( ui=TuiUi(session), log=logging.getLogger("test.contacts"), contact_store=None, devstate=SimpleNamespace(contacts=lambda: _contacts(remaining)), ) task = asyncio.ensure_future( open_contacts(ctx, "Us", "cc" * 32, list(remaining), 1, {}, _contacts_sort()) ) listed = await _step_until_screen(session, lambda s: isinstance(s, ContactsScreen)) assert "2 known" in listed.title listed.handle("down") # off our own node, onto the first contact listed.handle("enter") # The page deleted it, so the list that returns is a new list, read again without it. rebuilt = await _step_until_screen( session, lambda s: isinstance(s, ContactsScreen) and "1 known" in s.title ) assert rebuilt is not listed assert detailed == [alice] assert "Alice" not in _screen_text(rebuilt) assert "Bob" in _screen_text(rebuilt) rebuilt.handle("escape") await task def test_contacts_screen_tail_offers_archive_only_when_populated() -> None: """A list with contacts ends with the archive action. An empty list has no tail.""" from meshterm.core.models import Contact from meshterm.ui.contacts_screen import _ARCHIVE, ContactsScreen populated = ContactsScreen( "Us", "cc" * 32, [Contact(name="Alice", public_key="aa" * 32)], 1, {}, _contacts_sort() ) assert populated._choices()[-1].value == _ARCHIVE empty = ContactsScreen("Us", "cc" * 32, [], 1, {}, _contacts_sort()) values = [c.value for c in empty._choices()] assert _ARCHIVE not in values # nothing to archive, so no action row def test_the_archived_row_appears_only_when_something_is_archived() -> None: """The row for the archived list, with its count, shows only when there is an archived list. A screen must not offer a way into an empty list. The count is on the row, so the user can read it without opening the list. The user can see the output of the sweep on the screen of the sweep, and thus the user does not lose track of the archive. """ from meshterm.core.models import Contact from meshterm.ui.contacts_screen import _ARCHIVE, _ARCHIVED, ContactsScreen contacts = [Contact(name="Alice", public_key="aa" * 32)] none_yet = ContactsScreen("Us", "cc" * 32, contacts, 1, {}, _contacts_sort()) assert _ARCHIVED not in [c.value for c in none_yet._choices()] some = ContactsScreen("Us", "cc" * 32, contacts, 1, {}, _contacts_sort(), archived=12) values = [c.value for c in some._choices()] # It is under the archive action, because the output of the sweep goes there. assert values[-2:] == [_ARCHIVE, _ARCHIVED] row = next(c for c in some._choices() if c.value == _ARCHIVED) assert "View archived contacts" in row.label and "12" in row.label # It is alone for a device where the sweep took the whole table. There is nothing left # to archive, but there is much for the user to look at. swept = ContactsScreen("Us", "cc" * 32, [], 1, {}, _contacts_sort(), archived=3) values = [c.value for c in swept._choices()] assert _ARCHIVE not in values and values[-1] == _ARCHIVED def test_contacts_title_counts_the_archived_and_abbreviates_where_narrow() -> None: """The title has both counts. It removes the words before it removes a number. If nothing is archived, the title stays the plain ``… known`` and has no short form. If something is archived, the full ``198 known + 123 archived`` shows where it fits, and ``198 + 123`` shows where the rule is too narrow. This is true on both frames: the panel with a border and the title bar of one row. """ from meshterm.core.models import Contact from meshterm.ui.contacts_screen import ContactsScreen from meshterm.ui.tui.frame import _title_bar, fitted_title contacts = [Contact(name=f"n{i}", public_key=f"{i:02x}" * 32) for i in range(198)] plain = ContactsScreen("Us", "cc" * 32, contacts, 1, {}, _contacts_sort()) assert plain.title == "Contacts · 198 known" and not plain.short_title both = ContactsScreen("Us", "cc" * 32, contacts, 1, {}, _contacts_sort(), archived=123) assert both.title == "Contacts · 198 known + 123 archived" assert both.short_title == "Contacts · 198 + 123" assert fitted_title(both, 72) == both.title assert fitted_title(both, 30) == both.short_title assert "198 known + 123 archived" in _title_bar(both, 53, False, False).plain assert "Contacts · 198 + 123 " in _title_bar(both, 40, False, False).plain async def test_archiving_one_contact_takes_it_off_the_device_and_keeps_it() -> None: """The archive of one contact: an amber confirm, off the radio, and a stamp in the store. This is the reversible action of the pair that the detail page offers. It has ``danger`` and not ``destructive``, and it does not ask for a typed word, because the user can recover all that it takes. This is the reason that the app has two caution tiers. """ import logging from types import SimpleNamespace from meshterm.core.models import NODE_TYPE_REPEATER, Contact from meshterm.ui.node_detail_screen import _archive_contact from meshterm.ui.surface import TuiUi from meshterm.ui.theme import name_style from meshterm.ui.tui.session import TuiSession hub = Contact( name="Hub", public_key="3d" * 32, key_prefix="3d" * 6, node_type=NODE_TYPE_REPEATER ) removed: list[str] = [] archived: list[tuple[str, str]] = [] invalidated: list[bool] = [] class _Device: async def remove_contact(self, contact) -> None: # noqa: ANN001 removed.append(contact.name) ui = TuiUi(TuiSession()) asked: list[dict] = [] async def _dialog(prompt, buttons, **kw): # noqa: ANN001 asked.append({"prompt": prompt, "buttons": buttons, **kw}) return True ui.dialog = _dialog # type: ignore[method-assign] ctx = SimpleNamespace( ui=ui, log=logging.getLogger("test.archive"), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, archive=lambda dev, contact, when: archived.append((dev, contact.public_key)), ), devstate=SimpleNamespace(invalidate_contacts=lambda: invalidated.append(True)), device=lambda: _device(_Device()), ) assert await _archive_contact(ctx, hub, "cc" * 32, "Hub") is True assert asked[0]["danger"] is True and "destructive" not in asked[0] assert [label for label, _v in asked[0]["buttons"]] == ["Cancel", "Archive"] assert asked[0]["default"] == 1 prompt = asked[0]["prompt"] assert "restore it at any time" in prompt # The contact has the name that the Contacts list gives it: its type mark is to the left # of the name, and each has its own colour. The sentence around them keeps the amber. assert prompt.plain.startswith("Archive ▲ Hub? ") def styles_at(index: int) -> list[str]: return [str(span.style) for span in prompt.spans if span.start <= index < span.end] assert styles_at(prompt.plain.index("▲")) == ["type.repeater"] assert styles_at(prompt.plain.index("Hub")) == [name_style("Hub", "3d" * 32)] assert styles_at(0) == styles_at(prompt.plain.index("restore")) == ["warn"] assert not prompt.style, "a base style would merge the amber's bold into the mark" # The contact is off the radio and kept here. MeshTerm removes the cached contact list, # so the list reads again. assert removed == ["Hub"] assert archived == [("cc" * 32, "3d" * 32)] assert invalidated == [True] async def test_restoring_writes_the_contact_back_before_clearing_the_mark() -> None: """A refused write leaves the contact archived. The list never shows the contact as live on a radio that does not have it. """ import logging from types import SimpleNamespace from meshterm.core.connection import DeviceCommandError from meshterm.core.models import Contact from meshterm.ui.node_detail_screen import _restore_archived from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession hub = Contact(name="Hub", public_key="3d" * 32, key_prefix="3d" * 6) restored: list[str] = [] class _Refuses: async def add_contact(self, contact) -> None: # noqa: ANN001 raise DeviceCommandError("contact table full") session = TuiSession() ctx = SimpleNamespace( ui=TuiUi(session), log=logging.getLogger("test.restore"), contact_store=SimpleNamespace( locked_keys=lambda *a: frozenset(), is_locked=lambda *a: False, restore=lambda dev, key: restored.append(key), ), devstate=SimpleNamespace(invalidate_contacts=lambda: None), device=lambda: _device(_Refuses()), ) import asyncio task = asyncio.ensure_future(_restore_archived(ctx, hub, "cc" * 32, "Hub")) shown = await _step_until_screen(session, lambda s: "couldn't restore" in _screen_text(s)) shown.handle("escape") assert await task is False assert restored == [], "the archive mark survives a write the device refused" def test_non_strict_enum_accepts_unlisted_value() -> None: """multi_acks and adv_loc_policy list the common values but still accept other integers.""" spec = get_spec("multi_acks") assert spec.value_type == "enum" and not spec.strict_choices assert parse_value(spec, "5") == 5 # not in choices, but >= minimum: accepted with pytest.raises(DeviceConfigError, match=">= 0"): parse_value(spec, "-1") # the range is still checked def test_radio_presets_map_to_radio_settings() -> None: """Each preset gives the four radio fields with their registry keys.""" assert RADIO_PRESETS, "expected at least one standard radio preset" for preset in RADIO_PRESETS: settings = preset.as_settings() assert {"radio_freq", "radio_bw", "radio_sf", "radio_cr"} <= set(settings) # Each value parses with no error through its setting spec. for key, value in settings.items(): assert parse_value(get_spec(key), value) == value def test_radio_presets_match_meshcore_list() -> None: """Check some of the presets against the settings that MeshCore suggests. The values come from MeshCore, not from MeshTerm. A mesh hears a node only when the four radio parameters of the node match. Thus a table that drifted gives a radio that talks to nobody. These are the presets that a MeshTerm user is most likely to select. They include the two regions that MeshCore now marks as deprecated. They stay in the list, because nodes that the user never tuned again use them. """ by_name = {p.name: p for p in RADIO_PRESETS} expected = { "EU/UK (Narrow)": (869.618, 62.5, 8, 8), "EU/UK (Deprecated)": (869.525, 250.0, 11, 5), "USA/Canada (Recommended)": (910.525, 62.5, 7, 5), "Australia": (915.800, 250.0, 10, 5), "New Zealand (Narrow)": (917.375, 62.5, 7, 5), } for name, values in expected.items(): preset = by_name[name] assert (preset.freq, preset.bw, preset.sf, preset.cr) == values # A region that names a path hash size stages it as the mode of the firmware (size - 1). assert by_name["New Zealand (Narrow)"].as_settings()["path_hash_mode"] == 1 assert "path_hash_mode" not in by_name["EU/UK (Narrow)"].as_settings() def test_radio_preset_summary_reads_as_meshcore_prints_it() -> None: """The parameter lane of the row has the order and the precision that MeshCore uses.""" by_name = {p.name: p for p in RADIO_PRESETS} assert by_name["USA/Canada (Recommended)"].summary == "910.525 SF7 BW62.5 CR5" assert by_name["Australia"].summary == "915.800 SF10 BW250 CR5" async def test_current_preset_identifies_the_mock_radio() -> None: """The mock starts with the firmware defaults. These are EU/UK (Narrow), but not the CR.""" snapshot = await build_snapshot(await _connected_mock()) assert current_preset(snapshot) is None # CR5, where the EU preset asks CR8 assert current_preset({**snapshot, "radio_cr": 8}).name == "EU/UK (Narrow)" def test_current_preset_none_for_unmatched_radio() -> None: """A radio that the user tuned by hand matches no preset. A missing field does not crash it.""" assert ( current_preset({"radio_freq": 868.0, "radio_bw": 250.0, "radio_sf": 11, "radio_cr": 5}) is None ) assert current_preset({}) is None def test_get_spec_unknown_raises() -> None: """An unknown key gives an error that lists the valid keys.""" with pytest.raises(DeviceConfigError, match="unknown setting"): get_spec("not_a_setting") # -- apply to the mock device -------------------------------------------------- async def test_apply_simple_setting() -> None: """The node name that MeshTerm sets comes back from the device with no change.""" device = await _connected_mock() spec = get_spec("name") await spec.apply(device, parse_value(spec, "Yagi-Hub"), await build_snapshot(device)) assert (await device.get_self_info())["name"] == "Yagi-Hub" async def test_apply_coupled_radio_field_preserves_others() -> None: """A change of one radio field builds set_radio again from the current snapshot.""" device = await _connected_mock() spec = get_spec("radio_sf") await spec.apply(device, 9, await build_snapshot(device)) info = await device.get_self_info() assert info["radio_sf"] == 9 assert info["radio_freq"] == 869.618 # unchanged assert info["radio_bw"] == 62.5 assert info["radio_cr"] == 5 async def test_apply_telemetry_mode_preserves_siblings() -> None: """An edit of one telemetry mode does not change the other two.""" device = await _connected_mock() spec = get_spec("telemetry_mode_loc") await spec.apply(device, 2, await build_snapshot(device)) info = await device.get_self_info() assert info["telemetry_mode_loc"] == 2 assert info["telemetry_mode_base"] == 0 assert info["telemetry_mode_env"] == 0 async def test_custom_var_round_trip() -> None: """The device keeps custom variables.""" device = await _connected_mock() await device.set_custom_var("exp_flag", "1") assert (await device.get_custom_vars())["exp_flag"] == "1" # -- backup and restore -------------------------------------------------------- async def test_backup_and_read_round_trip(tmp_path: Path) -> None: """A backup writes all the settings, custom variables, and channels. Then it reads them back.""" device = await _connected_mock() await device.set_name("Foo") await device.set_channel(0, "Public", b"\x01" * 16) snapshot = await build_snapshot(device) path = backup_config(tmp_path / "cfg.toml", snapshot, {"exp": "1"}, await _channels(device)) backup = read_backup(path) assert backup.settings["name"] == "Foo" assert backup.settings["radio_sf"] == 8 assert backup.custom["exp"] == "1" assert backup.channels[0]["name"] == "Public" assert backup.channels[0]["secret"] == ("01" * 16) async def test_plan_restore_emits_only_differences(tmp_path: Path) -> None: """The restore planner returns operations only for values that are different from the device.""" source = await _connected_mock() await source.set_name("Foo") path = backup_config(tmp_path / "cfg.toml", await build_snapshot(source), {"exp": "1"}, []) backup = read_backup(path) target = await _connected_mock() # a new device: the name is still "MockCompanion" ops = plan_restore(backup, await build_snapshot(target), await target.get_custom_vars()) assert ("set", "name", "Foo") in ops assert ("set_custom", "exp", "1") in ops # Values that did not change (for example radio_sf 8 == 8) are not applied again. assert "radio_sf" not in [op[1] for op in ops if op[0] == "set"] # -- the TX power ceiling that the device reports ------------------------------- def test_tx_power_capped_by_device_reported_maximum() -> None: """With a snapshot, the max_tx_power of the board makes the static bound tighter.""" spec = get_spec("tx_power") snapshot = {"max_tx_power": 22} assert parse_value(spec, "22", snapshot) == 22 with pytest.raises(DeviceConfigError, match="<= 22"): parse_value(spec, "23", snapshot) # With no snapshot, only the static ceiling applies. assert parse_value(spec, "27") == 27 assert effective_maximum(spec, snapshot) == 22 assert effective_maximum(spec, None) == 30 def test_effective_maximum_never_loosens_the_static_bound() -> None: """A device that reports a maximum above the static ceiling does not raise the ceiling.""" spec = get_spec("tx_power") assert effective_maximum(spec, {"max_tx_power": 99}) == 30 assert effective_maximum(spec, {"max_tx_power": "bogus"}) == 30 # unparseable -> static # -- flood scope, auto-add, and TX delays ---------------------------------------- def test_flood_scope_length_limited_and_formats_empty() -> None: """The scope name keeps to its protocol slot of 31 bytes. An empty name shows readably.""" spec = get_spec("flood_scope") assert parse_value(spec, "alpha") == "alpha" with pytest.raises(DeviceConfigError, match="at most 30"): parse_value(spec, "x" * 31) assert format_value(spec, "") == "(not set)" assert format_value(spec, "alpha") == "alpha" with pytest.raises(DeviceConfigError, match="no spaces or commas"): parse_value(spec, "north shore") with pytest.raises(DeviceConfigError, match="30 bytes"): parse_value(spec, "é" * 16) # 16 characters, 32 bytes async def test_flood_scope_round_trips_with_hashtag_normalization() -> None: """A scope name is stored bare, as the firmware and apps store it. An empty name clears it.""" device = await _connected_mock() spec = get_spec("flood_scope") await spec.apply(device, "#alpha", await build_snapshot(device)) assert (await build_snapshot(device))["flood_scope"] == "alpha" await spec.apply(device, "", await build_snapshot(device)) assert (await build_snapshot(device))["flood_scope"] == "" async def test_autoadd_config_round_trips() -> None: """The auto-add bitmask goes to the device and comes back into the snapshot.""" device = await _connected_mock() spec = get_spec("autoadd_config") await spec.apply(device, parse_value(spec, "5"), await build_snapshot(device)) assert (await build_snapshot(device))["autoadd_config"] == 5 def test_tuning_fields_are_floats_with_firmware_ranges() -> None: """RX delay and airtime factor take real (not scaled) floats, with the firmware limits. The firmware stores both as floats (rx_delay_base 0–20 s, airtime_factor 0–9). Only the wire carries them ×1000. The settings must use the real units. """ assert parse_value(get_spec("rx_delay"), "0.5") == 0.5 assert parse_value(get_spec("airtime_factor"), "2.5") == 2.5 with pytest.raises(DeviceConfigError, match="<= 20"): parse_value(get_spec("rx_delay"), "500") # MeshTerm must reject a raw wire value with pytest.raises(DeviceConfigError, match="<= 9"): parse_value(get_spec("airtime_factor"), "1000") def test_rx_delay_is_rounded_to_one_place() -> None: """RX delay shows and applies with one decimal, as the delays of the repeater admin do.""" spec = get_spec("rx_delay") assert parse_value(spec, "0.3333") == 0.3 assert format_value(spec, 0.123) == "0.1" assert format_value(spec, 0.0) == "0.0" def test_tx_delay_factors_are_not_companion_settings() -> None: """The TX delay factors of repeaters must not appear as device settings. They do nothing. The CMD_SET_TUNING_PARAMS of the companion firmware reads only rx_delay and airtime_factor, and it ignores the bytes after them. Thus a setting that MeshTerm offers for these factors does nothing and gives no warning. """ with pytest.raises(DeviceConfigError, match="unknown setting"): get_spec("tx_delay_factor") with pytest.raises(DeviceConfigError, match="unknown setting"): get_spec("direct_tx_delay_factor") async def test_apply_tuning_field_preserves_the_other() -> None: """An edit of one tuning field sends the pair again and does not overwrite the other field.""" device = await _connected_mock() await device.set_tuning(0.5, 2.0) spec = get_spec("airtime_factor") await spec.apply(device, 3.5, await build_snapshot(device)) assert await device.get_tuning() == {"rx_delay": 0.5, "airtime_factor": 3.5} # -- clock sync ------------------------------------------------------------------- async def test_mock_clock_drifts_until_set_time_corrects_it() -> None: """The clock of the simulator is behind the host until set_time anchors it again.""" import time as _time device = await _connected_mock() before = await device.get_time() assert before is not None and before < int(_time.time()) # the drift that the mock starts with now = int(_time.time()) await device.set_time(now) after = await device.get_time() assert after is not None and abs(after - now) <= 1 # -- the Device info status panel ------------------------------------------------- async def test_status_panel_reports_role_battery_clock_and_stats() -> None: """The status panel of the info tool shows the read-only state of the device in one place.""" import io from rich.console import Console from meshterm.tools.info import _status_panel from meshterm.ui.theme import MESH_THEME device = await _connected_mock() panel = await _status_panel(device, await build_snapshot(device)) console = Console(theme=MESH_THEME, file=io.StringIO(), width=100) console.print(panel) out = console.file.getvalue() assert "companion" in out # adv_type 1, rendered as the node role assert "MeshCore Simulator" in out assert "4.10 V" in out assert "1d 2h 3m" in out # 93784 s of simulated uptime assert "s behind" in out # the clock drift that the mock sets at the start assert "-110 dBm noise floor" in out assert "210 sent" in out and "1234 received" in out def test_uptime_renders_compact_units() -> None: """Uptime shows only seconds under a minute. Then it shows the d, h, and m parts as needed.""" from meshterm.tools.info import _uptime assert _uptime(45) == "45 s" assert _uptime(3660) == "1h 1m" assert _uptime(93784) == "1d 2h 3m" assert _uptime(86400) == "1d 0m" # -- safety checks ------------------------------------------------------------- def test_require_yes_blocks_without_confirmation() -> None: """Destructive CLI commands stop unless the user passes --yes. The error is a usage error. A missing confirmation is a bad command line. Thus the exit status is the status of the parser, and the message is in the words of the parser on stderr. It is not a red line of its own on stdout (refer to :mod:`meshterm.core.exitcodes`). """ with pytest.raises(typer.BadParameter, match="--yes"): _require_yes(False, "factory reset erases all data") _require_yes(True, "factory reset erases all data") # does not raise async def _channels(device: MockDevice) -> list[dict]: """Get the configured channels of the mock for a backup.""" channels = [] for idx in range(8): ch = await device.get_channel(idx) if ch: channels.append(ch) return channels async def test_the_preview_opens_node_pages_with_no_management_verbs( monkeypatch: pytest.MonkeyPatch, ) -> None: """Enter on an archive candidate opens its page so that the user can look at it, not act on it. The rule (JP, 2026-09-01): the management verbs belong to the list where a contact lives. They never belong to a page that opens from a list that is about to act on the contact in bulk. The only job of a preview is to choose what to archive. It must not also offer an archive of one contact, or a delete, from inside its own list of candidates. These are two answers to one question. """ import asyncio from types import SimpleNamespace import meshterm.ui.node_detail_screen as nd from meshterm.core.contact_score import ContactSignals, ScoredContact from meshterm.core.models import Contact from meshterm.ui.surface import TuiUi from meshterm.ui.sweep_screen import _preview from meshterm.ui.tui.session import TuiSession opened: list[dict] = [] async def _fake_detail(ctx, contact, *, manage=True): # noqa: ANN001 opened.append({"name": contact.name, "manage": manage}) return False monkeypatch.setattr(nd, "open_node_detail", _fake_detail) key = "aa" * 32 victims = [ ScoredContact( contact=Contact(name="Solo", public_key=key, key_prefix=key[:12]), signals=ContactSignals(node=key[:12]), score=1.0, percentile=1, ) ] session = TuiSession() task = asyncio.ensure_future(_preview(SimpleNamespace(ui=TuiUi(session)), victims)) screen = await _step_until_screen(session, lambda s: "to archive" in getattr(s, "title", "")) screen.handle("home") # away from the default Apply row and onto the contact screen.handle("enter") await _step_until_screen(session, lambda s: bool(opened)) assert opened == [{"name": "Solo", "manage": False}] # The preview is still open under it. To look at a candidate is not to leave the sweep. assert session.top is screen screen.handle("escape") assert await task is False def test_a_locked_contact_leads_its_row_with_a_padlock_left_of_its_type_glyph() -> None: """A locked row starts with the padlock. Other rows keep the column blank, so glyphs line up.""" from meshterm.core.models import Contact from meshterm.ui.contacts_screen import ContactsScreen contacts = [ Contact(name="Al", public_key="aa" * 32), Contact(name="Bartholomew", public_key="bb" * 32), ] screen = ContactsScreen( "Us", "dd" * 32, contacts, 1, {}, _contacts_sort(), locked=frozenset({"aa" * 32}) ) screen.note_viewport(30) screen.render_body(72) lines = { name: choice.title.plain for choice in screen._choices() for name in ("Al", "Bartholomew") if getattr(choice.value, "name", None) == name } assert lines["Al"].startswith("\U0001f512 ● Al ") assert lines["Bartholomew"].startswith(" ● Bartholomew") # If the user unlocks a contact, the list paints again in place. A list with no locks does # not use cells for the column. screen.refresh_locks(frozenset()) screen.render_body(72) al = next(c for c in screen._choices() if getattr(c.value, "name", None) == "Al") assert al.title.plain.startswith("● Al ") async def test_a_locked_contact_is_never_offered_to_the_sweep() -> None: """The ranking reads the locks in the store, so a locked contact is protected, not a victim.""" import logging from types import SimpleNamespace from meshterm.core.contact_score import PROTECT_LOCKED, ContactSignals from meshterm.core.models import Contact from meshterm.ui.sweep_screen import _rank contacts = [ Contact(name="Kept", public_key="aa" * 32, key_prefix="aa" * 6), Contact(name="Loose", public_key="bb" * 32, key_prefix="bb" * 6), ] class _Repo: def contact_signals(self, nodes): # noqa: ANN001, ANN202 return { n: ContactSignals(node=n, heard_age_days=400.0, known_days=500.0) for n in nodes } def channel_post_counts(self): # noqa: ANN202 return {} async def _self_info(): # noqa: ANN202 return {} ctx = SimpleNamespace( repo=_Repo(), log=logging.getLogger("test.lock"), contact_store=SimpleNamespace(locked_keys=lambda dev: frozenset({"aa" * 32})), watch_store=None, admin_store=None, devstate=SimpleNamespace(self_info=_self_info), ) ranked = {s.contact.name: s for s in await _rank(ctx, contacts, "cc" * 32)} assert ranked["Kept"].protection == PROTECT_LOCKED assert not ranked["Loose"].protected async def test_contacts_screen_l_opens_the_map_on_the_highlighted_node() -> None: """^L asks for the map on the highlighted node. The hint and F6 name it only where it acts.""" from meshterm.core.models import Contact, MapMarker from meshterm.ui.contacts_screen import ContactsScreen from meshterm.ui.tui import KeyRequest, fkeys from tests.conftest import recording_locator contacts = [ Contact(name="Alice", public_key="aa" * 32), Contact(name="Bob", public_key="bb" * 32), ] alice = MapMarker(label="Alice", lat=45.5, lon=-73.6, key="aa" * 32) screen = ContactsScreen( "Us", "cc" * 32, contacts, 1, {}, _contacts_sort(), locator=recording_locator([alice]) ) resolved: list = [] screen.resolve = lambda value: resolved.append(value) # type: ignore[method-assign] # Our node is first, and it has no location here. assert "^L" not in screen.footer_hint lane = screen.picocalc_lyra_lane assert (lane[0].label, lane[0].opp_label) == ("Sect ↑", "Map") # Map is Shift+F1 (F6) assert lane[0].opp_enabled is False screen.handle("locate") assert resolved == [] screen.handle("down") # Alice assert "^L map" in screen.footer_hint assert screen.picocalc_lyra_lane[0].opp_enabled is True assert fkeys.PICOCALC_LYRA_DECK.action_for(screen.picocalc_lyra_lane, 6) == "locate" screen.handle("locate") assert resolved == [KeyRequest("locate", alice)] screen.handle("down") # Bob, with no location assert "^L" not in screen.footer_hint def test_k_on_the_contacts_list_asks_to_toggle_the_lock_of_the_highlighted_contact() -> None: """^K on a contact resolves a lock request, and Shift+F2 is ``Lock`` or ``Unlock``. It is the key of the chat (JP, 2026-10-19). Our node and the tail rows have nothing to lock. The hint names ^K only where it has room: a contact with a location already has ``^L map``, and then the hint is too full for it. """ from rich.cells import cell_len from meshterm.core.models import Contact, MapMarker from meshterm.ui.contacts_screen import ContactsScreen from meshterm.ui.tui import KeyRequest, fkeys from tests.conftest import recording_locator contacts = [ Contact(name="Alice", public_key="aa" * 32), Contact(name="Bob", public_key="bb" * 32), Contact(name="Carol", public_key="dd" * 32), ] alice = MapMarker(label="Alice", lat=45.5, lon=-73.6, key="aa" * 32) screen = ContactsScreen( "Us", "cc" * 32, contacts, 1, {}, _contacts_sort(), locked=frozenset({"bb" * 32}), locator=recording_locator([alice]), can_lock=True, ) resolved: list = [] screen.resolve = lambda value: resolved.append(value) # type: ignore[method-assign] # Our node: nothing to lock, and the chip is dim. lane = screen.picocalc_lyra_lane assert (lane[1].opp_label, lane[1].opp_action, lane[1].opp_enabled) == ("Lock", "lock", False) assert "^K" not in screen.footer_hint screen.handle("lock") assert resolved == [] # Alice has a location, so the hint has no room for ^K. The chip and the key still act. screen.handle("down") assert "^L map" in screen.footer_hint and "^K" not in screen.footer_hint lane = screen.picocalc_lyra_lane assert lane[1].opp_enabled is True assert fkeys.PICOCALC_LYRA_DECK.action_for(lane, 7) == "lock" screen.handle("lock") assert resolved == [KeyRequest("lock", contacts[0])] # Bob is locked: the chip says Unlock. screen.handle("down") assert screen.picocalc_lyra_lane[1].opp_label == "Unlock" # Carol has no location and no lock: the hint has room for ^K. screen.handle("down") assert "^K lock" in screen.footer_hint assert cell_len(screen.footer_hint) <= 72 # A lock that the list learns of changes the chip in place. screen.refresh_locks(frozenset({"dd" * 32})) assert screen.picocalc_lyra_lane[1].opp_label == "Unlock" # Where MeshTerm cannot store a lock, the Shift half of F2 is empty, and ^K does nothing. plain = ContactsScreen("Us", "cc" * 32, contacts, 1, {}, _contacts_sort()) plain.handle("down") slot = plain.picocalc_lyra_lane[1] assert slot is None or slot.opp_label == "" assert "^K" not in plain.footer_hint async def test_k_on_the_contacts_list_toggles_the_lock_in_place(tmp_path) -> None: # noqa: ANN001 """^K writes the lock to the store, and the row gains its padlock on the same screen. No confirm and no notice, as on the node page: the padlock is the acknowledgement, and the next ^K reverses it. """ import asyncio import logging from types import SimpleNamespace from meshterm.core.contact_store import ContactStore from meshterm.core.models import Contact from meshterm.ui.contacts_screen import ContactsScreen, open_contacts from meshterm.ui.surface import TuiUi from meshterm.ui.tui.session import TuiSession dev = "cc" * 32 store = ContactStore(tmp_path / "contacts.json") alice = Contact(name="Alice", public_key="aa" * 32) session = TuiSession() ctx = SimpleNamespace( ui=TuiUi(session), log=logging.getLogger("test.contacts"), contact_store=store, devstate=SimpleNamespace(contacts=lambda: _contacts([alice])), ) task = asyncio.ensure_future(open_contacts(ctx, "Us", dev, [alice], 1, {}, _contacts_sort())) listed = await _step_until_screen(session, lambda s: isinstance(s, ContactsScreen)) listed.handle("down") # off our own node, onto Alice def alice_row() -> str: listed.render_body(72) return next(c for c in listed._choices() if c.value == alice).title.plain listed.handle("lock") await _step_until_screen(session, lambda s: store.is_locked(dev, "aa" * 32)) await _step_until_screen(session, lambda s: alice_row().startswith("\U0001f512")) assert session.top is listed listed.handle("lock") await _step_until_screen(session, lambda s: not store.is_locked(dev, "aa" * 32)) await _step_until_screen(session, lambda s: not alice_row().startswith("\U0001f512")) listed.handle("escape") await task