diff --git a/meshterm/context.py b/meshterm/context.py index 40b6ed8..2697b15 100644 --- a/meshterm/context.py +++ b/meshterm/context.py @@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Optional from rich.console import Console from .core.admin_store import AdminStore +from .core.advert_store import AdvertStore from .core.config import DeviceProfile, Settings from .core.connection import Device, make_device from .core.device_store import DeviceStore @@ -22,6 +23,7 @@ from .persistence.logging import get_logger from .persistence.repository import Repository if TYPE_CHECKING: + from .services.advert_scheduler import AdvertScheduler from .services.chat_service import ChatService from .services.event_hub import EventHub from .services.monitor_service import MonitorService @@ -39,6 +41,8 @@ class AppContext: profile: Active device profile, if one was resolved. device_store: Store for the remembered "last known good" device. admin_store: Store for remembered remote-node admin passwords. + advert_store: Store for per-device background-advert schedules (defaults to + ``/adverts.json`` when not injected). mock: Whether the simulator device is in use. port_override: Explicit serial port (from ``--port`` or the interactive picker), overriding the profile. @@ -57,6 +61,7 @@ class AppContext: repo: Repository device_store: DeviceStore admin_store: AdminStore + advert_store: Optional[AdvertStore] = None profile: Optional[DeviceProfile] = None mock: bool = False port_override: Optional[str] = None @@ -81,8 +86,14 @@ class AppContext: _events: "Optional[EventHub]" = field(default=None, init=False, repr=False) _monitor: "Optional[MonitorService]" = field(default=None, init=False, repr=False) _chat: "Optional[ChatService]" = field(default=None, init=False, repr=False) + _adverts: "Optional[AdvertScheduler]" = field(default=None, init=False, repr=False) _ui: "Optional[Ui]" = field(default=None, init=False, repr=False) + def __post_init__(self) -> None: + """Derive the advert store's default location when one was not injected.""" + if self.advert_store is None: + self.advert_store = AdvertStore(self.settings.config_dir / "adverts.json") + @property def profile_name(self) -> Optional[str]: """Name of the active profile, if any.""" @@ -210,6 +221,20 @@ class AppContext: self._chat = ChatService(self) return self._chat + @property + def adverts(self) -> "AdvertScheduler": + """Return the session's background-advert scheduler, creating it on first use. + + Created idle here; the interactive session starts it alongside the other + always-on services. Scripted CLI runs never start it, so a one-shot command + can't fire a background transmission. + """ + if self._adverts is None: + from .services.advert_scheduler import AdvertScheduler + + self._adverts = AdvertScheduler(self) + return self._adverts + @property def log(self): # type: ignore[no-untyped-def] """The application logger.""" @@ -423,6 +448,8 @@ class AppContext: async def aclose(self) -> None: """Stop monitoring and chat, stop the event hub, disconnect, and close the repo.""" + if self._adverts is not None: + await self._adverts.aclose() if self._monitor is not None: await self._monitor.aclose() if self._chat is not None: diff --git a/meshterm/core/advert_store.py b/meshterm/core/advert_store.py new file mode 100644 index 0000000..3148036 --- /dev/null +++ b/meshterm/core/advert_store.py @@ -0,0 +1,221 @@ +"""Persistence for the background-advert schedule. + +MeshTerm announces the connected node on a cadence — a zero-hop (direct) advert for the +immediate neighbourhood and a flood advert for the wider mesh — so the node stays fresh in +other nodes' contact lists without anyone thinking about it. This store remembers, per +device (keyed by its public key), the chosen cadence for each advert type and when one was +last sent, so the countdown survives restarts and *every* advert counts: a manual send from +the advert menu resets the same clock the scheduler reads (see +:class:`~meshterm.services.advert_scheduler.AdvertScheduler`). + +Like the remembered devices (:mod:`meshterm.core.device_store`), this is global machine +state in a small JSON file (``/adverts.json``) rather than the per-invocation +SQLite database. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Optional + +from .models import utcnow + +#: The direct (zero-hop) cadence choices offered in the editor, in hours. +DIRECT_CADENCE_HOURS = (1, 2, 3, 4) + +#: The flood cadence choices offered in the editor, in hours (3 h .. 1 week). +FLOOD_CADENCE_HOURS = (3, 6, 12, 24, 48, 168) + +#: Defaults applied when a device has no stored policy: announce to neighbours hourly, +#: flood the wider mesh daily. +DEFAULT_DIRECT_HOURS = 1 +DEFAULT_FLOOD_HOURS = 24 + +#: The stored cadence meaning "this advert type is off" (0 hours). +OFF = 0 + + +@dataclass(slots=True) +class AdvertPolicy: + """One device's background-advert schedule and its last-sent marks. + + Attributes: + direct_hours: Hours between zero-hop adverts (:data:`OFF` disables them). + flood_hours: Hours between flood adverts (:data:`OFF` disables them). + last_direct: When a zero-hop advert (scheduled *or* manual) last went out. + last_flood: When a flood advert last went out. + """ + + direct_hours: int = DEFAULT_DIRECT_HOURS + flood_hours: int = DEFAULT_FLOOD_HOURS + last_direct: Optional[datetime] = None + last_flood: Optional[datetime] = None + + def cadence(self, flood: bool) -> int: + """The cadence in hours for one advert type (:data:`OFF` when disabled).""" + return self.flood_hours if flood else self.direct_hours + + def last_sent(self, flood: bool) -> Optional[datetime]: + """When an advert of one type last went out, or ``None`` if never recorded.""" + return self.last_flood if flood else self.last_direct + + def due(self, flood: bool, now: Optional[datetime] = None) -> bool: + """Whether an advert of one type is due at ``now``. + + Never-sent is *not* due: the scheduler arms the clock by recording "now" on its + first look (see :meth:`AdvertStore.arm`), so a fresh install waits a full cadence + instead of transmitting the moment it starts — adverts are routine, not urgent, + and the quiet default is the polite one on a shared mesh. + + Args: + flood: ``True`` for the flood clock, ``False`` for the zero-hop one. + now: The current time (defaults to :func:`~meshterm.core.models.utcnow`). + + Returns: + ``True`` when the type is enabled, armed, and its cadence has elapsed. + """ + hours = self.cadence(flood) + last = self.last_sent(flood) + if hours == OFF or last is None: + return False + return ((now or utcnow()) - last).total_seconds() >= hours * 3600 + + +def cadence_label(hours: int) -> str: + """Render a cadence for display: ``"off"``, ``"every hour"``, ``"every 2 h"``, ``"weekly"``.""" + if hours == OFF: + return "off" + if hours == 1: + return "every hour" + if hours == 24: + return "daily" + if hours == 168: + return "weekly" + return f"every {hours} h" + + +class AdvertStore: + """Reads and writes per-device background-advert schedules.""" + + def __init__(self, path: Path) -> None: + """Open the store against a JSON file location. + + Args: + path: Path to the JSON state file (created lazily on first write). + """ + self._path = path + + @staticmethod + def _key(public_key: str) -> str: + """Normalize a device public key into the storage key.""" + return public_key.lower().removeprefix("0x") + + def _load_all(self) -> dict[str, dict]: + """Return the raw key -> record mapping, or empty on missing/corrupt file.""" + try: + data = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + def load(self, public_key: str) -> AdvertPolicy: + """Return the stored policy for a device, or the on-by-default policy. + + Args: + public_key: The device's public key (hex). + + Returns: + The device's :class:`AdvertPolicy`; defaults apply for anything unset. + """ + record = self._load_all().get(self._key(public_key)) + if not isinstance(record, dict): + return AdvertPolicy() + return AdvertPolicy( + direct_hours=_as_hours(record.get("direct_hours"), DEFAULT_DIRECT_HOURS), + flood_hours=_as_hours(record.get("flood_hours"), DEFAULT_FLOOD_HOURS), + last_direct=_as_time(record.get("last_direct")), + last_flood=_as_time(record.get("last_flood")), + ) + + def set_cadence(self, public_key: str, *, flood: bool, hours: int) -> None: + """Store one advert type's cadence for a device. + + Args: + public_key: The device's public key (hex). + flood: ``True`` to set the flood cadence, ``False`` for the zero-hop one. + hours: The new cadence in hours (:data:`OFF` disables the type). + """ + self._update(public_key, {"flood_hours" if flood else "direct_hours": int(hours)}) + + def mark_sent(self, public_key: str, *, flood: bool, when: Optional[datetime] = None) -> None: + """Record that an advert went out, resetting that type's countdown. + + Called for scheduled *and* manual sends alike, so the next background advert + counts from the most recent announcement of that type, whoever triggered it. + + Args: + public_key: The device's public key (hex). + flood: Which advert type went out. + when: The send time (defaults to now). + """ + stamp = (when or utcnow()).isoformat() + self._update(public_key, {"last_flood" if flood else "last_direct": stamp}) + + def arm(self, public_key: str, *, flood: bool, when: Optional[datetime] = None) -> None: + """Start a never-sent advert type's countdown from ``when`` without sending. + + Only writes when no last-sent mark exists, so it is safe to call on every + scheduler pass. + + Args: + public_key: The device's public key (hex). + flood: Which advert type to arm. + when: The countdown start (defaults to now). + """ + if self.load(public_key).last_sent(flood) is None: + self.mark_sent(public_key, flood=flood, when=when) + + def _update(self, public_key: str, fields: dict) -> None: + """Merge ``fields`` into a device's record and persist the store.""" + records = self._load_all() + key = self._key(public_key) + record = records.get(key) + if not isinstance(record, dict): + record = {} + record.update(fields) + records[key] = record + self._write(records) + + def _write(self, records: dict[str, dict]) -> None: + """Persist ``records`` atomically (crash mid-write keeps the previous file).""" + self._path.parent.mkdir(parents=True, exist_ok=True) + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + tmp.write_text(json.dumps(records, indent=2), encoding="utf-8") + tmp.replace(self._path) + + +def _as_hours(value: object, default: int) -> int: + """Coerce a stored cadence to a non-negative int, falling back to ``default``.""" + try: + hours = int(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return default + return hours if hours >= 0 else default + + +def _as_time(value: object) -> Optional[datetime]: + """Parse a stored ISO-8601 timestamp, or ``None`` if absent/corrupt. + + A timestamp without a zone is treated as corrupt too: the store only ever writes + aware UTC stamps, and a naive one would poison the ``due`` arithmetic. + """ + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None diff --git a/meshterm/services/advert_scheduler.py b/meshterm/services/advert_scheduler.py new file mode 100644 index 0000000..315c08a --- /dev/null +++ b/meshterm/services/advert_scheduler.py @@ -0,0 +1,135 @@ +"""The background-advert scheduler: keep this node announced without anyone asking. + +A quiet session-long task that sends a zero-hop (direct) advert and a flood advert on +their configured cadences (see :class:`~meshterm.core.advert_store.AdvertStore`; the +cadence rows live in the device-configuration editor). The countdown for each type runs +from the *last* advert of that type — scheduled or manual — so using the Send advert menu +pushes the next background send out by a full cadence rather than doubling up. + +Deliberately transmission-shy, in keeping with the app's single-transmission rule: + +* At most **one** advert goes out per pass, even when both types are due (the other + follows a pass later), so the scheduler can never burst. +* A device with no recorded last-send is *armed* — the countdown starts now — instead of + advertised immediately, so a fresh install stays silent for its first full cadence. + +The scheduler holds no device subscription; each pass just checks whether a device is +connected and skips quietly otherwise. That makes it indifferent to reconnects — it keeps +ticking across a link drop and resumes sending once the session's reconnect flow has done +its job. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from ..context import AppContext + +#: Seconds between scheduler passes. Cadences are hours, so a coarse tick is plenty; it +#: also spaces consecutive sends (when both types come due together) a comfortable gap +#: apart on the shared mesh. +POLL_S = 30.0 + + +class AdvertScheduler: + """Owns the background-advert loop for an interactive session. + + Interact through the async lifecycle methods (:meth:`start`, :meth:`stop`, + :meth:`aclose`); everything else happens on the loop's own schedule. + """ + + def __init__(self, ctx: "AppContext") -> None: + """Initialize an idle scheduler bound to an application context. + + Args: + ctx: The shared application context, read each pass for the connected + device and the advert store. Nothing is touched until :meth:`start`. + """ + self._ctx = ctx + self._task: Optional[asyncio.Task] = None + # The connected device's public key, learned once per connection. Keyed by the + # device object's identity so a reconnect (a fresh Device) re-probes it. + self._key_for: Optional[tuple[int, str]] = None + + @property + def active(self) -> bool: + """Whether the scheduler loop is running.""" + return self._task is not None and not self._task.done() + + async def start(self) -> None: + """Start the scheduler loop. Idempotent; never touches the radio itself.""" + if self.active: + return + self._task = asyncio.ensure_future(self._run()) + self._ctx.log.info("advert scheduler started") + + async def stop(self) -> None: + """Stop the scheduler loop. Idempotent.""" + task, self._task = self._task, None + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._ctx.log.info("advert scheduler stopped") + + async def aclose(self) -> None: + """Stop the loop at session end (an alias for :meth:`stop`).""" + await self.stop() + + async def _run(self) -> None: + """Tick forever: sleep, then make one best-effort pass.""" + while True: + await asyncio.sleep(POLL_S) + try: + await self._pass() + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - a bad pass must not kill the loop + self._ctx.log.debug("advert scheduler: pass failed: %s", exc) + + async def _pass(self) -> None: + """One pass: arm never-sent clocks, and send the most overdue advert type if due.""" + ctx = self._ctx + if not ctx.is_connected: + return + key = await self._public_key() + if not key: + return + + store = ctx.advert_store + # Arm fresh clocks so "never sent" becomes "counting from now" (no-op otherwise). + store.arm(key, flood=False) + store.arm(key, flood=True) + + policy = store.load(key) + # One send per pass, direct first (the cheaper, zero-hop announcement); a flood + # due at the same moment follows on the next pass, POLL_S later. + for flood in (False, True): + if policy.due(flood): + device = await ctx.device() + await device.send_advert(flood) + store.mark_sent(key, flood=flood) + kind = "flood" if flood else "zero-hop" + ctx.log.info("advert scheduler: sent scheduled %s advert", kind) + return + + async def _public_key(self) -> str: + """The connected device's public key, probed once per connection (best-effort).""" + device = self._ctx._device + if device is None: + return "" + ident = id(device) + if self._key_for is not None and self._key_for[0] == ident: + return self._key_for[1] + try: + info = await device.get_self_info() + except Exception: # noqa: BLE001 - identity probe is best-effort; retry next pass + return "" + key = str(info.get("public_key") or "") + if key: + self._key_for = (ident, key) + return key diff --git a/meshterm/tools/config.py b/meshterm/tools/config.py index 0a0939b..c016044 100644 --- a/meshterm/tools/config.py +++ b/meshterm/tools/config.py @@ -237,8 +237,28 @@ async def apply_ops( elif kind == "advert": flood = len(op) > 1 and bool(op[1]) await device.send_advert(flood) + # A manual advert resets the background scheduler's countdown for its type, + # so the next scheduled send counts from this one (see AdvertScheduler). + public_key = str(snapshot.get("public_key") or "") + if public_key: + ctx.advert_store.mark_sent(public_key, flood=flood) kind_label = "flood" if flood else "zero-hop" ctx.ui.note(f"[ok]✓[/ok] {kind_label} advertisement sent") + elif kind == "advert_cadence": + from ..core.advert_store import cadence_label + + flood, hours = bool(op[1]), int(op[2]) + public_key = str(snapshot.get("public_key") or "") + if public_key: + ctx.advert_store.set_cadence(public_key, flood=flood, hours=hours) + kind_label = "flood" if flood else "zero-hop" + ctx.ui.note( + f"[ok]✓[/ok] background {kind_label} advert: " + f"[brand]{cadence_label(hours)}[/brand]" + ) + changes += 1 + else: # pragma: no cover - SELF_INFO always carries the key on real firmware + ctx.ui.note("[err]device reported no public key — cadence not saved[/err]") elif kind == "share": _share_contact(ctx, snapshot) elif kind == "sync_clock": diff --git a/meshterm/ui/config_editor.py b/meshterm/ui/config_editor.py index ba0b28e..32196c5 100644 --- a/meshterm/ui/config_editor.py +++ b/meshterm/ui/config_editor.py @@ -39,6 +39,13 @@ from rich.console import Group from rich.table import Table from rich.text import Text +from ..core.advert_store import ( + DIRECT_CADENCE_HOURS, + FLOOD_CADENCE_HOURS, + OFF, + AdvertPolicy, + cadence_label, +) from ..core.device_config import ( DeviceConfigError, SettingSpec, @@ -59,6 +66,8 @@ if TYPE_CHECKING: _LOCATION = "__location__" _PRESETS = "__presets__" _CUSTOM = "__custom__" +_ADVERT_DIRECT = "__advert_direct__" +_ADVERT_FLOOD = "__advert_flood__" _ADVERT = "__advert__" _SYNC_CLOCK = "__sync_clock__" _REBOOT = "__reboot__" @@ -99,6 +108,10 @@ async def edit_config(ctx: "AppContext") -> Optional[list[tuple]]: device = await ctx.device() snapshot = await build_snapshot(device) custom = await device.get_custom_vars() + # The background-advert cadences are app-side settings (MeshTerm sends the adverts, + # not the firmware), but they're staged and applied exactly like device values so + # the editor stays one coherent surface. + policy = ctx.advert_store.load(str(snapshot.get("public_key") or "")) # The editor is menu-only, so a full-screen session is always present. Keep the main # menu *pushed on the stack* for the whole session (rather than popping it between @@ -115,7 +128,7 @@ async def edit_config(ctx: "AppContext") -> Optional[list[tuple]]: while True: staged = len(pending) + len(extra_ops) - title, items = _menu_items(snapshot, pending, staged) + title, items = _menu_items(snapshot, pending, staged, policy) menu = SelectScreen( title, items, default=cursor, wrap=False, footer_hint="↑↓ move · type to filter · Enter select · Esc back", @@ -136,10 +149,17 @@ async def edit_config(ctx: "AppContext") -> Optional[list[tuple]]: continue # keep editing — the same menu is rebuilt next loop return None if choice == _APPLY: - ops: list[tuple] = [("set", k, v) for k, v in pending.items()] + ops: list[tuple] = [] + for k, v in pending.items(): + if k in (_ADVERT_DIRECT, _ADVERT_FLOOD): + ops.append(("advert_cadence", k == _ADVERT_FLOOD, v)) + else: + ops.append(("set", k, v)) ops.extend(extra_ops) return ops or None - if choice == _LOCATION: + if choice in (_ADVERT_DIRECT, _ADVERT_FLOOD): + await _stage_advert_cadence(ctx, choice == _ADVERT_FLOOD, policy, pending) + elif choice == _LOCATION: await _stage_location(ctx, snapshot, pending) elif choice == _PRESETS: await _stage_preset(ctx, pending) @@ -267,7 +287,7 @@ def _lane_row(label: str, value: Text, help_text: str, label_w: int, value_w: in def _menu_items( - snapshot: dict, pending: dict, staged: int + snapshot: dict, pending: dict, staged: int, policy: AdvertPolicy ) -> tuple[str, list]: """Build the editor menu's title and rows for the current snapshot + staged state. @@ -306,6 +326,27 @@ def _menu_items( )) sections.append((category, rows)) + # App-side rows: the background-advert cadences MeshTerm itself runs (see the + # advert scheduler). They stage and apply like device settings, so they sit in the + # same lanes under their own heading. + sections.append(( + "Background adverts", + [ + ( + "Direct advert", + _cadence_value(policy, False, pending), + "Scheduled zero-hop announce; any manual send resets it", + _ADVERT_DIRECT, + ), + ( + "Flood advert", + _cadence_value(policy, True, pending), + "Scheduled mesh-wide announce via repeaters", + _ADVERT_FLOOD, + ), + ], + )) + label_w = max(cell_len(label) for _, rows in sections for label, _, _, _ in rows) value_w = max(cell_len(value.plain) for _, rows in sections for _, value, _, _ in rows) @@ -352,6 +393,15 @@ def _changes(count: int) -> str: return f"{count} staged change{'' if count == 1 else 's'}" +def _cadence_value(policy: AdvertPolicy, flood: bool, pending: dict) -> Text: + """One background-advert row's VALUE lane: ``current [→ staged]``.""" + key = _ADVERT_FLOOD if flood else _ADVERT_DIRECT + value = Text(cadence_label(policy.cadence(flood))) + if key in pending: + value.append(f" → {cadence_label(pending[key])}", style="warn") + return value + + # --- staging individual changes ---------------------------------------------- @@ -443,6 +493,45 @@ async def _prompt_value( return None if raw is None else parse_value(spec, raw, snapshot) +async def _stage_advert_cadence( + ctx: "AppContext", flood: bool, policy: AdvertPolicy, pending: dict[str, Any] +) -> None: + """Pick one background-advert type's cadence and stage it. + + Staged under the type's sentinel key; Apply turns it into an ``advert_cadence`` op + (see :func:`~meshterm.tools.config.apply_ops`). Picking the value already in force + un-stages the row, matching :func:`_stage_setting`. + """ + key = _ADVERT_FLOOD if flood else _ADVERT_DIRECT + in_force = policy.cadence(flood) + current = pending.get(key, in_force) + hours_choices = FLOOD_CADENCE_HOURS if flood else DIRECT_CADENCE_HOURS + items: list = [ + Choice( + title=cadence_label(hours).capitalize() + + (" (current)" if hours == current else ""), + value=hours, + ) + for hours in (*hours_choices, OFF) + ] + if flood: + prompt = "How often repeaters rebroadcast this node across the mesh:" + else: + prompt = "How often this node announces itself to neighbours in range:" + selected = await ctx.ui.select( + "📡 Flood advert" if flood else "📡 Direct advert", + items, + prompt=prompt, + default=current, + ) + if selected is None: + return + if selected == in_force: + pending.pop(key, None) # set back to the value in force — nothing to change + else: + pending[key] = selected + + async def _stage_location( ctx: "AppContext", snapshot: dict, pending: dict[str, Any] ) -> None: diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index 684b64c..c99abb2 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -170,8 +170,9 @@ async def run_menu(ctx: AppContext) -> None: if await _startup(ctx): await _session_loop(ctx, session) finally: - # Stop history + chat recording (closing their run records) and the always-on - # event hub, even on an unexpected exit. + # Stop history + chat recording (closing their run records), the background + # advert scheduler, and the always-on event hub, even on an unexpected exit. + await ctx.adverts.aclose() await ctx.monitor.aclose() await ctx.chat.aclose() await ctx.events.aclose() @@ -410,6 +411,10 @@ async def _resume_monitor(ctx: AppContext) -> None: ctx: The shared application context. """ await ctx.monitor.start() + # The advert scheduler is safe to run from launch regardless of the connect policy: + # each pass checks for a connected device and skips quietly without one, so starting + # it here never opens the radio (and it simply waits out a deferred connect). + await ctx.adverts.start() if not ctx.settings.connect_on_start: return try: diff --git a/tests/test_advert.py b/tests/test_advert.py new file mode 100644 index 0000000..0fb460a --- /dev/null +++ b/tests/test_advert.py @@ -0,0 +1,202 @@ +"""Background-advert tests: the store's schedule arithmetic, the scheduler's +one-send-per-pass discipline, and the editor/executor integration that lets manual +adverts and staged cadence changes share one clock.""" + +from __future__ import annotations + +import io +from datetime import timedelta +from pathlib import Path +from typing import Any + +import pytest +from rich.console import Console + +from meshterm.context import AppContext +from meshterm.core.admin_store import AdminStore +from meshterm.core.advert_store import ( + DEFAULT_DIRECT_HOURS, + DEFAULT_FLOOD_HOURS, + OFF, + AdvertStore, + cadence_label, +) +from meshterm.core.config import Settings +from meshterm.core.device_store import DeviceStore +from meshterm.core.models import utcnow +from meshterm.persistence.repository import Repository +from meshterm.services.advert_scheduler import AdvertScheduler +from meshterm.tools.config import apply_ops + +KEY = "00" * 32 # the simulator's public key + + +# -- the store ----------------------------------------------------------------------- + + +def test_store_defaults_are_on(tmp_path: Path) -> None: + """An unknown device gets the on-by-default policy: hourly direct, daily flood.""" + policy = AdvertStore(tmp_path / "adverts.json").load(KEY) + assert policy.direct_hours == DEFAULT_DIRECT_HOURS + assert policy.flood_hours == DEFAULT_FLOOD_HOURS + assert policy.last_direct is None and policy.last_flood is None + + +def test_store_round_trips_cadence_and_marks(tmp_path: Path) -> None: + """Cadences and last-sent marks persist per device and per advert type.""" + store = AdvertStore(tmp_path / "adverts.json") + store.set_cadence(KEY, flood=True, hours=48) + store.set_cadence(KEY, flood=False, hours=OFF) + when = utcnow() + store.mark_sent(KEY, flood=True, when=when) + + policy = store.load(KEY) + assert policy.flood_hours == 48 + assert policy.direct_hours == OFF + assert policy.last_flood == when + assert policy.last_direct is None + # A different device is untouched. + assert AdvertStore(tmp_path / "adverts.json").load("ff" * 32).flood_hours == 24 + + +def test_never_sent_is_not_due_and_arm_starts_the_clock(tmp_path: Path) -> None: + """A fresh device is silent until armed; arming never overwrites a real mark.""" + store = AdvertStore(tmp_path / "adverts.json") + assert not store.load(KEY).due(flood=False) + + armed_at = utcnow() - timedelta(hours=2) + store.arm(KEY, flood=False, when=armed_at) + assert store.load(KEY).due(flood=False) # armed 2 h ago, hourly cadence — due + + store.arm(KEY, flood=False) # already armed: a no-op, not a reset to "now" + assert store.load(KEY).last_direct == armed_at + + +def test_due_respects_off_and_cadence(tmp_path: Path) -> None: + """The clock honours OFF and only fires once the full cadence has elapsed.""" + store = AdvertStore(tmp_path / "adverts.json") + store.mark_sent(KEY, flood=True, when=utcnow() - timedelta(hours=25)) + assert store.load(KEY).due(flood=True) # daily default, 25 h elapsed + + store.mark_sent(KEY, flood=True, when=utcnow() - timedelta(hours=23)) + assert not store.load(KEY).due(flood=True) + + store.set_cadence(KEY, flood=True, hours=OFF) + assert not store.load(KEY).due(flood=True) # overdue but switched off + + +def test_store_tolerates_corrupt_file(tmp_path: Path) -> None: + """A mangled state file reads as the defaults instead of raising.""" + path = tmp_path / "adverts.json" + path.write_text("{not json", encoding="utf-8") + assert AdvertStore(path).load(KEY).direct_hours == DEFAULT_DIRECT_HOURS + + +def test_cadence_labels() -> None: + """The display labels read naturally at every offered cadence.""" + assert cadence_label(OFF) == "off" + assert cadence_label(1) == "every hour" + assert cadence_label(3) == "every 3 h" + assert cadence_label(24) == "daily" + assert cadence_label(168) == "weekly" + + +# -- the scheduler and executor against the simulator -------------------------------- + + +@pytest.fixture() +def ctx(tmp_path: Path) -> AppContext: + """A mock-backed application context for scheduler/executor tests.""" + settings = Settings(config_dir=tmp_path, db_path=tmp_path / "adv.db") + context = AppContext( + console=Console(file=io.StringIO()), + settings=settings, + repo=Repository(settings.db_path), + device_store=DeviceStore(tmp_path / "devices.json"), + admin_store=AdminStore(tmp_path / "admin.json"), + mock=True, + ) + yield context + context.repo.close() + + +class _NoteUi: + """A minimal UI surface that only collects notes (all apply_ops needs).""" + + def __init__(self) -> None: + self.notes: list[str] = [] + + def note(self, markup: str) -> None: + self.notes.append(markup) + + +async def test_scheduler_first_pass_arms_without_sending(ctx: AppContext) -> None: + """The first look at a fresh device starts the countdown; nothing is transmitted.""" + await ctx.device() + scheduler = AdvertScheduler(ctx) + sent: list[bool] = [] + (await ctx.device()).send_advert = lambda flood=False: _record(sent, flood) # type: ignore[method-assign] + + await scheduler._pass() + policy = ctx.advert_store.load(KEY) + assert sent == [] + assert policy.last_direct is not None and policy.last_flood is not None + + +async def test_scheduler_sends_at_most_one_advert_per_pass(ctx: AppContext) -> None: + """With both types overdue, one pass sends only the direct; the next sends the flood.""" + await ctx.device() + store = ctx.advert_store + store.mark_sent(KEY, flood=False, when=utcnow() - timedelta(hours=2)) + store.mark_sent(KEY, flood=True, when=utcnow() - timedelta(hours=48)) + + scheduler = AdvertScheduler(ctx) + sent: list[bool] = [] + (await ctx.device()).send_advert = lambda flood=False: _record(sent, flood) # type: ignore[method-assign] + + await scheduler._pass() + assert sent == [False] # direct first, flood held for the next pass + assert not store.load(KEY).due(flood=False) + assert store.load(KEY).due(flood=True) + + await scheduler._pass() + assert sent == [False, True] + assert not store.load(KEY).due(flood=True) + + +async def test_scheduler_skips_quietly_when_disconnected(ctx: AppContext) -> None: + """A pass with no device connected does nothing (and records nothing).""" + scheduler = AdvertScheduler(ctx) + await scheduler._pass() + assert ctx.advert_store.load(KEY).last_direct is None + + +async def test_manual_advert_resets_the_background_clock(ctx: AppContext) -> None: + """An advert sent through apply_ops (any manual flow) re-marks its type's clock.""" + device = await ctx.device() + ctx.advert_store.mark_sent(KEY, flood=False, when=utcnow() - timedelta(hours=5)) + assert ctx.advert_store.load(KEY).due(flood=False) + + ctx._ui = _NoteUi() # type: ignore[assignment] + snapshot = dict(await device.get_self_info()) + await apply_ops(ctx, device, snapshot, [("advert", False)]) + assert not ctx.advert_store.load(KEY).due(flood=False) + + +async def test_advert_cadence_op_writes_the_store(ctx: AppContext) -> None: + """The editor's staged cadence change lands in the store via its op.""" + device = await ctx.device() + ctx._ui = _NoteUi() # type: ignore[assignment] + snapshot = dict(await device.get_self_info()) + changes, _ = await apply_ops( + ctx, device, snapshot, [("advert_cadence", True, 168), ("advert_cadence", False, OFF)] + ) + assert changes == 2 + policy = ctx.advert_store.load(KEY) + assert policy.flood_hours == 168 + assert policy.direct_hours == OFF + + +async def _record(sent: list[bool], flood: bool) -> None: + """Stand-in ``send_advert`` recording each transmission's type.""" + sent.append(flood) diff --git a/tests/test_config_editor.py b/tests/test_config_editor.py index 3f5b4fe..4c89b96 100644 --- a/tests/test_config_editor.py +++ b/tests/test_config_editor.py @@ -360,6 +360,29 @@ async def test_editor_restaging_the_current_value_clears_the_stage(ctx: AppConte assert await edit_config(ctx) is None +async def test_editor_stages_background_advert_cadence(ctx: AppContext) -> None: + """Picking a cadence stages it under its sentinel; Apply maps it to its own op.""" + _install(ctx, [ + ("select", "__advert_flood__"), + ("select", 48), # every 48 h + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ops == [("advert_cadence", True, 48)] + + +async def test_editor_repicking_the_cadence_in_force_clears_the_stage(ctx: AppContext) -> None: + """Choosing the cadence already in force un-stages the row (nothing to apply).""" + _install(ctx, [ + ("select", "__advert_direct__"), + ("select", 4), + ("select", "__advert_direct__"), + ("select", 1), # back to the default in force + ("select", "__cancel__"), # nothing staged now — closes without a discard dialog + ]) + assert await edit_config(ctx) is None + + async def test_editor_asks_before_discarding_staged_changes(ctx: AppContext) -> None: """Cancelling with staged changes confirms; Keep editing returns to the menu.""" _install(ctx, [