Jean-Pierre MartineauandClaude Opus 5.5
11b5274b31
Each test docstring states the behaviour that the test proves, in the simple present. No assertion message, test id, or other string literal changed. Four statements now agree with the code: the Trace highlight stays on the last row, a menu count, and two others. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
421 lines
17 KiB
Python
421 lines
17 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""Watchtower tests: the storage of the store and the rules of the sentinel.
|
|
|
|
The tests run the service synchronously through :meth:`note` and :meth:`evaluate`, with a
|
|
stub context (the same method as the monitor tests). Thus the tests do not use timers or
|
|
hardware.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from meshterm.core.models import Observation, utcnow
|
|
from meshterm.core.watch_store import ALERT_CAP, OFF, WatchStore
|
|
from meshterm.persistence.repository import Repository
|
|
from meshterm.services.watchtower import SNR_WINDOW, WatchtowerService
|
|
|
|
NODE = "3d" * 6
|
|
|
|
|
|
class _StubContext:
|
|
"""A minimal substitute for :class:`~meshterm.context.AppContext` in the rule tests."""
|
|
|
|
def __init__(self, tmp_path: Path) -> None:
|
|
self.repo = Repository(tmp_path / "wt.db")
|
|
self.watch_store = WatchStore(tmp_path / "watchtower.json")
|
|
self.log = logging.getLogger("test.watchtower")
|
|
|
|
|
|
def _service(tmp_path: Path) -> WatchtowerService:
|
|
ctx = _StubContext(tmp_path)
|
|
service = WatchtowerService(ctx)
|
|
service._known = set() # ``start()`` gets this from the DB. The history is empty here.
|
|
return service
|
|
|
|
|
|
def _obs(node=NODE, name="Hub", snr=None, age_s=0, kind="advert") -> Observation:
|
|
return Observation(
|
|
node=node,
|
|
name=name,
|
|
kind=kind,
|
|
snr=snr,
|
|
observed_at=utcnow() - timedelta(seconds=age_s),
|
|
)
|
|
|
|
|
|
# --- the store --------------------------------------------------------------------------
|
|
|
|
|
|
def test_store_watch_round_trips_and_persists(tmp_path: Path) -> None:
|
|
"""Watched nodes and their rules survive a new store instance."""
|
|
path = tmp_path / "watchtower.json"
|
|
store = WatchStore(path)
|
|
seen = utcnow() - timedelta(hours=2)
|
|
store.watch(NODE, "Hub", last_seen=seen)
|
|
store.set_silence(NODE, 24)
|
|
store.set_snr_watch(NODE, False)
|
|
|
|
again = WatchStore(path)
|
|
entry = again.watched()[NODE]
|
|
assert entry.name == "Hub" and entry.silence_hours == 24 and not entry.snr_watch
|
|
assert entry.last_heard == seen
|
|
again.unwatch(NODE)
|
|
assert not WatchStore(path).watched()
|
|
|
|
|
|
def test_store_alert_log_caps_acks_and_clears(tmp_path: Path) -> None:
|
|
"""The store adds alerts with the newest first.
|
|
|
|
It limits their number, acknowledges them, and clears them.
|
|
"""
|
|
store = WatchStore(tmp_path / "watchtower.json")
|
|
first = store.add_alert("silence", "Hub", "quiet")
|
|
store.add_alert("snr", "Hub", "sagging")
|
|
assert [a.kind for a in store.alerts()] == ["snr", "silence"]
|
|
assert store.unacked_count() == 2
|
|
|
|
store.ack(first.ident)
|
|
assert store.unacked_count() == 1
|
|
store.ack_all()
|
|
assert store.unacked_count() == 0
|
|
store.clear_acked()
|
|
assert store.alerts() == []
|
|
|
|
for i in range(ALERT_CAP + 10):
|
|
store.add_alert("new-node", f"n{i}", "hi")
|
|
assert len(store.alerts()) == ALERT_CAP
|
|
assert store.alerts()[0].label == f"n{ALERT_CAP + 9}" # the newest alert stays at the cap
|
|
|
|
|
|
def test_store_note_heard_updates_and_refreshes_name(tmp_path: Path) -> None:
|
|
"""When MeshTerm hears a watched node, the store moves its mark forward.
|
|
|
|
The store also takes the new name.
|
|
"""
|
|
store = WatchStore(tmp_path / "watchtower.json")
|
|
store.watch(NODE, NODE)
|
|
later = utcnow() + timedelta(minutes=5)
|
|
store.note_heard(NODE, when=later, name="Hilltop-Repeater")
|
|
entry = store.watched()[NODE]
|
|
assert entry.last_heard == later and entry.name == "Hilltop-Repeater"
|
|
|
|
|
|
# --- the silence rule ---------------------------------------------------------------------
|
|
|
|
|
|
def test_silence_fires_once_then_rearms_on_recovery(tmp_path: Path) -> None:
|
|
"""The silence rule alarms one time when the node is quiet past the threshold.
|
|
|
|
When MeshTerm hears the node again, the rule notes the recovery and arms again.
|
|
"""
|
|
service = _service(tmp_path)
|
|
store = service._ctx.watch_store
|
|
store.watch(NODE, "Hub", last_seen=utcnow() - timedelta(hours=13))
|
|
|
|
service.evaluate() # the default threshold is 12 h, and the node is past it
|
|
service.evaluate() # the alarm is latched, so there is no duplicate
|
|
alerts = store.alerts()
|
|
assert [a.kind for a in alerts] == ["silence"]
|
|
assert "12 h" in alerts[0].message
|
|
|
|
service.note(_obs()) # the node comes back
|
|
alerts = store.alerts()
|
|
assert [a.kind for a in alerts] == ["recovered", "silence"]
|
|
assert store.watched()[NODE].silent_since is None # armed again
|
|
|
|
service.evaluate() # the node was heard just now, so it is quiet again only after 12 h more
|
|
assert len(store.alerts()) == 2
|
|
|
|
|
|
def test_silence_respects_off_and_unheard(tmp_path: Path) -> None:
|
|
"""A rule that is OFF never alarms, also when the mark is very old."""
|
|
service = _service(tmp_path)
|
|
store = service._ctx.watch_store
|
|
store.watch(NODE, "Hub", last_seen=utcnow() - timedelta(days=30))
|
|
store.set_silence(NODE, OFF)
|
|
service.evaluate()
|
|
assert store.alerts() == []
|
|
|
|
|
|
# --- the SNR rule ---------------------------------------------------------------------
|
|
|
|
|
|
def test_snr_sag_alerts_with_cooldown(tmp_path: Path) -> None:
|
|
"""A clear drop in the median SNR gives one alert. The next packets do not give more alerts."""
|
|
service = _service(tmp_path)
|
|
store = service._ctx.watch_store
|
|
store.watch(NODE, "Hub")
|
|
|
|
for _ in range(SNR_WINDOW):
|
|
service.note(_obs(snr=8.0))
|
|
for _ in range(SNR_WINDOW):
|
|
service.note(_obs(snr=-2.0))
|
|
sags = [a for a in store.alerts() if a.kind == "snr"]
|
|
assert len(sags) == 1
|
|
assert "+8.0" in sags[0].message and "-2.0" in sags[0].message
|
|
|
|
service.note(_obs(snr=-2.0)) # the SNR is still low, but the cooldown is active
|
|
assert len([a for a in store.alerts() if a.kind == "snr"]) == 1
|
|
|
|
|
|
def test_snr_ignores_packet_rows_and_disabled_watch(tmp_path: Path) -> None:
|
|
"""The SNR that a relay measured on a packet keeps the rule quiet.
|
|
|
|
A switch that is off also keeps the rule quiet.
|
|
"""
|
|
service = _service(tmp_path)
|
|
store = service._ctx.watch_store
|
|
store.watch(NODE, "Hub")
|
|
store.set_snr_watch(NODE, False)
|
|
for snr in (9.0,) * SNR_WINDOW + (-9.0,) * SNR_WINDOW:
|
|
service.note(_obs(snr=snr))
|
|
assert [a for a in store.alerts() if a.kind == "snr"] == []
|
|
|
|
store.set_snr_watch(NODE, True)
|
|
for snr in (9.0,) * SNR_WINDOW + (-9.0,) * SNR_WINDOW:
|
|
service.note(_obs(snr=snr, kind="packet"))
|
|
assert [a for a in store.alerts() if a.kind == "snr"] == []
|
|
|
|
|
|
# --- the new-node rule ------------------------------------------------------------------
|
|
|
|
|
|
def test_new_node_announced_once_ever(tmp_path: Path) -> None:
|
|
"""An id that MeshTerm hears for the first time gives one alert.
|
|
|
|
Repeats and restarts give none.
|
|
"""
|
|
service = _service(tmp_path)
|
|
store = service._ctx.watch_store
|
|
service.note(_obs(node="f7" * 6, name="Newcomer"))
|
|
service.note(_obs(node="f7" * 6, name="Newcomer"))
|
|
news = [a for a in store.alerts() if a.kind == "new-node"]
|
|
assert len(news) == 1 and news[0].label == "Newcomer"
|
|
assert store.known_contains("f7" * 6) # the store keeps it, so restarts give no alert too
|
|
|
|
|
|
def test_new_node_rule_can_be_switched_off(tmp_path: Path) -> None:
|
|
"""If the toggle is off, the store remembers a node that is heard for the first time.
|
|
|
|
The store never announces it.
|
|
"""
|
|
service = _service(tmp_path)
|
|
store = service._ctx.watch_store
|
|
store.set_new_node_alerts(False)
|
|
service.note(_obs(node="c9" * 6))
|
|
assert store.alerts() == []
|
|
assert store.known_contains("c9" * 6)
|
|
|
|
|
|
# --- the screen's rows ---------------------------------------------------------------
|
|
|
|
|
|
def test_store_round_trips_node_type(tmp_path: Path) -> None:
|
|
"""The store keeps the advertised type of a starred node. An old entry loads as None."""
|
|
path = tmp_path / "watch.json"
|
|
store = WatchStore(path)
|
|
store.watch("aa" * 6, "Roof", node_type=2)
|
|
store.watch("bb" * 6, "Old-style") # no type, as in an entry from before the field existed
|
|
|
|
reloaded = WatchStore(path)
|
|
assert reloaded.watched()["aa" * 6].node_type == 2
|
|
assert reloaded.watched()["bb" * 6].node_type is None
|
|
|
|
|
|
def test_watched_row_type_glyph_and_hued_name(tmp_path: Path) -> None:
|
|
"""A watched row starts with its type glyph, and the name has a hue from the key.
|
|
|
|
The glyph keeps its own colour. The name takes the hue that comes from the key of the
|
|
entry. The silence still shows in the tail at the end.
|
|
"""
|
|
from meshterm.core.watch_store import WatchedNode
|
|
from meshterm.ui.theme import name_style
|
|
from meshterm.ui.watchtower_screen import _watched_row
|
|
from meshterm.ui.widgets import NODE_GLYPHS
|
|
|
|
entry = WatchedNode(key="a1" * 6, name="Roof", node_type=2, last_heard=utcnow())
|
|
row = _watched_row(entry, lambda key: None)
|
|
glyph, glyph_style = NODE_GLYPHS[2]
|
|
assert row.plain.startswith(f"{glyph} Roof")
|
|
assert any(s.style == glyph_style and s.start == 0 for s in row.spans)
|
|
name_at = row.plain.index("Roof")
|
|
assert any(
|
|
s.style == name_style("Roof", "a1" * 6) and s.start <= name_at < s.end for s in row.spans
|
|
)
|
|
|
|
# An old entry (with no stored type) uses the resolver of the contact table.
|
|
legacy = WatchedNode(key="a1" * 6, name="Roof")
|
|
resolved = _watched_row(legacy, lambda key: 2)
|
|
assert resolved.plain.startswith(f"{glyph} ")
|
|
|
|
# The ⚠ tail shows the silence. The colour of the glyph never shows it.
|
|
quiet = WatchedNode(key="a1" * 6, name="Roof", node_type=2, silent_since=utcnow())
|
|
assert "⚠ silent" in _watched_row(quiet, lambda key: None).plain
|
|
|
|
|
|
def test_alert_row_hues_the_label_by_resolved_key(tmp_path: Path) -> None:
|
|
"""The label of an alert that is not acknowledged takes its hue from the key.
|
|
|
|
An acknowledged alert is muted.
|
|
"""
|
|
from meshterm.core.watch_store import Alert
|
|
from meshterm.ui.theme import name_style
|
|
from meshterm.ui.watchtower_screen import _alert_lanes
|
|
|
|
def key_of(label):
|
|
return "d4" * 6 if label == "Roof" else None
|
|
|
|
alert = Alert(ident=1, when=utcnow(), kind="silence", label="Roof", message="quiet")
|
|
row = _alert_lanes(alert, key_of)
|
|
at = row.plain.index("Roof")
|
|
assert any(s.style == name_style("Roof", "d4" * 6) and s.start <= at < s.end for s in row.spans)
|
|
|
|
acked = Alert(ident=2, when=utcnow(), kind="silence", label="Roof", message="quiet", acked=True)
|
|
acked_row = _alert_lanes(acked, key_of)
|
|
at = acked_row.plain.index("Roof")
|
|
assert any(s.style == "muted" and s.start <= at < s.end for s in acked_row.spans)
|
|
|
|
|
|
def test_alert_row_leads_node_name_with_type_glyph(tmp_path: Path) -> None:
|
|
"""The type glyph that nodes share comes before the node name.
|
|
|
|
The glyph has its own colour, and it is muted when the alert is acknowledged.
|
|
"""
|
|
from meshterm.core.watch_store import Alert
|
|
from meshterm.ui.watchtower_screen import _alert_lanes
|
|
from meshterm.ui.widgets import DEFAULT_GLYPH, NODE_GLYPHS
|
|
|
|
def key_of(label):
|
|
return None
|
|
|
|
def type_of(label):
|
|
return 2 if label == "Roof" else None # Roof advertises itself as a repeater
|
|
|
|
glyph, glyph_style = NODE_GLYPHS[2]
|
|
|
|
alert = Alert(ident=1, when=utcnow(), kind="silence", label="Roof", message="quiet")
|
|
row = _alert_lanes(alert, key_of, type_of)
|
|
assert f"{glyph} Roof" in row.plain # the glyph is directly to the left of the name
|
|
gi = row.plain.index(glyph)
|
|
assert any(s.style == glyph_style and s.start <= gi < s.end for s in row.spans)
|
|
|
|
# An unknown type, and a kind with no key such as courier, use the glyph of a plain node.
|
|
ghost = Alert(ident=2, when=utcnow(), kind="courier", label="Ghost", message="gave up")
|
|
assert f"{DEFAULT_GLYPH[0]} Ghost" in _alert_lanes(ghost, key_of).plain
|
|
|
|
# An acknowledged alert mutes the glyph, with the rest of its history.
|
|
acked = Alert(ident=3, when=utcnow(), kind="silence", label="Roof", message="quiet", acked=True)
|
|
acked_row = _alert_lanes(acked, key_of, type_of)
|
|
gi = acked_row.plain.index(glyph)
|
|
assert any(s.style == "muted" and s.start <= gi < s.end for s in acked_row.spans)
|
|
|
|
|
|
def test_alert_rows_pin_their_lanes_and_scroll_only_the_message() -> None:
|
|
"""←→ slide the message of an alert. The marker, age, kind, glyph, and node stay in place.
|
|
|
|
The lanes before the ``—`` show which alert this is (JP, 2026-08-10). The user can read
|
|
a long message to its end, and the lanes must not go off the left edge. They are the
|
|
part that already fits. If they scrolled, there would be no gain, and the row would
|
|
lose its identity.
|
|
"""
|
|
from meshterm.core.watch_store import Alert
|
|
from meshterm.ui.tui import Choice
|
|
from meshterm.ui.watchtower_screen import _alert_lanes, _menu_items
|
|
|
|
alert = Alert(
|
|
ident=1,
|
|
when=utcnow(),
|
|
kind="silence",
|
|
label="Roof",
|
|
message="nothing heard for 6 h " + "and counting " * 6,
|
|
)
|
|
items = _menu_items([alert], {}, False)
|
|
row = next(it for it in items if isinstance(it, Choice) and it.value == ("ack", alert.ident))
|
|
|
|
lanes = _alert_lanes(alert, lambda label: None)
|
|
assert row.hscroll_from == lanes.cell_len
|
|
assert row.label.plain.startswith(lanes.plain)
|
|
assert lanes.plain.endswith(" — ") # the lead-in stays with the head that it introduces
|
|
assert row.label.plain[row.hscroll_from :] == alert.message # and the scroll run is the text
|
|
|
|
|
|
def _word_starts(items: list, words: dict) -> dict:
|
|
"""The display cell where the first word of each row starts, by row value.
|
|
|
|
The function counts cells and not characters. A ``⭐`` of two cells and a ``✓`` of one
|
|
cell are both one character. A count of characters hid a label that started one column
|
|
too early.
|
|
"""
|
|
from rich.cells import cell_len
|
|
|
|
from meshterm.ui.tui import Choice
|
|
|
|
starts = {}
|
|
for item in items:
|
|
if isinstance(item, Choice) and item.value in words:
|
|
plain = item.label.plain
|
|
starts[item.value] = cell_len(plain[: plain.index(words[item.value])])
|
|
assert starts.keys() == words.keys(), "every row the test names is on the screen"
|
|
return starts
|
|
|
|
|
|
def test_watchtower_actions_start_every_label_in_the_same_cell() -> None:
|
|
"""The action rows share one icon column, across the section headings of the list.
|
|
|
|
``✓`` and ``🗑`` use one cell, and ``⭐`` and ``🔔`` use two. Thus the bulk actions
|
|
started their words one column to the left of *Watch a node…* and the new-node toggle.
|
|
On the PicoCalc the icons are not there, and their padding must go with them.
|
|
"""
|
|
from rich.cells import cell_len
|
|
|
|
from meshterm.core.watch_store import Alert
|
|
from meshterm.platforms import PICOCALC_LYRA, REGULAR, set_platform
|
|
from meshterm.ui.watchtower_screen import _ACK_ALL, _CLEAR, _TOGGLE_NEW, _WATCH, _menu_items
|
|
|
|
assert {cell_len("✓"), cell_len("⭐")} == {1, 2}, "a list of one icon width proves nothing"
|
|
alerts = [
|
|
Alert(ident=1, when=utcnow(), kind="silence", label="Roof", message="quiet"),
|
|
Alert(ident=2, when=utcnow(), kind="snr", label="Roof", message="sag", acked=True),
|
|
]
|
|
words = {
|
|
_ACK_ALL: "Acknowledge",
|
|
_CLEAR: "Clear",
|
|
_WATCH: "Watch",
|
|
_TOGGLE_NEW: "New-node",
|
|
}
|
|
assert set(_word_starts(_menu_items(alerts, {}, True), words).values()) == {3}
|
|
try:
|
|
set_platform(PICOCALC_LYRA)
|
|
assert set(_word_starts(_menu_items(alerts, {}, True), words).values()) == {0}
|
|
finally:
|
|
set_platform(REGULAR)
|
|
|
|
|
|
def test_node_rules_start_every_label_and_value_in_the_same_cell() -> None:
|
|
"""The words of the rule dialog share one column, and its values share another.
|
|
|
|
This is true on both platforms. ``🕒`` and ``📶`` use two cells, and ``✗`` uses one.
|
|
Thus *Stop watching this node*
|
|
started one column too early. The code once aligned the values with spaces that a
|
|
person typed. Thus the test checks the values too: the values stay in one column, for
|
|
each width of the icon column.
|
|
"""
|
|
from meshterm.core.watch_store import WatchedNode
|
|
from meshterm.platforms import PICOCALC_LYRA, REGULAR, set_platform
|
|
from meshterm.ui.watchtower_screen import _rule_items
|
|
|
|
entry = WatchedNode(key="a1" * 6, name="Roof", silence_hours=12, snr_watch=True)
|
|
words = {"silence": "Silence", "snr": "SNR", "unwatch": "Stop"}
|
|
values = {"silence": "after 12 h", "snr": "on"}
|
|
try:
|
|
for platform, start in ((REGULAR, 3), (PICOCALC_LYRA, 0)):
|
|
set_platform(platform)
|
|
items = _rule_items(entry)
|
|
assert set(_word_starts(items, words).values()) == {start}
|
|
assert len(set(_word_starts(items, values).values())) == 1
|
|
finally:
|
|
set_platform(REGULAR)
|