mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-10-30 19:41:58 +01:00
Afri
ddae287ee6
* data: keep snapshot node ages when the radio clock is off * data: add the snapshot clock unit tests left out of the first commit
547 lines
19 KiB
Python
547 lines
19 KiB
Python
# Copyright © 2025-26 l5yth & contributors
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
"""Regression guard: a Meshtastic packet received now is posted with a receive
|
|
time the web app's read windows can see, even when the radio's clock is wrong.
|
|
|
|
The web app stores ``rx_time`` as posted (it clamps only future values) and
|
|
every read route filters on it: 7 days for ``GET /api/telemetry`` and the
|
|
charts, 28 days for ``GET /api/telemetry/:id`` (``CONTRACTS.md``, "GET
|
|
endpoint time windows"). A radio whose clock runs weeks behind therefore turns
|
|
every live record into a row no page can show, while the node row still takes
|
|
the packet's battery and uptime. Each case feeds one live packet through
|
|
:func:`handlers.on_receive`, the seam both the pubsub and the UDP transports
|
|
use, with the radio clock 60 days behind the ingestor host clock.
|
|
|
|
The cases after the regression guard pin the rule of the fix,
|
|
:mod:`data.mesh_ingestor.handlers.receive_time` (SPEC RK1-RK3): the one-hour
|
|
tolerance in both directions, the rate-limited warning, both Meshtastic
|
|
transports, the host-telemetry suppression window, and the MeshCore paths the
|
|
rule leaves alone.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from meshtastic import mesh_interface as lib_mesh_interface
|
|
from meshtastic.protobuf import mesh_pb2, portnums_pb2
|
|
from pubsub import pub
|
|
|
|
import data.mesh_ingestor.config as config
|
|
import data.mesh_ingestor.handlers as handlers
|
|
import data.mesh_ingestor.handlers._state as _state_mod
|
|
import data.mesh_ingestor.handlers.receive_time as receive_time
|
|
import data.mesh_ingestor.queue as queue
|
|
from data.mesh_ingestor.protocols import meshtastic_udp_decode as udp_decode
|
|
from data.mesh_ingestor.protocols.meshcore import (
|
|
_MeshcoreInterface,
|
|
_process_contacts,
|
|
)
|
|
from data.mesh_ingestor.protocols.meshtastic import MeshtasticProvider
|
|
from data.mesh_ingestor.protocols.meshtastic_udp import (
|
|
MeshtasticUdpProvider,
|
|
_UdpInterface,
|
|
)
|
|
from data.mesh_ingestor.serialization import _iso
|
|
|
|
NOW = 1_791_288_000
|
|
"""Ingestor host clock for every case: 2026-10-06T12:00:00Z."""
|
|
|
|
RADIO_CLOCK = NOW - 60 * 86_400
|
|
"""The radio's ``rxTime``: 60 days behind the host, past the 28-day window."""
|
|
|
|
BULK_WINDOW = 7 * 86_400
|
|
"""Narrowest read floor a live record must clear (``GET /api/telemetry``)."""
|
|
|
|
SENDER = {"from": 0xF0ACBFBA, "fromId": "!f0acbfba", "to": 0xFFFFFFFF, "toId": "^all"}
|
|
"""Broadcast addressing shared by every packet, as the meshtastic library emits it."""
|
|
|
|
CASES = {
|
|
"environment telemetry": (
|
|
"/api/telemetry",
|
|
{
|
|
"portnum": "TELEMETRY_APP",
|
|
"telemetry": {
|
|
"time": NOW - 5,
|
|
"environmentMetrics": {
|
|
"temperature": 21.5,
|
|
"relativeHumidity": 48.0,
|
|
"barometricPressure": 1012.3,
|
|
},
|
|
},
|
|
},
|
|
),
|
|
"device telemetry": (
|
|
"/api/telemetry",
|
|
{
|
|
"portnum": "TELEMETRY_APP",
|
|
"telemetry": {
|
|
"time": NOW - 5,
|
|
"deviceMetrics": {"batteryLevel": 101, "voltage": 4.2},
|
|
},
|
|
},
|
|
),
|
|
"position": (
|
|
"/api/positions",
|
|
{
|
|
"portnum": "POSITION_APP",
|
|
"position": {"latitude": 52.49, "longitude": 13.47, "time": NOW - 5},
|
|
},
|
|
),
|
|
"text message": (
|
|
"/api/messages",
|
|
{"portnum": "TEXT_MESSAGE_APP", "payload": {"text": "hello"}},
|
|
),
|
|
}
|
|
"""Record types the dashboard reads through rx_time windows, by POST route."""
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_handler_state():
|
|
"""Clear the handler module state other suites may have left behind."""
|
|
|
|
_state_mod._host_node_id = None
|
|
_state_mod._host_telemetry_last_rx = None
|
|
yield
|
|
_state_mod._host_node_id = None
|
|
_state_mod._host_telemetry_last_rx = None
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_warning_state(monkeypatch):
|
|
"""Start every case with no receive-time warning logged yet."""
|
|
|
|
monkeypatch.setattr(receive_time, "_last_warning_monotonic", None)
|
|
monkeypatch.setattr(receive_time, "_unlogged_replacements", 0)
|
|
|
|
|
|
@pytest.fixture
|
|
def host_now():
|
|
"""Start the shared ``host_clock`` fixture (``tests/conftest.py``) at :data:`NOW`."""
|
|
|
|
return NOW
|
|
|
|
|
|
@pytest.fixture
|
|
def posts(monkeypatch, host_clock):
|
|
"""Start the host clock at :data:`NOW` and capture every queued POST.
|
|
|
|
Returns:
|
|
A list that collects ``(path, payload)`` for each queued request.
|
|
"""
|
|
|
|
sent = []
|
|
monkeypatch.setattr(
|
|
queue,
|
|
"_queue_post_json",
|
|
lambda path, payload, *args, **kwargs: sent.append((path, payload)),
|
|
)
|
|
return sent
|
|
|
|
|
|
@pytest.mark.parametrize("case", sorted(CASES))
|
|
def test_live_packet_lands_inside_read_window(case, posts):
|
|
"""A packet received at :data:`NOW` is posted with ``rx_time`` inside the
|
|
7-day bulk window, not at the radio's clock 60 days earlier."""
|
|
|
|
route, decoded = CASES[case]
|
|
packet = {"id": 0x5EED0001, "rxTime": RADIO_CLOCK, "channel": 0, **SENDER}
|
|
packet["decoded"] = decoded
|
|
handlers.on_receive(packet, None)
|
|
|
|
payloads = [payload for path, payload in posts if path == route]
|
|
assert payloads, f"{case}: nothing queued for {route}"
|
|
rx_time = payloads[0]["rx_time"]
|
|
days_behind = (NOW - rx_time) / 86_400
|
|
assert NOW - BULK_WINDOW <= rx_time <= NOW, (
|
|
f"{case}: posted rx_time {rx_time} ({_iso(rx_time)}) is the radio clock, "
|
|
f"{days_behind:.0f} days behind the ingestor clock ({_iso(NOW)}); "
|
|
"no GET route returns it (7-day bulk and 28-day per-id windows)"
|
|
)
|
|
assert payloads[0]["rx_iso"] == _iso(rx_time)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared helpers for the rule's cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def warnings_logged(monkeypatch):
|
|
"""Capture the receive-time warnings instead of printing them.
|
|
|
|
Returns:
|
|
A list that collects each warning's severity and metadata.
|
|
"""
|
|
|
|
logged = []
|
|
|
|
def record(message, *, context=None, severity="debug", **metadata):
|
|
"""Keep the receive-time warnings and drop every other log line."""
|
|
|
|
if context == "handlers.receive_time":
|
|
logged.append({"severity": severity, **metadata})
|
|
|
|
monkeypatch.setattr(config, "_debug_log", record)
|
|
return logged
|
|
|
|
|
|
def _routed(posts, route):
|
|
"""Return the payloads queued for ``route``, in order."""
|
|
|
|
return [payload for path, payload in posts if path == route]
|
|
|
|
|
|
def _text_packet(radio_time, packet_id=0x5EED0010):
|
|
"""Return a broadcast text packet that the radio stamped ``radio_time``."""
|
|
|
|
return {
|
|
"id": packet_id,
|
|
"rxTime": radio_time,
|
|
"channel": 0,
|
|
**SENDER,
|
|
"decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": {"text": "hello"}},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tolerance: one hour, in both directions (SPEC RK2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_tolerance_and_warning_interval_are_named_constants():
|
|
"""The rule's two numbers live in named constants (SPEC RK2)."""
|
|
|
|
assert receive_time.RX_TIME_TOLERANCE_SECS == 60 * 60
|
|
assert receive_time.RX_TIME_WARNING_INTERVAL_SECS == 10 * 60
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("minutes_off", "kept"),
|
|
[(-59, True), (59, True), (-60, True), (60, True), (-61, False), (61, False)],
|
|
ids=[
|
|
"59 min behind kept",
|
|
"59 min ahead kept",
|
|
"60 min behind kept",
|
|
"60 min ahead kept",
|
|
"61 min behind replaced",
|
|
"61 min ahead replaced",
|
|
],
|
|
)
|
|
def test_radio_time_is_kept_only_within_the_hour(minutes_off, kept, posts):
|
|
"""A radio time up to an hour off the host clock is posted as stamped; one
|
|
further off, behind or ahead, is posted as the host clock."""
|
|
|
|
radio_time = NOW + minutes_off * 60
|
|
handlers.on_receive(_text_packet(radio_time), None)
|
|
|
|
[message] = _routed(posts, "/api/messages")
|
|
assert message["rx_time"] == (radio_time if kept else NOW)
|
|
assert message["rx_iso"] == _iso(message["rx_time"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Warning: names the node and the offset, rate-limited (SPEC RK2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def monotonic(monkeypatch):
|
|
"""Drive :func:`time.monotonic` by hand, starting at 1000 seconds.
|
|
|
|
Returns:
|
|
A namespace whose ``now`` attribute :func:`time.monotonic` returns.
|
|
"""
|
|
|
|
clock = SimpleNamespace(now=1_000.0)
|
|
monkeypatch.setattr(time, "monotonic", lambda: clock.now)
|
|
return clock
|
|
|
|
|
|
def test_replacement_warns_naming_the_node_and_the_offset(posts, warnings_logged):
|
|
"""The first replacement logs one warning with the sender and the offset."""
|
|
|
|
handlers.on_receive(_text_packet(RADIO_CLOCK), None)
|
|
|
|
assert warnings_logged == [
|
|
{
|
|
"severity": "warn",
|
|
"from_id": "!f0acbfba",
|
|
"host_node_id": None,
|
|
"offset_secs": -60 * 86_400,
|
|
"offset_hours": -1440.0,
|
|
"tolerance_secs": 3600,
|
|
"unlogged_replacements": 0,
|
|
}
|
|
]
|
|
|
|
|
|
def test_warning_is_logged_at_most_once_per_interval(
|
|
host_clock, monotonic, warnings_logged
|
|
):
|
|
"""Replacements inside the interval are counted, not logged, and the first
|
|
warning after it reports the count, which then starts again from zero. A
|
|
radio time inside the tolerance is neither logged nor counted."""
|
|
|
|
replace = receive_time.replace_skewed_rx_time
|
|
interval = receive_time.RX_TIME_WARNING_INTERVAL_SECS
|
|
replace(_text_packet(RADIO_CLOCK))
|
|
monotonic.now += interval - 1
|
|
replace(_text_packet(NOW + 2 * 3600))
|
|
replace(_text_packet(NOW - 2 * 3600))
|
|
replace(_text_packet(NOW - 30 * 60))
|
|
assert len(warnings_logged) == 1
|
|
|
|
monotonic.now += 1
|
|
replace(_text_packet(NOW + 39 * 3600))
|
|
# A third interval: one skipped replacement, then a warning that counts only
|
|
# that one, not the two the second warning already reported.
|
|
monotonic.now += interval - 1
|
|
replace(_text_packet(NOW - 3 * 3600))
|
|
monotonic.now += 1
|
|
replace(_text_packet(NOW - 4 * 3600))
|
|
|
|
assert [
|
|
(w["offset_secs"], w["unlogged_replacements"]) for w in warnings_logged
|
|
] == [
|
|
(-60 * 86_400, 0),
|
|
(39 * 3600, 2),
|
|
(-4 * 3600, 1),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Both Meshtastic transports pass the check (SPEC RK3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def open_filters(monkeypatch):
|
|
"""Neutral ingest filters and a primary channel ``MediumFast`` on ``AQ==``."""
|
|
|
|
for name, value in (
|
|
("ALLOWED_CHANNELS", ()),
|
|
("HIDDEN_CHANNELS", ()),
|
|
("PRIMARY_CHANNEL_ONLY", False),
|
|
("DROP_VIA_MQTT", False),
|
|
("PRIMARY_CHANNEL_NAME", "MediumFast"),
|
|
("PRIMARY_CHANNEL_KEY", "AQ=="),
|
|
):
|
|
monkeypatch.setattr(config, name, value, raising=False)
|
|
|
|
|
|
def _udp_datagram(radio_time):
|
|
"""Return a primary-channel text datagram that a gateway stamped ``radio_time``.
|
|
|
|
The payload is AES-CTR-encrypted with the default key under the firmware's
|
|
id/from nonce, so the UDP transport decrypts it as it would a real one;
|
|
``rx_time`` sits in the unencrypted header.
|
|
"""
|
|
|
|
packet_id, sender = 0x5EED0011, SENDER["from"]
|
|
nonce = packet_id.to_bytes(8, "little") + sender.to_bytes(8, "little")
|
|
key = udp_decode.expand_default_key("AQ==")
|
|
encryptor = Cipher(algorithms.AES(key), modes.CTR(nonce)).encryptor()
|
|
data = mesh_pb2.Data(
|
|
portnum=portnums_pb2.PortNum.TEXT_MESSAGE_APP, payload=b"hello"
|
|
)
|
|
packet = mesh_pb2.MeshPacket(
|
|
id=packet_id,
|
|
to=0xFFFFFFFF,
|
|
channel=udp_decode.channel_hash("MediumFast", "AQ=="),
|
|
rx_time=radio_time,
|
|
encrypted=encryptor.update(data.SerializeToString()) + encryptor.finalize(),
|
|
)
|
|
setattr(packet, "from", sender)
|
|
return packet.SerializeToString()
|
|
|
|
|
|
def test_udp_transport_posts_the_host_clock(posts, open_filters):
|
|
"""A datagram stamped by a gateway clock 60 days behind posts the host clock."""
|
|
|
|
MeshtasticUdpProvider()._handle_datagram(
|
|
_udp_datagram(RADIO_CLOCK), _UdpInterface()
|
|
)
|
|
|
|
assert [m["rx_time"] for m in _routed(posts, "/api/messages")] == [NOW]
|
|
|
|
|
|
def test_pubsub_transport_posts_the_host_clock(posts, open_filters, monkeypatch):
|
|
"""A packet the serial, TCP or BLE interface publishes posts the host clock.
|
|
|
|
Drives the pinned meshtastic library's own receive path into the
|
|
subscription that :class:`MeshtasticProvider` registers in production.
|
|
"""
|
|
|
|
monkeypatch.setattr(
|
|
lib_mesh_interface.publishingThread, "queueWork", lambda work: work()
|
|
)
|
|
iface = lib_mesh_interface.MeshInterface(noProto=True)
|
|
iface.nodes = {}
|
|
iface.nodesByNum = {}
|
|
packet = mesh_pb2.MeshPacket(id=0x5EED0012, to=0xFFFFFFFF, rx_time=RADIO_CLOCK)
|
|
setattr(packet, "from", SENDER["from"])
|
|
packet.decoded.portnum = portnums_pb2.PortNum.TEXT_MESSAGE_APP
|
|
packet.decoded.payload = b"hello"
|
|
topics = MeshtasticProvider().subscribe()
|
|
try:
|
|
iface._getOrCreateByNum(SENDER["from"])
|
|
iface._handlePacketFromRadio(packet)
|
|
finally:
|
|
for topic in topics:
|
|
pub.unsubscribe(handlers.on_receive, topic)
|
|
|
|
assert [m["rx_time"] for m in _routed(posts, "/api/messages")] == [NOW]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The host-telemetry suppression window runs on the posted receive time
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _host_telemetry(radio_time, packet_id):
|
|
"""Return a device-telemetry packet from the host radio itself."""
|
|
|
|
return {
|
|
"id": packet_id,
|
|
"rxTime": radio_time,
|
|
"channel": 0,
|
|
**SENDER,
|
|
"decoded": {
|
|
"portnum": "TELEMETRY_APP",
|
|
"telemetry": {"deviceMetrics": {"batteryLevel": 101, "voltage": 4.2}},
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"radio_offsets",
|
|
[
|
|
(-60 * 86_400, -60 * 86_400, -60 * 86_400),
|
|
((39 * 60 + 18) * 60, 0, 0),
|
|
],
|
|
ids=["radio 60 days behind", "radio 39.3 hours fast, then corrected"],
|
|
)
|
|
def test_host_telemetry_window_runs_on_the_host_clock(radio_offsets, posts, host_clock):
|
|
"""Host telemetry heard at 0, 30 and 61 minutes posts the first and the
|
|
last: the hourly window (``_state._host_telemetry_suppressed``) measures the
|
|
posted receive time. Before the fix, a radio clock 39.3 hours fast held
|
|
the window shut for 40 hours once the clock was corrected."""
|
|
|
|
_state_mod.register_host_node_id(SENDER["fromId"])
|
|
for index, (minutes, offset) in enumerate(zip((0, 30, 61), radio_offsets)):
|
|
host_clock.now = NOW + minutes * 60
|
|
packet = _host_telemetry(host_clock.now + offset, 0x5EED0020 + index)
|
|
handlers.on_receive(packet, None)
|
|
|
|
posted = [t["rx_time"] for t in _routed(posts, "/api/telemetry")]
|
|
assert posted == [NOW, NOW + 61 * 60]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Out of scope: MeshCore keeps the times it assigns (SPEC RK3, RS1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_meshcore_roster_position_keeps_last_advert(posts, warnings_logged):
|
|
"""MeshCore roster sync posts a contact's ``last_advert`` itself, outside
|
|
``on_receive``, so an advert 60 days old stays 60 days old (RS-A1)."""
|
|
|
|
last_advert = NOW - 60 * 86_400
|
|
stub = SimpleNamespace(
|
|
upsert_node=lambda *_args, **_kwargs: None,
|
|
host_node_id=lambda: None,
|
|
_mark_packet_activity=lambda: None,
|
|
)
|
|
pub_key = "f0acbfba" + "00" * 28
|
|
contact = {
|
|
"public_key": pub_key,
|
|
"adv_name": "RB",
|
|
"adv_lat": 52.49,
|
|
"adv_lon": 13.47,
|
|
"last_advert": last_advert,
|
|
}
|
|
_process_contacts({pub_key: contact}, _MeshcoreInterface(target=None), stub)
|
|
|
|
assert [p["rx_time"] for p in _routed(posts, "/api/positions")] == [last_advert]
|
|
assert warnings_logged == []
|
|
|
|
|
|
def test_store_packet_dict_posts_the_time_it_is_given(posts, warnings_logged):
|
|
"""The check runs in ``on_receive`` alone: ``store_packet_dict``, the entry
|
|
MeshCore messages and telemetry use, posts ``rx_time`` as given (C2 pins
|
|
the same through ``tests/test_mesh.py``)."""
|
|
|
|
handlers.store_packet_dict({**_text_packet(RADIO_CLOCK), "protocol": "meshcore"})
|
|
|
|
assert [m["rx_time"] for m in _routed(posts, "/api/messages")] == [RADIO_CLOCK]
|
|
assert warnings_logged == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# replace_skewed_rx_time: what it leaves alone and what it copies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"packet",
|
|
[
|
|
"<object object at 0x0>",
|
|
None,
|
|
{},
|
|
{"rxTime": None},
|
|
{"rxTime": "soon"},
|
|
{"rxTime": NOW - 60},
|
|
],
|
|
ids=[
|
|
"not a mapping",
|
|
"None",
|
|
"no receive time",
|
|
"null receive time",
|
|
"unparseable receive time",
|
|
"inside the tolerance",
|
|
],
|
|
)
|
|
def test_packet_without_a_skewed_rx_time_comes_back_as_is(
|
|
packet, host_clock, warnings_logged
|
|
):
|
|
"""With nothing to replace, the same object comes back and nothing is logged."""
|
|
|
|
assert receive_time.replace_skewed_rx_time(packet) is packet
|
|
assert warnings_logged == []
|
|
|
|
|
|
def test_replacement_copies_the_packet_and_covers_both_keys(
|
|
host_clock, warnings_logged
|
|
):
|
|
"""Both receive-time keys are replaced on a copy, so the meshtastic
|
|
library's own packet dict keeps the radio time; one packet, one warning."""
|
|
|
|
packet = {**_text_packet(RADIO_CLOCK), "rx_time": str(RADIO_CLOCK)}
|
|
corrected = receive_time.replace_skewed_rx_time(packet)
|
|
|
|
assert (corrected["rxTime"], corrected["rx_time"]) == (NOW, NOW)
|
|
assert (packet["rxTime"], packet["rx_time"]) == (RADIO_CLOCK, str(RADIO_CLOCK))
|
|
assert len(warnings_logged) == 1
|