diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index a80f270..c7affdb 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -22,6 +22,7 @@ from dataclasses import replace from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Callable, Optional +from . import transmit_gate from .channels import CHANNEL_SLOT_PROBE_CAP from .events import MeshEvent from .frames import frame_addressing, trace_link_snrs @@ -2038,6 +2039,7 @@ class MeshCoreDevice(Device): async def send_remote_command( # noqa: D102 - inherited docstring self, node: Contact, command: str, *, timeout: float = 8.0 ) -> Optional[str]: + transmit_gate.mark() return await self._send_admin_cmd(node, command, timeout=timeout) async def get_remote_tx_power(self, node: Contact) -> Optional[int]: # noqa: D102 @@ -2089,6 +2091,7 @@ class MeshCoreDevice(Device): path: Optional[str] = None, timeout: Optional[float] = None, ) -> TraceResult: + transmit_gate.mark() from meshcore import EventType # local import keeps mock path dependency-free from ..services.trace_runner import path_hash_flags, trace_timeout @@ -2426,6 +2429,7 @@ class MeshCoreDevice(Device): async def send_direct_message( # noqa: D102 - inherited docstring self, contact: Contact, text: str ) -> Optional[Ack]: + transmit_gate.mark() from meshcore import EventType mc = self._require() @@ -2457,6 +2461,7 @@ class MeshCoreDevice(Device): async def send_channel_message( # noqa: D102 - inherited docstring self, index: int, text: str ) -> None: + transmit_gate.mark() self._ok(await self._require().commands.send_chan_msg(index, text)) @staticmethod @@ -2642,6 +2647,7 @@ class MeshCoreDevice(Device): self._ok(await self._require().commands.set_time(epoch)) async def send_advert(self, flood: bool = False) -> None: # noqa: D102 + transmit_gate.mark(flood_advert=flood) self._ok(await self._require().commands.send_advert(flood)) async def reboot(self) -> None: # noqa: D102 - inherited docstring @@ -2835,6 +2841,7 @@ class MockDevice(Device): async def send_direct_message( # noqa: D102 - inherited docstring self, contact: Contact, text: str ) -> Optional[Ack]: + transmit_gate.mark() await asyncio.sleep(0) # Firmware can only address a contact it holds, so a recipient this simulated device # doesn't have is refused exactly as hardware refuses one — which is what makes the @@ -2848,6 +2855,7 @@ class MockDevice(Device): async def send_channel_message( # noqa: D102 - inherited docstring self, index: int, text: str ) -> None: + transmit_gate.mark() await asyncio.sleep(0) async def admin_login(self, node: Contact, password: str) -> LoginResult: # noqa: D102 @@ -2871,6 +2879,7 @@ class MockDevice(Device): async def send_remote_command( # noqa: D102 - inherited docstring self, node: Contact, command: str, *, timeout: float = 8.0 ) -> Optional[str]: + transmit_gate.mark() await asyncio.sleep(0) key = self._mock_key(node) if key not in self._admin_sessions: @@ -3088,6 +3097,7 @@ class MockDevice(Device): self._clock_offset = epoch - int(_time.time()) async def send_advert(self, flood: bool = False) -> None: # noqa: D102 + transmit_gate.mark(flood_advert=flood) await asyncio.sleep(0) async def reboot(self) -> None: # noqa: D102 - inherited docstring @@ -3271,6 +3281,7 @@ class MockDevice(Device): path: Optional[str] = None, timeout: Optional[float] = None, ) -> TraceResult: + transmit_gate.mark() await asyncio.sleep(0.05) # mimic radio latency so progress bars are visible forced = [h for h in path.split(",") if h.strip()] if path else None # Mirror the real device: the path-hash width is the byte length of a forced hop. diff --git a/meshterm/core/preferences.py b/meshterm/core/preferences.py index a817ecb..f0a1fcf 100644 --- a/meshterm/core/preferences.py +++ b/meshterm/core/preferences.py @@ -132,10 +132,21 @@ PREFERENCES: tuple[PrefSpec, ...] = ( group="Sending", value_type="float", default=5.0, - minimum=0.0, + minimum=0.1, # a floor, not a switch: there is no "off" for duty-cycle courtesy maximum=60.0, unit="s", ), + PrefSpec( + key="flood_advert_cooldown_s", + label="Flood cooldown", + help="Extra wait before another mesh-wide advert", + group="Sending", + value_type="float", + default=60.0, + minimum=5.0, # every repeater in range rebroadcasts one; five seconds is the floor + maximum=3600.0, + unit="s", + ), PrefSpec( key="direct_message_soft_retries", label="Message retries", diff --git a/meshterm/core/transmit_gate.py b/meshterm/core/transmit_gate.py new file mode 100644 index 0000000..0a091dc --- /dev/null +++ b/meshterm/core/transmit_gate.py @@ -0,0 +1,125 @@ +"""The shared transmit clock: how long ago we last put something on the air. + +MeshTerm has always paced its own bursts — a trace waits between samples, a TX sweep +between levels — but each did it privately, inside its own loop. Nothing knew what +*another* feature had just transmitted, so a trace could finish and an advert go out a +tenth of a second later, and no screen could say why it was making you wait. + +This is that missing clock. Every real transmission marks it (see the send methods in +:mod:`meshterm.core.connection`, which are the only place the radio is actually spoken +to), and anything about to transmit can ask :func:`remaining` how long is left before it +politely may. The waiting itself is the UI's job — see +:func:`meshterm.ui.cooldown.wait_for_cooldown`, which turns a wait worth noticing into a +countdown you can back out of. + +Two clocks, because two kinds of transmission cost the mesh very different amounts: + +* **every transmission**, spaced by the ``trace_cooldown_s`` preference. This is duty-cycle + courtesy: our own airtime, spread out. +* **flood adverts**, spaced by the longer ``flood_advert_cooldown_s``. A flood advert is + the most expensive thing a node can ask for — every repeater in range rebroadcasts it — + so it waits out both clocks, and its own is floored well above the general one. + +Timing is :func:`time.monotonic`, not the wall clock: this measures an elapsed interval, +and a clock sync mid-session (which MeshTerm can itself trigger) must not make a cooldown +appear to have elapsed, or to have years left. + +Like :func:`meshterm.core.preferences.current`, the gate is a process-wide singleton +rather than something passed down: the layer that transmits and the layer that waits are +far apart, and threading a clock between them would touch every call in between for +nothing. +""" + +from __future__ import annotations + +import time +from typing import Optional + + +class TransmitGate: + """The last-transmission clock, and the wait it implies. + + :meth:`mark` records that something went out; :meth:`remaining` says how long a + caller should hold off. A fresh gate has transmitted nothing, so nothing is owed. + """ + + def __init__(self) -> None: + """Start an idle gate — nothing sent, nothing owed.""" + self._last_sent: Optional[float] = None + self._last_flood_advert: Optional[float] = None + + def mark(self, *, flood_advert: bool = False) -> None: + """Record a transmission as having just finished. + + Called *after* the send returns, so the cooldown counts from the moment the air + was free again rather than from when we started talking. + + Args: + flood_advert: Whether the transmission was a flood advert, which starts its + own longer clock in addition to the general one. + """ + now = time.monotonic() + self._last_sent = now + if flood_advert: + self._last_flood_advert = now + + def remaining(self, *, flood_advert: bool = False) -> float: + """Seconds a caller should wait before transmitting; ``0.0`` when it may go now. + + Args: + flood_advert: Whether the transmission being considered is a flood advert. A + flood advert waits out *both* clocks — the general cooldown since anything + was sent, and the flood cooldown since the last flood advert — so this + returns whichever is longer. + + Returns: + The wait in seconds, never negative. + """ + from .preferences import current as current_preferences + + preferences = current_preferences() + wait = _left(self._last_sent, preferences.trace_cooldown_s) + if flood_advert: + wait = max( + wait, _left(self._last_flood_advert, preferences.flood_advert_cooldown_s) + ) + return wait + + def reset(self) -> None: + """Forget both clocks, as if nothing had ever been sent. + + For a fresh session and for tests; there is no user-facing way to clear a cooldown, + which would rather defeat it. + """ + self._last_sent = None + self._last_flood_advert = None + + +def _left(last: Optional[float], cooldown: float) -> float: + """Seconds left of ``cooldown`` since ``last``, or ``0.0`` if it never happened.""" + if last is None or cooldown <= 0: + return 0.0 + return max(0.0, cooldown - (time.monotonic() - last)) + + +#: The clock for this process. A module-level singleton for the reason given in the module +#: docstring: the transmitting layer and the waiting layer never meet. +_gate = TransmitGate() + + +def current() -> TransmitGate: + """The transmit clock in force for this process.""" + return _gate + + +def mark(*, flood_advert: bool = False) -> None: + """Record a transmission on the process-wide clock (see :meth:`TransmitGate.mark`).""" + _gate.mark(flood_advert=flood_advert) + + +def remaining(*, flood_advert: bool = False) -> float: + """The wait owed on the process-wide clock (see :meth:`TransmitGate.remaining`).""" + return _gate.remaining(flood_advert=flood_advert) + + +__all__ = ["TransmitGate", "current", "mark", "remaining"] diff --git a/meshterm/ui/config_editor.py b/meshterm/ui/config_editor.py index 94bb50e..17ea9ad 100644 --- a/meshterm/ui/config_editor.py +++ b/meshterm/ui/config_editor.py @@ -1003,7 +1003,10 @@ async def _run_now( async def send_advert(ctx: "AppContext") -> None: """Run the Send advert flow, behind the main menu's ``advert`` popup tool. - Send an advert (zero-hop or flood) or show this node's shareable contact card. Only + Send an advert (zero-hop or flood) or show this node's shareable contact card. Either + advert waits out the transmit cooldown first — a flood advert its own, longer one — + which is a countdown the reader can cancel where the wait is long enough to notice. + Only ``SELF_INFO`` is read here: the advert command itself never consults the snapshot, and the contact card needs just the name, public key, and advert type — so opening this skips the tuning/path-hash reads of a full @@ -1038,7 +1041,18 @@ async def send_advert(ctx: "AppContext") -> None: if choice == "share": await _show_contact_card(ctx, snapshot) return - await _run_now(ctx, device, snapshot, [("advert", choice == "flood")], "Advert") + + # An advert is the one transmission a reader fires by hand, over and over, so it is + # where the shared transmit cooldown is felt (see meshterm.ui.cooldown). A wait worth + # noticing becomes a countdown they can back out of; a short one just happens. + from .cooldown import wait_for_cooldown + + flood = choice == "flood" + if not await wait_for_cooldown( + ctx, action="Flood advert" if flood else "Zero-hop advert", flood_advert=flood + ): + return + await _run_now(ctx, device, snapshot, [("advert", flood)], "Advert") def contact_share_url(name: str, public_key: str, node_type: int = 1) -> str: diff --git a/meshterm/ui/cooldown.py b/meshterm/ui/cooldown.py new file mode 100644 index 0000000..c2f485b --- /dev/null +++ b/meshterm/ui/cooldown.py @@ -0,0 +1,98 @@ +"""Waiting out a transmit cooldown, visibly and cancellably. + +MeshTerm holds itself back between transmissions (see +:mod:`meshterm.core.transmit_gate`). A short hold is nothing anyone needs told about; a +long one, spent frozen, reads as a hung app on the very screen where "did the radio die?" +is the reader's first thought. So the wait has a threshold: under +:data:`SILENT_WAIT_S` it simply happens, and over it becomes a +:class:`~meshterm.ui.tui.prompt.CountdownDialog` — the seconds ticking down over a Cancel +chip, floating on the screen that asked, which the reader can back out of to abandon the +action entirely. + +One function, :func:`wait_for_cooldown`, and its answer is the only thing a caller tests: +``True`` means the air is clear and the action may go ahead, ``False`` means the reader +changed their mind. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING + +from ..core import transmit_gate + +if TYPE_CHECKING: + from ..context import AppContext + +#: A wait at or under this many seconds passes without a word. Long enough to cover the +#: ordinary spacing between two transmissions, short enough that a reader never sits in +#: front of an unexplained stillness — a dialog that flashes up for half a second would +#: itself be the interruption. +SILENT_WAIT_S = 2.0 + +#: How often the countdown redraws. Fast enough that the seconds look like they are +#: counting rather than jumping, slow enough to cost nothing. +_TICK_S = 0.2 + + +async def wait_for_cooldown( + ctx: "AppContext", *, action: str, flood_advert: bool = False +) -> bool: + """Hold until the transmit cooldown has passed; ``False`` if the reader gave up. + + Args: + ctx: Shared application context (for the UI surface). + action: What is waiting, named as the thing about to happen — it becomes the + dialog's title, so it reads "Flood advert", not "Waiting to send a flood + advert". + flood_advert: Whether the pending transmission is a flood advert, which waits out + its own longer clock as well as the general one. + + Returns: + ``True`` when the wait is over and the caller may transmit; ``False`` if the + countdown was cancelled. + """ + remaining = transmit_gate.remaining(flood_advert=flood_advert) + if remaining <= 0: + return True + if remaining <= SILENT_WAIT_S: + await asyncio.sleep(remaining) + return True + + session = getattr(ctx.ui, "session", None) + if session is None: + # No full-screen session to float a dialog over (the scripted CLI): the wait is + # still owed, so it is still taken — just silently, as every other CLI pause is. + await asyncio.sleep(remaining) + return True + + from .tui import CANCEL + from .tui.prompt import CountdownDialog + + reason = ( + "a flood advert reaches the whole mesh" + if flood_advert + else "spacing out our own transmissions" + ) + dialog = CountdownDialog(action, remaining, reason=reason) + ticker = asyncio.ensure_future(_tick(session, dialog, flood_advert)) + try: + return await session.run_screen(dialog) is not CANCEL + finally: + ticker.cancel() + + +async def _tick(session, dialog, flood_advert: bool) -> None: # noqa: ANN001 + """Feed the dialog the live remaining time until it resolves itself at zero.""" + try: + while True: + await asyncio.sleep(_TICK_S) + dialog.set_remaining(transmit_gate.remaining(flood_advert=flood_advert)) + session.invalidate() + except asyncio.CancelledError: + # Swallowed so the task finishes on its own final cycle rather than being torn + # down while pending — the same courtesy the progress dialog's animator takes. + pass + + +__all__ = ["SILENT_WAIT_S", "wait_for_cooldown"] diff --git a/meshterm/ui/tui/prompt.py b/meshterm/ui/tui/prompt.py index a2e296b..3bce656 100644 --- a/meshterm/ui/tui/prompt.py +++ b/meshterm/ui/tui/prompt.py @@ -603,6 +603,93 @@ class ButtonDialog(_KeylessDialog): super().handle("escape") +class CountdownDialog(_KeylessDialog): + """A wait you can watch, and back out of: the seconds left, over one Cancel chip. + + Shown while a cooldown holds a transmission back (see + :func:`meshterm.ui.cooldown.wait_for_cooldown`). The point is that a wait long enough + to notice should say *why* it is waiting and leave a way out — a frozen screen says + neither, and on a radio app it reads as a lost connection rather than as courtesy. + + It resolves itself: the caller ticks it down with :meth:`set_remaining` and the dialog + resolves ``True`` the moment the clock reaches zero. Esc — or Enter on its Cancel chip + — resolves CANCEL instead, and the caller abandons whatever was waiting. + + One chip, not two: there is nothing to choose between, only something to abandon, so + it takes the lone-button hint shape (no ``←→``) with Cancel as its verb. + """ + + def __init__(self, action: str, remaining: float, *, reason: str = "") -> None: + """Build a countdown over one Cancel chip. + + Args: + action: What is waiting, named as the thing that will happen ("Flood advert"). + remaining: Seconds left when the dialog opens. + reason: One muted line under the clock saying why the wait exists; empty for + no line at all. + """ + super().__init__() + self.title = action + # Both keys do the one thing this dialog offers, so they share an atom rather + # than each claiming "cancel" in turn (the hint line's shared-key rule). + self.footer_hint = "Enter/Esc cancel" + self.border_style = "warn" + self._remaining = max(0.0, remaining) + self._reason = reason + + def set_remaining(self, remaining: float) -> None: + """Update the clock, resolving the dialog once it reaches zero. + + Resolving twice is harmless — :meth:`~meshterm.ui.tui.screen.Screen.resolve` drops + a second result — so the tick never has to know whether Esc beat it to it. + """ + self._remaining = max(0.0, remaining) + if self._remaining <= 0: + self.resolve(True) + + @property + def dialog_width(self) -> int: + """Natural outer width, sized for the widest line it will ever draw. + + Measured against the *opening* clock rather than the current one, so the box does + not shrink a cell as the seconds tick from 10 to 9 — a dialog that resizes while + you read it is harder to read than a slightly wide one. + """ + inner = max( + cell_len(self.title), + cell_len(self.footer_hint), + cell_len(self._clock_text(self._remaining)), + cell_len(self._reason), + len(" Cancel "), + ) + return inner + 8 + + @staticmethod + def _clock_text(remaining: float) -> str: + """The clock line: whole seconds, counted the way a person would say them.""" + seconds = max(0, int(remaining + 0.999)) # 0.2s left still reads as "1s" + return f"Ready in {seconds}s" + + def render_body(self, width: int) -> list[str]: + """Draw the clock, its reason, and the Cancel chip.""" + lines: list[Text] = [Text(self._clock_text(self._remaining), style="warn")] + if self._reason: + lines.append(Text(self._reason, style="muted")) + lines.append(Text("")) + lines.append(Text(" Cancel ", style="selected")) + rendered: list[str] = [] + for line in lines: + # An empty Text renders to no lines at all on its own, so the spacer above the + # chip has to be asked for explicitly or the chip rides up against the reason. + rendered.extend(render_lines(line, width) or [""]) + return rendered + + def handle(self, action: str, data: str = "") -> None: + """Enter and Esc both abandon: the only chip on the dialog is Cancel.""" + if action in ("enter", "escape"): + super().handle("escape") + + class TypedConfirmDialog(_KeylessDialog): """A destructive-action gate: the user must type a confirmation word to proceed. diff --git a/tests/conftest.py b/tests/conftest.py index be61bdb..807b7d0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -71,6 +71,22 @@ def powerline(monkeypatch: pytest.MonkeyPatch) -> Callable[[bool], None]: return _set +@pytest.fixture(autouse=True) +def _reset_transmit_gate() -> Iterator[None]: + """Every test starts having transmitted nothing. + + The transmit clock is process-wide (see :mod:`meshterm.core.transmit_gate`), which is + right for a session and wrong for a suite: a test that sends anything would otherwise + leave a cooldown standing for whatever ran next, and the next test's flow would stop + on a countdown nobody scripted. + """ + from meshterm.core.transmit_gate import current as transmit_clock + + transmit_clock().reset() + yield + transmit_clock().reset() + + @pytest.fixture(autouse=True) def _reset_platform() -> Iterator[None]: """Every test starts and ends on :data:`~meshterm.platforms.REGULAR`. diff --git a/tests/test_cooldown.py b/tests/test_cooldown.py new file mode 100644 index 0000000..5402743 --- /dev/null +++ b/tests/test_cooldown.py @@ -0,0 +1,267 @@ +"""Tests for the transmit cooldown: the shared clock, the countdown, and the way out. + +Three layers again — what the clock owes (:mod:`meshterm.core.transmit_gate`), what the +dialog draws, and what :func:`~meshterm.ui.cooldown.wait_for_cooldown` decides at each +side of its threshold — plus the check that a real transmission is what starts the clock, +so the gate can never be a number nothing sets. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Optional + +import pytest + +from meshterm.core import transmit_gate +from meshterm.core.preferences import Preferences, get_spec, install +from meshterm.core.transmit_gate import TransmitGate +from meshterm.ui.cooldown import SILENT_WAIT_S, wait_for_cooldown +from meshterm.ui.tui.prompt import CountdownDialog +from meshterm.ui.tui.screen import CANCEL +from tests.conftest import plain as _plain + + +@pytest.fixture(autouse=True) +def _defaults(): + """Every test starts from an idle clock and the built-in preferences.""" + install(Preferences()) + transmit_gate.current().reset() + yield + install(Preferences()) + transmit_gate.current().reset() + + +# -- the clock ------------------------------------------------------------------- + + +def test_an_idle_gate_owes_nothing() -> None: + """Nothing sent, nothing to wait for — a fresh session transmits at once.""" + gate = TransmitGate() + assert gate.remaining() == 0.0 + assert gate.remaining(flood_advert=True) == 0.0 + + +def test_a_send_starts_the_general_clock() -> None: + """Any transmission holds the next one off by the transmit cooldown.""" + gate = TransmitGate() + gate.mark() + assert gate.remaining() == pytest.approx(5.0, abs=0.5) + + +def test_a_flood_advert_waits_out_both_clocks() -> None: + """A flood advert owes its own longer wait; an ordinary send still owes only the short one.""" + gate = TransmitGate() + gate.mark(flood_advert=True) + assert gate.remaining() == pytest.approx(5.0, abs=0.5) + assert gate.remaining(flood_advert=True) == pytest.approx(60.0, abs=0.5) + + +def test_an_ordinary_send_does_not_arm_the_flood_clock() -> None: + """Sending a message shouldn't cost you a minute before you may advertise.""" + gate = TransmitGate() + gate.mark() + assert gate.remaining(flood_advert=True) == pytest.approx(5.0, abs=0.5) + + +def test_the_clocks_follow_the_preferences() -> None: + """Both waits are read live, so a change on the Preferences page applies at once.""" + prefs = Preferences() + prefs.set("trace_cooldown_s", 0.1) + prefs.set("flood_advert_cooldown_s", 30.0) + install(prefs) + gate = TransmitGate() + gate.mark(flood_advert=True) + assert gate.remaining() == pytest.approx(0.1, abs=0.05) + assert gate.remaining(flood_advert=True) == pytest.approx(30.0, abs=0.5) + + +def test_reset_clears_both_clocks() -> None: + """Reset returns a gate to never-sent (for a fresh session, and for these tests).""" + gate = TransmitGate() + gate.mark(flood_advert=True) + gate.reset() + assert gate.remaining(flood_advert=True) == 0.0 + + +def test_the_bounds_say_what_may_be_asked_for() -> None: + """The general cooldown has a floor rather than an off switch; a flood advert's is higher.""" + assert get_spec("trace_cooldown_s").minimum == 0.1 + assert get_spec("flood_advert_cooldown_s").minimum == 5.0 + + +# -- the dialog ------------------------------------------------------------------ + + +def _lines(dialog: CountdownDialog) -> list[str]: + """The dialog's body as plain text.""" + return _plain(dialog.render_body(dialog.dialog_width - 8)).splitlines() + + +def test_the_countdown_shows_the_clock_its_reason_and_the_way_out() -> None: + """What is waiting, how long is left, why, and the one chip that abandons it.""" + dialog = CountdownDialog("Flood advert", 47.0, reason="reaches the whole mesh") + assert dialog.title == "Flood advert" + body = _lines(dialog) + assert "Ready in 47s" in body[0] + assert "reaches the whole mesh" in body[1] + assert any("Cancel" in line for line in body) + # Both keys do the one thing on offer, so they share an atom instead of repeating it. + assert dialog.footer_hint == "Enter/Esc cancel" + + +def test_a_part_second_still_reads_as_a_second() -> None: + """The clock counts the way a person would say it: 0.2s left is still "1s".""" + assert "Ready in 1s" in _lines(CountdownDialog("Advert", 0.2))[0] + assert "Ready in 3s" in _lines(CountdownDialog("Advert", 2.4))[0] + + +def test_the_box_does_not_resize_as_it_counts() -> None: + """Sized once, from the opening clock — a box that shrinks under you is harder to read.""" + dialog = CountdownDialog("Flood advert", 47.0, reason="reaches the whole mesh") + width = dialog.dialog_width + dialog.set_remaining(3.0) + assert dialog.dialog_width == width + + +def test_the_countdown_resolves_itself_at_zero() -> None: + """Reaching zero is the dialog's own answer: the wait is over, the action goes ahead.""" + dialog = CountdownDialog("Advert", 5.0) + dialog.future = asyncio.get_event_loop_policy().new_event_loop().create_future() + dialog.set_remaining(3.0) + assert not dialog.future.done() + dialog.set_remaining(0.0) + assert dialog.future.done() and dialog.future.result() is True + + +@pytest.mark.parametrize("key", ["enter", "escape"]) +def test_either_key_abandons_the_wait(key: str) -> None: + """There is nothing to choose between, so Enter and Esc both mean cancel.""" + dialog = CountdownDialog("Advert", 30.0) + dialog.future = asyncio.get_event_loop_policy().new_event_loop().create_future() + dialog.handle(key) + assert dialog.future.done() and dialog.future.result() is CANCEL + + +# -- the wait -------------------------------------------------------------------- + + +class _Session: + """A session that answers ``run_screen`` from a script and records what it was shown.""" + + def __init__(self, answer: Any = True) -> None: + self.answer = answer + self.shown: list[CountdownDialog] = [] + + async def run_screen(self, screen: Any) -> Any: + self.shown.append(screen) + return self.answer + + def invalidate(self) -> None: + pass + + +class _Ui: + """A UI surface carrying nothing but a session (all the wait helper reads).""" + + def __init__(self, session: Optional[_Session]) -> None: + self.session = session + + +class _Ctx: + """The slice of :class:`~meshterm.context.AppContext` the wait helper touches.""" + + def __init__(self, session: Optional[_Session]) -> None: + self.ui = _Ui(session) + + +async def test_no_cooldown_goes_straight_through() -> None: + """Nothing owed, nothing shown: the action fires with no dialog in the way.""" + session = _Session() + assert await wait_for_cooldown(_Ctx(session), action="Advert") is True + assert session.shown == [] + + +async def test_a_short_wait_passes_without_a_word() -> None: + """Under the threshold the wait just happens: a dialog that flashes up *is* the interruption.""" + prefs = Preferences() + prefs.set("trace_cooldown_s", 0.3) # comfortably under SILENT_WAIT_S + install(prefs) + transmit_gate.mark() + session = _Session() + assert await wait_for_cooldown(_Ctx(session), action="Advert") is True + assert session.shown == [] + # And it really did wait it out. Not exactly zero: the sleep is for the wait read a + # few instructions earlier, so a millisecond or two of it is still nominally owed. + assert transmit_gate.remaining() < 0.05 + + +async def test_a_long_wait_opens_a_countdown() -> None: + """Over the threshold the reader is told what is happening, and for how long.""" + transmit_gate.mark(flood_advert=True) # a minute owed on the flood clock + session = _Session(answer=True) + assert await wait_for_cooldown(_Ctx(session), action="Flood advert", flood_advert=True) + assert len(session.shown) == 1 + dialog = session.shown[0] + assert dialog.title == "Flood advert" + assert "whole mesh" in _lines(dialog)[1] + + +async def test_cancelling_the_countdown_abandons_the_action() -> None: + """Backing out of the wait is backing out of what was waiting.""" + transmit_gate.mark(flood_advert=True) + session = _Session(answer=CANCEL) + assert await wait_for_cooldown( + _Ctx(session), action="Flood advert", flood_advert=True + ) is False + + +async def test_the_threshold_is_where_it_says_it_is() -> None: + """SILENT_WAIT_S is the whole rule: at or under it, silence; over it, a dialog.""" + assert SILENT_WAIT_S == 2.0 + prefs = Preferences() + prefs.set("trace_cooldown_s", SILENT_WAIT_S + 3) + install(prefs) + transmit_gate.mark() + session = _Session() + await wait_for_cooldown(_Ctx(session), action="Advert") + assert len(session.shown) == 1 + + +async def test_a_scripted_run_waits_without_a_dialog() -> None: + """The CLI has no session to float a dialog over, so it takes the wait silently.""" + prefs = Preferences() + prefs.set("trace_cooldown_s", 0.2) + install(prefs) + transmit_gate.mark() + assert await wait_for_cooldown(_Ctx(None), action="Advert") is True + + +# -- what starts the clock ------------------------------------------------------- + + +async def test_a_real_send_is_what_arms_the_gate() -> None: + """The clock is set by transmitting, not by anything remembering to set it. + + Driven against the simulator, which marks the gate at exactly the points the radio + does — so the countdown is walkable (and this is checkable) with no hardware. + """ + from meshterm.core.connection import MockDevice + + device = MockDevice() + await device.connect() + + assert transmit_gate.remaining() == 0.0 + await device.send_advert(False) + assert transmit_gate.remaining() > 0 + assert transmit_gate.remaining(flood_advert=True) == pytest.approx( + transmit_gate.remaining(), abs=0.1 + ) # a zero-hop advert does not arm the flood clock + + transmit_gate.current().reset() + await device.send_advert(True) + assert transmit_gate.remaining(flood_advert=True) > transmit_gate.remaining() + + transmit_gate.current().reset() + await device.run_trace("Alice") + assert transmit_gate.remaining() > 0 diff --git a/tests/test_gallery.py b/tests/test_gallery.py index a515dad..501986d 100644 --- a/tests/test_gallery.py +++ b/tests/test_gallery.py @@ -78,6 +78,7 @@ from meshterm.ui.node_detail_screen import NodeDetailScreen, _Action, _RoutesVie from meshterm.ui.packet_viewer import PacketEntry, PacketViewer from meshterm.ui.path_composer import PathComposerScreen from meshterm.ui.preferences import _menu_items as _preference_items +from meshterm.ui.tui.prompt import CountdownDialog from meshterm.ui.records_screen import RecordDialog from meshterm.ui.remote_cli import RemoteCliScreen from meshterm.ui.timemachine_screen import TimeMachineScreen @@ -564,6 +565,18 @@ def _preferences(cols: int, rows: int) -> Screen: ) +def _cooldown_countdown(cols: int, rows: int) -> Screen: + """The transmit-cooldown countdown at its widest: a flood advert's wait and its reason. + + A dialog rather than a menu screen, and here for the one thing only this harness does: + its box is sized from its own content, so a longer reason line would overflow the + PicoCalc's 53 columns with nothing else to catch it. + """ + return CountdownDialog( + "Flood advert", 47.0, reason="a flood advert reaches the whole mesh" + ) + + def _about_meshterm(cols: int, rows: int) -> Screen: return AboutPage("About MeshTerm", about_meshterm()) @@ -605,6 +618,7 @@ _ENTRIES: list[_Entry] = [ _Entry("config_editor", _config_editor), _Entry("config_editor_revealed", _config_editor_revealed), _Entry("preferences", _preferences), + _Entry("cooldown_countdown", _cooldown_countdown), _Entry("about_meshterm", _about_meshterm), _Entry("about_author", _about_author), _Entry("join_discord", _join_discord),