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>
438 lines
16 KiB
Python
438 lines
16 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""Tests for the passive mesh monitor: listening, aggregation, and storage.
|
|
|
|
All the tests run against the :class:`MockDevice` simulator and an SQLite database on disk.
|
|
They do not need hardware.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from meshterm.core.connection import (
|
|
_MOCK_MONITOR_INTERVAL_S as _MOCK_INTERVAL,
|
|
)
|
|
from meshterm.core.connection import (
|
|
MockDevice,
|
|
ack_from_event,
|
|
message_from_event,
|
|
observation_from_event,
|
|
)
|
|
from meshterm.core.models import ChatMessage, HeardNode, Observation, utcnow
|
|
from meshterm.persistence.repository import Repository
|
|
from meshterm.services.monitor_service import MonitorService
|
|
|
|
|
|
class _StubContext:
|
|
"""A minimal substitute for :class:`~meshterm.context.AppContext` in the service tests."""
|
|
|
|
def __init__(self, repo: Repository, device: MockDevice) -> None:
|
|
self.repo = repo
|
|
self._device = device
|
|
self._events = None
|
|
self.profile_name = None
|
|
self.log = logging.getLogger("test.monitor")
|
|
|
|
async def device(self) -> MockDevice:
|
|
await self._device.connect()
|
|
return self._device
|
|
|
|
@property
|
|
def events(self):
|
|
"""Build a real event hub for this stub context at the first use, as AppContext does."""
|
|
from meshterm.services.event_hub import EventHub
|
|
|
|
if self._events is None:
|
|
self._events = EventHub(self)
|
|
return self._events
|
|
|
|
|
|
async def _collect_observations(device: MockDevice, duration_s: float) -> list[Observation]:
|
|
"""Subscribe, collect observations for a period, then unsubscribe. This is a test helper."""
|
|
seen: list[Observation] = []
|
|
|
|
def on_event(event) -> None:
|
|
if event.observation is not None:
|
|
seen.append(event.observation)
|
|
|
|
unsubscribe = await device.subscribe_events(on_event)
|
|
try:
|
|
await asyncio.sleep(duration_s)
|
|
finally:
|
|
unsubscribe()
|
|
return seen
|
|
|
|
|
|
async def test_subscribe_events_streams_observations_then_stops() -> None:
|
|
"""The event stream sends a burst at once, continues to send, and stops at unsubscribe."""
|
|
device = MockDevice()
|
|
await device.connect()
|
|
|
|
seen: list[Observation] = []
|
|
|
|
def on_event(event) -> None:
|
|
if event.observation is not None:
|
|
seen.append(event.observation)
|
|
|
|
unsubscribe = await device.subscribe_events(on_event)
|
|
try:
|
|
# The stream sends the first burst synchronously, before any await.
|
|
first_burst = len(seen)
|
|
assert first_burst >= 1
|
|
assert any(o.lat is not None for o in seen) # the burst has a repeater with a location
|
|
|
|
await asyncio.sleep(_MOCK_INTERVAL * 3)
|
|
assert len(seen) > first_burst # more packets arrived while the test did nothing
|
|
finally:
|
|
unsubscribe()
|
|
|
|
frozen = len(seen)
|
|
await asyncio.sleep(_MOCK_INTERVAL * 3)
|
|
assert len(seen) == frozen # no more packets after the unsubscribe
|
|
|
|
|
|
async def test_disconnect_stops_background_emitter() -> None:
|
|
"""A disconnect cancels each live subscription task. The task does not stay running."""
|
|
device = MockDevice()
|
|
await device.connect()
|
|
await device.subscribe_events(lambda _event: None)
|
|
assert device._bg_tasks # a background sender is running
|
|
await device.disconnect()
|
|
await asyncio.sleep(0) # let the cancellation finish
|
|
assert not device._bg_tasks
|
|
|
|
|
|
def test_heard_node_aggregation() -> None:
|
|
"""``from_observations`` gives the count, a robust SNR, the latest RSSI, and the location."""
|
|
now = utcnow()
|
|
obs = [
|
|
Observation(node="a1", name="Yagi", snr=4.0, rssi=-100.0, observed_at=now),
|
|
Observation(
|
|
node="a1",
|
|
name="Yagi",
|
|
snr=8.0,
|
|
rssi=-90.0,
|
|
lat=45.5,
|
|
lon=-73.5,
|
|
observed_at=now + timedelta(seconds=5),
|
|
),
|
|
Observation(
|
|
node="a1", name="Yagi", snr=6.0, rssi=-95.0, observed_at=now + timedelta(seconds=2)
|
|
),
|
|
]
|
|
node = HeardNode.from_observations("a1", obs)
|
|
assert node.count == 3
|
|
assert node.median_snr == 6.0 # median of [4, 8, 6]
|
|
assert node.best_snr == 8.0
|
|
assert node.last_seen == now + timedelta(seconds=5) # the newest observation
|
|
assert node.last_rssi == -90.0 # the RSSI of that newest observation
|
|
assert node.has_location # the location of one observation stays
|
|
|
|
|
|
def test_heard_node_latest_wins_for_rssi_and_location() -> None:
|
|
"""The newest observation gives the RSSI.
|
|
|
|
The newest observation with a location gives the coordinates.
|
|
"""
|
|
now = utcnow()
|
|
obs = [
|
|
Observation(node="b2", snr=1.0, rssi=-80.0, lat=10.0, lon=20.0, observed_at=now),
|
|
Observation(node="b2", snr=2.0, rssi=-70.0, observed_at=now + timedelta(seconds=10)),
|
|
]
|
|
node = HeardNode.from_observations("b2", obs)
|
|
assert node.last_rssi == -70.0 # the latest reading
|
|
assert (node.lat, node.lon) == (10.0, 20.0) # the latest observation that had a fix
|
|
|
|
|
|
def test_observation_from_event_parses_advert() -> None:
|
|
"""The event parser gets the node id, name, SNR, RSSI, and location from a payload."""
|
|
|
|
class _Event:
|
|
payload = {
|
|
"public_key": "AABBCC0011",
|
|
"adv_name": "Repeater",
|
|
"snr": "7.5",
|
|
"rssi": -92,
|
|
"adv_lat": 45.5,
|
|
"adv_lon": -73.6,
|
|
}
|
|
|
|
obs = observation_from_event(_Event(), "advert")
|
|
assert obs is not None
|
|
assert obs.node == "aabbcc0011"
|
|
assert obs.name == "Repeater"
|
|
assert obs.snr == 7.5
|
|
assert obs.rssi == -92.0
|
|
assert (obs.lat, obs.lon) == (45.5, -73.6)
|
|
|
|
|
|
def test_observation_from_event_keeps_full_public_key() -> None:
|
|
"""An advert with the whole key keeps it in ``public_key``, and ``node`` stays the 12-hex id."""
|
|
full = "aabbccddee11223344556677" + "00" * 20 # 64 hex
|
|
|
|
class _Event:
|
|
payload = {"public_key": full.upper(), "adv_name": "Repeater", "snr": 7.5}
|
|
|
|
obs = observation_from_event(_Event(), "advert")
|
|
assert obs is not None
|
|
assert obs.node == full[:12] # the stored 12-hex canonical id that all joins use
|
|
assert obs.public_key == full # the whole key, kept for a longer hash display
|
|
|
|
|
|
def test_observation_from_event_short_hash_leaves_no_full_key() -> None:
|
|
"""An advert with only a short hash stores only the id. The code does not make a full key."""
|
|
|
|
class _Event:
|
|
payload = {"hash": "aabbcc0011", "adv_name": "Rep"}
|
|
|
|
obs = observation_from_event(_Event(), "advert")
|
|
assert obs is not None and obs.node == "aabbcc0011" and obs.public_key is None
|
|
|
|
|
|
def test_observation_from_event_without_node_is_skipped() -> None:
|
|
"""A payload with no node identifier gives no observation."""
|
|
|
|
class _Event:
|
|
payload = {"snr": 3.0}
|
|
|
|
assert observation_from_event(_Event(), "advert") is None
|
|
|
|
|
|
def test_message_from_event_parses_direct_message() -> None:
|
|
"""A CONTACT_MSG_RECV payload becomes a direct message with a sender and a timestamp."""
|
|
from datetime import datetime, timezone
|
|
|
|
class _Event:
|
|
payload = {
|
|
"type": "PRIV",
|
|
"pubkey_prefix": "aabbccddeeff",
|
|
"text": "hello there",
|
|
"sender_timestamp": 1_700_000_000,
|
|
"txt_type": 0,
|
|
}
|
|
|
|
msg = message_from_event(_Event())
|
|
assert msg is not None
|
|
assert msg.text == "hello there"
|
|
assert msg.sender == "aabbccddeeff"
|
|
assert msg.is_channel is False
|
|
assert msg.channel is None
|
|
assert msg.sender_timestamp == datetime.fromtimestamp(1_700_000_000, tz=timezone.utc)
|
|
|
|
|
|
def test_message_from_event_parses_channel_message() -> None:
|
|
"""A CHANNEL_MSG_RECV payload becomes a channel message with no sender contact."""
|
|
|
|
class _Event:
|
|
payload = {"type": "CHAN", "channel_idx": 2, "text": "net tonight", "SNR": 5.0}
|
|
|
|
msg = message_from_event(_Event())
|
|
assert msg is not None
|
|
assert msg.is_channel is True
|
|
assert msg.channel == 2
|
|
assert msg.sender is None
|
|
assert msg.snr == 5.0
|
|
|
|
|
|
def test_message_from_event_without_text_is_skipped() -> None:
|
|
"""A payload with no text body gives no message."""
|
|
|
|
class _Event:
|
|
payload = {"pubkey_prefix": "aabb"}
|
|
|
|
assert message_from_event(_Event()) is None
|
|
|
|
|
|
def test_ack_from_event_extracts_code() -> None:
|
|
"""An ACK payload becomes an Ack with the correlation code."""
|
|
|
|
class _Event:
|
|
payload = {"code": "deadbeef"}
|
|
|
|
assert ack_from_event(_Event()).code == "deadbeef"
|
|
|
|
|
|
async def test_observations_round_trip_and_aggregate(tmp_path: Path) -> None:
|
|
"""The store keeps observations, and ``heard_nodes`` combines them again across runs."""
|
|
repo = Repository(tmp_path / "obs.db")
|
|
run_id = repo.start_run("monitor", {"duration": 1})
|
|
|
|
device = MockDevice()
|
|
await device.connect()
|
|
observations = await _collect_observations(device, 0.2)
|
|
for obs in observations:
|
|
repo.record_observation(run_id, obs)
|
|
assert observations
|
|
|
|
nodes = repo.heard_nodes()
|
|
assert nodes
|
|
# The aggregate must count each reception, except the packet-log rows. The SNR of such
|
|
# a row belongs to the last relay and not to the node, so these rows go to the topology.
|
|
receptions = [o for o in observations if o.kind != "packet"]
|
|
assert sum(n.count for n in nodes) == len(receptions)
|
|
# The simulator also overhears relayed packets. The store keeps them with their path.
|
|
packets = repo.packet_paths()
|
|
assert packets and all(p.hops for p in packets)
|
|
# The nodes are in order, with the most recently heard node first.
|
|
assert nodes == sorted(nodes, key=lambda n: n.last_seen, reverse=True)
|
|
repo.close()
|
|
|
|
|
|
def test_observation_count_totals_every_run(tmp_path: Path) -> None:
|
|
"""``observation_count`` adds the observations of all runs.
|
|
|
|
It also works for an empty database.
|
|
"""
|
|
repo = Repository(tmp_path / "count.db")
|
|
assert repo.observation_count() == 0
|
|
run_id = repo.start_run("monitor", {})
|
|
repo.record_observation(run_id, Observation(node="a1", snr=1.0))
|
|
repo.record_observation(run_id, Observation(node="b2", snr=2.0))
|
|
assert repo.observation_count() == 2
|
|
repo.close()
|
|
|
|
|
|
def test_observation_full_key_round_trips_into_heard_nodes(tmp_path: Path) -> None:
|
|
"""The store keeps a captured full public key, and it shows on the combined HeardNode."""
|
|
full = "3d63c6429436" + "ab" * 26 # 64 hex digits. The node is the first 12.
|
|
repo = Repository(tmp_path / "keys.db")
|
|
run_id = repo.start_run("monitor", {})
|
|
# One advert has the whole key. A later advert for the same node has only the prefix.
|
|
repo.record_observation(run_id, Observation(node=full[:12], public_key=full, snr=5.0))
|
|
repo.record_observation(run_id, Observation(node=full[:12], snr=6.0))
|
|
(node,) = repo.heard_nodes()
|
|
assert node.node == full[:12] # the id stays the 12-hex prefix
|
|
assert node.public_key == full # the full key comes from the row that had it
|
|
assert node.count == 2
|
|
repo.close()
|
|
|
|
|
|
def test_last_heard_by_node_is_the_latest_non_packet_reception(tmp_path: Path) -> None:
|
|
"""The last heard time of a node is its latest reception that is not a packet.
|
|
|
|
This is the evidence of the heard time, with one stamp for each node. It does not use
|
|
``packet`` rows. The contacts merge uses it, and compares it with the advert time that
|
|
the sender stamped on the device. The query excludes ``packet`` rows in the same way as
|
|
``heard_nodes``, because when a node hears a relayed packet, it hears the last relay and
|
|
not the originator.
|
|
"""
|
|
repo = Repository(tmp_path / "heard.db")
|
|
run_id = repo.start_run("monitor", {})
|
|
early = utcnow() - timedelta(days=4)
|
|
late = utcnow() - timedelta(minutes=5)
|
|
repo.record_observation(run_id, Observation(node="aa" * 6, observed_at=early))
|
|
repo.record_observation(run_id, Observation(node="aa" * 6, observed_at=late))
|
|
repo.record_observation(
|
|
run_id, Observation(node="bb" * 6, kind="packet", path="c1", observed_at=late)
|
|
)
|
|
heard = repo.last_heard_by_node()
|
|
assert heard == {"aa" * 6: late} # the latest wins, and the relayed packet credits no node
|
|
repo.close()
|
|
|
|
|
|
def test_last_message_by_peer_credits_only_inbound_direct_messages(tmp_path: Path) -> None:
|
|
"""When MeshTerm receives a DM, it hears the sender.
|
|
|
|
A sent DM and channel traffic do not count.
|
|
"""
|
|
repo = Repository(tmp_path / "msgs.db")
|
|
early = utcnow() - timedelta(days=4)
|
|
late = utcnow() - timedelta(minutes=5)
|
|
repo.record_chat_message(ChatMessage(text="hi", peer="AbAbAb", created_at=early))
|
|
repo.record_chat_message(ChatMessage(text="again", peer="ababab", created_at=late))
|
|
repo.record_chat_message(
|
|
ChatMessage(text="mine", peer="cdcdcd", outbound=True, created_at=late)
|
|
)
|
|
repo.record_chat_message(
|
|
ChatMessage(text="all", is_channel=True, channel_id="pub", created_at=late)
|
|
)
|
|
assert repo.last_message_by_peer() == {"ababab": late} # the store keeps peers in lower case
|
|
repo.close()
|
|
|
|
|
|
async def test_monitor_service_records_while_hub_pumps(tmp_path: Path) -> None:
|
|
"""``start()`` subscribes before the hub opens. It logs observations when the hub pumps."""
|
|
repo = Repository(tmp_path / "svc.db")
|
|
ctx = _StubContext(repo, MockDevice())
|
|
service = MonitorService(ctx)
|
|
|
|
assert not service.active
|
|
await service.start() # no device is necessary: no connection is open yet
|
|
assert service.active
|
|
await ctx.events.start() # the hub opens, and the recording catches the first burst
|
|
|
|
await asyncio.sleep(_MOCK_INTERVAL * 3) # let packets stream in during "other work"
|
|
assert service.session_count > 0
|
|
# The database was empty at the start, so the total is what this session captured.
|
|
assert service.total_count() == service.session_count
|
|
|
|
await service.stop()
|
|
assert not service.active
|
|
assert repo.observation_count() == service.session_count # all the observations were logged
|
|
|
|
frozen = service.session_count
|
|
await asyncio.sleep(_MOCK_INTERVAL * 3)
|
|
assert service.session_count == frozen # the recording did stop
|
|
|
|
# The hub is always on, and it continues to listen after the recording stops. Shut it
|
|
# down cleanly, so that the background sender task of the simulator does not continue
|
|
# after the test.
|
|
await ctx.events.stop()
|
|
await ctx._device.disconnect()
|
|
repo.close()
|
|
|
|
|
|
async def test_monitor_service_counts_every_packet_kind(tmp_path: Path) -> None:
|
|
"""The activity histogram counts observations, messages, and acks in the same way.
|
|
|
|
The newest bucket is first.
|
|
"""
|
|
from meshterm.core.events import MeshEvent
|
|
from meshterm.core.models import Ack, Message
|
|
from meshterm.services.monitor_service import ACTIVITY_BUCKETS
|
|
|
|
repo = Repository(tmp_path / "act.db")
|
|
ctx = _StubContext(repo, MockDevice())
|
|
service = MonitorService(ctx)
|
|
await service.start()
|
|
|
|
assert service.activity_histogram() == (0,) * ACTIVITY_BUCKETS
|
|
hub = ctx.events
|
|
hub.publish(MeshEvent.observation_event(Observation(node="n1", name="n1")))
|
|
hub.publish(MeshEvent.message_event(Message(text="hi", sender="n1")))
|
|
hub.publish(MeshEvent.ack_event(Ack(code="01")))
|
|
|
|
histogram = service.activity_histogram()
|
|
assert len(histogram) == ACTIVITY_BUCKETS
|
|
# All three are in the newest bucket. If a slot rolls over, they are in the newest two.
|
|
assert sum(histogram[:2]) == 3 and sum(histogram) == 3
|
|
|
|
await service.stop()
|
|
hub.publish(MeshEvent.ack_event(Ack(code=2)))
|
|
assert sum(service.activity_histogram()) == 3 # the counter stopped when the service stopped
|
|
repo.close()
|
|
|
|
|
|
async def test_monitor_service_start_without_device_is_safe(tmp_path: Path) -> None:
|
|
"""The recording can start before a device exists. A silent session leaves no run."""
|
|
|
|
class _NoDevice(_StubContext):
|
|
async def device(self) -> MockDevice:
|
|
raise ValueError("no companion device selected")
|
|
|
|
repo = Repository(tmp_path / "nodev.db")
|
|
service = MonitorService(_NoDevice(repo, MockDevice()))
|
|
|
|
await service.start() # the call does not use the device, so it cannot fail
|
|
assert service.active
|
|
assert service.session_count == 0
|
|
|
|
await service.stop()
|
|
assert not service.active
|
|
assert repo.observation_count() == 0
|
|
assert repo.list_runs() == [] # the service opens the run row only if needed, so none
|
|
repo.close()
|