Added NodeDB fixtures and refactored to use std maps for better memory efficiency (#10464)
* Added NodeDB fixtures and refactored to use std maps for better efficiency * Defer NodeDB save during xmodem transfer to prevent mid-transfer fsFormat
This commit is contained in:
Ben Meadors
GitHub
@@ -0,0 +1,382 @@
|
||||
"""Fake NodeDB fixture push — Portduino file copy + hardware XModem upload.
|
||||
|
||||
The fixture pipeline is two-stage:
|
||||
1. `bin/gen-fake-nodedb-seed.py` produces a deterministic JSONL describing N
|
||||
fake-but-realistic peers. Committed under `test/fixtures/nodedb/`.
|
||||
2. `bin/seed-json-to-proto.py` compiles JSONL → binary v25 NodeDatabase
|
||||
protobuf with fresh wall-clock timestamps.
|
||||
|
||||
This module exposes `push_fake_nodedb(...)`, the MCP tool that:
|
||||
- target="portduino": compiles the JSONL into the device's prefs dir on
|
||||
the local filesystem (`~/.portduino/<config>/prefs/nodes.proto`).
|
||||
- target="hardware": compiles to a temp file, then streams it over the
|
||||
XModem protocol (via the meshtastic SerialInterface/BLEInterface +
|
||||
`meshtastic.xmodempacket` pubsub topic) to `/prefs/nodes.proto` on the
|
||||
device. Triggers a reboot so the firmware loads the new state on next
|
||||
boot.
|
||||
|
||||
XModem wire details (mirrors firmware impl at src/xmodem.cpp:115-260):
|
||||
* 128-byte chunks; final chunk padded to 128 B with 0x1A (SUB) bytes.
|
||||
* CRC16-CCITT (poly 0x1021, init 0x0000).
|
||||
* SOH/seq=0 carries the destination filename in `buffer.bytes`. ACK if
|
||||
`FSCom.open(filename, FILE_O_WRITE)` succeeds; NAK otherwise.
|
||||
* SOH/seq≥1 carries a 128-byte chunk. ACK = advance; NAK = retransmit.
|
||||
* EOT after the last chunk flushes + closes the file on-device.
|
||||
|
||||
Hardware push requires `confirm=True` (mirrors factory_reset / erase_and_flash
|
||||
in the .github/copilot-instructions.md "never do these without asking" list).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import hashlib
|
||||
import pathlib
|
||||
import queue
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from typing import Any, Literal
|
||||
|
||||
from .connection import connect, is_tcp_port
|
||||
|
||||
# Resolve repo root so the tool works regardless of mcp-server cwd.
|
||||
_REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]
|
||||
_SEED_DIR = _REPO_ROOT / "test" / "fixtures" / "nodedb"
|
||||
_COMPILE_SCRIPT = _REPO_ROOT / "bin" / "seed-json-to-proto.py"
|
||||
|
||||
_DEFAULT_NODES_FILENAME = "/prefs/nodes.proto"
|
||||
_XMODEM_CHUNK = 128
|
||||
_XMODEM_SUB = 0x1A
|
||||
_ACK_TIMEOUT_INIT_S = 5.0
|
||||
_ACK_TIMEOUT_CHUNK_S = 2.0
|
||||
_MAX_CHUNK_RETRIES = 5
|
||||
|
||||
_VALID_SIZES = (250, 500, 1000, 2000)
|
||||
|
||||
|
||||
class FixtureError(RuntimeError):
|
||||
"""Raised for any fixture-push failure (compile, transport, ack timeout, …)."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CRC16-CCITT (poly 0x1021, init 0x0000). Matches the firmware's `crc16_ccitt`.
|
||||
# Hand-rolled to avoid the optional `crcmod` dep.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _crc16_ccitt(data: bytes, *, init: int = 0x0000) -> int:
|
||||
crc = init
|
||||
for b in data:
|
||||
crc ^= b << 8
|
||||
for _ in range(8):
|
||||
if crc & 0x8000:
|
||||
crc = ((crc << 1) ^ 0x1021) & 0xFFFF
|
||||
else:
|
||||
crc = (crc << 1) & 0xFFFF
|
||||
return crc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Compile step — shells out to bin/seed-json-to-proto.py so the MCP module
|
||||
# doesn't have to duplicate the proto-encoding logic.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _compile_proto(jsonl_path: pathlib.Path, out_path: pathlib.Path) -> None:
|
||||
if not _COMPILE_SCRIPT.is_file():
|
||||
raise FixtureError(f"compile script missing at {_COMPILE_SCRIPT}")
|
||||
cmd = [
|
||||
sys.executable,
|
||||
str(_COMPILE_SCRIPT),
|
||||
"--in",
|
||||
str(jsonl_path),
|
||||
"--out",
|
||||
str(out_path),
|
||||
]
|
||||
try:
|
||||
subprocess.run(cmd, check=True, capture_output=True, text=True)
|
||||
except subprocess.CalledProcessError as exc:
|
||||
raise FixtureError(
|
||||
f"seed-json-to-proto.py failed (exit {exc.returncode}):\n"
|
||||
f" stdout: {exc.stdout}\n stderr: {exc.stderr}"
|
||||
) from exc
|
||||
|
||||
|
||||
def _resolve_seed_jsonl(size: int, custom: str | None) -> pathlib.Path:
|
||||
if custom is not None:
|
||||
p = pathlib.Path(custom).expanduser().resolve()
|
||||
if not p.is_file():
|
||||
raise FixtureError(f"custom_seed_jsonl not found: {p}")
|
||||
return p
|
||||
p = _SEED_DIR / f"seed_v25_{size:04d}.jsonl"
|
||||
if not p.is_file():
|
||||
raise FixtureError(
|
||||
f"missing committed seed at {p}. "
|
||||
f"Run `./bin/regen-fake-nodedbs.sh` to generate it."
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Portduino push — file copy into ~/.portduino/<config>/prefs/
|
||||
# ---------------------------------------------------------------------------
|
||||
def _portduino_prefs_dir(config_name: str) -> pathlib.Path:
|
||||
home = pathlib.Path.home()
|
||||
return home / ".portduino" / config_name / "prefs"
|
||||
|
||||
|
||||
def _push_portduino(
|
||||
size: int,
|
||||
jsonl: pathlib.Path,
|
||||
portduino_config: str,
|
||||
backup_existing: bool,
|
||||
) -> dict[str, Any]:
|
||||
prefs = _portduino_prefs_dir(portduino_config)
|
||||
prefs.mkdir(parents=True, exist_ok=True)
|
||||
target = prefs / "nodes.proto"
|
||||
backed_up_to: str | None = None
|
||||
if backup_existing and target.is_file():
|
||||
ts = int(time.time())
|
||||
backup = prefs / f"nodes.proto.bak.{ts}"
|
||||
shutil.move(str(target), str(backup))
|
||||
backed_up_to = str(backup)
|
||||
_compile_proto(jsonl, target)
|
||||
raw = target.read_bytes()
|
||||
return {
|
||||
"transport": "portduino",
|
||||
"path": str(target),
|
||||
"bytes": len(raw),
|
||||
"sha256": hashlib.sha256(raw).hexdigest(),
|
||||
"jsonl_source": str(jsonl),
|
||||
"backed_up_to": backed_up_to,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hardware push — XModem over BLE/serial via the meshtastic Python interface.
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclasses.dataclass
|
||||
class _AckEvent:
|
||||
control: int
|
||||
seq: int
|
||||
|
||||
|
||||
def _wait_for_response(q: "queue.Queue[_AckEvent]", timeout_s: float) -> _AckEvent:
|
||||
try:
|
||||
return q.get(timeout=timeout_s)
|
||||
except queue.Empty as exc:
|
||||
raise FixtureError(
|
||||
f"XModem response timeout after {timeout_s:.1f}s — device not responding"
|
||||
) from exc
|
||||
|
||||
|
||||
def _push_hardware(
|
||||
size: int,
|
||||
jsonl: pathlib.Path,
|
||||
port: str | None,
|
||||
reboot_after: bool,
|
||||
) -> dict[str, Any]:
|
||||
# Lazy imports so the module loads even when the meshtastic deps aren't
|
||||
# available (e.g. CI in a Python env without the package installed).
|
||||
try:
|
||||
from meshtastic.protobuf import mesh_pb2, xmodem_pb2
|
||||
from pubsub import pub
|
||||
except ImportError as exc: # pragma: no cover — dep missing
|
||||
raise FixtureError(
|
||||
f"hardware push requires the meshtastic + pypubsub packages: {exc}"
|
||||
) from exc
|
||||
|
||||
if is_tcp_port(port):
|
||||
raise FixtureError(
|
||||
"hardware push over TCP/portduino is not supported — use "
|
||||
"target='portduino' to drop the fixture directly into the prefs dir."
|
||||
)
|
||||
|
||||
# Compile the fixture to a temp file with fresh timestamps.
|
||||
with tempfile.NamedTemporaryFile(suffix=".proto", delete=False) as tf:
|
||||
proto_path = pathlib.Path(tf.name)
|
||||
try:
|
||||
_compile_proto(jsonl, proto_path)
|
||||
payload = proto_path.read_bytes()
|
||||
finally:
|
||||
proto_path.unlink(missing_ok=True)
|
||||
|
||||
sha256 = hashlib.sha256(payload).hexdigest()
|
||||
total_bytes = len(payload)
|
||||
|
||||
# Subscribe to XModem responses BEFORE we open the interface, so we don't
|
||||
# race the first ACK that arrives during the SOH/seq=0 handshake.
|
||||
#
|
||||
# NB: the signature MUST declare every kwarg pypubsub will see for this
|
||||
# topic, or pubsub locks the topic spec to a smaller set (whichever
|
||||
# subscribe arrives first) and then *rejects* the meshtastic library's
|
||||
# publish call with `SenderUnknownMsgDataError: unknown ... interface`.
|
||||
# The meshtastic lib publishes both `packet=` and `interface=`
|
||||
# (mesh_interface.py:1389-1395), so both must appear here.
|
||||
response_q: "queue.Queue[_AckEvent]" = queue.Queue()
|
||||
|
||||
def _on_xmodem(packet: Any = None, interface: Any = None, **_kw: Any) -> None:
|
||||
if packet is None:
|
||||
return
|
||||
response_q.put(_AckEvent(control=int(packet.control), seq=int(packet.seq)))
|
||||
|
||||
pub.subscribe(_on_xmodem, "meshtastic.xmodempacket")
|
||||
|
||||
chunks_sent = 0
|
||||
retried = 0
|
||||
rebooted = False
|
||||
|
||||
XMC = xmodem_pb2.XModem.Control
|
||||
try:
|
||||
with connect(port=port) as iface:
|
||||
# 1) Send the filename (SOH, seq=0).
|
||||
init_pkt = xmodem_pb2.XModem(
|
||||
control=XMC.Value("SOH"),
|
||||
seq=0,
|
||||
buffer=_DEFAULT_NODES_FILENAME.encode("utf-8"),
|
||||
)
|
||||
iface._sendToRadio(mesh_pb2.ToRadio(xmodemPacket=init_pkt))
|
||||
ack = _wait_for_response(response_q, _ACK_TIMEOUT_INIT_S)
|
||||
if ack.control != XMC.Value("ACK"):
|
||||
raise FixtureError(
|
||||
f"device refused filename {_DEFAULT_NODES_FILENAME!r} "
|
||||
f"(got control={ack.control}, expected ACK). "
|
||||
f"Filesystem full or permissions issue?"
|
||||
)
|
||||
|
||||
# 2) Stream the payload in 128 B chunks.
|
||||
for offset in range(0, total_bytes, _XMODEM_CHUNK):
|
||||
chunk = payload[offset : offset + _XMODEM_CHUNK]
|
||||
if len(chunk) < _XMODEM_CHUNK:
|
||||
# Pad final chunk to 128 B with SUB. The trailing 0x1A bytes
|
||||
# become part of the file on-device, but nanopb ignores
|
||||
# bytes past the end of the top-level message.
|
||||
chunk = chunk + bytes([_XMODEM_SUB] * (_XMODEM_CHUNK - len(chunk)))
|
||||
seq = ((offset // _XMODEM_CHUNK) + 1) % 256
|
||||
# Retry loop on NAK / timeout.
|
||||
attempts = 0
|
||||
while True:
|
||||
pkt = xmodem_pb2.XModem(
|
||||
control=XMC.Value("SOH"),
|
||||
seq=seq,
|
||||
buffer=chunk,
|
||||
crc16=_crc16_ccitt(chunk),
|
||||
)
|
||||
iface._sendToRadio(mesh_pb2.ToRadio(xmodemPacket=pkt))
|
||||
ack = _wait_for_response(response_q, _ACK_TIMEOUT_CHUNK_S)
|
||||
if ack.control == XMC.Value("ACK"):
|
||||
chunks_sent += 1
|
||||
break
|
||||
if ack.control == XMC.Value("NAK"):
|
||||
attempts += 1
|
||||
retried += 1
|
||||
if attempts >= _MAX_CHUNK_RETRIES:
|
||||
# Abort: send CAN so the firmware removes the half-
|
||||
# written file via FSCom.remove(filename).
|
||||
iface._sendToRadio(
|
||||
mesh_pb2.ToRadio(
|
||||
xmodemPacket=xmodem_pb2.XModem(
|
||||
control=XMC.Value("CAN")
|
||||
)
|
||||
)
|
||||
)
|
||||
raise FixtureError(
|
||||
f"chunk seq={seq} NAK'd {attempts} times; "
|
||||
f"aborted transfer (file removed on-device)."
|
||||
)
|
||||
continue # retry the same chunk
|
||||
raise FixtureError(
|
||||
f"unexpected XModem control={ack.control} on seq={seq}"
|
||||
)
|
||||
|
||||
# 3) Tell the device we're done.
|
||||
iface._sendToRadio(
|
||||
mesh_pb2.ToRadio(
|
||||
xmodemPacket=xmodem_pb2.XModem(control=XMC.Value("EOT"))
|
||||
)
|
||||
)
|
||||
ack = _wait_for_response(response_q, _ACK_TIMEOUT_CHUNK_S)
|
||||
if ack.control != XMC.Value("ACK"):
|
||||
raise FixtureError(f"EOT not ACKed (got control={ack.control})")
|
||||
|
||||
# 4) Reboot so loadFromDisk picks up the new file.
|
||||
if reboot_after:
|
||||
iface.localNode.reboot(secs=1)
|
||||
rebooted = True
|
||||
finally:
|
||||
try:
|
||||
pub.unsubscribe(_on_xmodem, "meshtastic.xmodempacket")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return {
|
||||
"transport": "hardware",
|
||||
"port": port,
|
||||
"filename_on_device": _DEFAULT_NODES_FILENAME,
|
||||
"bytes": total_bytes,
|
||||
"chunks_sent": chunks_sent,
|
||||
"retried": retried,
|
||||
"sha256": sha256,
|
||||
"jsonl_source": str(jsonl),
|
||||
"rebooted": rebooted,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public entry point — registered as an MCP tool in server.py.
|
||||
# ---------------------------------------------------------------------------
|
||||
def push_fake_nodedb(
|
||||
size: int,
|
||||
target: Literal["portduino", "hardware"] = "portduino",
|
||||
*,
|
||||
port: str | None = None,
|
||||
portduino_config: str = "default",
|
||||
backup_existing: bool = True,
|
||||
confirm: bool = False,
|
||||
reboot_after: bool = True,
|
||||
custom_seed_jsonl: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Compile a fresh-timestamp NodeDatabase fixture and push it to a device.
|
||||
|
||||
Args:
|
||||
size: 250, 500, 1000, or 2000 — selects which committed seed JSONL to use.
|
||||
target: "portduino" (file copy to ~/.portduino/<config>/prefs/) or
|
||||
"hardware" (XModem upload to /prefs/nodes.proto + reboot).
|
||||
port: required for target="hardware". Serial path (e.g. /dev/cu.usbmodemXXXX)
|
||||
or BLE identifier. TCP endpoints are rejected — use target="portduino"
|
||||
instead.
|
||||
portduino_config: which Portduino instance dir under ~/.portduino/. Default "default".
|
||||
backup_existing: portduino only. Move nodes.proto -> nodes.proto.bak.<ts>
|
||||
if present, so you can roll back.
|
||||
confirm: required True for target="hardware" (writes flash + reboots).
|
||||
reboot_after: hardware only. If True, send a 1-second reboot after the
|
||||
final ACK so loadFromDisk picks up the new file at next boot.
|
||||
custom_seed_jsonl: override the committed JSONL. Use to push a hand-edited
|
||||
test scenario.
|
||||
|
||||
Returns:
|
||||
dict with transport, bytes, sha256, etc. — depends on target.
|
||||
|
||||
"""
|
||||
if size not in _VALID_SIZES:
|
||||
raise FixtureError(
|
||||
f"size must be one of {_VALID_SIZES}; got {size!r}. "
|
||||
f"Add a new committed seed if you need a different cardinality."
|
||||
)
|
||||
|
||||
jsonl = _resolve_seed_jsonl(size, custom_seed_jsonl)
|
||||
|
||||
if target == "portduino":
|
||||
return _push_portduino(size, jsonl, portduino_config, backup_existing)
|
||||
|
||||
if target == "hardware":
|
||||
if not confirm:
|
||||
raise FixtureError(
|
||||
"hardware push writes flash and triggers a reboot — pass confirm=True."
|
||||
)
|
||||
if not port:
|
||||
raise FixtureError(
|
||||
"target='hardware' requires a port (e.g. /dev/cu.usbmodemXXXX)."
|
||||
)
|
||||
return _push_hardware(size, jsonl, port, reboot_after)
|
||||
|
||||
raise FixtureError(f"unknown target {target!r}; expected 'portduino' or 'hardware'")
|
||||
@@ -15,6 +15,7 @@ from . import (
|
||||
admin,
|
||||
boards,
|
||||
devices,
|
||||
fixtures,
|
||||
flash,
|
||||
hw_tools,
|
||||
info,
|
||||
@@ -963,3 +964,45 @@ def recorder_export(
|
||||
dest_dir=dest_dir,
|
||||
streams=streams,
|
||||
)
|
||||
|
||||
|
||||
# ---------- Fixture / test-data push --------------------------------------
|
||||
|
||||
|
||||
@app.tool()
|
||||
def push_fake_nodedb(
|
||||
size: int,
|
||||
target: str = "portduino",
|
||||
port: str | None = None,
|
||||
portduino_config: str = "default",
|
||||
backup_existing: bool = True,
|
||||
confirm: bool = False,
|
||||
reboot_after: bool = True,
|
||||
custom_seed_jsonl: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Push a fake-NodeDB v25 fixture (250/500/1000/2000 nodes) onto a device.
|
||||
|
||||
Two transports:
|
||||
target="portduino" — file copy to ~/.portduino/<portduino_config>/prefs/nodes.proto.
|
||||
Fast, no device connection needed.
|
||||
target="hardware" — XModem upload over serial/BLE to /prefs/nodes.proto.
|
||||
Requires `port` + `confirm=True`. Triggers a reboot
|
||||
so loadFromDisk picks up the new file at next boot.
|
||||
|
||||
Compiles a fresh-timestamp proto from the committed JSONL seed under
|
||||
test/fixtures/nodedb/seed_v25_<N>.jsonl each invocation, so the loaded
|
||||
NodeDB always looks "recent" to the connecting phone. Structural data
|
||||
(names, IDs, positions, telemetries) is deterministic per the seed.
|
||||
|
||||
Override the JSONL via `custom_seed_jsonl` to push a hand-edited scenario.
|
||||
"""
|
||||
return fixtures.push_fake_nodedb(
|
||||
size=size,
|
||||
target=target, # type: ignore[arg-type]
|
||||
port=port,
|
||||
portduino_config=portduino_config,
|
||||
backup_existing=backup_existing,
|
||||
confirm=confirm,
|
||||
reboot_after=reboot_after,
|
||||
custom_seed_jsonl=custom_seed_jsonl,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,364 @@
|
||||
"""Tests for the fake-NodeDB fixture pipeline (bin/gen-fake-nodedb-seed.py
|
||||
+ bin/seed-json-to-proto.py + mcp-server fixtures.push_fake_nodedb).
|
||||
|
||||
Lives under tests/unit/ because none of these touch real hardware — they
|
||||
shell out to the bin/ scripts and decode the resulting protobufs in-process.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import pathlib
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]
|
||||
SEED_GEN = REPO_ROOT / "bin" / "gen-fake-nodedb-seed.py"
|
||||
COMPILE = REPO_ROOT / "bin" / "seed-json-to-proto.py"
|
||||
FIXTURES_DIR = REPO_ROOT / "test" / "fixtures" / "nodedb"
|
||||
|
||||
# Ensure the locally-generated Python protobuf bindings are importable.
|
||||
# These live under `meshtastic_v25` (not `meshtastic`) so they don't shadow
|
||||
# the PyPI `meshtastic` package that the rest of the mcp-server depends on.
|
||||
_BINDINGS_DIR = REPO_ROOT / "bin" / "_generated"
|
||||
if _BINDINGS_DIR.is_dir() and str(_BINDINGS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(_BINDINGS_DIR))
|
||||
|
||||
try:
|
||||
from meshtastic_v25.deviceonly_pb2 import (
|
||||
NodeDatabase, # type: ignore[import-not-found]
|
||||
)
|
||||
except ImportError:
|
||||
NodeDatabase = None # type: ignore[assignment]
|
||||
|
||||
|
||||
def _require_v25_bindings() -> None:
|
||||
if NodeDatabase is None:
|
||||
pytest.skip(
|
||||
"v25 Python protobuf bindings missing; run `./bin/regen-py-protos.sh`."
|
||||
)
|
||||
if "positions" not in NodeDatabase.DESCRIPTOR.fields_by_name:
|
||||
pytest.skip(
|
||||
"Loaded NodeDatabase predates v25 — run `./bin/regen-py-protos.sh`."
|
||||
)
|
||||
|
||||
|
||||
def _run(cmd: list[str]) -> None:
|
||||
subprocess.check_call(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Seed generator: deterministic for given --seed (no wall-clock dependence).
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_seed_generator_is_deterministic(tmp_path: pathlib.Path) -> None:
|
||||
a = tmp_path / "a.jsonl"
|
||||
b = tmp_path / "b.jsonl"
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SEED_GEN),
|
||||
"--count",
|
||||
"100",
|
||||
"--seed",
|
||||
"42",
|
||||
"--out",
|
||||
str(a),
|
||||
]
|
||||
)
|
||||
# Sleep so any sneaky wall-clock leak in the generator would surface as
|
||||
# a byte diff between the two runs.
|
||||
time.sleep(0.8)
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SEED_GEN),
|
||||
"--count",
|
||||
"100",
|
||||
"--seed",
|
||||
"42",
|
||||
"--out",
|
||||
str(b),
|
||||
]
|
||||
)
|
||||
assert a.read_bytes() == b.read_bytes()
|
||||
|
||||
|
||||
def test_seed_generator_meta_line(tmp_path: pathlib.Path) -> None:
|
||||
out = tmp_path / "seed.jsonl"
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SEED_GEN),
|
||||
"--count",
|
||||
"50",
|
||||
"--seed",
|
||||
"1",
|
||||
"--out",
|
||||
str(out),
|
||||
]
|
||||
)
|
||||
lines = out.read_text(encoding="utf-8").splitlines()
|
||||
assert len(lines) == 51 # 1 meta + 50 nodes
|
||||
meta = json.loads(lines[0])
|
||||
assert "_meta" in meta
|
||||
assert meta["_meta"]["version"] == 25
|
||||
assert meta["_meta"]["count"] == 50
|
||||
assert meta["_meta"]["seed"] == 1
|
||||
|
||||
|
||||
def test_seed_only_uses_active_hardware_and_roles(tmp_path: pathlib.Path) -> None:
|
||||
"""Confirm no deprecated roles + no off-list HW models leak through."""
|
||||
out = tmp_path / "seed.jsonl"
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SEED_GEN),
|
||||
"--count",
|
||||
"500",
|
||||
"--seed",
|
||||
"7",
|
||||
"--out",
|
||||
str(out),
|
||||
]
|
||||
)
|
||||
forbidden_roles = {"ROUTER_CLIENT", "REPEATER"}
|
||||
forbidden_hw = {
|
||||
"TLORA_V1",
|
||||
"TLORA_V2",
|
||||
"TLORA_V1_1P3",
|
||||
"TLORA_V2_1_1P6",
|
||||
"TLORA_V2_1_1P8",
|
||||
"HELTEC_V1",
|
||||
"HELTEC_V2_0",
|
||||
"HELTEC_V2_1",
|
||||
"TBEAM",
|
||||
"TBEAM_V0P7",
|
||||
"NANO_G1",
|
||||
"NANO_G1_EXPLORER",
|
||||
"NANO_G2_ULTRA",
|
||||
"STATION_G1",
|
||||
"STATION_G2",
|
||||
"PORTDUINO",
|
||||
"ANDROID_SIM",
|
||||
"DIY_V1",
|
||||
"LORA_RELAY_V1",
|
||||
"NRF52840_PCA10059",
|
||||
"NRF52_UNKNOWN",
|
||||
"DR_DEV",
|
||||
"GENIEBLOCKS",
|
||||
"M5STACK",
|
||||
"RP2040_LORA",
|
||||
"PPR",
|
||||
}
|
||||
for raw in out.read_text(encoding="utf-8").splitlines()[1:]:
|
||||
node = json.loads(raw)
|
||||
assert node["role"] not in forbidden_roles, f"deprecated role: {node['role']}"
|
||||
assert (
|
||||
node["hw_model"] not in forbidden_hw
|
||||
), f"non-tier-1 HW: {node['hw_model']}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Compile step + committed seeds.
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.parametrize("size", [250, 500, 1000, 2000])
|
||||
def test_committed_seed_compiles_and_decodes(size: int, tmp_path: pathlib.Path) -> None:
|
||||
_require_v25_bindings()
|
||||
proto = tmp_path / "out.proto"
|
||||
jsonl = FIXTURES_DIR / f"seed_v25_{size:04d}.jsonl"
|
||||
if not jsonl.is_file():
|
||||
pytest.skip(f"{jsonl} not present — run ./bin/regen-fake-nodedbs.sh")
|
||||
_run([sys.executable, str(COMPILE), "--in", str(jsonl), "--out", str(proto)])
|
||||
|
||||
db = NodeDatabase()
|
||||
db.ParseFromString(proto.read_bytes())
|
||||
assert db.version == 25
|
||||
assert len(db.nodes) == size
|
||||
nums = {n.num for n in db.nodes}
|
||||
assert len(nums) == size, "node numbers must be unique"
|
||||
assert all(n.long_name and n.short_name for n in db.nodes)
|
||||
assert all(len(n.long_name) <= 24 for n in db.nodes) # max_size:25 - NUL
|
||||
|
||||
# Coverage sanity (±10pp tolerance for binomial fluctuation).
|
||||
def in_range(actual: int, expected_ratio: float, tol_pp: float = 0.10) -> bool:
|
||||
lo = max(0, int((expected_ratio - tol_pp) * size))
|
||||
hi = min(size, int((expected_ratio + tol_pp) * size))
|
||||
return lo <= actual <= hi
|
||||
|
||||
assert in_range(len(db.positions), 0.85)
|
||||
assert in_range(len(db.telemetry), 0.70)
|
||||
assert in_range(len(db.environment), 0.25)
|
||||
assert in_range(len(db.status), 0.40)
|
||||
|
||||
|
||||
def test_compile_freshens_timestamps(tmp_path: pathlib.Path) -> None:
|
||||
"""Same JSONL compiled twice → identical structure, different timestamps."""
|
||||
_require_v25_bindings()
|
||||
jsonl = FIXTURES_DIR / "seed_v25_0250.jsonl"
|
||||
if not jsonl.is_file():
|
||||
pytest.skip("250-node seed not present — run ./bin/regen-fake-nodedbs.sh")
|
||||
a = tmp_path / "a.proto"
|
||||
b = tmp_path / "b.proto"
|
||||
_run([sys.executable, str(COMPILE), "--in", str(jsonl), "--out", str(a)])
|
||||
time.sleep(1.2)
|
||||
_run([sys.executable, str(COMPILE), "--in", str(jsonl), "--out", str(b)])
|
||||
|
||||
da = NodeDatabase()
|
||||
db_ = NodeDatabase()
|
||||
da.ParseFromString(a.read_bytes())
|
||||
db_.ParseFromString(b.read_bytes())
|
||||
|
||||
# Zero out timestamp fields and confirm everything else is byte-identical.
|
||||
for d in (da, db_):
|
||||
for n in d.nodes:
|
||||
n.last_heard = 0
|
||||
for p in d.positions:
|
||||
p.position.time = 0
|
||||
assert da.SerializeToString() == db_.SerializeToString()
|
||||
|
||||
# Re-load fresh copies to confirm timestamps actually moved.
|
||||
aa = NodeDatabase()
|
||||
bb = NodeDatabase()
|
||||
aa.ParseFromString(a.read_bytes())
|
||||
bb.ParseFromString(b.read_bytes())
|
||||
aa_max = max(n.last_heard for n in aa.nodes if n.last_heard)
|
||||
bb_max = max(n.last_heard for n in bb.nodes if n.last_heard)
|
||||
assert bb_max >= aa_max
|
||||
assert bb_max - aa_max < 5 # within a few seconds
|
||||
|
||||
|
||||
def test_compile_pinned_now_epoch_is_byte_identical(tmp_path: pathlib.Path) -> None:
|
||||
"""With --now-epoch pinned, two compiles produce identical bytes."""
|
||||
_require_v25_bindings()
|
||||
jsonl = FIXTURES_DIR / "seed_v25_0250.jsonl"
|
||||
if not jsonl.is_file():
|
||||
pytest.skip("250-node seed not present")
|
||||
a = tmp_path / "a.proto"
|
||||
b = tmp_path / "b.proto"
|
||||
for o in (a, b):
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
str(COMPILE),
|
||||
"--in",
|
||||
str(jsonl),
|
||||
"--now-epoch",
|
||||
"1700000000",
|
||||
"--out",
|
||||
str(o),
|
||||
]
|
||||
)
|
||||
assert a.read_bytes() == b.read_bytes()
|
||||
|
||||
|
||||
def test_compile_timestamps_are_recent(tmp_path: pathlib.Path) -> None:
|
||||
_require_v25_bindings()
|
||||
jsonl = FIXTURES_DIR / "seed_v25_0250.jsonl"
|
||||
if not jsonl.is_file():
|
||||
pytest.skip("250-node seed not present")
|
||||
out = tmp_path / "out.proto"
|
||||
_run([sys.executable, str(COMPILE), "--in", str(jsonl), "--out", str(out)])
|
||||
db = NodeDatabase()
|
||||
db.ParseFromString(out.read_bytes())
|
||||
now = int(time.time())
|
||||
# No timestamp older than 7 days, none in the future.
|
||||
for n in db.nodes:
|
||||
if n.last_heard:
|
||||
assert now - 7 * 86400 <= n.last_heard <= now
|
||||
# At least half should be within the last hour
|
||||
# (matches expovariate(mean=3600s)).
|
||||
recent = sum(1 for n in db.nodes if n.last_heard and n.last_heard >= now - 3600)
|
||||
assert recent >= 0.4 * len(db.nodes)
|
||||
|
||||
|
||||
def test_compile_hand_edit_round_trip(tmp_path: pathlib.Path) -> None:
|
||||
"""Edit one JSONL line, recompile, confirm edit appears in the proto."""
|
||||
_require_v25_bindings()
|
||||
src = FIXTURES_DIR / "seed_v25_0250.jsonl"
|
||||
if not src.is_file():
|
||||
pytest.skip("250-node seed not present")
|
||||
dst = tmp_path / "edited.jsonl"
|
||||
lines = src.read_text(encoding="utf-8").splitlines()
|
||||
|
||||
# Find a node that already has telemetry so the index relationship is
|
||||
# easy to assert on the other side.
|
||||
edit_idx = None
|
||||
for i, raw in enumerate(lines[1:], start=1):
|
||||
node = json.loads(raw)
|
||||
if node.get("telemetry") is not None:
|
||||
edit_idx = i
|
||||
break
|
||||
assert edit_idx is not None, "expected at least one node with telemetry"
|
||||
|
||||
node = json.loads(lines[edit_idx])
|
||||
target_num = int(node["num"], 16)
|
||||
node["long_name"] = "Hand Edited Node"
|
||||
node["telemetry"] = {
|
||||
"battery_level": 42,
|
||||
"voltage": 3.71,
|
||||
"channel_utilization": 0.0,
|
||||
"air_util_tx": 0.0,
|
||||
"uptime_seconds": 1,
|
||||
}
|
||||
lines[edit_idx] = json.dumps(node, ensure_ascii=False, sort_keys=True)
|
||||
dst.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
|
||||
out = tmp_path / "out.proto"
|
||||
_run([sys.executable, str(COMPILE), "--in", str(dst), "--out", str(out)])
|
||||
db = NodeDatabase()
|
||||
db.ParseFromString(out.read_bytes())
|
||||
edited = next((n for n in db.nodes if n.num == target_num), None)
|
||||
assert edited is not None
|
||||
assert edited.long_name == "Hand Edited Node"
|
||||
tel = next((t for t in db.telemetry if t.num == target_num), None)
|
||||
assert tel is not None
|
||||
assert tel.device_metrics.battery_level == 42
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Misc smoke checks on the module surface.
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_crc16_ccitt_matches_known_vectors() -> None:
|
||||
"""Sanity-check the hand-rolled CRC16-CCITT matches well-known vectors.
|
||||
|
||||
Test vectors from the XModem-CRC spec (init=0, poly=0x1021):
|
||||
crc16("123456789") = 0x31C3
|
||||
crc16("") = 0x0000
|
||||
"""
|
||||
from meshtastic_mcp.fixtures import _crc16_ccitt
|
||||
|
||||
assert _crc16_ccitt(b"") == 0x0000
|
||||
assert _crc16_ccitt(b"123456789") == 0x31C3
|
||||
|
||||
|
||||
def test_push_fake_nodedb_rejects_invalid_size() -> None:
|
||||
from meshtastic_mcp.fixtures import FixtureError, push_fake_nodedb
|
||||
|
||||
with pytest.raises(FixtureError, match="size must be one of"):
|
||||
push_fake_nodedb(size=999, target="portduino") # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_push_fake_nodedb_hardware_requires_confirm() -> None:
|
||||
from meshtastic_mcp.fixtures import FixtureError, push_fake_nodedb
|
||||
|
||||
with pytest.raises(FixtureError, match="confirm=True"):
|
||||
push_fake_nodedb(size=250, target="hardware", port="/dev/cu.fake")
|
||||
|
||||
|
||||
def test_push_fake_nodedb_hardware_requires_port() -> None:
|
||||
from meshtastic_mcp.fixtures import FixtureError, push_fake_nodedb
|
||||
|
||||
with pytest.raises(FixtureError, match="requires a port"):
|
||||
push_fake_nodedb(size=250, target="hardware", confirm=True)
|
||||
|
||||
|
||||
def test_push_fake_nodedb_hardware_rejects_tcp_port() -> None:
|
||||
from meshtastic_mcp.fixtures import FixtureError, push_fake_nodedb
|
||||
|
||||
with pytest.raises(FixtureError, match="not supported"):
|
||||
push_fake_nodedb(
|
||||
size=250, target="hardware", confirm=True, port="tcp://localhost:4403"
|
||||
)
|
||||
Reference in New Issue
Block a user