Jean-Pierre MartineauandClaude Opus 5.5
efc02ddced
A final pass over the docstrings and comments: "the own X of Y" is now "Y's own X" (84 places), a navigation entry is a screen on the stack, our node is our node, a paint is a paint, the quoted UX labels are in code font, and a list of punctuation glyphs is in code font. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1018 lines
39 KiB
Python
1018 lines
39 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
||
"""Smoke tests for the skeleton: models, mock device, service, and persistence.
|
||
|
||
These tests run without hardware. They use the :class:`MockDevice` simulator.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from meshterm.core.connection import DeviceCommandError, MockDevice, clamp_tx_power
|
||
from meshterm.core.models import Contact, TraceResult, TraceStats
|
||
from meshterm.persistence.repository import Repository
|
||
from meshterm.services import trace_runner
|
||
|
||
|
||
def test_clamp_tx_power() -> None:
|
||
"""The function clamps TX power to the supported range."""
|
||
assert clamp_tx_power(-5) == 1
|
||
assert clamp_tx_power(99) == 22
|
||
assert clamp_tx_power(14) == 14
|
||
|
||
|
||
def test_advert_time_refuses_a_future_sender_clock() -> None:
|
||
"""The stamp of an advert is from the clock of the sender. A future stamp is refused.
|
||
|
||
A plausible stamp converts. A node with a wrong clock advertises timestamps from the
|
||
future. If the code accepts such a timestamp, the contact shows "heard now" and is at
|
||
the top of each list that is sorted by heard time, until the wall clock catches up.
|
||
Normal skew (less than the
|
||
tolerance) stays: its age clamps to "now" for a short time and corrects itself. The
|
||
usual never-heard forms (zero, absent, or garbage) still become None.
|
||
"""
|
||
from meshterm.core.models import ADVERT_FUTURE_SKEW_S, advert_time, utcnow
|
||
|
||
past = int(utcnow().timestamp()) - 3600
|
||
when = advert_time(past)
|
||
assert when is not None and int(when.timestamp()) == past
|
||
assert advert_time(int(utcnow().timestamp()) + 60) is not None # normal skew
|
||
assert advert_time(int(utcnow().timestamp()) + ADVERT_FUTURE_SKEW_S + 60) is None
|
||
assert advert_time(0) is None
|
||
assert advert_time(None) is None
|
||
assert advert_time("garbage") is None
|
||
|
||
|
||
def test_trace_stats_aggregation() -> None:
|
||
"""Robust statistics show the successes and the bottleneck SNR."""
|
||
from meshterm.core.models import Hop
|
||
|
||
traces = [
|
||
TraceResult(target="x", success=True, hops=[Hop(0, "a", 5.0), Hop(1, "b", -2.0)]),
|
||
TraceResult(target="x", success=False),
|
||
TraceResult(target="x", success=True, hops=[Hop(0, "a", 7.0), Hop(1, "b", 0.0)]),
|
||
]
|
||
stats = TraceStats.from_traces("x", traces)
|
||
assert stats.samples == 3
|
||
assert stats.successes == 2
|
||
assert stats.success_rate == pytest.approx(2 / 3)
|
||
assert stats.median_min_snr == pytest.approx(-1.0) # median of [-2.0, 0.0]
|
||
|
||
|
||
def test_parse_trace_path_mixes_names_and_hex() -> None:
|
||
"""Names and hex can be mixed. The hex width stays, and the names are cut to match."""
|
||
contacts = [Contact(name="Alice", key_prefix="d4e5f6a7")]
|
||
# 1-byte hops: hex, a contact name (the case does not matter), hex again
|
||
assert trace_runner.parse_trace_path("3d, alice ,f2", contacts) == "3d,d4,f2"
|
||
# the width comes from the hex tokens that the user types, with no cut to 1 byte
|
||
assert trace_runner.parse_trace_path("a1b2c3", contacts) == "a1b2c3"
|
||
|
||
|
||
def test_parse_trace_path_preserves_three_byte_width() -> None:
|
||
"""A width of 3 bytes (6 hex digits) stays, and the names are cut to that width."""
|
||
contacts = [Contact(name="Alice", key_prefix="d4e5f6a7b8c9")]
|
||
assert trace_runner.parse_trace_path("3d5f7a,Alice,f2a1b3", contacts) == "3d5f7a,d4e5f6,f2a1b3"
|
||
|
||
|
||
def test_parse_trace_path_rejects_mixed_widths() -> None:
|
||
"""All the hex hops must have the same number of bytes."""
|
||
with pytest.raises(ValueError):
|
||
trace_runner.parse_trace_path("3d,a1b2c3", contacts=[])
|
||
|
||
|
||
def test_last_traced_by_name_folds_hex_to_contacts_and_keeps_latest() -> None:
|
||
"""The TRACED lane of the Trace picker joins each stored target to its contact.
|
||
|
||
A node can be traced as "alice" on one day and by its key prefix on another day. The
|
||
lane shows one last-traced time for it: the later of the two times. A name that looks
|
||
like hex stays a name (the code matches it by name, not by address). A prefix that
|
||
names no contact is removed.
|
||
"""
|
||
from datetime import datetime, timezone
|
||
|
||
from meshterm.tools.trace import _last_traced_by_name
|
||
|
||
contacts = [
|
||
Contact(name="Alice", public_key="d4e5f6a7" + "00" * 28, key_prefix="d4e5f6a7"),
|
||
Contact(name="cafe", public_key="12ab34cd" + "00" * 28, key_prefix="12ab34cd"),
|
||
]
|
||
|
||
def when(day: int) -> datetime:
|
||
return datetime(2026, 7, day, tzinfo=timezone.utc)
|
||
|
||
traced = {
|
||
"d4e5f6": when(1), # Alice, by an early trace with the key prefix
|
||
"Alice": when(9), # Alice by name, at a later time: the latest time must win
|
||
"cafe": when(5), # a name that looks like hex, matched as a name
|
||
"beefbeef": when(3), # a prefix that names no contact: removed
|
||
}
|
||
assert _last_traced_by_name(contacts, traced) == {"Alice": when(9), "cafe": when(5)}
|
||
|
||
|
||
def test_node_type_resolver_matches_hop_hash_to_contact_type() -> None:
|
||
"""The hash of a relay resolves to the node type of its contact (a prefix in either)."""
|
||
from meshterm.core.models import NODE_TYPE_REPEATER
|
||
|
||
type_of = trace_runner.make_node_type_resolver(
|
||
[
|
||
Contact(
|
||
name="Repeater",
|
||
public_key="3d63c6" + "00" * 26,
|
||
key_prefix="3d63c6429436",
|
||
node_type=NODE_TYPE_REPEATER,
|
||
),
|
||
Contact(name="Typeless", public_key="a1b2c3" + "00" * 26, key_prefix="a1b2c3d4"),
|
||
]
|
||
)
|
||
assert type_of("3d63") == NODE_TYPE_REPEATER # a short hash is the prefix of the full key
|
||
assert type_of("a1b2") is None # a known contact, but it advertised no type
|
||
assert type_of("ffff") is None # an unknown node
|
||
assert type_of(None) is None # our device, passed through
|
||
|
||
|
||
def test_key_resolver_expands_a_stored_prefix_to_the_full_key() -> None:
|
||
"""The stored 12-hex prefix of a heard node resolves to the whole public key of the contact."""
|
||
full = "3d63c6429436" + "ab" * 26 # 64 hex
|
||
resolve = trace_runner.make_key_resolver(
|
||
[
|
||
Contact(name="Repeater", public_key=full, key_prefix="3d63c6429436"),
|
||
Contact(name="Keyless", public_key="", key_prefix="a1b2c3d4"),
|
||
]
|
||
)
|
||
assert resolve("3d63c6429436") == full # the stored prefix starts the key
|
||
assert resolve("3d63") == full # each shorter slice of it, too
|
||
assert resolve("ffffffffffff") == "ffffffffffff" # no contact: the prefix stays
|
||
assert resolve("a1b2c3d4") == "a1b2c3d4" # the contact has no full key to expand to
|
||
assert resolve(None) is None
|
||
|
||
|
||
def test_path_hash_flags_power_of_two_only() -> None:
|
||
"""The trace flags encode the hash width as 1 << s, so only 1, 2, 4, and 8 bytes map."""
|
||
assert trace_runner.path_hash_flags(1) == 0
|
||
assert trace_runner.path_hash_flags(2) == 1
|
||
assert trace_runner.path_hash_flags(4) == 2
|
||
assert trace_runner.path_hash_flags(8) == 3
|
||
assert trace_runner.path_hash_flags(3) is None
|
||
|
||
|
||
def test_parse_trace_hops_reads_path_snr() -> None:
|
||
"""The SNR of each hop is in the payload['path'] dicts. The last node has no hash. It counts."""
|
||
from meshterm.core.connection import parse_trace_hops
|
||
|
||
payload = {
|
||
"path": [
|
||
{"hash": "3d", "snr": 4.5},
|
||
{"hash": "f2", "snr": -2.0},
|
||
{"snr": 1.25}, # last node: our device, no hash
|
||
]
|
||
}
|
||
hops = parse_trace_hops(payload)
|
||
assert [h.node for h in hops] == ["3d", "f2", None]
|
||
assert [h.snr for h in hops] == [4.5, -2.0, 1.25]
|
||
assert [h.index for h in hops] == [0, 1, 2]
|
||
|
||
|
||
def test_parse_trace_hops_empty_without_path() -> None:
|
||
"""A reply with no parsed path gives no hops (the SNR shows as n/a)."""
|
||
from meshterm.core.connection import parse_trace_hops
|
||
|
||
assert parse_trace_hops({}) == []
|
||
|
||
|
||
def test_route_text_annotates_nodes_with_command_width_hash() -> None:
|
||
"""The route shows the hash of each node at the path-hash width of the command."""
|
||
from meshterm.core.models import Hop
|
||
from meshterm.ui.widgets import _route_text
|
||
|
||
resolve = trace_runner.make_node_resolver(
|
||
[Contact(name="Alice", public_key="3d63c6" + "00" * 26, key_prefix="3d63c6429436")]
|
||
)
|
||
# A known node (resolves to Alice), then the reply that returns to our device (node=None).
|
||
result = TraceResult(
|
||
target="Alice",
|
||
success=True,
|
||
hops=[Hop(0, "3d63c6", 12.0), Hop(1, None, 12.0)],
|
||
path_hash_bytes=2, # the command used 2-byte hashes
|
||
)
|
||
plain = _route_text(result, "Me", resolve).plain
|
||
assert "Alice" in plain
|
||
assert "(3d63)" in plain # cut to 2 bytes, not the full 3d63c6
|
||
assert "3d63c6" not in plain
|
||
assert plain.count("Me") == 2 # our device at both ends, with no hash
|
||
assert "Me (" not in plain.replace(" ", " ") # the device has no annotation
|
||
|
||
|
||
def test_route_text_annotates_our_device_with_hash() -> None:
|
||
"""If the function has our key, both endpoints (us) have our hash at the command width."""
|
||
from meshterm.core.models import Hop
|
||
from meshterm.ui.widgets import _route_text
|
||
|
||
result = TraceResult(
|
||
target="x",
|
||
success=True,
|
||
hops=[Hop(0, "3d63", 12.0), Hop(1, None, 12.0)],
|
||
path_hash_bytes=2, # the command used 2-byte hashes
|
||
)
|
||
plain = _route_text(result, "Me", device_hash="a1b2c3" + "00" * 29).plain.replace("\xa0", " ")
|
||
assert plain.count("Me (a1b2)") == 2 # our device at both ends, with our hash
|
||
assert "a1b2c3" not in plain # cut to the 2-byte width of the command
|
||
|
||
|
||
def test_traces_table_annotates_links_with_hashes() -> None:
|
||
"""The From→To column of each trace shows ``name (hash)``, also for our device."""
|
||
from meshterm.core.models import Hop
|
||
from meshterm.ui.widgets import traces_table
|
||
|
||
resolve = trace_runner.make_node_resolver(
|
||
[Contact(name="Alice", public_key="3d63c6" + "00" * 26, key_prefix="3d63c6429436")]
|
||
)
|
||
traces = [
|
||
TraceResult(
|
||
target="Alice",
|
||
success=True,
|
||
hops=[Hop(0, "3d63c6", 12.0), Hop(1, None, 9.0)],
|
||
path_hash_bytes=2, # the command used 2-byte hashes
|
||
)
|
||
]
|
||
table = traces_table(traces, "Me", resolve, device_hash="a1b2c3" + "00" * 29)
|
||
links = [cell.plain.replace("\xa0", " ") for cell in table.columns[1].cells]
|
||
assert any("Me (a1b2)" in link for link in links) # our device has its hash
|
||
assert any("Alice (3d63)" in link for link in links) # cut to the 2-byte width
|
||
assert not any("3d63c6" in link for link in links)
|
||
|
||
|
||
def test_route_text_unknown_node_shows_hash_only() -> None:
|
||
"""An unresolved node shows only its hash, cut to the command width."""
|
||
from meshterm.core.models import Hop
|
||
from meshterm.ui.widgets import _route_text
|
||
|
||
result = TraceResult(
|
||
target="x",
|
||
success=True,
|
||
hops=[Hop(0, "aa11bb", 12.0), Hop(1, None, 12.0)],
|
||
path_hash_bytes=1, # a width of 1 byte
|
||
)
|
||
plain = _route_text(result, "Me").plain.replace(" ", " ")
|
||
assert "→ aa →" in plain # a hash of 1 byte, no name, no parentheses
|
||
assert "aa11bb" not in plain
|
||
|
||
|
||
def test_route_text_contains_no_non_breaking_spaces() -> None:
|
||
"""prompt_toolkit draws U+00A0 as an underscore, so route text must never have one."""
|
||
from meshterm.core.models import Hop
|
||
from meshterm.ui.widgets import _route_text
|
||
|
||
result = TraceResult(
|
||
target="x",
|
||
success=True,
|
||
hops=[Hop(0, "3d63", 12.0), Hop(1, None, 12.0)],
|
||
path_hash_bytes=2,
|
||
)
|
||
plain = _route_text(result, "Me", device_hash="a1b2" + "00" * 30).plain
|
||
assert "\xa0" not in plain
|
||
assert plain == "Me (a1b2) → 3d63 → Me (a1b2)"
|
||
|
||
|
||
def test_parse_trace_path_rejects_unknown_token() -> None:
|
||
"""A token that is not a known contact and is not valid hex is an error."""
|
||
with pytest.raises(ValueError):
|
||
trace_runner.parse_trace_path("nope", contacts=[])
|
||
with pytest.raises(ValueError):
|
||
trace_runner.parse_trace_path(" ", contacts=[])
|
||
|
||
|
||
async def test_adopt_device_is_reused_without_reopening(tmp_path: Path) -> None:
|
||
"""``device()`` returns a device that the startup probe adopted, without a change."""
|
||
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.device_store import DeviceStore
|
||
|
||
settings = Settings(config_dir=tmp_path, db_path=tmp_path / "adopt.db")
|
||
ctx = AppContext(
|
||
console=Console(file=__import__("io").StringIO()),
|
||
settings=settings,
|
||
repo=Repository(settings.db_path),
|
||
device_store=DeviceStore(tmp_path / "devices.json"),
|
||
admin_store=AdminStore(tmp_path / "admin.json"),
|
||
)
|
||
try:
|
||
adopted = MockDevice()
|
||
await adopted.connect()
|
||
ctx.adopt_device(adopted)
|
||
# ``device()`` must return the same connection that we adopted. It must not open
|
||
# another connection.
|
||
assert await ctx.device() is adopted
|
||
finally:
|
||
ctx.repo.close()
|
||
|
||
|
||
async def test_mock_device_honors_forced_path() -> None:
|
||
"""A forced path sets the repeater hops, and the device adds a hop that returns to us."""
|
||
device = MockDevice()
|
||
await device.connect()
|
||
result = await device.run_trace("Alice", path="3d,f2,3d")
|
||
# Three forced repeater hops, and the hop with no hash for the reply that returns to us.
|
||
assert result.hop_count == 4
|
||
assert [h.node for h in result.hops] == ["3d", "f2", "3d", None]
|
||
|
||
|
||
async def test_mock_device_remove_contact_drops_it_from_the_table() -> None:
|
||
"""If the code removes a contact, the device deletes it from its table (by public key)."""
|
||
from meshterm.core.connection import ContactNotOnDeviceError
|
||
|
||
device = MockDevice()
|
||
await device.connect()
|
||
before = await device.get_contacts()
|
||
alice = next(c for c in before if c.name == "Alice")
|
||
await device.remove_contact(alice)
|
||
after = await device.get_contacts()
|
||
assert "Alice" not in {c.name for c in after}
|
||
assert len(after) == len(before) - 1
|
||
# If the code removes a contact that the device does not hold, the device refuses, as the
|
||
# firmware refuses. Thus the path "wasn't on the device, removed here anyway" of the
|
||
# screen can run on --mock.
|
||
with pytest.raises(ContactNotOnDeviceError):
|
||
await device.remove_contact(alice)
|
||
assert len(await device.get_contacts()) == len(after)
|
||
|
||
|
||
async def test_mock_device_blank_path_traces() -> None:
|
||
"""A blank or automatic path still gives a successful trace that has hops."""
|
||
device = MockDevice()
|
||
await device.connect()
|
||
# Try several times, because the simulator sometimes drops very weak links.
|
||
results = [await device.run_trace("Alice") for _ in range(10)]
|
||
assert any(r.success and r.hops for r in results)
|
||
|
||
|
||
def test_trace_timeout_scales_with_hops() -> None:
|
||
"""The reply wait grows by one allowance for each hop, within its documented limits."""
|
||
from meshterm.services.trace_runner import (
|
||
TRACE_TIMEOUT_BASE_S,
|
||
TRACE_TIMEOUT_CEILING_S,
|
||
TRACE_TIMEOUT_FLOOD_S,
|
||
TRACE_TIMEOUT_PER_HOP_S,
|
||
trace_timeout,
|
||
)
|
||
|
||
# The code cannot set a size for an unknown hop count (a flood with no path). Thus it
|
||
# uses the flat budget that it always used.
|
||
assert trace_timeout(0) == TRACE_TIMEOUT_FLOOD_S
|
||
assert trace_timeout(-3) == TRACE_TIMEOUT_FLOOD_S
|
||
|
||
# Each hop adds exactly one allowance for a hop to the fixed base.
|
||
assert trace_timeout(1) == TRACE_TIMEOUT_BASE_S + TRACE_TIMEOUT_PER_HOP_S
|
||
assert trace_timeout(5) == TRACE_TIMEOUT_BASE_S + 5 * TRACE_TIMEOUT_PER_HOP_S
|
||
# Thus a longer walk always has more time to come home than a shorter walk.
|
||
assert trace_timeout(8) > trace_timeout(3)
|
||
# The ceiling is the limit for a route that otherwise grows with no end.
|
||
assert trace_timeout(10_000) == TRACE_TIMEOUT_CEILING_S
|
||
|
||
|
||
async def test_run_trace_sizes_wait_to_the_forced_route(monkeypatch) -> None: # noqa: ANN001
|
||
"""The reply wait of a forced path has a size that depends on its hop count, not a flat default.
|
||
|
||
The real device must give the reply the budget that grows with the route. Thus a long
|
||
walk is not cut off before it can travel out and back.
|
||
"""
|
||
import asyncio as _asyncio
|
||
|
||
from meshterm.core import connection as connection_module
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
from meshterm.services.trace_runner import trace_timeout
|
||
|
||
seen: dict[str, float] = {}
|
||
real_wait_for = _asyncio.wait_for
|
||
|
||
async def spy_wait_for(awaitable, timeout): # noqa: ANN001, ANN202
|
||
seen["timeout"] = timeout
|
||
return await real_wait_for(awaitable, 0) # a miss, and the budget is not used
|
||
|
||
monkeypatch.setattr(connection_module.asyncio, "wait_for", spy_wait_for)
|
||
|
||
class _Commands:
|
||
async def send_trace(self, *, auth_code, tag, flags, path): # noqa: ANN001, ANN201
|
||
return None
|
||
|
||
class _MC:
|
||
commands = _Commands()
|
||
|
||
def subscribe(self, event_type, callback, attribute_filters=None): # noqa: ANN001, ANN201
|
||
return _Subscription()
|
||
|
||
class _Subscription:
|
||
def unsubscribe(self) -> None:
|
||
return None
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
device._mc = _MC()
|
||
|
||
# Five hops of one byte give a walk of five hops (the spec is already the mirrored
|
||
# round trip).
|
||
result = await device.run_trace("Alice", path="3d,f2,3d,f2,3d")
|
||
assert result.success is False
|
||
assert seen["timeout"] == trace_timeout(5)
|
||
|
||
# An explicit timeout still has priority over the automatic size.
|
||
await device.run_trace("Alice", path="3d,f2,3d,f2,3d", timeout=2.0)
|
||
assert seen["timeout"] == 2.0
|
||
|
||
|
||
async def test_run_trace_catches_a_reply_that_beats_the_send_acknowledgement() -> None:
|
||
"""A reply that arrives while ``send_trace`` is still blocked is still the reply to our trace.
|
||
|
||
The send acknowledgement of the companion has only its event type as a correlation. Thus
|
||
a command at the same time can consume our acknowledgement, and the send waits for its
|
||
own default time. In the past, the code listened for the tag only after the send
|
||
returned. It dropped each reply that arrived in that window, and the trace showed "no
|
||
reply" although the mesh answered it.
|
||
"""
|
||
import asyncio as _asyncio
|
||
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
class _Event:
|
||
payload = {"tag": 1, "path": [{"hash": "3d", "snr": 4.0}, {"snr": 2.0}]}
|
||
|
||
listeners: list = []
|
||
|
||
class _Commands:
|
||
async def send_trace(self, *, auth_code, tag, flags, path): # noqa: ANN001, ANN201
|
||
# The reply arrives before the acknowledgement: deliver it, then keep waiting.
|
||
for callback in listeners:
|
||
callback(_Event())
|
||
await _asyncio.sleep(0)
|
||
return None
|
||
|
||
async def send_appstart(self): # noqa: ANN201 - the read of the tx power after a success
|
||
return _SelfInfo()
|
||
|
||
class _SelfInfo:
|
||
payload = {"tx_power": 20}
|
||
|
||
class _MC:
|
||
commands = _Commands()
|
||
|
||
def subscribe(self, event_type, callback, attribute_filters=None): # noqa: ANN001, ANN201
|
||
listeners.append(callback)
|
||
return _Subscription()
|
||
|
||
class _Subscription:
|
||
def unsubscribe(self) -> None:
|
||
listeners.clear()
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
device._mc = _MC()
|
||
result = await device.run_trace("Alice", path="3d,f2,3d")
|
||
assert result.success is True
|
||
assert [h.node for h in result.hops] == ["3d", None]
|
||
|
||
|
||
def test_trace_edges_endpoints_are_our_device() -> None:
|
||
"""The first edge starts at us and the last edge returns to us (#3 and #4)."""
|
||
from meshterm.core.models import Hop
|
||
|
||
result = TraceResult(
|
||
target="Alice",
|
||
success=True,
|
||
hops=[Hop(0, "3d", 5.0), Hop(1, "f2", 1.0), Hop(2, None, -1.0)],
|
||
)
|
||
edges = result.edges(device_label="us")
|
||
assert [(e.origin, e.destination) for e in edges] == [
|
||
("us", "3d"),
|
||
("3d", "f2"),
|
||
("f2", "us"),
|
||
]
|
||
assert [e.snr for e in edges] == [5.0, 1.0, -1.0]
|
||
|
||
|
||
def test_trace_stats_reports_per_hop_medians() -> None:
|
||
"""The aggregation gives a median SNR for each hop position (#5)."""
|
||
from meshterm.core.models import Hop
|
||
|
||
traces = [
|
||
TraceResult(target="x", success=True, hops=[Hop(0, "a", 4.0), Hop(1, None, 0.0)]),
|
||
TraceResult(target="x", success=True, hops=[Hop(0, "a", 8.0), Hop(1, None, 2.0)]),
|
||
]
|
||
stats = TraceStats.from_traces("x", traces)
|
||
# The node identities are raw (None = our device). The code adds a label at render time.
|
||
assert [(h.origin, h.destination, h.median_snr) for h in stats.hop_snrs] == [
|
||
(None, "a", 6.0),
|
||
("a", None, 1.0),
|
||
]
|
||
|
||
|
||
async def test_mock_device_trace_is_unimodal_in_tx() -> None:
|
||
"""The simulator has its peak near its optimal TX power (the average over the noise)."""
|
||
device = MockDevice(optimal_tx=14)
|
||
await device.connect()
|
||
|
||
async def avg_snr(tx: int) -> float:
|
||
await device.set_tx_power(tx)
|
||
results = await trace_runner.run_traces(device, "Alice", samples=15, cooldown_s=0)
|
||
stats = TraceStats.from_traces("Alice", results)
|
||
return stats.median_min_snr or -99.0
|
||
|
||
low, peak, high = await avg_snr(2), await avg_snr(14), await avg_snr(22)
|
||
assert peak > low
|
||
assert peak > high
|
||
|
||
|
||
async def _setup_link(optimal_remote_tx: int = 20):
|
||
"""Return a connected mock device, the (admin, target) contacts, and the forced path.
|
||
|
||
The path goes admin → target. Thus the optimizer tunes the admin node and reads the
|
||
SNR that the target receives from it.
|
||
"""
|
||
device = MockDevice(optimal_remote_tx=optimal_remote_tx)
|
||
await device.connect()
|
||
contacts = await device.get_contacts()
|
||
admin = next(c for c in contacts if c.name == "Yagi-Repeater")
|
||
target = next(c for c in contacts if c.name == "Alice")
|
||
path = trace_runner.parse_trace_path(f"{admin.name},{target.name}", contacts)
|
||
assert await device.admin_login(admin, "admin")
|
||
return device, admin, target, path
|
||
|
||
|
||
async def test_admin_login_rejects_wrong_password() -> None:
|
||
"""The device refuses a wrong admin password, and the node stays closed to tuning."""
|
||
from meshterm.core.connection import LoginResult
|
||
|
||
device = MockDevice(admin_password="secret")
|
||
await device.connect()
|
||
admin = (await device.get_contacts())[0]
|
||
|
||
refused = await device.admin_login(admin, "nope")
|
||
assert refused is LoginResult.REFUSED and not refused
|
||
with pytest.raises(DeviceCommandError):
|
||
await device.set_remote_tx_power(admin, 18) # not logged in
|
||
accepted = await device.admin_login(admin, "secret")
|
||
assert accepted is LoginResult.ACCEPTED and accepted
|
||
await device.set_remote_tx_power(admin, 18) # allowed now
|
||
assert await device.get_remote_tx_power(admin) == 18
|
||
|
||
|
||
async def test_tx_optimizer_finds_simulator_peak(tmp_path: Path) -> None:
|
||
"""The optimizer converges near the simulated remote optimum, and it applies the value."""
|
||
from meshterm.services import tx_optimizer
|
||
|
||
device, admin, target, path = await _setup_link(optimal_remote_tx=20)
|
||
|
||
result = await tx_optimizer.optimize_tx_power(
|
||
device,
|
||
target.name,
|
||
admin,
|
||
path,
|
||
samples_per_level=8,
|
||
coarse_step=3,
|
||
cooldown_s=0,
|
||
)
|
||
assert abs(result.best_tx - 20) <= 3 # near the true remote peak
|
||
assert result.best_success_rate == 1.0 # reliability first: the winner never drops
|
||
assert result.applied
|
||
assert await device.get_remote_tx_power(admin) == result.best_tx # the node stays tuned
|
||
assert result.admin_node == "Yagi-Repeater"
|
||
|
||
|
||
async def test_tx_optimizer_no_apply_restores_original(tmp_path: Path) -> None:
|
||
"""If apply is off, the node stays at the power that it had at the start."""
|
||
from meshterm.services import tx_optimizer
|
||
|
||
device, admin, target, path = await _setup_link()
|
||
await device.set_remote_tx_power(admin, 15) # a power at the start that is known
|
||
|
||
result = await tx_optimizer.optimize_tx_power(
|
||
device,
|
||
target.name,
|
||
admin,
|
||
path,
|
||
samples_per_level=4,
|
||
coarse_step=4,
|
||
refine=False,
|
||
verify=False,
|
||
apply=False,
|
||
cooldown_s=0,
|
||
)
|
||
assert result.original_tx == 15
|
||
assert not result.applied
|
||
assert await device.get_remote_tx_power(admin) == 15 # restored, not the winner
|
||
|
||
|
||
async def test_tx_optimizer_reports_phases_in_order() -> None:
|
||
"""on_phase announces each enabled search stage, in the order coarse → refine → verify."""
|
||
from meshterm.services import tx_optimizer
|
||
|
||
device, admin, target, path = await _setup_link()
|
||
phases: list[str] = []
|
||
|
||
await tx_optimizer.optimize_tx_power(
|
||
device,
|
||
target.name,
|
||
admin,
|
||
path,
|
||
samples_per_level=2,
|
||
coarse_step=6,
|
||
apply=False,
|
||
cooldown_s=0,
|
||
on_phase=phases.append,
|
||
)
|
||
assert phases == list(tx_optimizer.PHASES)
|
||
|
||
phases.clear()
|
||
await tx_optimizer.optimize_tx_power(
|
||
device,
|
||
target.name,
|
||
admin,
|
||
path,
|
||
samples_per_level=2,
|
||
coarse_step=6,
|
||
refine=False,
|
||
verify=False,
|
||
apply=False,
|
||
cooldown_s=0,
|
||
on_phase=phases.append,
|
||
)
|
||
assert phases == ["coarse"] # the optimizer does not announce a stage that is off
|
||
|
||
|
||
def test_select_best_prefers_reliability_then_lowest_power() -> None:
|
||
"""A level with 100% reliability wins over a less reliable level with a higher SNR.
|
||
|
||
If the levels are equal, the lower TX wins.
|
||
"""
|
||
from meshterm.core.models import TraceStats, TxLevelResult
|
||
from meshterm.services.tx_optimizer import select_best
|
||
|
||
def level(tx: int, snr: float, successes: int, samples: int = 5) -> TxLevelResult:
|
||
return TxLevelResult(
|
||
tx_power=tx,
|
||
samples=samples,
|
||
successes=successes,
|
||
target_snr=snr,
|
||
score=snr,
|
||
stats=TraceStats(
|
||
target="t",
|
||
samples=samples,
|
||
successes=successes,
|
||
median_min_snr=snr,
|
||
median_rtt_ms=None,
|
||
),
|
||
)
|
||
|
||
# A level that is not reliable but has a great SNR must not win over a level that is
|
||
# fully reliable.
|
||
flaky = level(26, snr=12.0, successes=3)
|
||
solid = level(20, snr=7.0, successes=5)
|
||
assert select_best([flaky, solid]).tx_power == 20
|
||
|
||
# Among reliable levels with an SNR in the tolerance, the lowest TX wins.
|
||
a = level(18, snr=7.4, successes=5)
|
||
b = level(22, snr=8.0, successes=5)
|
||
assert select_best([a, b], snr_tolerance=1.0).tx_power == 18
|
||
|
||
|
||
def test_round_trip_path_mirrors_back_to_a_reachable_node() -> None:
|
||
"""The trace path goes out to the target and back, so a node that is near can answer."""
|
||
from meshterm.services.tx_optimizer import _round_trip_path
|
||
|
||
assert _round_trip_path(["3f", "f2"]) == "3f,f2,3f"
|
||
assert _round_trip_path(["3d", "3f", "f2"]) == "3d,3f,f2,3f,3d"
|
||
assert _round_trip_path(["f2"]) == "f2" # one hop: nothing to mirror
|
||
|
||
|
||
def test_trace_target_snr_matches_target_hop_not_position() -> None:
|
||
"""The code reads the SNR of the target from its hash, also when it is the turn-around hop."""
|
||
from meshterm.core.models import Hop
|
||
from meshterm.services.tx_optimizer import trace_target_snr
|
||
|
||
# Round trip 3f -> f2 -> 3f -> us: we want the f2 hop (index 1). We do not want the
|
||
# later 3f hop or the return hop that has no hash.
|
||
trace = TraceResult(
|
||
target="f2",
|
||
success=True,
|
||
hops=[Hop(0, "3f", 4.0), Hop(1, "f2", 7.5), Hop(2, "3f", 3.0), Hop(3, None, 4.0)],
|
||
)
|
||
assert trace_target_snr(trace, "f2") == 7.5
|
||
assert trace_target_snr(TraceResult(target="f2", success=False), "f2") is None
|
||
|
||
|
||
class _Event:
|
||
def __init__(self, payload: dict) -> None:
|
||
self.payload = payload
|
||
|
||
def is_error(self) -> bool:
|
||
return False
|
||
|
||
|
||
def _mc_with(contacts: dict, path_hash_mode: int = 2):
|
||
"""Build a minimal mock ``MeshCore`` client that has the trace commands that the tests use.
|
||
|
||
The contacts arrive in the same way as the radio sends them: one record frame for each
|
||
contact, then the closing frame with the whole table (refer to :class:`_FakeMeshCore`).
|
||
"""
|
||
from meshcore import EventType
|
||
|
||
mc = _FakeMeshCore(
|
||
[(EventType.NEXT_CONTACT, info) for info in contacts.values()]
|
||
+ [(EventType.CONTACTS, contacts)]
|
||
)
|
||
|
||
async def get_path_hash_mode() -> int:
|
||
return path_hash_mode
|
||
|
||
mc.commands.get_path_hash_mode = get_path_hash_mode
|
||
return mc
|
||
|
||
|
||
async def test_contact_with_future_advert_stamp_reads_as_never_heard() -> None:
|
||
"""A contact with an advert that a wrong (future) clock stamped arrives with no heard time.
|
||
|
||
The firmware stores the timestamp in the advert without a change, and it only
|
||
increases it (its replay filter). Thus a wrong future stamp stays until real time
|
||
passes it. The conversion refuses the stamp here, so the contact does not show as
|
||
"heard in the future".
|
||
"""
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
from meshterm.core.models import utcnow
|
||
|
||
now = int(utcnow().timestamp())
|
||
mc = _mc_with(
|
||
{
|
||
"Bogus-Clock": {
|
||
"adv_name": "Bogus-Clock",
|
||
"public_key": "aa" * 32,
|
||
"last_advert": now + 7 * 86400,
|
||
},
|
||
"Honest": {"adv_name": "Honest", "public_key": "bb" * 32, "last_advert": now - 300},
|
||
}
|
||
)
|
||
device = MeshCoreDevice(port="COM-test")
|
||
device._mc = mc
|
||
by_name = {c.name: c for c in await device.get_contacts()}
|
||
assert by_name["Bogus-Clock"].last_seen is None # refused, so not "heard in the future"
|
||
assert by_name["Honest"].last_seen is not None
|
||
|
||
|
||
async def test_three_byte_route_appends_destination_hash() -> None:
|
||
"""A learned route with many hops goes through the repeaters, the target, then back (#1).
|
||
|
||
Two things are important for a region that is not a power of two (hashes of 3 bytes).
|
||
First, each routing hash of 3 bytes becomes its first 2 bytes (the widest trace width
|
||
that the protocol can represent). Second, the destination's own hash is added as
|
||
the last outbound hop, because a trace replies only when its destination is the last
|
||
outbound hop. The trace protocol has no separate return-path field, so the code
|
||
mirrors the same repeaters back afterwards. If it does not, no node relays the reply
|
||
home.
|
||
"""
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
mc = _mc_with(
|
||
{
|
||
"Repeater": {
|
||
"adv_name": "Repeater",
|
||
"public_key": "aabbcc" + "00" * 29,
|
||
# Two hops, and each hop is a routing hash of 3 bytes (mode 2 => size 3).
|
||
"out_path": "112233445566",
|
||
"out_path_len": 2,
|
||
"out_path_hash_mode": 2,
|
||
}
|
||
}
|
||
)
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
resolved = await device._trace_path_to_contact(mc, "Repeater")
|
||
assert resolved is not None
|
||
path_bytes, flags = resolved
|
||
# Two repeater hops cut to 2 bytes, the target's own hash (2 bytes), then the same
|
||
# two repeaters mirrored back to us.
|
||
assert path_bytes == bytes.fromhex("11224455aabb44551122")
|
||
assert flags == trace_runner.path_hash_flags(2) # a width of 2 bytes
|
||
|
||
|
||
async def test_direct_neighbor_resolves_to_destination_hash() -> None:
|
||
"""A direct neighbor (no learned route) resolves to one destination hop.
|
||
|
||
This was checked on hardware. These contacts report ``out_path_len == -1`` (no stored
|
||
route), but they answer a trace with one hop that is addressed to their own hash. If
|
||
the mode of the contact is unknown (-1), the path-hash mode of the region (3 bytes)
|
||
sets the width. The code cuts it to the 2 bytes that the protocol can represent.
|
||
"""
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
mc = _mc_with(
|
||
{
|
||
"Neighbor": {
|
||
"adv_name": "Neighbor",
|
||
"public_key": "3d63c6" + "00" * 29,
|
||
# A direct neighbor, as the firmware reports it: no learned route.
|
||
"out_path": "",
|
||
"out_path_len": -1,
|
||
"out_path_hash_mode": -1,
|
||
}
|
||
},
|
||
path_hash_mode=2,
|
||
)
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
resolved = await device._trace_path_to_contact(mc, "Neighbor")
|
||
assert resolved is not None
|
||
path_bytes, flags = resolved
|
||
# Only the target's own hash. The region width of 3 is cut to a trace hop of 2 bytes.
|
||
assert path_bytes == bytes.fromhex("3d63")
|
||
assert flags == trace_runner.path_hash_flags(2)
|
||
|
||
|
||
async def test_unknown_contact_resolves_to_none() -> None:
|
||
"""A target that is not known gives ``None``, so the trace can use no path."""
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
mc = _mc_with(
|
||
{
|
||
"Somebody": {
|
||
"adv_name": "Somebody",
|
||
"public_key": "abcdef" + "00" * 29,
|
||
"out_path": "",
|
||
"out_path_len": -1,
|
||
"out_path_hash_mode": -1,
|
||
}
|
||
}
|
||
)
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
resolved = await device._trace_path_to_contact(mc, "Nobody")
|
||
assert resolved is None
|
||
|
||
|
||
class _FakeSubscription:
|
||
"""One :meth:`_FakeMeshCore.subscribe` registration. ``unsubscribe`` cancels it."""
|
||
|
||
def __init__(self, bus: dict, event_type, callback) -> None: # noqa: ANN001
|
||
"""Register ``callback`` for ``event_type`` on ``bus``."""
|
||
self._bus = bus
|
||
self._key = event_type
|
||
self._callback = callback
|
||
bus.setdefault(event_type, []).append(callback)
|
||
|
||
def unsubscribe(self) -> None:
|
||
"""Remove the registration, as the subscription of the library does."""
|
||
self._bus.get(self._key, []).remove(self._callback)
|
||
|
||
|
||
class _FakeMeshCore:
|
||
"""A companion that answers a contacts request with a script of frames.
|
||
|
||
Each script entry is one ``(EventType, payload)`` frame. The class sends it to each
|
||
listener that subscribed to it. Thus a test can stream records, close the dump, refuse
|
||
the dump, or say nothing.
|
||
"""
|
||
|
||
def __init__(self, *scripts) -> None:
|
||
"""Take one frame script for each contacts request. Answer them in order."""
|
||
from types import SimpleNamespace
|
||
|
||
self._bus: dict = {}
|
||
self._scripts = list(scripts)
|
||
self.requests = 0
|
||
self.commands = SimpleNamespace(get_contacts_async=self._request)
|
||
|
||
def subscribe(self, event_type, callback, attribute_filters=None): # noqa: ANN001, ANN201
|
||
"""Register a listener. Return an object that can unsubscribe it."""
|
||
return _FakeSubscription(self._bus, event_type, callback)
|
||
|
||
async def _request(self, lastmod: int = 0) -> None:
|
||
"""Answer with the next script of frames."""
|
||
from meshcore.events import Event
|
||
|
||
self.requests += 1
|
||
script = self._scripts.pop(0) if self._scripts else []
|
||
for event_type, payload in script:
|
||
for callback in list(self._bus.get(event_type, [])):
|
||
callback(Event(event_type, payload))
|
||
|
||
|
||
async def test_contacts_payload_retries_a_refused_read() -> None:
|
||
"""If a companion refuses the dump, the code tries again. It does not show the error."""
|
||
from meshcore import EventType
|
||
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
record = {"adv_name": "Repeater", "public_key": "aabbcc" + "00" * 29}
|
||
mc = _FakeMeshCore(
|
||
[(EventType.ERROR, {"reason": "ERR_CODE_BAD_STATE"})],
|
||
[(EventType.ERROR, {"reason": "ERR_CODE_BAD_STATE"})],
|
||
[(EventType.NEXT_CONTACT, record), (EventType.CONTACTS, {"Repeater": record})],
|
||
)
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
payload = await device._contacts_payload(mc, retries=3, delay=0, idle=0.05)
|
||
assert "Repeater" in payload
|
||
assert mc.requests == 3 # refused two times, served on the third try
|
||
|
||
|
||
async def test_contacts_payload_waits_out_a_slow_dump() -> None:
|
||
"""A table that streams for longer than a fixed deadline still completes.
|
||
|
||
The idle clock starts again at each record. Thus the read ends only when the radio is
|
||
quiet. The time of the whole dump does not end it.
|
||
"""
|
||
from meshcore import EventType
|
||
from meshcore.events import Event
|
||
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
class _SlowMeshCore(_FakeMeshCore):
|
||
async def _request(self, lastmod: int = 0) -> None:
|
||
self.requests += 1
|
||
table = {}
|
||
for index in range(8):
|
||
await asyncio.sleep(0.04) # each gap is in the idle window
|
||
record = {"adv_name": f"Node{index}", "public_key": f"{index:02x}" + "00" * 31}
|
||
table[record["adv_name"]] = record
|
||
for callback in list(self._bus.get(EventType.NEXT_CONTACT, [])):
|
||
callback(Event(EventType.NEXT_CONTACT, record))
|
||
for callback in list(self._bus.get(EventType.CONTACTS, [])):
|
||
callback(Event(EventType.CONTACTS, table))
|
||
|
||
mc = _SlowMeshCore()
|
||
device = MeshCoreDevice(port="COM-test")
|
||
payload = await device._contacts_payload(mc, retries=0, delay=0, idle=0.1)
|
||
assert len(payload) == 8 # 0.32 s of streaming, with an idle window of 0.1 s
|
||
|
||
|
||
async def test_contacts_payload_ignores_an_error_meant_for_another_command() -> None:
|
||
"""An ERROR with no correlation in the dump is for another command. It does not end the read."""
|
||
from meshcore import EventType
|
||
|
||
from meshterm.core.connection import MeshCoreDevice
|
||
|
||
record = {"adv_name": "Repeater", "public_key": "aabbcc" + "00" * 29}
|
||
mc = _FakeMeshCore(
|
||
[
|
||
(EventType.NEXT_CONTACT, record),
|
||
(EventType.ERROR, {"reason": "ERR_CODE_BAD_STATE"}), # for a battery poll, not for us
|
||
(EventType.CONTACTS, {"Repeater": record}),
|
||
]
|
||
)
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
payload = await device._contacts_payload(mc, retries=0, delay=0, idle=0.05)
|
||
assert "Repeater" in payload
|
||
assert mc.requests == 1
|
||
|
||
|
||
async def test_contacts_payload_raises_clean_error_after_retries() -> None:
|
||
"""A companion that says nothing causes a clear error that the user can act on."""
|
||
from meshterm.core.connection import DeviceCommandError, MeshCoreDevice
|
||
|
||
mc = _FakeMeshCore() # the companion answers each request with silence
|
||
|
||
device = MeshCoreDevice(port="COM-test")
|
||
with pytest.raises(DeviceCommandError) as exc:
|
||
await device._contacts_payload(mc, retries=2, delay=0, idle=0.05)
|
||
assert "no event received" in str(exc.value)
|
||
assert mc.requests == 3
|
||
|
||
|
||
async def test_latest_trace_returns_previous_run(tmp_path: Path) -> None:
|
||
"""latest_trace rebuilds the most recent stored trace, and it skips a given run."""
|
||
from meshterm.core.models import Hop
|
||
|
||
repo = Repository(tmp_path / "prev.db")
|
||
old_run = repo.start_run("trace", {"target": "Alice"})
|
||
repo.record_trace(
|
||
old_run,
|
||
TraceResult(target="Alice", success=True, hops=[Hop(0, "3d", 4.0), Hop(1, None, 1.0)]),
|
||
)
|
||
new_run = repo.start_run("trace", {"target": "Alice"})
|
||
|
||
prev = repo.latest_trace("Alice", exclude_run_id=new_run)
|
||
assert prev is not None
|
||
assert [h.node for h in prev.hops] == ["3d", None]
|
||
assert prev.edges("us")[-1].destination == "us"
|
||
assert repo.latest_trace("Nobody") is None
|
||
repo.close()
|
||
|
||
|
||
async def test_repository_round_trip(tmp_path: Path) -> None:
|
||
"""The code stores a run and its traces, and it can read them back."""
|
||
repo = Repository(tmp_path / "test.db")
|
||
run_id = repo.start_run("trace", {"target": "Alice", "samples": 2})
|
||
|
||
device = MockDevice()
|
||
await device.connect()
|
||
results = await trace_runner.run_traces(
|
||
device, "Alice", samples=2, cooldown_s=0, persist=lambda t: repo.record_trace(run_id, t)
|
||
)
|
||
repo.finish_run(run_id, "ok", {"count": len(results)})
|
||
|
||
runs = repo.list_runs()
|
||
assert runs[0].tool == "trace"
|
||
assert runs[0].status == "ok"
|
||
repo.close()
|