mirror of
https://github.com/alexhopeoconnor/firmware.git
synced 2026-10-14 23:41:29 +10:00
Consolidate type imports and remove placeholder test files
This commit is contained in:
Ben Meadors
@@ -27,8 +27,7 @@ import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Sequence
|
||||
from typing import TextIO
|
||||
from typing import Sequence, TextIO
|
||||
|
||||
from . import config
|
||||
|
||||
|
||||
@@ -1,28 +0,0 @@
|
||||
"""Admin placeholders: v2 roadmap entries, skipped in v2.0."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="placeholder — implement in v2.1")
|
||||
|
||||
|
||||
def test_set_channel_url_imports_all() -> None:
|
||||
"""Synthesize a URL with primary + 2 secondary channels; `set_channel_url`;
|
||||
verify all 3 show up in `localNode.channels` with correct PSKs + names.
|
||||
Common operator flow: deploy a new channel scheme across a fleet."""
|
||||
pass
|
||||
|
||||
|
||||
def test_telemetry_interval_configurable() -> None:
|
||||
"""`set_config("telemetry.device_update_interval", 60)`; wait 2× interval;
|
||||
verify telemetry packets arrived at ~60s cadence (within ±20% jitter
|
||||
tolerance). A frequent 'my telemetry isn't working' user issue."""
|
||||
pass
|
||||
|
||||
|
||||
def test_fixed_position_set_and_reported() -> None:
|
||||
"""`set_fixed_position(lat, lon, alt)`; verify `position_config.fixed_position=true`
|
||||
and `list_nodes()[self].position` reports lat/lon within 1e-6 degree tolerance.
|
||||
The privacy-vs-utility trade-off position-precision users care about."""
|
||||
pass
|
||||
@@ -1,21 +0,0 @@
|
||||
"""Fleet placeholders: v2 roadmap entries, skipped in v2.0."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="placeholder — implement in v2.1")
|
||||
|
||||
|
||||
def test_reflash_is_idempotent() -> None:
|
||||
"""`erase_and_flash` + wait-for-boot; `erase_and_flash` again with same
|
||||
env+profile; second pass completes with the device reachable and
|
||||
`device_info` identical to first pass. Operational CI workflow check."""
|
||||
pass
|
||||
|
||||
|
||||
def test_concurrent_admin_ops_dont_stomp() -> None:
|
||||
"""Launch two threads: one calls `device_info(nrf52)` repeatedly, the other
|
||||
calls `device_info(esp32s3)`. Both succeed, no cross-port contamination.
|
||||
Validates `connection.py`'s per-port lock."""
|
||||
pass
|
||||
@@ -7,6 +7,18 @@ accessible through `meshtastic.SerialInterface`'s pubsub mechanism.
|
||||
`ReceiveCollector` opens a long-lived SerialInterface on a port, subscribes
|
||||
to the pubsub topic of interest, and exposes an atomic `wait_for(predicate)`
|
||||
that mesh tests use to verify end-to-end delivery.
|
||||
|
||||
This module also exposes two module-level helpers for forcing a device to
|
||||
broadcast a fresh NodeInfo — the on-demand path that sidesteps the
|
||||
firmware's 10-minute NodeInfo rate-limit. Tests doing directed PKI-encrypted
|
||||
sends need BOTH endpoints to hold current pubkeys for each other:
|
||||
|
||||
nudge_nodeinfo(iface) # nudge an already-open SerialInterface
|
||||
nudge_nodeinfo_port(port) # open briefly, nudge, close
|
||||
|
||||
See `ReceiveCollector.broadcast_nodeinfo_ping` for the firmware-side
|
||||
rationale (PKI staleness → directed sends NAK with Routing.Error=35
|
||||
PKI_UNKNOWN_PUBKEY or 39 PKI_SEND_FAIL_PUBLIC_KEY).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -16,6 +28,39 @@ import time
|
||||
from typing import Any, Callable
|
||||
|
||||
|
||||
def nudge_nodeinfo(iface: Any) -> None:
|
||||
"""Force the device behind ``iface`` to broadcast a fresh NodeInfo.
|
||||
|
||||
Sends a ``ToRadio.Heartbeat(nonce=1)`` — the firmware's documented
|
||||
on-demand NodeInfo trigger (see `src/mesh/api/PacketAPI.cpp:74-79`
|
||||
for TCP/UDP and `src/mesh/PhoneAPI.cpp::handleToRadio` for serial,
|
||||
both routed to `NodeInfoModule::sendOurNodeInfo(..., shorterTimeout=true)`
|
||||
with the 60-s window rather than the 10-min rate-limit).
|
||||
|
||||
Call on BOTH TX and RX ifaces before a directed PKI-encrypted send.
|
||||
Nudging only one side leaves the other with a stale pubkey cache and
|
||||
makes the directed send NAK with PKI_UNKNOWN_PUBKEY.
|
||||
"""
|
||||
from meshtastic.protobuf import mesh_pb2 # type: ignore[import-untyped]
|
||||
|
||||
tr = mesh_pb2.ToRadio()
|
||||
tr.heartbeat.nonce = 1
|
||||
iface._sendToRadio(tr)
|
||||
|
||||
|
||||
def nudge_nodeinfo_port(port: str) -> None:
|
||||
"""Open ``port`` briefly, nudge, close — for when no iface is open yet.
|
||||
|
||||
Uses the meshtastic_mcp port-lock-aware `connect()` context manager
|
||||
so we don't race ReceiveCollector or other long-lived handles on
|
||||
the same port.
|
||||
"""
|
||||
from meshtastic_mcp.connection import connect
|
||||
|
||||
with connect(port=port) as iface:
|
||||
nudge_nodeinfo(iface)
|
||||
|
||||
|
||||
class ReceiveCollector:
|
||||
"""Listen for meshtastic packets on `port` and let tests wait for a match.
|
||||
|
||||
@@ -139,29 +184,18 @@ class ReceiveCollector:
|
||||
def broadcast_nodeinfo_ping(self) -> None:
|
||||
"""Force the firmware on `port` to broadcast a fresh NodeInfo.
|
||||
|
||||
Why this exists: firmware rate-limits NodeInfo broadcasts to every
|
||||
10 min (and 12 h for reply suppression). After a reboot, an existing
|
||||
cooldown window can leave peers with a stale nodesByNum entry that
|
||||
lacks `publicKey`, which makes directed PKI-encrypted sends fail
|
||||
with Routing.Error=39 (PKI_SEND_FAIL_PUBLIC_KEY). But a ToRadio
|
||||
`Heartbeat` with `nonce == 1` is treated as a special "nodeinfo
|
||||
ping" trigger in `src/mesh/api/PacketAPI.cpp:74-79`:
|
||||
Thin wrapper around the module-level :func:`nudge_nodeinfo` that
|
||||
also validates the context-manager invariant. Delegates so tests
|
||||
that need to nudge BOTH sides (bilateral PKI warmup) share one
|
||||
implementation — the caller just passes each iface in turn.
|
||||
|
||||
if (mr->heartbeat.nonce == 1) {
|
||||
nodeInfoModule->sendOurNodeInfo(NODENUM_BROADCAST, true, 0, true);
|
||||
}
|
||||
|
||||
The trailing `true` puts it on the 60-second shorterTimeout path
|
||||
rather than the 10-minute one — so tests can force a fresh NodeInfo
|
||||
broadcast (with public key) on demand.
|
||||
Firmware-side details (rate-limit bypass, nonce==1 trigger path,
|
||||
shorterTimeout=true window) are documented on the module-level
|
||||
helper.
|
||||
"""
|
||||
from meshtastic.protobuf import mesh_pb2 # type: ignore[import-untyped]
|
||||
|
||||
if self._iface is None:
|
||||
raise RuntimeError("ReceiveCollector not started; use as context manager")
|
||||
tr = mesh_pb2.ToRadio()
|
||||
tr.heartbeat.nonce = 1
|
||||
self._iface._sendToRadio(tr)
|
||||
nudge_nodeinfo(self._iface)
|
||||
|
||||
def wait_for(
|
||||
self,
|
||||
|
||||
@@ -15,7 +15,7 @@ from typing import Any
|
||||
import pytest
|
||||
from meshtastic_mcp.connection import connect
|
||||
|
||||
from ._receive import ReceiveCollector
|
||||
from ._receive import ReceiveCollector, nudge_nodeinfo
|
||||
|
||||
|
||||
@pytest.mark.timeout(240)
|
||||
@@ -44,24 +44,37 @@ def test_direct_with_ack_roundtrip(
|
||||
# SerialInterface immediately after sendText() races that retry loop —
|
||||
# empirically the packet never reaches RX.
|
||||
#
|
||||
# Why we ping RX for a fresh NodeInfo before polling:
|
||||
# Why we ping BOTH RX and TX for a fresh NodeInfo before polling:
|
||||
# Directed packets are PKI-encrypted with the destination's public key.
|
||||
# After a factory_reset or reboot, a peer's entry in the sender's
|
||||
# nodeDB can still contain that peer's OLD public key — a directed
|
||||
# send then fails with Routing.Error=39 (PKI_SEND_FAIL_PUBLIC_KEY) or
|
||||
# decryption fails on the receiver side. NodeInfo broadcasts are the
|
||||
# sole source of fresh pubkeys, and firmware rate-limits them to
|
||||
# every 10 min organically. ToRadio.heartbeat(nonce=1) bypasses that
|
||||
# — it triggers an on-demand NodeInfo broadcast via
|
||||
# `src/mesh/PhoneAPI.cpp::handleToRadio` (serial) and
|
||||
# `src/mesh/api/PacketAPI.cpp::handlePacket` (TCP/UDP), both sharing
|
||||
# the 60s shorterTimeout path in `src/modules/NodeInfoModule.cpp`.
|
||||
# After ping, poll TX's nodesByNum until publicKey propagates, then
|
||||
# send. A small retry loop guards against transient LoRa collisions.
|
||||
# The ENCRYPT path needs TX to hold RX's current pubkey; the DECRYPT
|
||||
# path needs RX to hold TX's current pubkey. After a factory_reset or
|
||||
# reboot, either side's nodeDB entry for the other can still carry
|
||||
# a stale pubkey — directed sends then NAK with Routing.Error=35
|
||||
# (PKI_UNKNOWN_PUBKEY, receiver can't decrypt) or 39
|
||||
# (PKI_SEND_FAIL_PUBLIC_KEY, sender has no pubkey at all). NodeInfo
|
||||
# broadcasts are the sole source of fresh pubkeys and the firmware
|
||||
# rate-limits them to every 10 min. ToRadio.heartbeat(nonce=1)
|
||||
# bypasses that via the 60-s shorterTimeout path
|
||||
# (`src/mesh/PhoneAPI.cpp::handleToRadio` for serial,
|
||||
# `src/mesh/api/PacketAPI.cpp::handlePacket` for TCP/UDP, both
|
||||
# calling `NodeInfoModule::sendOurNodeInfo(..., true)`).
|
||||
#
|
||||
# Earlier revisions of this test only nudged RX — which covers the
|
||||
# common case of a recently-baked RX whose TX doesn't know its new
|
||||
# key yet. But when the OPPOSITE side is the one with stale state
|
||||
# (RX holds an old TX pubkey), the test would silently fail with
|
||||
# err=35 in the firmware log. Bilateral nudge eliminates that blind
|
||||
# spot. Poll TX's nodesByNum for RX's publicKey as a proxy for "the
|
||||
# exchange has propagated"; a matching symmetry on RX's side is
|
||||
# implied by the firmware's NodeInfo-on-receipt update path.
|
||||
with ReceiveCollector(rx_port, topic="meshtastic.receive.text") as rx:
|
||||
rx.broadcast_nodeinfo_ping()
|
||||
|
||||
with connect(port=tx_port) as tx_iface:
|
||||
# Bilateral warmup: nudge TX to broadcast too, so RX's nodeDB
|
||||
# also gets refreshed with TX's current pubkey.
|
||||
nudge_nodeinfo(tx_iface)
|
||||
|
||||
pk_deadline = time.monotonic() + 45.0
|
||||
last_nudge = time.monotonic()
|
||||
last_rec: dict[str, Any] = {}
|
||||
@@ -70,10 +83,13 @@ def test_direct_with_ack_roundtrip(
|
||||
user = last_rec.get("user", {})
|
||||
if user.get("publicKey"):
|
||||
break
|
||||
# Re-nudge every 15s in case the first NodeInfo was lost to
|
||||
# a LoRa collision with concurrent traffic.
|
||||
# Re-nudge every 15s in case the first NodeInfo broadcast
|
||||
# was lost to a LoRa collision with concurrent traffic. Both
|
||||
# sides re-broadcast for the same reason they were nudged
|
||||
# initially — stale pubkeys can live on either side.
|
||||
if time.monotonic() - last_nudge > 15.0:
|
||||
rx.broadcast_nodeinfo_ping()
|
||||
nudge_nodeinfo(tx_iface)
|
||||
last_nudge = time.monotonic()
|
||||
time.sleep(1.0)
|
||||
else:
|
||||
|
||||
@@ -1,50 +0,0 @@
|
||||
"""Mesh placeholders: v2 roadmap entries, skipped in v2.0."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="placeholder — implement in v2.1")
|
||||
|
||||
|
||||
def test_direct_to_ghost_times_out_or_naks() -> None:
|
||||
"""Send a direct message to a fabricated node_num (e.g. 0xDEADBEEF) with
|
||||
want_ack=True; assert caller observes NAK or timeout within 60s. Operator
|
||||
fleet-health check — 'is my offline-peer detection working'."""
|
||||
pass
|
||||
|
||||
|
||||
def test_hop_limit_decrements() -> None:
|
||||
"""Send a message with hop_limit=3; inspect received packet on B via
|
||||
serial decoder; assert hop_limit=2 (1-hop decrement across A→B). Prevents
|
||||
runaway rebroadcasting that destroys mesh throughput."""
|
||||
pass
|
||||
|
||||
|
||||
def test_snr_rssi_populated() -> None:
|
||||
"""After 2 min of idle mesh traffic, B's entry in A's node DB must have
|
||||
`snr ∈ [-25, 15]` and `rssi ∈ [-140, 0]`. The most common 'my mesh is
|
||||
broken' misdiagnosis is actually just an empty node DB."""
|
||||
pass
|
||||
|
||||
|
||||
def test_secondary_channel_stays_private() -> None:
|
||||
"""Bake A with channels [primary + secondary 'Secret' PSK K1]; bake B
|
||||
with primary only. A sends on ch_index=1; B's decoded log must NOT contain
|
||||
the secret payload. The privacy-by-channel promise."""
|
||||
pass
|
||||
|
||||
|
||||
def test_psk_mismatch_no_decode() -> None:
|
||||
"""Bake A with psk_seed='run-A', B with psk_seed='run-B' (same
|
||||
channel_name). A sends broadcast; B's log shows packet-received-but-
|
||||
rejected at decode (PSK mismatch). Proves session isolation between
|
||||
concurrent test labs."""
|
||||
pass
|
||||
|
||||
|
||||
def test_traceroute_one_hop() -> None:
|
||||
"""A → traceroute(B); response path contains exactly [A, B] with no
|
||||
intermediate routers (since there aren't any in a 2-device mesh).
|
||||
Validates `traceroute` packet round-trip."""
|
||||
pass
|
||||
@@ -0,0 +1,147 @@
|
||||
"""Mesh: traceroute from TX to RX round-trips with no intermediate hops.
|
||||
|
||||
TX sends a `TRACEROUTE_APP` request (RouteDiscovery with `want_response=True`)
|
||||
addressed to RX's node_num. RX's firmware (`modules/TraceRouteModule.cpp`)
|
||||
replies with a RouteDiscovery payload whose `route` / `route_back` lists
|
||||
contain any intermediate relays and `snr_towards` / `snr_back` carry per-hop
|
||||
SNRs. In a 2-device direct mesh there are no relays between TX and RX, so
|
||||
both route lists must be empty and each SNR list carries exactly one entry
|
||||
for the direct TX↔RX link.
|
||||
|
||||
Validates the full TRACEROUTE_APP portnum round-trip: request encoding, RX
|
||||
firmware dispatch, RouteDiscovery payload construction, wire response, and
|
||||
client-side decode through `meshtastic.__init__.py::protocols[TRACEROUTE_APP]`
|
||||
(which is what publishes the `meshtastic.receive.traceroute` pubsub topic).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic.mesh_interface import MeshInterface
|
||||
|
||||
from ._receive import ReceiveCollector, nudge_nodeinfo_port
|
||||
|
||||
|
||||
@pytest.mark.timeout(240)
|
||||
def test_traceroute_one_hop(mesh_pair: dict[str, Any]) -> None:
|
||||
"""Runs for every directed pair. Asserts TX sends + RX responds, then
|
||||
inspects the captured RouteDiscovery to confirm the path is direct.
|
||||
|
||||
Why the listener is on TX (not RX):
|
||||
The traceroute RESPONSE is addressed to TX (the original requester).
|
||||
The meshtastic Python client publishes `meshtastic.receive.traceroute`
|
||||
on the interface that received that response — which is TX's iface.
|
||||
A listener on RX would only see the inbound REQUEST, which lacks
|
||||
the SNR-towards / SNR-back fields the firmware only fills on reply.
|
||||
|
||||
Why we ping RX's NodeInfo before sending:
|
||||
Traceroute requests are directed sends (wantResponse=True, specific
|
||||
destinationId) — subject to the same PKI_SEND_FAIL_PUBLIC_KEY trap
|
||||
as `test_direct_with_ack`. We open RX briefly to trigger the
|
||||
on-demand NodeInfo broadcast, then wait for TX's nodesByNum to
|
||||
populate RX's publicKey before calling sendTraceRoute.
|
||||
"""
|
||||
tx_port = mesh_pair["tx"]["port"]
|
||||
rx_port = mesh_pair["rx"]["port"]
|
||||
rx_node_num = mesh_pair["rx"]["my_node_num"]
|
||||
tx_role = mesh_pair["tx_role"]
|
||||
rx_role = mesh_pair["rx_role"]
|
||||
assert rx_node_num is not None, f"{rx_role} my_node_num missing"
|
||||
|
||||
with ReceiveCollector(
|
||||
tx_port, topic="meshtastic.receive.traceroute"
|
||||
) as tx_listener:
|
||||
# Bilateral PKI warmup — traceroute requests are directed and
|
||||
# PKI-encrypted, so both sides need current pubkeys. See
|
||||
# `_receive.py::nudge_nodeinfo` and the test_direct_with_ack
|
||||
# comment for the full rationale (one-sided nudge lets err=35
|
||||
# PKI_UNKNOWN_PUBKEY slip through in whichever direction had
|
||||
# stale RX-side cache).
|
||||
nudge_nodeinfo_port(rx_port) # RX via brief side-connection
|
||||
tx_listener.broadcast_nodeinfo_ping() # TX via already-open iface
|
||||
|
||||
# Poll TX's view of RX until the publicKey propagates. 45 s matches
|
||||
# the cap used in `test_direct_with_ack`; the re-nudge at 15 s
|
||||
# covers a LoRa collision on the first NodeInfo broadcast.
|
||||
pk_deadline = time.monotonic() + 45.0
|
||||
last_nudge = time.monotonic()
|
||||
last_rec: dict[str, Any] = {}
|
||||
while time.monotonic() < pk_deadline:
|
||||
last_rec = (tx_listener._iface.nodesByNum or {}).get(rx_node_num, {})
|
||||
if last_rec.get("user", {}).get("publicKey"):
|
||||
break
|
||||
if time.monotonic() - last_nudge > 15.0:
|
||||
nudge_nodeinfo_port(rx_port)
|
||||
tx_listener.broadcast_nodeinfo_ping()
|
||||
last_nudge = time.monotonic()
|
||||
time.sleep(1.0)
|
||||
else:
|
||||
pytest.fail(
|
||||
f"TX ({tx_role}) never saw RX ({rx_role}) public key within "
|
||||
f"45s; nodesByNum entry={last_rec!r}"
|
||||
)
|
||||
|
||||
# sendTraceRoute blocks internally on `waitForTraceRoute` and raises
|
||||
# `MeshInterface.MeshInterfaceError` on timeout. One retry covers a
|
||||
# transient LoRa collision on either the request or the reply.
|
||||
ok = False
|
||||
for _attempt in range(2):
|
||||
try:
|
||||
tx_listener._iface.sendTraceRoute(
|
||||
dest=rx_node_num,
|
||||
hopLimit=3,
|
||||
)
|
||||
ok = True
|
||||
break
|
||||
except MeshInterface.MeshInterfaceError:
|
||||
time.sleep(5.0)
|
||||
assert ok, (
|
||||
f"sendTraceRoute {tx_role}→{rx_role} timed out twice; the mesh "
|
||||
f"may be saturated or RX's TraceRouteModule is misrouting the "
|
||||
f"reply"
|
||||
)
|
||||
|
||||
# sendTraceRoute already waited for the response internally, but
|
||||
# pubsub dispatch runs on the meshtastic-python reader thread —
|
||||
# give it a short grace window to queue the packet.
|
||||
packet = tx_listener.wait_for(
|
||||
lambda p: p.get("from") == rx_node_num,
|
||||
timeout=5.0,
|
||||
)
|
||||
assert packet is not None, (
|
||||
f"sendTraceRoute returned OK but no `receive.traceroute` packet "
|
||||
f"from RX (0x{rx_node_num:08x}) arrived via pubsub. Captured: "
|
||||
f"{tx_listener.snapshot()!r}"
|
||||
)
|
||||
|
||||
# Inspect the decoded RouteDiscovery. The meshtastic client stores
|
||||
# the parsed protobuf (as a plain dict via MessageToDict) under
|
||||
# `decoded.traceroute` for this portnum; keys are camelCase because
|
||||
# protobuf JSON conversion uses `preserving_proto_field_name=False`
|
||||
# by default.
|
||||
decoded = packet.get("decoded", {})
|
||||
route_info = decoded.get("traceroute") or {}
|
||||
|
||||
forward_hops = route_info.get("route") or []
|
||||
back_hops = route_info.get("routeBack") or []
|
||||
snr_towards = route_info.get("snrTowards") or []
|
||||
|
||||
assert forward_hops == [], (
|
||||
f"traceroute forward `route` should be empty on a 2-device direct "
|
||||
f"mesh (no intermediaries between {tx_role} and {rx_role}); got "
|
||||
f"{forward_hops!r}"
|
||||
)
|
||||
assert back_hops == [], (
|
||||
f"traceroute `routeBack` should be empty on a 2-device direct "
|
||||
f"mesh; got {back_hops!r}"
|
||||
)
|
||||
# `snr_towards` has len(route) + 1 entries — one per hop plus a final
|
||||
# entry for the destination's receive SNR. Direct mesh → len(route)
|
||||
# is 0 → exactly 1 SNR entry.
|
||||
assert len(snr_towards) == 1, (
|
||||
f"traceroute `snrTowards` should carry exactly 1 entry (direct "
|
||||
f"link SNR) on a 2-device mesh; got {snr_towards!r}"
|
||||
)
|
||||
@@ -1,22 +0,0 @@
|
||||
"""Monitor placeholders: v2 roadmap entries, skipped in v2.0."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="placeholder — implement in v2.1")
|
||||
|
||||
|
||||
def test_env_selects_correct_filter() -> None:
|
||||
"""Open `serial_open(env='t-beam-1w')`; assert `resolved_filters` includes
|
||||
`esp32_exception_decoder`. Proves `pio device monitor -e <env>` correctly
|
||||
applies per-variant monitor_filters from platformio.ini."""
|
||||
pass
|
||||
|
||||
|
||||
def test_dropped_lines_reported() -> None:
|
||||
"""Force a flood of log output (e.g. via a debug flag); read the session
|
||||
with a long delay; assert `serial_read.dropped > 0` surfaces correctly
|
||||
when the ring buffer overflows. Operators relying on serial capture for
|
||||
post-mortem need to know when they missed data."""
|
||||
pass
|
||||
@@ -1,37 +0,0 @@
|
||||
"""Provisioning placeholders: v2 roadmap entries, skipped in v2.
|
||||
|
||||
Each test documents intent so `pytest --collect-only` surfaces the complete
|
||||
planned suite. Implementations land in v2.1+ as hardware coverage expands.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="placeholder — implement in v2.1")
|
||||
|
||||
|
||||
def test_firmware_edition_surfaces() -> None:
|
||||
"""Bake `USERPREFS_FIRMWARE_EDITION=meshtastic_FirmwareEdition_BURNING_MAN`;
|
||||
verify `iface.myInfo.firmware_edition == BURNING_MAN`. Relevant for event
|
||||
firmware drops."""
|
||||
pass
|
||||
|
||||
|
||||
def test_oem_text_in_boot_log() -> None:
|
||||
"""Bake `USERPREFS_OEM_TEXT="Lab Bench"`; open serial at boot; assert the
|
||||
OEM string appears in the splash log lines."""
|
||||
pass
|
||||
|
||||
|
||||
def test_tz_applied() -> None:
|
||||
"""Bake `USERPREFS_TZ_STRING="PST8PDT,M3.2.0,M11.1.0"`; verify
|
||||
`localConfig.device.tzdef` matches and clock-display lines reflect the TZ."""
|
||||
pass
|
||||
|
||||
|
||||
def test_full_factory_reset_regenerates_identity() -> None:
|
||||
"""Before: capture `my_node_num`. Call `factory_reset(full=True)`. After
|
||||
reboot: `my_node_num` must differ (new identity), AND baked USERPREFS
|
||||
defaults (region/channel) still come back."""
|
||||
pass
|
||||
@@ -1,22 +0,0 @@
|
||||
"""Telemetry placeholders: v2 roadmap entries, skipped in v2.0."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="placeholder — implement in v2.1")
|
||||
|
||||
|
||||
def test_position_broadcast_precision() -> None:
|
||||
"""Bake A with `USERPREFS_CHANNEL_0_PRECISION=10` (low-precision channel);
|
||||
A sets a fixed position; B's `list_nodes` entry for A shows truncated
|
||||
coordinates (lower bits zeroed relative to what A transmitted). Validates
|
||||
the privacy-precision knob operators tune for event deployments."""
|
||||
pass
|
||||
|
||||
|
||||
def test_fixed_position_round_trip() -> None:
|
||||
"""`set_fixed_position(lat=40.7128, lon=-74.0060, alt=10)` on A; wait for
|
||||
broadcast; assert B's node DB shows A's position within 1e-6° tolerance
|
||||
AND `position_config.fixed_position=true` on A persists across reboot."""
|
||||
pass
|
||||
@@ -0,0 +1,187 @@
|
||||
"""Telemetry: on-demand device-metrics request gets a prompt reply.
|
||||
|
||||
Complementary to ``test_device_telemetry_broadcast`` — that one witnesses the
|
||||
firmware's *periodic* broadcast (900 s default interval, up to ~7 min worst
|
||||
case). This one exercises the *request/reply* path: TX sends a
|
||||
``meshtastic_Telemetry`` with the ``device_metrics`` variant-tag set and
|
||||
``want_response=True`` on ``TELEMETRY_APP`` to RX, and RX's
|
||||
``modules/Telemetry/DeviceTelemetry.cpp::allocReply`` fires immediately with
|
||||
populated ``DeviceMetrics``. On a direct 2-device mesh the whole round-trip
|
||||
finishes in under a minute even from a cold boot.
|
||||
|
||||
Validates:
|
||||
* ``sendData(portNum=TELEMETRY_APP, want_response=True)`` encodes + routes
|
||||
to RX (directed, PKI-encrypted to RX's pubkey)
|
||||
* RX's ``DeviceTelemetryModule::handleReceivedProtobuf`` dispatches to
|
||||
``allocReply`` — which is only invoked by the framework when
|
||||
``want_response`` is set on the incoming packet
|
||||
* The reply carries a ``DeviceMetrics`` sub-message with at least one
|
||||
non-zero field (uptime_seconds is guaranteed non-zero a few seconds
|
||||
after boot, so it reliably survives protobuf's default-value
|
||||
serialization stripping)
|
||||
* The reply routes back to TX and gets matched against the original
|
||||
request via ``request_id`` — using the library's ``onResponse``
|
||||
callback mechanism, which stores the handler at
|
||||
``responseHandlers[sent_packet.id]`` and dispatches when a packet
|
||||
arrives with ``decoded.request_id == sent_packet.id``. This is more
|
||||
precise than a pubsub ``from==rx_node_num`` filter, which can
|
||||
accidentally match RX's periodic broadcast or a stale reply to a
|
||||
different prior request.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic.protobuf import ( # type: ignore[import-untyped]
|
||||
portnums_pb2,
|
||||
telemetry_pb2,
|
||||
)
|
||||
|
||||
from ..mesh._receive import ReceiveCollector, nudge_nodeinfo_port
|
||||
|
||||
# Fields on the DeviceMetrics sub-message. The camelCase versions are what
|
||||
# `google.protobuf.json_format.MessageToDict` emits (preserving_proto_field_name
|
||||
# defaults to False); the snake_case names are the proto-source spellings.
|
||||
_DEVICE_METRICS_FIELDS = (
|
||||
"batteryLevel",
|
||||
"voltage",
|
||||
"channelUtilization",
|
||||
"airUtilTx",
|
||||
"uptimeSeconds",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.timeout(240)
|
||||
def test_telemetry_request_reply(mesh_pair: dict[str, Any]) -> None:
|
||||
"""Runs for every directed pair. TX requests RX's telemetry via
|
||||
``want_response=True`` and asserts the reply arrives with populated
|
||||
DeviceMetrics.
|
||||
"""
|
||||
tx_port = mesh_pair["tx"]["port"]
|
||||
rx_port = mesh_pair["rx"]["port"]
|
||||
rx_node_num = mesh_pair["rx"]["my_node_num"]
|
||||
tx_role = mesh_pair["tx_role"]
|
||||
rx_role = mesh_pair["rx_role"]
|
||||
assert rx_node_num is not None, f"{rx_role} my_node_num missing"
|
||||
|
||||
# ReceiveCollector is still used to hold TX's SerialInterface open and
|
||||
# give us `tx_listener._iface` for sendData / nodesByNum polling. The
|
||||
# subscribed topic is irrelevant for this test (we match via
|
||||
# onResponse, not pubsub), but keeping a concrete topic avoids the
|
||||
# surprise of a pubsub wildcard receiving every packet type.
|
||||
with ReceiveCollector(tx_port, topic="meshtastic.receive.telemetry") as tx_listener:
|
||||
# Bilateral PKI warmup — nudge BOTH sides to rebroadcast their
|
||||
# NodeInfo (with current pubkey) before the directed send.
|
||||
# * Nudging only RX gets RX's key → TX, but leaves RX with a
|
||||
# potentially stale TX pubkey → RX NAKs our request with
|
||||
# err=35 (PKI_UNKNOWN_PUBKEY) and we see no reply.
|
||||
# * Nudging only TX is the mirror failure.
|
||||
# See `tests/mesh/_receive.py::nudge_nodeinfo` for firmware path.
|
||||
nudge_nodeinfo_port(rx_port) # briefly opens RX to send heartbeat
|
||||
tx_listener.broadcast_nodeinfo_ping() # TX via the already-open iface
|
||||
|
||||
pk_deadline = time.monotonic() + 45.0
|
||||
last_nudge = time.monotonic()
|
||||
last_rec: dict[str, Any] = {}
|
||||
while time.monotonic() < pk_deadline:
|
||||
last_rec = (tx_listener._iface.nodesByNum or {}).get(rx_node_num, {})
|
||||
if last_rec.get("user", {}).get("publicKey"):
|
||||
break
|
||||
if time.monotonic() - last_nudge > 15.0:
|
||||
# Re-nudge both sides — LoRa collisions can drop either
|
||||
# direction's NodeInfo broadcast independently.
|
||||
nudge_nodeinfo_port(rx_port)
|
||||
tx_listener.broadcast_nodeinfo_ping()
|
||||
last_nudge = time.monotonic()
|
||||
time.sleep(1.0)
|
||||
else:
|
||||
pytest.fail(
|
||||
f"TX ({tx_role}) never saw RX ({rx_role}) public key within "
|
||||
f"45s; nodesByNum entry={last_rec!r}"
|
||||
)
|
||||
|
||||
# Send the request. The Telemetry protobuf has a `which_variant`
|
||||
# oneof tag that the firmware uses to decide which reply to build
|
||||
# (see `src/modules/Telemetry/DeviceTelemetry.cpp::allocReply`):
|
||||
# device_metrics_tag → getDeviceTelemetry()
|
||||
# local_stats_tag → getLocalStatsTelemetry()
|
||||
# anything else → return NULL (request silently dropped)
|
||||
# An empty `Telemetry()` has `which_variant = UNSET (0)`, so we MUST
|
||||
# explicitly set the variant. `CopyFrom(DeviceMetrics())` with a
|
||||
# default-constructed sub-message is the canonical Python-protobuf
|
||||
# idiom for "set the oneof tag without populating fields" — matching
|
||||
# how `MeshInterface.sendTelemetry()` constructs requests for the
|
||||
# other variants.
|
||||
#
|
||||
# Matching the reply: the meshtastic client's `onResponse` callback
|
||||
# mechanism fires ONLY for packets whose `decoded.request_id` equals
|
||||
# the original outgoing packet's `id`. That's exactly the semantic
|
||||
# we want — rejects periodic broadcasts (no request_id), rejects
|
||||
# stale replies to prior requests (different request_id), and
|
||||
# tolerates the firmware's reply_id/request_id naming quirk
|
||||
# (firmware's `setReplyTo` writes the original packet's id into
|
||||
# `decoded.request_id`, not `decoded.reply_id`).
|
||||
#
|
||||
# One retry covers transient LoRa collisions on request or reply.
|
||||
reply_holder: list[dict[str, Any]] = []
|
||||
got_reply = threading.Event()
|
||||
|
||||
def _on_reply(packet: dict[str, Any]) -> None:
|
||||
reply_holder.append(packet)
|
||||
got_reply.set()
|
||||
|
||||
got = None
|
||||
for _attempt in range(2):
|
||||
got_reply.clear()
|
||||
del reply_holder[:]
|
||||
req = telemetry_pb2.Telemetry()
|
||||
req.device_metrics.CopyFrom(telemetry_pb2.DeviceMetrics())
|
||||
tx_listener._iface.sendData(
|
||||
req,
|
||||
destinationId=rx_node_num,
|
||||
portNum=portnums_pb2.PortNum.TELEMETRY_APP,
|
||||
wantResponse=True,
|
||||
onResponse=_on_reply,
|
||||
hopLimit=3,
|
||||
)
|
||||
if got_reply.wait(timeout=45.0):
|
||||
got = reply_holder[0]
|
||||
break
|
||||
time.sleep(5.0)
|
||||
|
||||
assert got is not None, (
|
||||
f"no telemetry reply from {rx_role} (0x{rx_node_num:08x}) within "
|
||||
f"90s of 2 requests; onResponse callback never fired. Captured "
|
||||
f"{len(tx_listener.snapshot())} unrelated telemetry packet(s): "
|
||||
f"{[hex(p.get('from') or 0) for p in tx_listener.snapshot()]!r}"
|
||||
)
|
||||
|
||||
# Sanity: the reply's origin matches — a firmware bug that routed
|
||||
# the response to the wrong sender would make onResponse fire on
|
||||
# the wrong packet.
|
||||
assert got.get("from") == rx_node_num, (
|
||||
f"telemetry reply origin mismatch: from=0x{got.get('from'):08x}, "
|
||||
f"expected 0x{rx_node_num:08x}"
|
||||
)
|
||||
|
||||
# Inspect the decoded Telemetry payload. The meshtastic client stores
|
||||
# it under `decoded.telemetry`; DeviceMetrics under `.deviceMetrics`.
|
||||
decoded = got.get("decoded", {})
|
||||
telem = decoded.get("telemetry") or {}
|
||||
dm = telem.get("deviceMetrics") or {}
|
||||
|
||||
# A populated reply must contain at least one DeviceMetrics field.
|
||||
# Protobuf's JSON serializer strips default-valued (zero) fields,
|
||||
# so a bare `deviceMetrics: {}` would mean the firmware wrote the
|
||||
# sub-message but every field was zero — plausible right at boot
|
||||
# but not for a device that's been running long enough for a test
|
||||
# session's warmup + NodeInfo exchange (~10-30 s uptime minimum).
|
||||
populated = [k for k in _DEVICE_METRICS_FIELDS if k in dm]
|
||||
assert populated, (
|
||||
f"telemetry reply from {rx_role} carried no DeviceMetrics fields; "
|
||||
f"decoded.telemetry={telem!r}"
|
||||
)
|
||||
Reference in New Issue
Block a user