MeshTerm/tests/test_device_config.py

# 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