From 3e59db6f3e886b71ed3162846fee5899c39845bb Mon Sep 17 00:00:00 2001 From: Ben Meadors Date: Fri, 17 Apr 2026 21:20:02 -0500 Subject: [PATCH] Consolidate type imports and remove placeholder test files --- mcp-server/src/meshtastic_mcp/pio.py | 3 +- mcp-server/tests/admin/test_placeholders.py | 28 --- mcp-server/tests/fleet/test_placeholders.py | 21 -- mcp-server/tests/mesh/_receive.py | 72 +++++-- mcp-server/tests/mesh/test_direct_with_ack.py | 48 +++-- mcp-server/tests/mesh/test_placeholders.py | 50 ----- mcp-server/tests/mesh/test_traceroute.py | 147 ++++++++++++++ mcp-server/tests/monitor/test_placeholders.py | 22 --- .../tests/provisioning/test_placeholders.py | 37 ---- .../tests/telemetry/test_placeholders.py | 22 --- .../telemetry/test_telemetry_request_reply.py | 187 ++++++++++++++++++ 11 files changed, 420 insertions(+), 217 deletions(-) delete mode 100644 mcp-server/tests/admin/test_placeholders.py delete mode 100644 mcp-server/tests/fleet/test_placeholders.py delete mode 100644 mcp-server/tests/mesh/test_placeholders.py create mode 100644 mcp-server/tests/mesh/test_traceroute.py delete mode 100644 mcp-server/tests/monitor/test_placeholders.py delete mode 100644 mcp-server/tests/provisioning/test_placeholders.py delete mode 100644 mcp-server/tests/telemetry/test_placeholders.py create mode 100644 mcp-server/tests/telemetry/test_telemetry_request_reply.py diff --git a/mcp-server/src/meshtastic_mcp/pio.py b/mcp-server/src/meshtastic_mcp/pio.py index be9ca1d0a..c0c23f9bb 100644 --- a/mcp-server/src/meshtastic_mcp/pio.py +++ b/mcp-server/src/meshtastic_mcp/pio.py @@ -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 diff --git a/mcp-server/tests/admin/test_placeholders.py b/mcp-server/tests/admin/test_placeholders.py deleted file mode 100644 index b3230ef32..000000000 --- a/mcp-server/tests/admin/test_placeholders.py +++ /dev/null @@ -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 diff --git a/mcp-server/tests/fleet/test_placeholders.py b/mcp-server/tests/fleet/test_placeholders.py deleted file mode 100644 index 5f9266da3..000000000 --- a/mcp-server/tests/fleet/test_placeholders.py +++ /dev/null @@ -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 diff --git a/mcp-server/tests/mesh/_receive.py b/mcp-server/tests/mesh/_receive.py index d037f103b..a1e5bb3e8 100644 --- a/mcp-server/tests/mesh/_receive.py +++ b/mcp-server/tests/mesh/_receive.py @@ -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, diff --git a/mcp-server/tests/mesh/test_direct_with_ack.py b/mcp-server/tests/mesh/test_direct_with_ack.py index 38ea7df36..1b3186ac5 100644 --- a/mcp-server/tests/mesh/test_direct_with_ack.py +++ b/mcp-server/tests/mesh/test_direct_with_ack.py @@ -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: diff --git a/mcp-server/tests/mesh/test_placeholders.py b/mcp-server/tests/mesh/test_placeholders.py deleted file mode 100644 index dc9b57761..000000000 --- a/mcp-server/tests/mesh/test_placeholders.py +++ /dev/null @@ -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 diff --git a/mcp-server/tests/mesh/test_traceroute.py b/mcp-server/tests/mesh/test_traceroute.py new file mode 100644 index 000000000..01ef4c39b --- /dev/null +++ b/mcp-server/tests/mesh/test_traceroute.py @@ -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}" + ) diff --git a/mcp-server/tests/monitor/test_placeholders.py b/mcp-server/tests/monitor/test_placeholders.py deleted file mode 100644 index 388065b82..000000000 --- a/mcp-server/tests/monitor/test_placeholders.py +++ /dev/null @@ -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 ` 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 diff --git a/mcp-server/tests/provisioning/test_placeholders.py b/mcp-server/tests/provisioning/test_placeholders.py deleted file mode 100644 index 81acfadc2..000000000 --- a/mcp-server/tests/provisioning/test_placeholders.py +++ /dev/null @@ -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 diff --git a/mcp-server/tests/telemetry/test_placeholders.py b/mcp-server/tests/telemetry/test_placeholders.py deleted file mode 100644 index 66fe7065e..000000000 --- a/mcp-server/tests/telemetry/test_placeholders.py +++ /dev/null @@ -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 diff --git a/mcp-server/tests/telemetry/test_telemetry_request_reply.py b/mcp-server/tests/telemetry/test_telemetry_request_reply.py new file mode 100644 index 000000000..8914c758d --- /dev/null +++ b/mcp-server/tests/telemetry/test_telemetry_request_reply.py @@ -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}" + )