Settings: remember MeshTerm-changed settings across sessions
A firmware-less radio bridge holds its configuration only in RAM, so the node name, radio parameters, and tuning set through MeshTerm are back to firmware defaults when the bridge process restarts. Record every setting changed through MeshTerm (editor, config CLI, and TOML restore all funnel through _apply_setting), keyed by the device's own public key. Unlike channels — silently replayed because filling an empty slot overwrites nothing — a setting is a single canonical value, so replaying a stale one could clobber a change made elsewhere. So the store never writes on its own: on connect it detects drift between what it remembers and what the device reports, and a startup offer lets the operator restore the saved values, adopt the device's current ones, or stop remembering the device (the "don't ask again" exit). The offer is quiet unless a connected device actually drifts, so a firmware radio you don't manage through MeshTerm is never interrupted. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
711621d69e
commit
842a58a515
6 files changed
+645
No files matched your search
@@ -19,6 +19,7 @@ from .core.config import DeviceProfile, Settings
|
||||
from .core.courier_store import CourierStore
|
||||
from .core.mute_store import MuteStore
|
||||
from .core.remote_store import RemoteStore
|
||||
from .core.settings_store import SettingsStore
|
||||
from .core.watch_store import WatchStore
|
||||
from .core.connection import Device, make_device
|
||||
from .core.device_store import DeviceStore
|
||||
@@ -65,6 +66,9 @@ class AppContext:
|
||||
channel_store: Store for channels created through MeshTerm, replayed into a device
|
||||
that forgot them (a firmware-less radio bridge). Keyed by device public key;
|
||||
defaults to ``<config_dir>/channels.json`` when not injected.
|
||||
settings_store: Store for settings changed through MeshTerm, offered back on connect to
|
||||
a device that forgot them (a firmware-less radio bridge). Keyed by device public
|
||||
key; defaults to ``<config_dir>/settings.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.
|
||||
@@ -92,6 +96,7 @@ class AppContext:
|
||||
courier_store: Optional[CourierStore] = None
|
||||
mute_store: Optional[MuteStore] = None
|
||||
channel_store: Optional[ChannelStore] = None
|
||||
settings_store: Optional[SettingsStore] = None
|
||||
profile: Optional[DeviceProfile] = None
|
||||
mock: bool = False
|
||||
port_override: Optional[str] = None
|
||||
@@ -140,6 +145,8 @@ class AppContext:
|
||||
self.mute_store = MuteStore(self.settings.config_dir / "mutes.json")
|
||||
if self.channel_store is None:
|
||||
self.channel_store = ChannelStore(self.settings.config_dir / "channels.json")
|
||||
if self.settings_store is None:
|
||||
self.settings_store = SettingsStore(self.settings.config_dir / "settings.json")
|
||||
|
||||
@property
|
||||
def profile_name(self) -> Optional[str]:
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
"""Persistence for device settings changed through MeshTerm, so they survive a forgetful device.
|
||||
|
||||
Most companions keep their configuration in firmware, so MeshTerm reads settings live from the
|
||||
device each session and never has to remember them. A firmware-less radio bridge is the
|
||||
exception: it holds its configuration only in RAM, so the node name, radio parameters, and
|
||||
tuning you set last session are back to firmware defaults when the bridge process restarts.
|
||||
|
||||
This store is the durable backup for exactly that case. It records every setting *changed
|
||||
through MeshTerm* (the editor, the ``config`` CLI, and a TOML restore all funnel through
|
||||
:func:`~meshterm.tools.config._apply_setting`), keyed by the device's own public key so two
|
||||
radios keep separate sets. Unlike channels — which are silently replayed because a slot can be
|
||||
filled without overwriting anything — a setting is a single canonical value, so replaying a
|
||||
stale one could clobber a change made elsewhere. So the store never writes on its own: on
|
||||
connect it only *detects drift* between what it remembers and what the device now reports, and
|
||||
the startup offer (see :mod:`meshterm.ui.settings_offer`) lets the operator choose to restore
|
||||
the saved values, adopt the device's current ones, or stop remembering the device entirely.
|
||||
|
||||
The gate is provenance, not device type: only settings you changed through MeshTerm are
|
||||
recorded, so a firmware radio you never edit through MeshTerm keeps an empty record and is never
|
||||
flagged. A firmware-less bridge, where the values necessarily came from MeshTerm, gets them back.
|
||||
|
||||
Like the other operator state (mutes, watched nodes, remembered channels), this is global
|
||||
machine state in a small JSON file (``<config_dir>/settings.json``), not the per-invocation
|
||||
SQLite database. Reads are served from memory after the first load; writes flush immediately
|
||||
and atomically.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, Optional
|
||||
|
||||
from .device_config import DeviceConfigError, format_value, get_spec, parse_value
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .connection import Device
|
||||
|
||||
# The JSON-native scalar types a setting value may take; anything else in a loaded file is
|
||||
# dropped as malformed (settings are strings, numbers, and booleans — never containers).
|
||||
_SCALARS = (str, int, float, bool)
|
||||
|
||||
|
||||
def _norm(pubkey: str) -> str:
|
||||
"""Normalise a device public key to the lowercase hex used as the store's device key."""
|
||||
return (pubkey or "").lower().removeprefix("0x")
|
||||
|
||||
|
||||
class SettingsStore:
|
||||
"""Reads and writes the per-device set of MeshTerm-changed settings, memory-first.
|
||||
|
||||
Interact through :meth:`settings` (a device's remembered ``key -> value`` map),
|
||||
:meth:`remember` (record one changed setting), :meth:`forget` (drop one), and
|
||||
:meth:`forget_all` (stop remembering a device). The backing map is loaded once on first
|
||||
access and kept in memory; each mutation persists the whole map atomically.
|
||||
"""
|
||||
|
||||
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 the first remembered setting).
|
||||
"""
|
||||
self._path = path
|
||||
self._devices: Optional[dict[str, dict[str, Any]]] = None
|
||||
|
||||
@property
|
||||
def _state(self) -> dict[str, dict[str, Any]]:
|
||||
"""The device -> remembered-settings map, loaded from disk on first access."""
|
||||
if self._devices is None:
|
||||
self._devices = self._load()
|
||||
return self._devices
|
||||
|
||||
def _load(self) -> dict[str, dict[str, Any]]:
|
||||
"""Parse the file into the device map, or empty on a missing/corrupt file."""
|
||||
try:
|
||||
data = json.loads(self._path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
if not isinstance(data, dict):
|
||||
return {}
|
||||
devices: dict[str, dict[str, Any]] = {}
|
||||
for pubkey, entries in (data.get("devices") or {}).items():
|
||||
if not isinstance(entries, dict):
|
||||
continue
|
||||
# Keep only plain scalar values; a container or null is a malformed entry.
|
||||
values = {
|
||||
str(key): value
|
||||
for key, value in entries.items()
|
||||
if isinstance(value, _SCALARS)
|
||||
}
|
||||
if values:
|
||||
devices[_norm(str(pubkey))] = values
|
||||
return devices
|
||||
|
||||
def settings(self, pubkey: str) -> dict[str, Any]:
|
||||
"""The settings remembered for a device (a copy), empty if none."""
|
||||
return dict(self._state.get(_norm(pubkey), {}))
|
||||
|
||||
def remember(self, pubkey: str, key: str, value: Any) -> None:
|
||||
"""Record the value a setting was changed to, replacing any prior value for that key.
|
||||
|
||||
Called after a setting is applied to the device through MeshTerm. Re-recording the
|
||||
same value is a no-op (no needless write).
|
||||
|
||||
Args:
|
||||
pubkey: The device's own public key.
|
||||
key: The setting key (a :class:`~meshterm.core.device_config.SettingSpec` key).
|
||||
value: The applied, already-typed scalar value.
|
||||
"""
|
||||
pub = _norm(pubkey)
|
||||
if not pub or not isinstance(value, _SCALARS):
|
||||
return
|
||||
values = self._state.get(pub, {})
|
||||
if values.get(key) == value and key in values:
|
||||
return
|
||||
values = {**values, key: value}
|
||||
self._state[pub] = values
|
||||
self._save()
|
||||
|
||||
def forget(self, pubkey: str, key: str) -> None:
|
||||
"""Drop one remembered setting; persist only a real change."""
|
||||
pub = _norm(pubkey)
|
||||
values = self._state.get(pub)
|
||||
if not values or key not in values:
|
||||
return
|
||||
values = {k: v for k, v in values.items() if k != key}
|
||||
if values:
|
||||
self._state[pub] = values
|
||||
else:
|
||||
del self._state[pub]
|
||||
self._save()
|
||||
|
||||
def forget_all(self, pubkey: str) -> None:
|
||||
"""Stop remembering a device's settings entirely; persist only a real change."""
|
||||
pub = _norm(pubkey)
|
||||
if pub not in self._state:
|
||||
return
|
||||
del self._state[pub]
|
||||
self._save()
|
||||
|
||||
def _save(self) -> None:
|
||||
"""Persist the whole device map atomically (a crash mid-write keeps the old file)."""
|
||||
data = {
|
||||
"devices": {
|
||||
pubkey: dict(sorted(values.items()))
|
||||
for pubkey, values in sorted(self._state.items())
|
||||
if values
|
||||
}
|
||||
}
|
||||
self._path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = self._path.with_suffix(self._path.suffix + ".tmp")
|
||||
tmp.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
||||
tmp.replace(self._path)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SettingDrift:
|
||||
"""One setting whose remembered value differs from what the device now reports.
|
||||
|
||||
Attributes:
|
||||
key: The setting key.
|
||||
remembered: The value MeshTerm has saved for this device.
|
||||
current: The value the device currently reports (``None`` if it reports none).
|
||||
"""
|
||||
|
||||
key: str
|
||||
remembered: Any
|
||||
current: Any
|
||||
|
||||
|
||||
def settings_drift(store: SettingsStore, pubkey: str, snapshot: dict) -> list[SettingDrift]:
|
||||
"""Return the remembered settings that differ from a device's current snapshot.
|
||||
|
||||
Values are compared by their *formatted* form (via the setting's spec), so a difference
|
||||
only in representation — an integer that reads the same, a float the firmware rounds
|
||||
identically — is not reported as drift. A remembered key the registry no longer knows is
|
||||
skipped. With nothing remembered this returns an empty list, so a firmware radio you don't
|
||||
manage through MeshTerm never shows drift.
|
||||
|
||||
Args:
|
||||
store: The settings store to read remembered values from.
|
||||
pubkey: The device's own public key.
|
||||
snapshot: The device's current configuration (see
|
||||
:func:`~meshterm.core.device_config.build_snapshot`).
|
||||
|
||||
Returns:
|
||||
The drifted settings, in the registry's key order.
|
||||
"""
|
||||
remembered = store.settings(pubkey)
|
||||
if not remembered:
|
||||
return []
|
||||
drifted: list[SettingDrift] = []
|
||||
for key, saved in remembered.items():
|
||||
try:
|
||||
spec = get_spec(key)
|
||||
except DeviceConfigError:
|
||||
continue # a key the current registry no longer defines
|
||||
current = snapshot.get(key)
|
||||
if format_value(spec, saved) != format_value(spec, current):
|
||||
drifted.append(SettingDrift(key=key, remembered=saved, current=current))
|
||||
order = {spec.key: i for i, spec in enumerate(_ordered_specs())}
|
||||
return sorted(drifted, key=lambda d: order.get(d.key, len(order)))
|
||||
|
||||
|
||||
def _ordered_specs() -> list:
|
||||
"""The registry's settings in display order (imported lazily to avoid a load-time cost)."""
|
||||
from .device_config import DEVICE_SETTINGS
|
||||
|
||||
return DEVICE_SETTINGS
|
||||
|
||||
|
||||
async def restore(
|
||||
store: SettingsStore, device: "Device", snapshot: dict, keys: list[str]
|
||||
) -> int:
|
||||
"""Write remembered values for ``keys`` back onto the device, updating ``snapshot`` in place.
|
||||
|
||||
Applies each setting through its registry spec — the same path the editor uses — so coupled
|
||||
commands (radio, coordinates, tuning, telemetry) rebuild from the running snapshot and a
|
||||
setting can be restored in isolation. Best-effort per key: a setter a given firmware lacks is
|
||||
skipped rather than aborting the rest.
|
||||
|
||||
Args:
|
||||
store: The store holding the remembered values.
|
||||
device: The connected device to write to.
|
||||
snapshot: The device's current snapshot; mutated in place as values are applied so a
|
||||
later coupled key is rebuilt from current values.
|
||||
keys: The setting keys to restore (typically the drifted ones).
|
||||
|
||||
Returns:
|
||||
The number of settings successfully written.
|
||||
"""
|
||||
pubkey = str(snapshot.get("public_key") or "")
|
||||
remembered = store.settings(pubkey)
|
||||
restored = 0
|
||||
for key in keys:
|
||||
if key not in remembered:
|
||||
continue
|
||||
try:
|
||||
spec = get_spec(key)
|
||||
value = parse_value(spec, remembered[key], snapshot)
|
||||
await spec.apply(device, value, snapshot)
|
||||
except Exception: # noqa: BLE001 - a firmware-unsupported setting must not abort the rest
|
||||
continue
|
||||
snapshot[key] = value
|
||||
restored += 1
|
||||
return restored
|
||||
|
||||
|
||||
def adopt(store: SettingsStore, pubkey: str, snapshot: dict, keys: list[str]) -> None:
|
||||
"""Update the remembered copy of ``keys`` to the device's current values.
|
||||
|
||||
The other side of a drift: rather than restore the saved values, take the device's current
|
||||
ones as the new truth — so a change made outside MeshTerm is kept, not clobbered. A key the
|
||||
device no longer reports is forgotten instead of stored.
|
||||
|
||||
Args:
|
||||
store: The store to update.
|
||||
pubkey: The device's own public key.
|
||||
snapshot: The device's current snapshot to read values from.
|
||||
keys: The setting keys to adopt.
|
||||
"""
|
||||
for key in keys:
|
||||
current = snapshot.get(key)
|
||||
if current is None:
|
||||
store.forget(pubkey, key)
|
||||
else:
|
||||
store.remember(pubkey, key, current)
|
||||
@@ -329,6 +329,11 @@ async def _apply_setting(
|
||||
value = parse_value(spec, raw, snapshot)
|
||||
await spec.apply(device, value, snapshot)
|
||||
snapshot[key] = value
|
||||
# Remember what we set, keyed by the device's own public key, so a forgetful device (a
|
||||
# firmware-less radio bridge) can be offered its settings back on the next connect (see
|
||||
# meshterm.core.settings_store). Provenance-gated: only values changed through MeshTerm.
|
||||
if ctx.settings_store is not None:
|
||||
ctx.settings_store.remember(str(snapshot.get("public_key") or ""), key, value)
|
||||
ctx.ui.note(f"[ok]✓[/ok] [brand]{key}[/brand] = {format_value(spec, value)}")
|
||||
return 1
|
||||
|
||||
|
||||
@@ -578,6 +578,12 @@ async def _startup(ctx: AppContext) -> bool:
|
||||
# Float the skeleton card across that gap on any real link (see _busy_over_link).
|
||||
async with _busy_over_link(ctx, title="Starting up"):
|
||||
await _resume_monitor(ctx)
|
||||
# A forgetful device (a firmware-less radio bridge) may have lost settings you saved through
|
||||
# MeshTerm; offer to reconcile them on the splash before the menu paints. Quiet unless a
|
||||
# connected device actually drifts from what's remembered (see settings_offer).
|
||||
from .settings_offer import offer_remembered_settings
|
||||
|
||||
await offer_remembered_settings(ctx)
|
||||
return True
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
"""The connect-time offer to reconcile a device's settings with what MeshTerm remembers.
|
||||
|
||||
A firmware-less radio bridge forgets its configuration whenever it restarts, so the settings you
|
||||
changed through MeshTerm last session come back as firmware defaults. Unlike channels — silently
|
||||
replayed because filling an empty slot overwrites nothing — a setting is a single canonical
|
||||
value, so restoring a stale one could clobber a change made elsewhere. This surface therefore
|
||||
*asks* rather than acting: on connect, when the device's current settings drift from what
|
||||
MeshTerm saved for it, it offers to restore the saved values, adopt the device's current ones, or
|
||||
stop remembering the device (see :mod:`meshterm.core.settings_store`).
|
||||
|
||||
It runs once at startup, on the splash, only when a device is connected and something is actually
|
||||
remembered for it — so a firmware radio you don't manage through MeshTerm is never interrupted.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from rich.text import Text
|
||||
|
||||
from ..context import AppContext
|
||||
from ..core.device_config import build_snapshot, get_spec
|
||||
from ..core.settings_store import SettingDrift, adopt, restore, settings_drift
|
||||
from .menus import menu_rows
|
||||
from .tui import Separator
|
||||
|
||||
# The offer's actions, returned by the startup select.
|
||||
_RESTORE = "restore"
|
||||
_ADOPT = "adopt"
|
||||
_FORGET = "forget"
|
||||
|
||||
|
||||
async def offer_remembered_settings(ctx: AppContext) -> None:
|
||||
"""Offer to reconcile the connected device's settings with MeshTerm's saved copy.
|
||||
|
||||
Best-effort and quiet: does nothing without a live link, without anything remembered for the
|
||||
device, or when the device already matches. Never opens the radio itself — a deferred connect
|
||||
(``connect_on_start`` off) simply skips this, since ``is_connected`` is false. A failure here
|
||||
must not block reaching the menu, so everything is guarded.
|
||||
|
||||
Args:
|
||||
ctx: The shared application context.
|
||||
"""
|
||||
store = ctx.settings_store
|
||||
if store is None or not ctx.is_connected:
|
||||
return
|
||||
try:
|
||||
device = await ctx.device()
|
||||
info = await device.get_self_info()
|
||||
pubkey = str(info.get("public_key") or "")
|
||||
if not store.settings(pubkey):
|
||||
return # nothing remembered — a firmware radio pays only this cheap probe
|
||||
snapshot = await build_snapshot(device)
|
||||
drifted = settings_drift(store, pubkey, snapshot)
|
||||
except Exception as exc: # noqa: BLE001 - a probe failure must not block the menu
|
||||
ctx.log.debug("settings: drift check on connect failed: %s", exc)
|
||||
return
|
||||
if not drifted:
|
||||
return
|
||||
|
||||
choice = await _prompt(ctx, drifted)
|
||||
keys = [d.key for d in drifted]
|
||||
try:
|
||||
if choice == _RESTORE:
|
||||
restored = await restore(store, device, snapshot, keys)
|
||||
ctx.devstate.invalidate_config()
|
||||
ctx.log.info("settings: restored %d remembered setting(s) to the device", restored)
|
||||
elif choice == _ADOPT:
|
||||
adopt(store, pubkey, snapshot, keys)
|
||||
ctx.log.info("settings: adopted the device's values for %d setting(s)", len(keys))
|
||||
elif choice == _FORGET:
|
||||
store.forget_all(pubkey)
|
||||
ctx.log.info("settings: stopped remembering this device's settings")
|
||||
except Exception as exc: # noqa: BLE001 - acting on the choice must not block the menu
|
||||
ctx.log.debug("settings: reconciling on connect failed: %s", exc)
|
||||
|
||||
|
||||
async def _prompt(ctx: AppContext, drifted: list[SettingDrift]) -> object:
|
||||
"""Show the startup offer and return the chosen action, or ``None`` if dismissed."""
|
||||
from .. import copyright_notice
|
||||
from .logo import load_logo
|
||||
|
||||
count = len(drifted)
|
||||
plural = "" if count == 1 else "s"
|
||||
rows = menu_rows(
|
||||
[
|
||||
(
|
||||
"📂 Restore my saved settings",
|
||||
f"Write the {count} saved value{plural} back onto the device",
|
||||
_RESTORE,
|
||||
),
|
||||
(
|
||||
"💾 Keep the device's settings",
|
||||
"Update my saved copy to match the device now",
|
||||
_ADOPT,
|
||||
),
|
||||
(
|
||||
"🗑 Stop remembering this device",
|
||||
"Forget the saved settings and don't ask again",
|
||||
_FORGET,
|
||||
),
|
||||
]
|
||||
)
|
||||
items = [
|
||||
Separator(
|
||||
f" This device reports {count} setting{plural} that differ from what "
|
||||
"you last saved through MeshTerm:"
|
||||
),
|
||||
Separator(_drift_summary(drifted)),
|
||||
Separator(" "),
|
||||
*rows,
|
||||
]
|
||||
return await ctx.ui.select_startup(
|
||||
"Saved settings differ",
|
||||
items,
|
||||
default=_RESTORE,
|
||||
banner=load_logo(),
|
||||
footnote=copyright_notice(),
|
||||
)
|
||||
|
||||
|
||||
def _drift_summary(drifted: list[SettingDrift], *, cap: int = 6) -> Text:
|
||||
"""A muted one-line list of the drifted settings' labels, capped with a ``+N more`` tail."""
|
||||
labels = [get_spec(d.key).label for d in drifted]
|
||||
shown = labels[:cap]
|
||||
text = " " + ", ".join(shown)
|
||||
if len(labels) > cap:
|
||||
text += f", +{len(labels) - cap} more"
|
||||
return Text(text, style="muted")
|
||||
@@ -0,0 +1,231 @@
|
||||
"""Tests for remembering device settings across sessions (``core.settings_store``).
|
||||
|
||||
Covers the durable per-device store, the drift detection that compares remembered values to a
|
||||
device's live snapshot, and the ``restore``/``adopt`` reconcile actions the startup offer drives.
|
||||
The reconcile tests run against the :class:`MockDevice` simulator, whose configuration resets to
|
||||
firmware defaults each construction — standing in for the firmware-less radio bridge that forgets
|
||||
its settings on restart.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from rich.console import Console
|
||||
|
||||
from meshterm.context import AppContext
|
||||
from meshterm.core.admin_store import AdminStore
|
||||
from meshterm.core.config import Settings
|
||||
from meshterm.core.connection import make_device
|
||||
from meshterm.core.device_config import build_snapshot
|
||||
from meshterm.core.device_store import DeviceStore
|
||||
from meshterm.core.settings_store import (
|
||||
SettingsStore,
|
||||
adopt,
|
||||
restore,
|
||||
settings_drift,
|
||||
)
|
||||
from meshterm.persistence.repository import Repository
|
||||
from meshterm.tools.config import apply_ops
|
||||
|
||||
PUB_A = "aa" * 32
|
||||
PUB_B = "bb" * 32
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def ctx(tmp_path: Path) -> AppContext:
|
||||
"""A mock-backed application context with the plain (console) UI surface."""
|
||||
settings = Settings(config_dir=tmp_path, db_path=tmp_path / "cfg.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()
|
||||
|
||||
|
||||
# --- the store ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_store_round_trips_and_persists(tmp_path: Path) -> None:
|
||||
"""Remembered settings survive a fresh store instance, kept per device."""
|
||||
path = tmp_path / "settings.json"
|
||||
store = SettingsStore(path)
|
||||
store.remember(PUB_A, "name", "Ops-Node")
|
||||
store.remember(PUB_A, "radio_freq", 915.0)
|
||||
store.remember(PUB_B, "tx_power", 20)
|
||||
|
||||
reloaded = SettingsStore(path) # a new process reads the same file
|
||||
assert reloaded.settings(PUB_A) == {"name": "Ops-Node", "radio_freq": 915.0}
|
||||
assert reloaded.settings(PUB_B) == {"tx_power": 20} # a second device keeps its own set
|
||||
|
||||
|
||||
def test_store_replace_forget_and_forget_all(tmp_path: Path) -> None:
|
||||
"""Re-remembering a key replaces it; forget drops one; forget_all clears the device."""
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(PUB_A, "name", "First")
|
||||
store.remember(PUB_A, "name", "Second") # same key, new value
|
||||
store.remember(PUB_A, "tx_power", 20)
|
||||
assert store.settings(PUB_A) == {"name": "Second", "tx_power": 20}
|
||||
|
||||
store.forget(PUB_A, "name")
|
||||
assert store.settings(PUB_A) == {"tx_power": 20}
|
||||
|
||||
store.forget_all(PUB_A)
|
||||
assert store.settings(PUB_A) == {}
|
||||
# The device drops out of the persisted file entirely once it has nothing remembered.
|
||||
assert json.loads((tmp_path / "settings.json").read_text())["devices"] == {}
|
||||
|
||||
|
||||
def test_store_normalises_device_key(tmp_path: Path) -> None:
|
||||
"""A device key is matched case-insensitively and with an optional 0x prefix stripped."""
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember("AABB", "name", "Ops")
|
||||
assert store.settings("0xaabb") == {"name": "Ops"}
|
||||
|
||||
|
||||
def test_store_ignores_corrupt_file(tmp_path: Path) -> None:
|
||||
"""A garbage file reads as empty rather than raising, and stays writable."""
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text("not json at all", encoding="utf-8")
|
||||
store = SettingsStore(path)
|
||||
assert store.settings(PUB_A) == {}
|
||||
store.remember(PUB_A, "name", "Ops") # recovers and persists
|
||||
assert SettingsStore(path).settings(PUB_A) == {"name": "Ops"}
|
||||
|
||||
|
||||
def test_store_drops_malformed_values(tmp_path: Path) -> None:
|
||||
"""Non-scalar values (a container, a null) are skipped; scalar ones are kept."""
|
||||
path = tmp_path / "settings.json"
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"devices": {
|
||||
PUB_A: {"name": "Good", "bad_list": [1, 2], "bad_null": None, "tx_power": 20}
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert SettingsStore(path).settings(PUB_A) == {"name": "Good", "tx_power": 20}
|
||||
|
||||
|
||||
def test_store_ignores_non_scalar_remember(tmp_path: Path) -> None:
|
||||
"""Remembering a non-scalar value is a no-op — only strings, numbers, and bools persist."""
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(PUB_A, "name", ["not", "a", "scalar"])
|
||||
assert store.settings(PUB_A) == {}
|
||||
|
||||
|
||||
# --- drift -------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _mock_device():
|
||||
device = make_device(mock=True, port=None)
|
||||
await device.connect()
|
||||
pubkey = (await device.get_self_info())["public_key"]
|
||||
return device, pubkey
|
||||
|
||||
|
||||
async def test_drift_reports_only_changed_settings(tmp_path: Path) -> None:
|
||||
"""Drift lists remembered settings that differ from the device, in registry order."""
|
||||
device, pubkey = await _mock_device()
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(pubkey, "radio_freq", 915.0) # device default is 869.525 — drifts
|
||||
store.remember(pubkey, "name", "MockCompanion") # equals the device default — no drift
|
||||
|
||||
snapshot = await build_snapshot(device)
|
||||
drifted = settings_drift(store, pubkey, snapshot)
|
||||
assert [d.key for d in drifted] == ["radio_freq"]
|
||||
assert drifted[0].remembered == 915.0 and drifted[0].current == 869.525
|
||||
|
||||
|
||||
async def test_drift_empty_when_nothing_remembered(tmp_path: Path) -> None:
|
||||
"""A device with no remembered settings never shows drift (firmware radios pay nothing)."""
|
||||
device, pubkey = await _mock_device()
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
snapshot = await build_snapshot(device)
|
||||
assert settings_drift(store, pubkey, snapshot) == []
|
||||
|
||||
|
||||
async def test_drift_skips_unknown_keys(tmp_path: Path) -> None:
|
||||
"""A remembered key the registry no longer defines is ignored, not reported as drift."""
|
||||
device, pubkey = await _mock_device()
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(pubkey, "gone_from_registry", "whatever")
|
||||
snapshot = await build_snapshot(device)
|
||||
assert settings_drift(store, pubkey, snapshot) == []
|
||||
|
||||
|
||||
# --- restore / adopt ---------------------------------------------------------
|
||||
|
||||
|
||||
async def test_restore_writes_remembered_values_including_coupled(tmp_path: Path) -> None:
|
||||
"""Restore replays saved values onto the device, coupled radio fields rebuilt together."""
|
||||
device, pubkey = await _mock_device()
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(pubkey, "name", "Ops-Node")
|
||||
store.remember(pubkey, "radio_freq", 915.0)
|
||||
store.remember(pubkey, "radio_sf", 12) # a second radio field: exercises the coupled apply
|
||||
|
||||
snapshot = await build_snapshot(device)
|
||||
drifted = settings_drift(store, pubkey, snapshot)
|
||||
written = await restore(store, device, snapshot, [d.key for d in drifted])
|
||||
assert written == 3
|
||||
|
||||
after = await build_snapshot(device)
|
||||
assert after["name"] == "Ops-Node"
|
||||
assert after["radio_freq"] == 915.0
|
||||
assert after["radio_sf"] == 12
|
||||
assert after["radio_bw"] == 250.0 # untouched radio field preserved by the coupled rebuild
|
||||
# Idempotent: with the device now matching, there is nothing left to restore.
|
||||
assert settings_drift(store, pubkey, after) == []
|
||||
|
||||
|
||||
async def test_adopt_updates_store_to_device_values(tmp_path: Path) -> None:
|
||||
"""Adopt takes the device's current values as the new saved truth, clearing the drift."""
|
||||
device, pubkey = await _mock_device()
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(pubkey, "radio_freq", 915.0) # differs from the device's 869.525
|
||||
|
||||
snapshot = await build_snapshot(device)
|
||||
drifted = settings_drift(store, pubkey, snapshot)
|
||||
adopt(store, pubkey, snapshot, [d.key for d in drifted])
|
||||
|
||||
assert store.settings(pubkey)["radio_freq"] == 869.525 # now matches the device
|
||||
assert settings_drift(store, pubkey, snapshot) == []
|
||||
|
||||
|
||||
async def test_adopt_forgets_a_setting_the_device_no_longer_reports(tmp_path: Path) -> None:
|
||||
"""Adopting a key the device reports no value for forgets it rather than storing a null."""
|
||||
device, pubkey = await _mock_device()
|
||||
store = SettingsStore(tmp_path / "settings.json")
|
||||
store.remember(pubkey, "flood_scope", "#ops")
|
||||
|
||||
snapshot = await build_snapshot(device)
|
||||
snapshot.pop("flood_scope", None) # simulate firmware that doesn't report this field
|
||||
adopt(store, pubkey, snapshot, ["flood_scope"])
|
||||
assert "flood_scope" not in store.settings(pubkey)
|
||||
|
||||
|
||||
# --- write-through -----------------------------------------------------------
|
||||
|
||||
|
||||
async def test_apply_setting_remembers_through_the_config_executor(ctx) -> None:
|
||||
"""Every setting changed through the config executor is recorded under the device key."""
|
||||
device = await ctx.device()
|
||||
pubkey = (await device.get_self_info())["public_key"]
|
||||
snapshot = await build_snapshot(device)
|
||||
|
||||
changes, _ = await apply_ops(
|
||||
ctx, device, snapshot, [("set", "name", "Ops-Node"), ("set", "tx_power", "18")]
|
||||
)
|
||||
assert changes == 2
|
||||
assert ctx.settings_store.settings(pubkey) == {"name": "Ops-Node", "tx_power": 18}
|
||||
Reference in new issue
Block a user