mirror of
https://github.com/alexhopeoconnor/firmware.git
synced 2026-10-22 11:17:49 +10:00
Start of MCP server and test suite
This commit is contained in:
Ben Meadors
@@ -0,0 +1,116 @@
|
||||
# Meshtastic MCP Server — Test Harness
|
||||
|
||||
Automated test suite for the MCP server, organized around real operator
|
||||
concerns rather than generic "unit vs hardware".
|
||||
|
||||
## Tiers
|
||||
|
||||
| Dir | Hardware | Question this tier answers |
|
||||
| --------------- | ----------------------- | --------------------------------------------------------------------- |
|
||||
| `unit/` | none | Do the parsing / filtering / profile-generation primitives work? |
|
||||
| `provisioning/` | 1 device, per-test bake | Did my pre-bake recipe stick? Does it survive a factory reset? |
|
||||
| `admin/` | 1 device, shared bake | Do my daily admin ops (owner, channel URL, config writes) round-trip? |
|
||||
| `mesh/` | 2 devices, shared bake | Do my devices actually form a mesh? Send + receive? ACKs? |
|
||||
| `telemetry/` | 2 devices, shared bake | Is telemetry reporting? Is position broadcast correct? |
|
||||
| `monitor/` | 1 device, shared bake | Is the boot log clean (no panics)? |
|
||||
| `fleet/` | varies | Are my CI runs isolated from each other? Are reflashes idempotent? |
|
||||
|
||||
## Quick start
|
||||
|
||||
```bash
|
||||
cd mcp-server
|
||||
pip install -e ".[test]"
|
||||
|
||||
# No hardware — 33 unit tests, ~3 seconds
|
||||
pytest tests/unit -v
|
||||
|
||||
# Hub attached (nRF52840 + ESP32-S3) — first run bakes, then exercises everything
|
||||
pytest tests/ --html=report.html
|
||||
|
||||
# Hub already baked with session profile (dev loop) — skip bake
|
||||
pytest tests/ --assume-baked --html=report.html
|
||||
|
||||
# Force a rebake (new firmware, new seed, etc.)
|
||||
pytest tests/ --force-bake --html=report.html
|
||||
```
|
||||
|
||||
## CLI flags
|
||||
|
||||
- `--force-bake` — always reflash both roles at session start, even if the
|
||||
current state matches the session profile.
|
||||
- `--assume-baked` — skip `test_00_bake.py` entirely. Use when you know the
|
||||
devices are already baked and want a fast dev loop.
|
||||
- `--hub-profile=<yaml>` — point at a YAML file for non-default hub hardware.
|
||||
Default targets VID `0x239a` (nRF52) and `0x303a`/`0x10c4` (ESP32-S3).
|
||||
- `--no-teardown-rebake` — skip the session-end rebake that `provisioning/`
|
||||
and `fleet/` tests perform. Useful in rapid iteration.
|
||||
|
||||
## Environment variables
|
||||
|
||||
- `MESHTASTIC_FIRMWARE_ROOT` — firmware repo path (defaults to `../` from tests/)
|
||||
- `MESHTASTIC_MCP_ENV_NRF52` — PlatformIO env for the nRF52 role (default
|
||||
`heltec-mesh-node-t114`)
|
||||
- `MESHTASTIC_MCP_ENV_ESP32S3` — PlatformIO env for the ESP32-S3 role (default
|
||||
`t-beam-1w`)
|
||||
- `MESHTASTIC_MCP_SEED` — override the session PSK seed (default:
|
||||
`pytest-<unix-ts>`). Set this to reproduce a specific failing run.
|
||||
|
||||
## Fixtures you'll use when adding tests
|
||||
|
||||
All defined in `conftest.py`:
|
||||
|
||||
- **`hub_devices`** → `{"nrf52": "/dev/cu.X", "esp32s3": "/dev/cu.Y"}`. Auto-
|
||||
skips the test if a required role isn't present.
|
||||
- **`test_profile`** → USERPREFS dict for the session (`build_testing_profile`).
|
||||
- **`no_region_profile`** → variant without `USERPREFS_CONFIG_LORA_REGION`.
|
||||
- **`baked_mesh`** → verifies both devices are baked with the session profile
|
||||
(does NOT reflash — that's `test_00_bake.py`'s job).
|
||||
- **`baked_single`** → single verified baked device; parametrize `request.param`
|
||||
to pick role.
|
||||
- **`serial_capture`** → factory; `cap = serial_capture("esp32s3")` starts a
|
||||
pio device monitor session, drains into a per-test buffer, attaches the
|
||||
buffer to the pytest-html report on failure.
|
||||
- **`wait_until`** → exponential-backoff polling helper; `wait_until(lambda:
|
||||
predicate(), timeout=60)` replaces flaky `time.sleep()` patterns.
|
||||
|
||||
## Reports
|
||||
|
||||
`pytest --html=report.html` produces a self-contained HTML with:
|
||||
|
||||
- Per-test pass/fail/skip with timings
|
||||
- On failure: serial log capture from any `serial_capture` fixture used
|
||||
- On failure: `device_info` + lora config JSON for every role on the hub
|
||||
- Session seed and session start time (for reproducibility)
|
||||
|
||||
`pytest --junitxml=junit.xml` produces CI-integration XML.
|
||||
|
||||
`tool_coverage.json` is emitted at session end in the tests directory — shows
|
||||
which of the 37 MCP tools the run exercised. Useful for closing test gaps.
|
||||
|
||||
## Adding a new test
|
||||
|
||||
1. Pick the category that matches the operator concern (not the technical
|
||||
surface). "Does my fleet's owner name persist" is `admin/`, not `unit/`.
|
||||
2. If you need both devices, depend on `baked_mesh`. If you need one, depend
|
||||
on `baked_single`. If you need to mutate hardware state, put it in
|
||||
`provisioning/` or `fleet/` and add a `try/finally` teardown that re-bakes
|
||||
the session profile.
|
||||
3. Use `wait_until` for anything involving LoRa timing — fixed `sleep()`
|
||||
produces flakes.
|
||||
4. Use `serial_capture` when you need to observe firmware log output (e.g.
|
||||
"did the packet get decoded?").
|
||||
5. Add a `@pytest.mark.timeout(N)` — mesh tests routinely hit LoRa-airtime
|
||||
waits; default pytest timeout is infinite.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
- **All hardware tests SKIP** → hub not detected. Plug in the USB hub, verify
|
||||
with `pytest tests/ --collect-only` or `python -c "from meshtastic_mcp import
|
||||
devices; print(devices.list_devices())"`.
|
||||
- **`baked_mesh` fails with "devices not baked"** → run `pytest
|
||||
tests/test_00_bake.py` first, or pass `--force-bake` on the full run.
|
||||
- **Mesh formation tests time out** → check that both devices are on the same
|
||||
session profile (`--force-bake` forces both to the current seed).
|
||||
- **Provisioning tests leave device in bad state** → teardowns re-bake, but
|
||||
if a test crashes between "bake broken state" and "bake good state", run
|
||||
`pytest tests/test_00_bake.py --force-bake` to recover.
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Admin: channel URL export and re-import round-trip.
|
||||
|
||||
Real operator workflow: "I have two fleets, I want them to share a channel
|
||||
config. Export URL from fleet A's bootstrap device, paste into fleet B's
|
||||
onboarding tool, expect identical channels." Proves `getURL` + `setURL`
|
||||
round-trip without data loss.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, info
|
||||
|
||||
|
||||
@pytest.mark.timeout(60)
|
||||
def test_channel_url_roundtrip(
|
||||
baked_mesh: dict[str, Any],
|
||||
test_profile: dict[str, Any],
|
||||
) -> None:
|
||||
"""Verify:
|
||||
1. `get_channel_url()` on a baked device returns a non-empty URL.
|
||||
2. The URL parses — `set_channel_url(url)` accepts it without error.
|
||||
3. After set, `get_channel_url()` returns the same (canonicalized) URL.
|
||||
4. Primary channel name survives round-trip.
|
||||
"""
|
||||
target = "esp32s3"
|
||||
if target not in baked_mesh:
|
||||
pytest.skip(f"role {target!r} not on hub")
|
||||
port = baked_mesh[target]["port"]
|
||||
|
||||
url_before = admin.get_channel_url(include_all=False, port=port)["url"]
|
||||
assert url_before, "device returned empty channel URL"
|
||||
assert (
|
||||
"meshtastic" in url_before.lower() or "#" in url_before
|
||||
), f"URL does not look like a Meshtastic channel URL: {url_before!r}"
|
||||
|
||||
# Re-apply the same URL — no-op in content but exercises the setURL path.
|
||||
applied = admin.set_channel_url(url=url_before, port=port)
|
||||
assert applied["ok"] is True
|
||||
assert applied["channels_imported"] >= 1
|
||||
time.sleep(2.0)
|
||||
|
||||
# Confirm the primary channel name survived
|
||||
live = info.device_info(port=port, timeout_s=8.0)
|
||||
assert live["primary_channel"] == test_profile["USERPREFS_CHANNEL_0_NAME"]
|
||||
|
||||
url_after = admin.get_channel_url(include_all=False, port=port)["url"]
|
||||
# Canonicalization: URLs should match bit-for-bit after a no-op set.
|
||||
assert (
|
||||
url_after == url_before
|
||||
), f"URL changed across setURL round-trip:\nbefore: {url_before}\nafter: {url_after}"
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Admin: owner name persists across a reboot.
|
||||
|
||||
The single most common "did my admin change stick?" test. Proves
|
||||
`localNode.setOwner()` + `writeConfig("device")` commits to non-volatile
|
||||
storage before the reboot.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, info
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_owner_survives_reboot(
|
||||
baked_mesh: dict[str, Any],
|
||||
wait_until,
|
||||
) -> None:
|
||||
target = "esp32s3"
|
||||
if target not in baked_mesh:
|
||||
pytest.skip(f"role {target!r} not on hub")
|
||||
port = baked_mesh[target]["port"]
|
||||
|
||||
pre = info.device_info(port=port, timeout_s=8.0)
|
||||
original = pre.get("long_name") or ""
|
||||
marker = "RebootSurvive"
|
||||
try:
|
||||
admin.set_owner(long_name=marker, short_name="RS", port=port)
|
||||
time.sleep(1.5)
|
||||
|
||||
# Confirm pre-reboot
|
||||
confirmed = info.device_info(port=port, timeout_s=8.0)
|
||||
assert confirmed["long_name"] == marker
|
||||
|
||||
# Reboot (short delay)
|
||||
admin.reboot(port=port, confirm=True, seconds=3)
|
||||
|
||||
# Wait for device to come back
|
||||
time.sleep(8.0)
|
||||
wait_until(
|
||||
lambda: info.device_info(port=port, timeout_s=5.0).get("my_node_num")
|
||||
is not None,
|
||||
timeout=60,
|
||||
backoff_start=1.0,
|
||||
)
|
||||
|
||||
post = info.device_info(port=port, timeout_s=8.0)
|
||||
assert post["long_name"] == marker, (
|
||||
f"owner name did not persist across reboot: "
|
||||
f"expected {marker!r}, got {post['long_name']!r}"
|
||||
)
|
||||
finally:
|
||||
# Restore original (best-effort)
|
||||
try:
|
||||
admin.set_owner(long_name=original or "TestNode", port=port)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -0,0 +1,28 @@
|
||||
"""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
|
||||
@@ -0,0 +1,466 @@
|
||||
"""Root conftest for the MCP server test harness.
|
||||
|
||||
Organizes the fixture graph used by every test tier:
|
||||
|
||||
session_seed ── test_profile ─┐
|
||||
hub_devices ──────────────────┴─ baked_mesh (verifies) ── baked_single (parametrized)
|
||||
hub_devices ──────────────────── no_region_profile (provisioning negative test)
|
||||
(per-test) ──────────────────── serial_capture, device_state_dump, wait_until
|
||||
|
||||
CLI flags (see `pytest_addoption`):
|
||||
--force-bake always reflash at session start, even if state matches
|
||||
--assume-baked trust the operator; skip test_00_bake collection entirely
|
||||
--hub-profile=... path to a YAML file mapping role → {vid, pid_contains}
|
||||
--no-teardown-rebake skip the session-end rebake that provisioning/fleet perform
|
||||
|
||||
Coverage hooks:
|
||||
- Failure artifacts (serial capture, device_info, get_config) are attached
|
||||
to pytest-html reports via `pytest_runtest_makereport`.
|
||||
- Tool-surface coverage (which of the 37 MCP tools got exercised) is
|
||||
accumulated in `tests/tool_coverage.py` and written to
|
||||
`tool_coverage.json` at session end.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import sys
|
||||
import time
|
||||
from typing import Any, Callable
|
||||
|
||||
import pytest
|
||||
|
||||
# Ensure the MCP server is on `sys.path` without requiring installation in
|
||||
# development mode for every checkout (we DO install in .venv but this makes
|
||||
# `pytest tests/` work from a fresh clone too).
|
||||
_HERE = pathlib.Path(__file__).resolve().parent
|
||||
_MCP_SRC = _HERE.parent / "src"
|
||||
if str(_MCP_SRC) not in sys.path:
|
||||
sys.path.insert(0, str(_MCP_SRC))
|
||||
|
||||
# Default firmware root: the repo this mcp-server/ lives inside.
|
||||
os.environ.setdefault("MESHTASTIC_FIRMWARE_ROOT", str(_HERE.parent.parent))
|
||||
|
||||
from meshtastic_mcp import (
|
||||
admin,
|
||||
)
|
||||
from meshtastic_mcp import ( # noqa: E402 (import after path setup)
|
||||
devices as devices_module,
|
||||
)
|
||||
from meshtastic_mcp import (
|
||||
flash,
|
||||
info,
|
||||
serial_session,
|
||||
userprefs,
|
||||
)
|
||||
|
||||
from . import tool_coverage # noqa: E402
|
||||
|
||||
# ---------- CLI options ---------------------------------------------------
|
||||
|
||||
|
||||
def pytest_addoption(parser: pytest.Parser) -> None:
|
||||
group = parser.getgroup("meshtastic", "Meshtastic MCP test options")
|
||||
group.addoption(
|
||||
"--force-bake",
|
||||
action="store_true",
|
||||
help="Flash both hub roles at session start, even if devices appear baked.",
|
||||
)
|
||||
group.addoption(
|
||||
"--assume-baked",
|
||||
action="store_true",
|
||||
help="Skip `test_00_bake.py` and trust devices are already baked.",
|
||||
)
|
||||
group.addoption(
|
||||
"--hub-profile",
|
||||
default=None,
|
||||
help="YAML file mapping role → {vid, pid_contains} for non-default hardware.",
|
||||
)
|
||||
group.addoption(
|
||||
"--no-teardown-rebake",
|
||||
action="store_true",
|
||||
help="Skip session-end rebake after provisioning/fleet tests mutate state.",
|
||||
)
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(
|
||||
config: pytest.Config, items: list[pytest.Item]
|
||||
) -> None:
|
||||
"""Deselect `test_00_bake.py` when --assume-baked is passed."""
|
||||
if config.getoption("--assume-baked"):
|
||||
keep, skip = [], []
|
||||
for item in items:
|
||||
if "test_00_bake" in item.nodeid:
|
||||
skip.append(item)
|
||||
else:
|
||||
keep.append(item)
|
||||
if skip:
|
||||
for item in skip:
|
||||
item.add_marker(pytest.mark.skip(reason="skipped by --assume-baked"))
|
||||
|
||||
|
||||
# ---------- Session-scoped fixtures ---------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def session_seed(request: pytest.FixtureRequest) -> str:
|
||||
"""Deterministic PSK seed for this pytest session.
|
||||
|
||||
Logged in the HTML report header so two runs can be correlated — and so a
|
||||
flaky-looking test can be reproduced exactly by passing the seed back via
|
||||
an env var (future extension).
|
||||
"""
|
||||
# Pytest session `starttime` isn't directly exposed on the pytest API we
|
||||
# care about, so derive from process start time — unique enough for human
|
||||
# purposes and stable across the session.
|
||||
seed = os.environ.get("MESHTASTIC_MCP_SEED") or f"pytest-{int(time.time())}"
|
||||
return seed
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_profile(session_seed: str) -> dict[str, Any]:
|
||||
"""The canonical isolated-mesh test profile for this session."""
|
||||
return userprefs.build_testing_profile(
|
||||
psk_seed=session_seed,
|
||||
channel_name="McpTest",
|
||||
channel_num=88,
|
||||
region="US",
|
||||
modem_preset="LONG_FAST",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def no_region_profile(session_seed: str) -> dict[str, Any]:
|
||||
"""Variant of `test_profile` with the LoRa region stripped.
|
||||
|
||||
Used only by the negative `unset_region_blocks_tx` test. That test MUST
|
||||
re-bake `test_profile` in its own teardown so downstream shared-state
|
||||
tests still see a correctly-configured mesh.
|
||||
"""
|
||||
profile = userprefs.build_testing_profile(
|
||||
psk_seed=session_seed,
|
||||
channel_name="McpTest",
|
||||
channel_num=88,
|
||||
region="US", # placeholder; we delete the key below
|
||||
modem_preset="LONG_FAST",
|
||||
)
|
||||
profile.pop("USERPREFS_CONFIG_LORA_REGION", None)
|
||||
return profile
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def hub_profile(request: pytest.FixtureRequest) -> dict[str, dict[str, Any]]:
|
||||
"""Role → {vid, pid_contains} map for detecting connected hardware.
|
||||
|
||||
Default covers the common nRF52840 + ESP32-S3 lab hub. Override via
|
||||
`--hub-profile=path/to/hub.yaml`. Example YAML:
|
||||
|
||||
nrf52:
|
||||
vid: 0x239a
|
||||
pid_contains: null
|
||||
esp32s3:
|
||||
vid: 0x303a
|
||||
pid_contains: null
|
||||
"""
|
||||
path = request.config.getoption("--hub-profile")
|
||||
if path:
|
||||
import yaml
|
||||
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
return {
|
||||
"nrf52": {"vid": 0x239A, "pid_contains": None},
|
||||
# ESP32-S3 can enumerate under Espressif native USB (0x303a) or via a
|
||||
# CP2102 USB-serial chip (0x10c4). Both should match for
|
||||
# Meshtastic-compatible boards.
|
||||
"esp32s3": {"vid": 0x303A, "pid_contains": None},
|
||||
"esp32s3_alt": {"vid": 0x10C4, "pid_contains": None},
|
||||
}
|
||||
|
||||
|
||||
def _hex_to_int(value: Any) -> int | None:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return int(value, 16) if value.startswith("0x") else int(value)
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def hub_devices(hub_profile: dict[str, dict[str, Any]]) -> dict[str, str]:
|
||||
"""Map of `role → port` for devices detected on the hub.
|
||||
|
||||
Excludes `*_alt` roles from the returned map (they're additional VID
|
||||
matchers for the same logical role). If a role isn't detected, an entry is
|
||||
absent from the return value; fixtures that require specific roles should
|
||||
check presence and `pytest.skip` with an actionable message.
|
||||
"""
|
||||
# include_unknown=True so non-whitelisted VIDs (e.g. CP2102 at 0x10c4) that
|
||||
# are configured as hub roles still match. The hub_profile itself gates
|
||||
# which VIDs we consider — no risk of unrelated serial ports sneaking in.
|
||||
found = devices_module.list_devices(include_unknown=True)
|
||||
# Coalesce alt roles into their base name (esp32s3_alt → esp32s3)
|
||||
resolved: dict[str, str] = {}
|
||||
for role, spec in hub_profile.items():
|
||||
target_vid = spec["vid"]
|
||||
pid_contains = spec.get("pid_contains")
|
||||
canonical = role.split("_alt", 1)[0]
|
||||
if canonical in resolved:
|
||||
continue
|
||||
for dev in found:
|
||||
vid = _hex_to_int(dev.get("vid"))
|
||||
pid = _hex_to_int(dev.get("pid"))
|
||||
if vid != target_vid:
|
||||
continue
|
||||
if pid_contains is not None and (pid is None or pid_contains not in pid):
|
||||
continue
|
||||
resolved[canonical] = dev["port"]
|
||||
break
|
||||
return resolved
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def baked_mesh(
|
||||
hub_devices: dict[str, str],
|
||||
test_profile: dict[str, Any],
|
||||
session_seed: str,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> dict[str, Any]:
|
||||
"""Verifies that both roles are baked with the session `test_profile`.
|
||||
|
||||
Does NOT reflash. `test_00_bake.py` is responsible for applying the bake;
|
||||
this fixture just checks the result by connecting to each device and
|
||||
comparing the live config to the expected profile.
|
||||
|
||||
Raises with an actionable error if state is missing or mismatched:
|
||||
"device nrf52 at /dev/cu.X not baked with session profile —
|
||||
run test_00_bake.py first or pass --force-bake"
|
||||
|
||||
Returns a per-role dict with `{port, iface_fresh: callable, my_node_num}`.
|
||||
"""
|
||||
required = {"nrf52", "esp32s3"}
|
||||
missing = required - set(hub_devices)
|
||||
if missing:
|
||||
pytest.skip(
|
||||
f"hub missing required role(s): {sorted(missing)}. "
|
||||
f"Attach the hub or override with --hub-profile."
|
||||
)
|
||||
|
||||
expected_region = test_profile["USERPREFS_CONFIG_LORA_REGION"]
|
||||
expected_preset = test_profile["USERPREFS_LORACONFIG_MODEM_PRESET"]
|
||||
expected_slot = test_profile["USERPREFS_LORACONFIG_CHANNEL_NUM"]
|
||||
expected_channel_name = test_profile["USERPREFS_CHANNEL_0_NAME"]
|
||||
|
||||
out: dict[str, Any] = {}
|
||||
for role in ("nrf52", "esp32s3"):
|
||||
port = hub_devices[role]
|
||||
try:
|
||||
live = info.device_info(port=port, timeout_s=12.0)
|
||||
except Exception as exc:
|
||||
pytest.fail(
|
||||
f"device {role} at {port}: could not query device_info "
|
||||
f"({exc!r}). Run test_00_bake.py or pass --force-bake."
|
||||
)
|
||||
# `device_info` surfaces region/primary_channel but not modem preset
|
||||
# or channel_num directly; pull those via a separate get_config call.
|
||||
lora_cfg = admin.get_config(section="lora", port=port)["config"]["lora"]
|
||||
channel_num = int(lora_cfg.get("channel_num", 0))
|
||||
modem_preset = lora_cfg.get("modem_preset")
|
||||
region_short = live.get("region")
|
||||
primary = live.get("primary_channel")
|
||||
|
||||
mismatches = []
|
||||
if region_short and not expected_region.endswith(str(region_short)):
|
||||
mismatches.append(f"region={region_short} (expected {expected_region})")
|
||||
if modem_preset and not expected_preset.endswith(str(modem_preset)):
|
||||
mismatches.append(
|
||||
f"modem_preset={modem_preset} (expected {expected_preset})"
|
||||
)
|
||||
if channel_num != expected_slot:
|
||||
mismatches.append(f"channel_num={channel_num} (expected {expected_slot})")
|
||||
if primary and primary != expected_channel_name:
|
||||
mismatches.append(
|
||||
f"primary_channel={primary!r} (expected {expected_channel_name!r})"
|
||||
)
|
||||
|
||||
if mismatches:
|
||||
pytest.fail(
|
||||
f"device {role} at {port} not baked with session profile:\n "
|
||||
+ "\n ".join(mismatches)
|
||||
+ "\nRun `pytest tests/test_00_bake.py` first or pass --force-bake."
|
||||
)
|
||||
|
||||
out[role] = {
|
||||
"port": port,
|
||||
"my_node_num": live.get("my_node_num"),
|
||||
"firmware_version": live.get("firmware_version"),
|
||||
}
|
||||
|
||||
return out
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def baked_single(
|
||||
baked_mesh: dict[str, Any], request: pytest.FixtureRequest
|
||||
) -> dict[str, Any]:
|
||||
"""Function-scoped: a single verified baked device.
|
||||
|
||||
Parametrize over `request.param` = role name. Defaults to "esp32s3"
|
||||
because it's typically more stable as an admin target (no UF2 transitions).
|
||||
"""
|
||||
role = getattr(request, "param", "esp32s3")
|
||||
if role not in baked_mesh:
|
||||
pytest.skip(f"role {role!r} not present on the hub")
|
||||
return {"role": role, **baked_mesh[role]}
|
||||
|
||||
|
||||
# ---------- Failure-artifact fixtures -------------------------------------
|
||||
|
||||
|
||||
class _SerialCapture:
|
||||
"""Active-session wrapper that lazily opens + closes a pio monitor."""
|
||||
|
||||
def __init__(self, port: str, env: str | None = None) -> None:
|
||||
self._port = port
|
||||
self._env = env
|
||||
self._session = None
|
||||
self._last_cursor: int | None = None
|
||||
|
||||
def start(self) -> None:
|
||||
self._session = serial_session.open_session(port=self._port, env=self._env)
|
||||
|
||||
def snapshot(self, max_lines: int = 500) -> list[str]:
|
||||
if self._session is None:
|
||||
return []
|
||||
out = serial_session.read_session(
|
||||
self._session, max_lines=max_lines, since_cursor=0
|
||||
)
|
||||
return out.get("lines", [])
|
||||
|
||||
def stop(self) -> None:
|
||||
if self._session is not None:
|
||||
try:
|
||||
serial_session.close_session(self._session)
|
||||
except Exception:
|
||||
pass
|
||||
self._session = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def serial_capture(hub_devices: dict[str, str], request: pytest.FixtureRequest) -> Any:
|
||||
"""Returns a `_SerialCapture` factory.
|
||||
|
||||
Usage:
|
||||
cap = serial_capture("esp32s3")
|
||||
cap.start()
|
||||
... run test ...
|
||||
# on failure, serial buffer is attached via pytest_runtest_makereport
|
||||
"""
|
||||
captures: list[_SerialCapture] = []
|
||||
|
||||
def factory(role: str, env: str | None = None) -> _SerialCapture:
|
||||
if role not in hub_devices:
|
||||
pytest.skip(f"role {role!r} not present on the hub")
|
||||
cap = _SerialCapture(port=hub_devices[role], env=env)
|
||||
cap.start()
|
||||
captures.append(cap)
|
||||
request.node._serial_captures = captures # type: ignore[attr-defined]
|
||||
return cap
|
||||
|
||||
yield factory
|
||||
|
||||
for cap in captures:
|
||||
cap.stop()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wait_until() -> Callable[..., Any]:
|
||||
"""Exponential-backoff polling helper.
|
||||
|
||||
Usage:
|
||||
wait_until(lambda: b.node_num in a.iface.nodesByNum, timeout=60)
|
||||
"""
|
||||
|
||||
def _impl(
|
||||
predicate: Callable[[], Any],
|
||||
timeout: float = 60.0,
|
||||
backoff_start: float = 0.5,
|
||||
backoff_max: float = 5.0,
|
||||
) -> Any:
|
||||
deadline = time.monotonic() + timeout
|
||||
delay = backoff_start
|
||||
last: Any = None
|
||||
while time.monotonic() < deadline:
|
||||
last = predicate()
|
||||
if last:
|
||||
return last
|
||||
time.sleep(delay)
|
||||
delay = min(delay * 1.5, backoff_max)
|
||||
raise AssertionError(
|
||||
f"predicate did not return truthy within {timeout}s (last={last!r})"
|
||||
)
|
||||
|
||||
return _impl
|
||||
|
||||
|
||||
# ---------- pytest hooks: report attachments + coverage -------------------
|
||||
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[Any]) -> Any:
|
||||
"""On test failure, attach serial capture + device state as report artifacts."""
|
||||
outcome = yield
|
||||
report = outcome.get_result()
|
||||
|
||||
if report.when != "call" or report.outcome != "failed":
|
||||
return
|
||||
|
||||
extras: list[str] = []
|
||||
|
||||
# Attach serial captures (if the test used `serial_capture`)
|
||||
caps = getattr(item, "_serial_captures", None)
|
||||
if caps:
|
||||
for i, cap in enumerate(caps):
|
||||
lines = cap.snapshot(max_lines=2000)
|
||||
extras.append(
|
||||
f"--- serial capture [{cap._port}] ({len(lines)} lines) ---\n"
|
||||
+ "\n".join(lines[-200:])
|
||||
)
|
||||
|
||||
# Dump device state for any role in hub_devices (if fixture available)
|
||||
hub_fixture = (
|
||||
item.funcargs.get("hub_devices") if hasattr(item, "funcargs") else None
|
||||
)
|
||||
if hub_fixture:
|
||||
for role, port in hub_fixture.items():
|
||||
state: dict[str, Any] = {"role": role, "port": port}
|
||||
try:
|
||||
state["device_info"] = info.device_info(port=port, timeout_s=5.0)
|
||||
except Exception as exc:
|
||||
state["device_info_error"] = repr(exc)
|
||||
try:
|
||||
state["config"] = admin.get_config(section="lora", port=port)
|
||||
except Exception as exc:
|
||||
state["config_error"] = repr(exc)
|
||||
extras.append(
|
||||
f"--- device state [{role}] ---\n{json.dumps(state, indent=2, default=str)}"
|
||||
)
|
||||
|
||||
if extras:
|
||||
# Attach to pytest-html via `report.sections`; pytest-html renders these
|
||||
report.sections.append(("Meshtastic debug", "\n\n".join(extras)))
|
||||
|
||||
|
||||
def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
|
||||
"""Emit `tool_coverage.json` at session end."""
|
||||
out_path = pathlib.Path(__file__).parent / "tool_coverage.json"
|
||||
tool_coverage.write_report(out_path)
|
||||
|
||||
|
||||
# Activate the tool-coverage tracker at import time so imports in fixtures are
|
||||
# also counted.
|
||||
tool_coverage.install()
|
||||
@@ -0,0 +1,21 @@
|
||||
"""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
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Fleet: different session seeds produce non-overlapping PSKs.
|
||||
|
||||
No hardware needed — this is a pure property check on the test profile
|
||||
generator, elevated into the `fleet/` tier because it's the critical
|
||||
invariant for running concurrent CI labs without cross-contamination.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from meshtastic_mcp import userprefs
|
||||
|
||||
|
||||
def test_psk_seed_isolates_runs() -> None:
|
||||
"""Two labs running simultaneously with different seeds must end up with
|
||||
different PSKs — which means firmware baked in lab A cannot decode lab B's
|
||||
traffic, and vice versa.
|
||||
|
||||
This is the formal statement of the isolation claim that
|
||||
`testing_profile` promises operators.
|
||||
"""
|
||||
lab_a_morning = userprefs.build_testing_profile(psk_seed="lab-a-2026-04-16-morning")
|
||||
lab_a_evening = userprefs.build_testing_profile(psk_seed="lab-a-2026-04-16-evening")
|
||||
lab_b_morning = userprefs.build_testing_profile(psk_seed="lab-b-2026-04-16-morning")
|
||||
|
||||
# Same lab, same date, different time-of-day → different PSKs
|
||||
assert (
|
||||
lab_a_morning["USERPREFS_CHANNEL_0_PSK"]
|
||||
!= lab_a_evening["USERPREFS_CHANNEL_0_PSK"]
|
||||
)
|
||||
# Different labs, same time-of-day → different PSKs
|
||||
assert (
|
||||
lab_a_morning["USERPREFS_CHANNEL_0_PSK"]
|
||||
!= lab_b_morning["USERPREFS_CHANNEL_0_PSK"]
|
||||
)
|
||||
|
||||
# Re-deriving with the same seed yields the same PSK (reproducibility)
|
||||
lab_a_morning_again = userprefs.build_testing_profile(
|
||||
psk_seed="lab-a-2026-04-16-morning"
|
||||
)
|
||||
assert (
|
||||
lab_a_morning["USERPREFS_CHANNEL_0_PSK"]
|
||||
== lab_a_morning_again["USERPREFS_CHANNEL_0_PSK"]
|
||||
)
|
||||
@@ -0,0 +1,46 @@
|
||||
"""Mesh: broadcast text from A arrives at B.
|
||||
|
||||
Proves end-to-end send → receive path across a 2-device mesh. Uses serial log
|
||||
capture on B to observe the decoded message rather than the meshtastic Python
|
||||
`onReceive` callback (which would require long-lived iface subscription).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_broadcast_delivers(
|
||||
baked_mesh: dict[str, Any],
|
||||
serial_capture,
|
||||
wait_until,
|
||||
) -> None:
|
||||
"""Flow:
|
||||
1. Start a serial capture on B before sending.
|
||||
2. From A, send a uniquely-tagged text broadcast.
|
||||
3. Poll B's serial buffer for the unique tag.
|
||||
"""
|
||||
if "nrf52" not in baked_mesh or "esp32s3" not in baked_mesh:
|
||||
pytest.skip("both roles required")
|
||||
|
||||
# Capture on B (esp32s3) — pio device monitor shows decoded mesh packets
|
||||
b_env = os.environ.get("MESHTASTIC_MCP_ENV_ESP32S3", "t-beam-1w")
|
||||
cap = serial_capture("esp32s3", env=b_env)
|
||||
time.sleep(2.0) # let monitor settle
|
||||
|
||||
unique = f"mcp-test-{int(time.time())}"
|
||||
admin.send_text(
|
||||
text=unique,
|
||||
port=baked_mesh["nrf52"]["port"],
|
||||
)
|
||||
|
||||
def unique_in_log() -> bool:
|
||||
return any(unique in line for line in cap.snapshot(max_lines=4000))
|
||||
|
||||
wait_until(unique_in_log, timeout=90, backoff_start=2.0, backoff_max=10.0)
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Mesh: direct message with want_ack=True returns a real ACK.
|
||||
|
||||
Real operator concern: "did my message actually arrive?" — want_ack exists
|
||||
precisely to answer that question. A silent drop is the single most common
|
||||
"my mesh is broken" user complaint; this test proves the happy-path ACK
|
||||
round-trip works on a well-formed mesh.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.connection import connect
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_direct_with_ack_roundtrip(baked_mesh: dict[str, Any], wait_until) -> None:
|
||||
"""Wait for mesh formation, then send A → B with want_ack=True via the
|
||||
raw SerialInterface (so we can observe the ACK bookkeeping on the sender
|
||||
iface). The meshtastic SDK exposes `iface.sendText` which returns the
|
||||
outbound packet; the ACK is accounted by the firmware but not directly
|
||||
surfaced to the caller — so we fall back to checking that the send did
|
||||
not raise, and that B's node record has `last_heard` bumped."""
|
||||
if "nrf52" not in baked_mesh or "esp32s3" not in baked_mesh:
|
||||
pytest.skip("both roles required")
|
||||
|
||||
a_port = baked_mesh["nrf52"]["port"]
|
||||
b_node_num = baked_mesh["esp32s3"]["my_node_num"]
|
||||
|
||||
# Wait for mesh formation first (B in A's DB)
|
||||
def b_in_a() -> bool:
|
||||
with connect(port=a_port) as iface:
|
||||
return b_node_num in (iface.nodesByNum or {})
|
||||
|
||||
wait_until(b_in_a, timeout=120, backoff_start=2.0, backoff_max=10.0)
|
||||
|
||||
# Send with want_ack and record lastHeard before/after
|
||||
with connect(port=a_port) as iface:
|
||||
b_record_before = iface.nodesByNum.get(b_node_num, {})
|
||||
last_heard_before = b_record_before.get("lastHeard", 0) or 0
|
||||
|
||||
packet = iface.sendText(
|
||||
"ack-check",
|
||||
destinationId=b_node_num,
|
||||
wantAck=True,
|
||||
)
|
||||
assert packet is not None, "sendText returned None"
|
||||
assert hasattr(packet, "id") or isinstance(
|
||||
packet, dict
|
||||
), "sendText did not return a recognizable packet object"
|
||||
|
||||
# Within a few ACK round-trips on LONG_FAST, lastHeard should tick forward
|
||||
def last_heard_advanced() -> bool:
|
||||
with connect(port=a_port) as iface:
|
||||
current = (iface.nodesByNum.get(b_node_num) or {}).get("lastHeard", 0) or 0
|
||||
return current > last_heard_before
|
||||
|
||||
wait_until(last_heard_advanced, timeout=60, backoff_start=2.0)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Mesh: two devices baked with the same session profile discover each other.
|
||||
|
||||
The fundamental "does my mesh work" test. If both devices share a PSK, LoRa
|
||||
region, modem preset, and channel slot, they should hear each other's
|
||||
NodeInfo packets within ~60s of boot and appear in each other's `nodesByNum`
|
||||
DB.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import info
|
||||
from meshtastic_mcp.connection import connect
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_mesh_formation_within_60s(baked_mesh: dict[str, Any], wait_until) -> None:
|
||||
"""Connect to A, poll its node DB until B's node_num appears. If both
|
||||
devices were freshly baked, NodeInfo broadcast should happen within
|
||||
~30-60s on LONG_FAST."""
|
||||
if "nrf52" not in baked_mesh or "esp32s3" not in baked_mesh:
|
||||
pytest.skip("both roles required")
|
||||
|
||||
a_port = baked_mesh["nrf52"]["port"]
|
||||
b_node_num = baked_mesh["esp32s3"]["my_node_num"]
|
||||
assert b_node_num is not None, "esp32s3 my_node_num not populated"
|
||||
|
||||
def b_visible_from_a() -> bool:
|
||||
with connect(port=a_port) as iface:
|
||||
nodes = iface.nodesByNum or {}
|
||||
return b_node_num in nodes
|
||||
|
||||
wait_until(b_visible_from_a, timeout=120, backoff_start=2.0, backoff_max=10.0)
|
||||
@@ -0,0 +1,50 @@
|
||||
"""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,66 @@
|
||||
"""Monitor: boot log is clean — no panic markers in the first 60 seconds.
|
||||
|
||||
This is the single highest-signal test for catching firmware regressions.
|
||||
If a commit broke something critical at boot (stack overflow, NULL deref, HAL
|
||||
misconfig), this test fails within a minute of reboot.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin
|
||||
|
||||
# Substrings that indicate a panic/assert/crash. Case-insensitive.
|
||||
_PANIC_MARKERS = [
|
||||
"guru meditation",
|
||||
"corrupt heap",
|
||||
"abort()",
|
||||
"assertion failed",
|
||||
"***", # ESP-IDF "*** something" panic prefix
|
||||
"panic",
|
||||
"stack overflow",
|
||||
"load prohibited",
|
||||
"store prohibited",
|
||||
"illegalinstr",
|
||||
"watchdog got triggered",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_boot_log_no_panic(
|
||||
baked_mesh: dict[str, Any],
|
||||
serial_capture,
|
||||
wait_until,
|
||||
) -> None:
|
||||
"""Reboot the device, then watch ~60s of boot log for panic markers."""
|
||||
target = "esp32s3"
|
||||
if target not in baked_mesh:
|
||||
pytest.skip(f"role {target!r} not on hub")
|
||||
port = baked_mesh[target]["port"]
|
||||
|
||||
env = os.environ.get("MESHTASTIC_MCP_ENV_ESP32S3", "t-beam-1w")
|
||||
|
||||
# Start monitor BEFORE reboot so we catch the reset banner + early boot
|
||||
cap = serial_capture(target, env=env)
|
||||
time.sleep(1.0)
|
||||
|
||||
# Trigger reboot
|
||||
admin.reboot(port=port, confirm=True, seconds=3)
|
||||
# Wait through the reboot+boot window
|
||||
time.sleep(60.0)
|
||||
|
||||
lines = cap.snapshot(max_lines=4000)
|
||||
assert lines, "serial capture returned no log lines — monitor may have failed"
|
||||
blob = "\n".join(lines).lower()
|
||||
|
||||
hits = [marker for marker in _PANIC_MARKERS if marker in blob]
|
||||
assert (
|
||||
not hits
|
||||
), f"panic markers in boot log: {hits!r}\n\n" f"last 60 lines:\n" + "\n".join(
|
||||
lines[-60:]
|
||||
)
|
||||
@@ -0,0 +1,22 @@
|
||||
"""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
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Provisioning: baked admin keys end up in the device's security config.
|
||||
|
||||
Fleet operators pre-bake an `USERPREFS_USE_ADMIN_KEY_0` into firmware so that
|
||||
remote-admin messages from a central controller are accepted. This test
|
||||
verifies the key bytes make the round-trip: USERPREFS → build-time `-D` flag
|
||||
→ firmware → `localConfig.security.admin_key`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, flash
|
||||
|
||||
# Deterministic 32-byte "admin key" — just the byte values 0..31 for easy
|
||||
# recognition in the output, formatted as a C brace-init.
|
||||
_ADMIN_KEY_BYTES = list(range(32))
|
||||
_ADMIN_KEY_BRACE = "{ " + ", ".join(f"0x{b:02x}" for b in _ADMIN_KEY_BYTES) + " }"
|
||||
|
||||
|
||||
@pytest.mark.timeout(600)
|
||||
def test_admin_key_baked(
|
||||
hub_devices: dict[str, str],
|
||||
test_profile: dict[str, Any],
|
||||
) -> None:
|
||||
"""Bake test_profile + admin key 0; verify `security.admin_key` contains
|
||||
the baked bytes after boot. Re-bakes session profile (without admin key)
|
||||
on teardown so downstream tests see baseline state.
|
||||
"""
|
||||
target = "esp32s3"
|
||||
if target not in hub_devices:
|
||||
pytest.skip(f"role {target!r} not on hub")
|
||||
port = hub_devices[target]
|
||||
env = os.environ.get("MESHTASTIC_MCP_ENV_ESP32S3", "t-beam-1w")
|
||||
|
||||
augmented = dict(test_profile)
|
||||
augmented["USERPREFS_USE_ADMIN_KEY_0"] = _ADMIN_KEY_BRACE
|
||||
|
||||
try:
|
||||
result = flash.erase_and_flash(
|
||||
env=env,
|
||||
port=port,
|
||||
confirm=True,
|
||||
userprefs_overrides=augmented,
|
||||
)
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
security = admin.get_config(section="security", port=port)["config"]["security"]
|
||||
# `admin_key` may be a list of byte-sequences under newer protobuf, or
|
||||
# a single bytes field under older. We accept either as long as the
|
||||
# baked bytes appear somewhere in the serialization.
|
||||
key_field = security.get("admin_key")
|
||||
import base64
|
||||
import json
|
||||
|
||||
serialized = json.dumps(security)
|
||||
|
||||
# Protobuf→JSON typically base64-encodes bytes fields. Encode our
|
||||
# expected bytes and look for them (or a substring) in the serialized
|
||||
# security config.
|
||||
b64 = base64.b64encode(bytes(_ADMIN_KEY_BYTES)).decode("ascii").rstrip("=")
|
||||
assert (
|
||||
b64[:40] in serialized or "admin_key" in serialized
|
||||
), f"admin_key bytes not visible in security config: {security!r}"
|
||||
assert (
|
||||
key_field is not None
|
||||
), "security.admin_key field absent — baking key 0 didn't stick"
|
||||
finally:
|
||||
# Restore session profile (no admin key)
|
||||
restore = flash.erase_and_flash(
|
||||
env=env,
|
||||
port=port,
|
||||
confirm=True,
|
||||
userprefs_overrides=test_profile,
|
||||
)
|
||||
assert restore["exit_code"] == 0
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Provisioning: the pre-bake recipe (US/LONG_FAST/slot 88/private channel)
|
||||
lands on the device exactly as specified.
|
||||
|
||||
This is THE test that proves the MCP's core value prop — flashing firmware
|
||||
with a preset USERPREFS produces a device in the expected radio config without
|
||||
any post-flash admin steps.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, info
|
||||
|
||||
|
||||
@pytest.mark.timeout(60)
|
||||
def test_bake_sets_region_preset_and_slot(
|
||||
baked_mesh: dict[str, Any],
|
||||
test_profile: dict[str, Any],
|
||||
) -> None:
|
||||
"""After test_00_bake, both devices must report the exact region, modem
|
||||
preset, slot, and channel name that the profile specified."""
|
||||
for role, state in baked_mesh.items():
|
||||
port = state["port"]
|
||||
live = info.device_info(port=port, timeout_s=8.0)
|
||||
lora = admin.get_config(section="lora", port=port)["config"]["lora"]
|
||||
|
||||
expected_region = test_profile["USERPREFS_CONFIG_LORA_REGION"].rsplit("_", 1)[
|
||||
-1
|
||||
]
|
||||
expected_preset = test_profile["USERPREFS_LORACONFIG_MODEM_PRESET"].rsplit(
|
||||
"_", 2
|
||||
)[-2:]
|
||||
expected_preset_str = "_".join(expected_preset)
|
||||
|
||||
assert (
|
||||
live["region"] == expected_region
|
||||
), f"{role}: region={live['region']!r}, expected {expected_region!r}"
|
||||
assert lora.get("modem_preset") in (
|
||||
expected_preset_str,
|
||||
expected_preset_str.upper(),
|
||||
), f"{role}: modem_preset={lora.get('modem_preset')!r}, expected {expected_preset_str!r}"
|
||||
assert (
|
||||
int(lora.get("channel_num", 0))
|
||||
== test_profile["USERPREFS_LORACONFIG_CHANNEL_NUM"]
|
||||
), f"{role}: channel_num={lora.get('channel_num')!r}"
|
||||
assert live["primary_channel"] == test_profile["USERPREFS_CHANNEL_0_NAME"]
|
||||
@@ -0,0 +1,37 @@
|
||||
"""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
|
||||
@@ -0,0 +1,103 @@
|
||||
"""Provisioning (negative): firmware baked WITHOUT
|
||||
`USERPREFS_CONFIG_LORA_REGION` must refuse to transmit.
|
||||
|
||||
Real operator concern: FCC compliance. A device shipped without an explicit
|
||||
region setting must not emit RF until the operator sets a region — this test
|
||||
proves the firmware honors that invariant when the USERPREFS bake deliberately
|
||||
omits the region key.
|
||||
|
||||
Teardown re-bakes the session `test_profile` so downstream shared-state
|
||||
tiers (admin/mesh/telemetry) still see a correctly configured mesh.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, flash, info
|
||||
|
||||
|
||||
@pytest.mark.timeout(600)
|
||||
def test_unset_region_blocks_tx(
|
||||
hub_devices: dict[str, str],
|
||||
no_region_profile: dict[str, Any],
|
||||
test_profile: dict[str, Any],
|
||||
serial_capture,
|
||||
) -> None:
|
||||
"""Bake a device with no LoRa region, then assert:
|
||||
1. `config.lora.region` reads as "UNSET" (or 0).
|
||||
2. An attempt to `send_text` surfaces a refusal — either the meshtastic
|
||||
SDK raises, or the serial log contains a clear "region unset" marker.
|
||||
|
||||
Always re-bakes the session test_profile in the finalizer so downstream
|
||||
categories are not left with a broken device.
|
||||
"""
|
||||
target = "esp32s3"
|
||||
if target not in hub_devices:
|
||||
pytest.skip(f"role {target!r} not on hub")
|
||||
port = hub_devices[target]
|
||||
|
||||
# Pick the right env for this role — must match what test_00_bake used.
|
||||
import os
|
||||
|
||||
env = os.environ.get("MESHTASTIC_MCP_ENV_ESP32S3", "t-beam-1w")
|
||||
|
||||
# Capture serial before the bake to see the "region unset" log line on boot
|
||||
cap = serial_capture(target, env=env)
|
||||
|
||||
# Bake without region
|
||||
result = flash.erase_and_flash(
|
||||
env=env,
|
||||
port=port,
|
||||
confirm=True,
|
||||
userprefs_overrides=no_region_profile,
|
||||
)
|
||||
assert (
|
||||
result["exit_code"] == 0
|
||||
), f"bake of no_region_profile failed:\n{result.get('stderr_tail', '')}"
|
||||
|
||||
try:
|
||||
# After bake, device should boot with region=UNSET
|
||||
live = info.device_info(port=port, timeout_s=12.0)
|
||||
assert live.get("region") in (None, "UNSET", "UNSET_0", ""), (
|
||||
f"expected region UNSET after baking without region pref; "
|
||||
f"got {live.get('region')!r}"
|
||||
)
|
||||
|
||||
# Attempting to send a message should either raise or be logged as
|
||||
# refused. The meshtastic SDK's sendText may raise in this condition,
|
||||
# or it may accept the call but the firmware rejects on air.
|
||||
send_failed = False
|
||||
try:
|
||||
admin.send_text(text="should not transmit", port=port)
|
||||
except Exception:
|
||||
send_failed = True
|
||||
|
||||
# Give the firmware a moment to log anything
|
||||
import time as _time
|
||||
|
||||
_time.sleep(3.0)
|
||||
log = "\n".join(cap.snapshot(max_lines=2000))
|
||||
# We expect EITHER the send raised at the Python layer, OR the serial
|
||||
# log explicitly says region is unset.
|
||||
log_says_unset = any(
|
||||
marker in log.lower()
|
||||
for marker in ("region unset", "region is unset", "no region set")
|
||||
)
|
||||
assert send_failed or log_says_unset, (
|
||||
"expected send to fail or log 'region unset'; neither happened.\n"
|
||||
f"log tail:\n{log[-2000:]}"
|
||||
)
|
||||
finally:
|
||||
# Re-bake the session profile so downstream tests work.
|
||||
restore = flash.erase_and_flash(
|
||||
env=env,
|
||||
port=port,
|
||||
confirm=True,
|
||||
userprefs_overrides=test_profile,
|
||||
)
|
||||
assert restore["exit_code"] == 0, (
|
||||
"CRITICAL: failed to re-bake session profile after "
|
||||
"no-region test; downstream tests will fail."
|
||||
)
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Provisioning: after a non-full factory_reset, USERPREFS defaults come back.
|
||||
|
||||
Real operator concern: "if someone resets my fleet device, will it come back
|
||||
on my private mesh or on Meshtastic defaults?" A baked USERPREFS recipe
|
||||
should be the factory floor for the device — reset goes back to THAT state,
|
||||
not to stock Meshtastic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, info
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_baked_prefs_survive_factory_reset(
|
||||
baked_mesh: dict[str, Any],
|
||||
test_profile: dict[str, Any],
|
||||
wait_until,
|
||||
) -> None:
|
||||
"""Flow:
|
||||
1. Change owner name to a known-non-default value.
|
||||
2. Trigger factory_reset(full=False).
|
||||
3. Wait for device to come back.
|
||||
4. Confirm owner is back to USERPREFS-baked default (or blank default if
|
||||
not baked), and primary channel/region/slot are still the baked values.
|
||||
"""
|
||||
# Use esp32s3 — typically more robust across reset cycles.
|
||||
target = "esp32s3"
|
||||
if target not in baked_mesh:
|
||||
pytest.skip(f"role {target!r} not on hub")
|
||||
port = baked_mesh[target]["port"]
|
||||
|
||||
# Snapshot pre-reset config
|
||||
pre_reset = info.device_info(port=port, timeout_s=8.0)
|
||||
original_long_name = pre_reset.get("long_name")
|
||||
|
||||
# Poison the owner name with a non-default marker
|
||||
admin.set_owner(long_name="PoisonMarker", short_name="POIZ", port=port)
|
||||
time.sleep(2.0)
|
||||
|
||||
# Confirm poison stuck before reset
|
||||
poisoned = info.device_info(port=port, timeout_s=8.0)
|
||||
assert poisoned.get("long_name") == "PoisonMarker"
|
||||
|
||||
# Trigger non-full factory reset
|
||||
admin.factory_reset(port=port, confirm=True, full=False)
|
||||
|
||||
# Wait for device to come back (serial reappears)
|
||||
time.sleep(10.0) # reset takes a moment
|
||||
wait_until(
|
||||
lambda: info.device_info(port=port, timeout_s=5.0).get("my_node_num")
|
||||
is not None,
|
||||
timeout=60,
|
||||
backoff_start=1.0,
|
||||
)
|
||||
|
||||
post = info.device_info(port=port, timeout_s=8.0)
|
||||
# The key assertion: channel + region are STILL the USERPREFS-baked values,
|
||||
# NOT Meshtastic stock defaults (which would be "LongFast" and the region
|
||||
# the device shipped with).
|
||||
assert post["primary_channel"] == test_profile["USERPREFS_CHANNEL_0_NAME"], (
|
||||
f"after factory_reset, primary_channel reverted to "
|
||||
f"{post['primary_channel']!r}, not baked {test_profile['USERPREFS_CHANNEL_0_NAME']!r}"
|
||||
)
|
||||
expected_region = test_profile["USERPREFS_CONFIG_LORA_REGION"].rsplit("_", 1)[-1]
|
||||
assert post.get("region") == expected_region
|
||||
|
||||
# Owner name should NOT be "PoisonMarker" anymore
|
||||
assert (
|
||||
post.get("long_name") != "PoisonMarker"
|
||||
), "factory_reset did not wipe the poisoned owner name"
|
||||
|
||||
# If we had an original_long_name, restore it so downstream tests see
|
||||
# the same baseline.
|
||||
if original_long_name and post.get("long_name") != original_long_name:
|
||||
admin.set_owner(long_name=original_long_name, port=port)
|
||||
@@ -0,0 +1,42 @@
|
||||
"""Telemetry: device metrics (battery, voltage, channel util) arrive at the peer.
|
||||
|
||||
After ~2× the telemetry interval, B's entry in A's node DB should carry a
|
||||
populated `deviceMetrics` block. This is the happy-path "my fleet is
|
||||
reporting health data" operator test.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.connection import connect
|
||||
|
||||
|
||||
@pytest.mark.timeout(360)
|
||||
def test_device_telemetry_broadcast(baked_mesh: dict[str, Any], wait_until) -> None:
|
||||
"""Wait up to 5 minutes for B's device telemetry to land in A's DB.
|
||||
|
||||
Firmware default telemetry interval is 900s; on a fresh mesh the first
|
||||
device-metrics broadcast happens within ~30-120s of boot because devices
|
||||
broadcast once on startup. We only require that some telemetry is present,
|
||||
not that we see multiple cycles.
|
||||
"""
|
||||
if "nrf52" not in baked_mesh or "esp32s3" not in baked_mesh:
|
||||
pytest.skip("both roles required")
|
||||
|
||||
a_port = baked_mesh["nrf52"]["port"]
|
||||
b_node_num = baked_mesh["esp32s3"]["my_node_num"]
|
||||
|
||||
def b_has_telemetry() -> bool:
|
||||
with connect(port=a_port) as iface:
|
||||
rec = (iface.nodesByNum or {}).get(b_node_num, {})
|
||||
metrics = rec.get("deviceMetrics") or {}
|
||||
# Any one of these being non-None is sufficient evidence that
|
||||
# telemetry arrived.
|
||||
return any(
|
||||
metrics.get(k) is not None
|
||||
for k in ("batteryLevel", "voltage", "channelUtilization", "airUtilTx")
|
||||
)
|
||||
|
||||
wait_until(b_has_telemetry, timeout=300, backoff_start=5.0, backoff_max=15.0)
|
||||
@@ -0,0 +1,22 @@
|
||||
"""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,117 @@
|
||||
"""Session-bake module — runs first (alphabetical collection) to flash both hub
|
||||
roles with the session `test_profile`.
|
||||
|
||||
Skipped entirely when `--assume-baked` is passed. All downstream hardware
|
||||
tests either depend on `baked_mesh` (which verifies state) or do their own
|
||||
per-test bake (provisioning/fleet tiers), so failing here gives one clear
|
||||
actionable failure instead of a cascade of mismatches.
|
||||
|
||||
Hardware-specific env names live in a small role→env map at the top of this
|
||||
file; override by setting `MESHTASTIC_MCP_ENV_<ROLE>` env vars (e.g.
|
||||
`MESHTASTIC_MCP_ENV_NRF52=heltec-mesh-node-t114`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import flash, info
|
||||
|
||||
# Default envs for a common lab setup. Override per-role via env var.
|
||||
_DEFAULT_ENVS = {
|
||||
"nrf52": "heltec-mesh-node-t114",
|
||||
"esp32s3": "t-beam-1w",
|
||||
}
|
||||
|
||||
|
||||
def _env_for(role: str) -> str:
|
||||
override = os.environ.get(f"MESHTASTIC_MCP_ENV_{role.upper()}")
|
||||
if override:
|
||||
return override
|
||||
if role not in _DEFAULT_ENVS:
|
||||
pytest.fail(
|
||||
f"no default PlatformIO env for role {role!r}. "
|
||||
f"Set MESHTASTIC_MCP_ENV_{role.upper()} to the env name."
|
||||
)
|
||||
return _DEFAULT_ENVS[role]
|
||||
|
||||
|
||||
def _bake_role(
|
||||
role: str,
|
||||
port: str,
|
||||
test_profile: dict[str, Any],
|
||||
force_bake: bool,
|
||||
) -> None:
|
||||
"""Bake + boot + verify for a single role. Skips if already baked unless
|
||||
`--force-bake` was passed."""
|
||||
env = _env_for(role)
|
||||
|
||||
# If not forcing, check if already baked with session profile.
|
||||
if not force_bake:
|
||||
try:
|
||||
live = info.device_info(port=port, timeout_s=8.0)
|
||||
# Quick heuristic: region matches and primary channel matches.
|
||||
expected_region_short = test_profile["USERPREFS_CONFIG_LORA_REGION"].rsplit(
|
||||
"_", 1
|
||||
)[-1]
|
||||
if (
|
||||
live.get("region") == expected_region_short
|
||||
and live.get("primary_channel")
|
||||
== test_profile["USERPREFS_CHANNEL_0_NAME"]
|
||||
):
|
||||
pytest.skip(
|
||||
f"{role} at {port} already baked with session profile "
|
||||
f"(pass --force-bake to reflash)"
|
||||
)
|
||||
except Exception:
|
||||
# If we can't query, fall through and bake anyway.
|
||||
pass
|
||||
|
||||
result = flash.erase_and_flash(
|
||||
env=env,
|
||||
port=port,
|
||||
confirm=True,
|
||||
skip_build=False,
|
||||
userprefs_overrides=test_profile,
|
||||
)
|
||||
assert result["exit_code"] == 0, (
|
||||
f"{role} bake failed: exit={result['exit_code']}\n"
|
||||
f"stdout tail:\n{result.get('stdout_tail', '')}\n"
|
||||
f"stderr tail:\n{result.get('stderr_tail', '')}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.timeout(600)
|
||||
def test_bake_nrf52(
|
||||
hub_devices: dict[str, str],
|
||||
test_profile: dict[str, Any],
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
"""Flash the nRF52840 role with the session test profile."""
|
||||
if "nrf52" not in hub_devices:
|
||||
pytest.skip("nRF52 not detected on hub")
|
||||
_bake_role(
|
||||
role="nrf52",
|
||||
port=hub_devices["nrf52"],
|
||||
test_profile=test_profile,
|
||||
force_bake=request.config.getoption("--force-bake"),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.timeout(600)
|
||||
def test_bake_esp32s3(
|
||||
hub_devices: dict[str, str],
|
||||
test_profile: dict[str, Any],
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
"""Flash the ESP32-S3 role with the session test profile."""
|
||||
if "esp32s3" not in hub_devices:
|
||||
pytest.skip("ESP32-S3 not detected on hub")
|
||||
_bake_role(
|
||||
role="esp32s3",
|
||||
port=hub_devices["esp32s3"],
|
||||
test_profile=test_profile,
|
||||
force_bake=request.config.getoption("--force-bake"),
|
||||
)
|
||||
@@ -0,0 +1,124 @@
|
||||
"""Tool-surface coverage: track which MCP tools the test suite actually exercises.
|
||||
|
||||
This is NOT line coverage (that's `coverage.py`). This measures which of the
|
||||
37 public MCP tools in `meshtastic_mcp.server` got invoked during a pytest
|
||||
run — a quick signal for "where are the test-coverage gaps".
|
||||
|
||||
Approach: introspect `meshtastic_mcp.server.app` for registered tools, find
|
||||
the underlying handler functions in their source modules, and wrap each with
|
||||
a counting shim. At session end, emit `tool_coverage.json` mapping each tool
|
||||
name to its call count. Tools never called show `count=0`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import pathlib
|
||||
from typing import Any, Callable
|
||||
|
||||
_counts: dict[str, int] = {}
|
||||
_installed = False
|
||||
|
||||
|
||||
def _bump(name: str) -> None:
|
||||
_counts[name] = _counts.get(name, 0) + 1
|
||||
|
||||
|
||||
def _wrap(module: Any, attr: str, tool_name: str) -> None:
|
||||
original = getattr(module, attr, None)
|
||||
if original is None or not callable(original):
|
||||
return
|
||||
|
||||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
_bump(tool_name)
|
||||
return original(*args, **kwargs)
|
||||
|
||||
wrapper.__wrapped__ = original # type: ignore[attr-defined]
|
||||
wrapper.__name__ = attr
|
||||
wrapper.__doc__ = original.__doc__
|
||||
setattr(module, attr, wrapper)
|
||||
|
||||
|
||||
# Mapping: MCP tool name → (module, function name). Mirrors the wiring in
|
||||
# `meshtastic_mcp.server`. Keep synchronized manually — adding a tool without
|
||||
# updating this map means it shows as count=0 in reports even if exercised.
|
||||
_TOOL_MAP: dict[str, tuple[str, str]] = {
|
||||
# Discovery & metadata
|
||||
"list_devices": ("meshtastic_mcp.devices", "list_devices"),
|
||||
"list_boards": ("meshtastic_mcp.boards", "list_boards"),
|
||||
"get_board": ("meshtastic_mcp.boards", "get_board"),
|
||||
# Build & flash
|
||||
"build": ("meshtastic_mcp.flash", "build"),
|
||||
"clean": ("meshtastic_mcp.flash", "clean"),
|
||||
"pio_flash": ("meshtastic_mcp.flash", "flash"),
|
||||
"erase_and_flash": ("meshtastic_mcp.flash", "erase_and_flash"),
|
||||
"update_flash": ("meshtastic_mcp.flash", "update_flash"),
|
||||
"touch_1200bps": ("meshtastic_mcp.flash", "touch_1200bps"),
|
||||
# Serial log sessions — module-level functions on serial_session
|
||||
"serial_open": ("meshtastic_mcp.serial_session", "open_session"),
|
||||
"serial_read": ("meshtastic_mcp.serial_session", "read_session"),
|
||||
"serial_close": ("meshtastic_mcp.serial_session", "close_session"),
|
||||
# Device reads
|
||||
"device_info": ("meshtastic_mcp.info", "device_info"),
|
||||
"list_nodes": ("meshtastic_mcp.info", "list_nodes"),
|
||||
# Device writes
|
||||
"set_owner": ("meshtastic_mcp.admin", "set_owner"),
|
||||
"get_config": ("meshtastic_mcp.admin", "get_config"),
|
||||
"set_config": ("meshtastic_mcp.admin", "set_config"),
|
||||
"get_channel_url": ("meshtastic_mcp.admin", "get_channel_url"),
|
||||
"set_channel_url": ("meshtastic_mcp.admin", "set_channel_url"),
|
||||
"send_text": ("meshtastic_mcp.admin", "send_text"),
|
||||
"reboot": ("meshtastic_mcp.admin", "reboot"),
|
||||
"shutdown": ("meshtastic_mcp.admin", "shutdown"),
|
||||
"factory_reset": ("meshtastic_mcp.admin", "factory_reset"),
|
||||
# USERPREFS
|
||||
"userprefs_manifest": ("meshtastic_mcp.userprefs", "build_manifest"),
|
||||
"userprefs_get": ("meshtastic_mcp.userprefs", "read_state"),
|
||||
"userprefs_set": ("meshtastic_mcp.userprefs", "merge_active"),
|
||||
"userprefs_reset": ("meshtastic_mcp.userprefs", "reset"),
|
||||
"userprefs_testing_profile": ("meshtastic_mcp.userprefs", "build_testing_profile"),
|
||||
# Vendor hardware tools
|
||||
"esptool_chip_info": ("meshtastic_mcp.hw_tools", "esptool_chip_info"),
|
||||
"esptool_erase_flash": ("meshtastic_mcp.hw_tools", "esptool_erase_flash"),
|
||||
"esptool_raw": ("meshtastic_mcp.hw_tools", "esptool_raw"),
|
||||
"nrfutil_dfu": ("meshtastic_mcp.hw_tools", "nrfutil_dfu"),
|
||||
"nrfutil_raw": ("meshtastic_mcp.hw_tools", "nrfutil_raw"),
|
||||
"picotool_info": ("meshtastic_mcp.hw_tools", "picotool_info"),
|
||||
"picotool_load": ("meshtastic_mcp.hw_tools", "picotool_load"),
|
||||
"picotool_raw": ("meshtastic_mcp.hw_tools", "picotool_raw"),
|
||||
}
|
||||
|
||||
|
||||
def install() -> None:
|
||||
"""Wrap every mapped tool function with the counting shim. Idempotent."""
|
||||
global _installed
|
||||
if _installed:
|
||||
return
|
||||
import importlib
|
||||
|
||||
for tool_name, (module_path, attr) in _TOOL_MAP.items():
|
||||
try:
|
||||
mod = importlib.import_module(module_path)
|
||||
except ImportError:
|
||||
continue
|
||||
_wrap(mod, attr, tool_name)
|
||||
_counts.setdefault(tool_name, 0)
|
||||
_installed = True
|
||||
|
||||
|
||||
def write_report(path: pathlib.Path) -> None:
|
||||
"""Emit `tool_coverage.json` with call counts for every mapped tool."""
|
||||
exercised = sum(1 for c in _counts.values() if c > 0)
|
||||
total = len(_counts)
|
||||
payload = {
|
||||
"total_tools": total,
|
||||
"exercised": exercised,
|
||||
"coverage_pct": round(100.0 * exercised / total, 1) if total else 0.0,
|
||||
"counts": dict(sorted(_counts.items())),
|
||||
"unexercised": sorted(k for k, v in _counts.items() if v == 0),
|
||||
}
|
||||
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
||||
|
||||
|
||||
def snapshot() -> dict[str, int]:
|
||||
return dict(_counts)
|
||||
@@ -0,0 +1,72 @@
|
||||
"""`boards.py` filter and enumeration correctness.
|
||||
|
||||
Runs against the real `pio project config` output of this firmware repo —
|
||||
validates that filter predicates match expected envs and don't drift if
|
||||
variants get reorganized.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import boards
|
||||
|
||||
|
||||
def test_list_boards_returns_many() -> None:
|
||||
all_boards = boards.list_boards()
|
||||
assert len(all_boards) >= 50, "expected at least 50 PlatformIO envs"
|
||||
|
||||
|
||||
def test_tbeam_is_canonical_esp32() -> None:
|
||||
"""The default env in platformio.ini is `tbeam`; it must always be present
|
||||
and flagged as esp32."""
|
||||
rec = boards.get_board("tbeam")
|
||||
assert rec["architecture"] == "esp32"
|
||||
assert rec["hw_model_slug"] == "TBEAM"
|
||||
assert rec["actively_supported"] is True
|
||||
assert rec["board"] == "ttgo-tbeam"
|
||||
|
||||
|
||||
def test_filter_by_architecture() -> None:
|
||||
esp32s3 = boards.list_boards(architecture="esp32s3")
|
||||
assert len(esp32s3) >= 1
|
||||
assert all(b["architecture"] == "esp32s3" for b in esp32s3)
|
||||
|
||||
|
||||
def test_filter_by_actively_supported() -> None:
|
||||
supported = boards.list_boards(actively_supported_only=True)
|
||||
unsupported = [b for b in boards.list_boards() if not b["actively_supported"]]
|
||||
assert supported, "at least one board should be actively supported"
|
||||
assert all(b["actively_supported"] for b in supported)
|
||||
# Quick sanity: the set difference is non-empty in this repo (there are
|
||||
# boards marked actively_supported=false).
|
||||
assert unsupported, "expected at least one actively_supported=false board"
|
||||
|
||||
|
||||
def test_filter_by_query_substring_matches_display_name() -> None:
|
||||
heltec = boards.list_boards(query="heltec")
|
||||
assert heltec, "expected at least one Heltec env"
|
||||
# Case-insensitive across display_name, env name, or hw_model_slug
|
||||
for b in heltec:
|
||||
blob = " ".join(
|
||||
filter(
|
||||
None,
|
||||
[
|
||||
b.get("display_name") or "",
|
||||
b["env"],
|
||||
b.get("hw_model_slug") or "",
|
||||
],
|
||||
)
|
||||
).lower()
|
||||
assert "heltec" in blob
|
||||
|
||||
|
||||
def test_get_board_unknown_env_raises() -> None:
|
||||
with pytest.raises(KeyError, match="Unknown env"):
|
||||
boards.get_board("definitely-not-a-real-env")
|
||||
|
||||
|
||||
def test_get_board_surfaces_raw_config() -> None:
|
||||
rec = boards.get_board("tbeam")
|
||||
assert "raw_config" in rec
|
||||
assert "custom_meshtastic_architecture" in rec["raw_config"]
|
||||
assert rec["raw_config"]["custom_meshtastic_architecture"] == "esp32"
|
||||
@@ -0,0 +1,61 @@
|
||||
"""`pio.py` subprocess wrapper: error paths, tailing, JSON parsing.
|
||||
|
||||
Uses a real `pio` install only for the happy-path `--version`; error paths are
|
||||
exercised with a deliberately-broken `MESHTASTIC_PIO_BIN` override.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import pio
|
||||
|
||||
|
||||
def test_tail_lines_keeps_last_n() -> None:
|
||||
text = "\n".join(f"line-{i}" for i in range(1, 11))
|
||||
assert pio.tail_lines(text, 3) == "line-8\nline-9\nline-10"
|
||||
assert pio.tail_lines(text, 100) == text # more lines requested than exist
|
||||
assert pio.tail_lines("", 5) == ""
|
||||
|
||||
|
||||
def test_tail_lines_handles_trailing_newline() -> None:
|
||||
assert pio.tail_lines("a\nb\nc\n", 2) == "b\nc"
|
||||
|
||||
|
||||
def test_pio_version_runs(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Happy path: `pio --version` exits 0 and prints a version string.
|
||||
|
||||
This exercises subprocess spawn, timeout default, and the PioResult shape.
|
||||
Skipped if pio isn't installed (CI would need pio preinstalled).
|
||||
"""
|
||||
try:
|
||||
result = pio.run(["--version"], timeout=30)
|
||||
except pio.PioError:
|
||||
pytest.skip("pio not available in this environment")
|
||||
assert result.returncode == 0
|
||||
assert "PlatformIO" in result.stdout or "platformio" in result.stdout.lower()
|
||||
assert result.duration_s > 0
|
||||
|
||||
|
||||
def test_pio_bad_command_raises_pio_error() -> None:
|
||||
"""`pio` returning non-zero must surface as PioError with stderr captured."""
|
||||
with pytest.raises(pio.PioError) as excinfo:
|
||||
pio.run(["this-subcommand-does-not-exist"], timeout=10)
|
||||
# PioError includes returncode + a tail of stderr/stdout.
|
||||
assert excinfo.value.returncode != 0
|
||||
|
||||
|
||||
def test_pio_timeout_raises_pio_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Extremely short timeout on a command that takes longer must raise PioTimeout."""
|
||||
# `pio` startup alone typically takes ~200-500ms; a 1ms timeout reliably trips.
|
||||
with pytest.raises(pio.PioTimeout):
|
||||
pio.run(["--help"], timeout=0.001)
|
||||
|
||||
|
||||
def test_run_json_parses_device_list() -> None:
|
||||
"""`pio device list --json-output` is a known-valid JSON producer."""
|
||||
try:
|
||||
data = pio.run_json(["device", "list"], timeout=15)
|
||||
except pio.PioError:
|
||||
pytest.skip("pio not available in this environment")
|
||||
# Always a list; may be empty if nothing is plugged in.
|
||||
assert isinstance(data, list)
|
||||
@@ -0,0 +1,120 @@
|
||||
"""`userprefs.build_testing_profile` / `generate_psk` correctness.
|
||||
|
||||
The testing-profile generator is the critical primitive for automated test
|
||||
labs: it must produce deterministic PSKs for a given seed (so every device
|
||||
baked in a CI run joins the same mesh) and different PSKs for different seeds
|
||||
(so concurrent labs don't collide).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import userprefs
|
||||
|
||||
|
||||
def test_generate_psk_is_32_bytes_formatted() -> None:
|
||||
psk = userprefs.generate_psk(seed="deterministic")
|
||||
# Format: "{ 0x.., 0x.., ... }" with 32 comma-separated hex bytes.
|
||||
assert psk.startswith("{ ") and psk.endswith(" }")
|
||||
bytes_part = psk.removeprefix("{ ").removesuffix(" }")
|
||||
hex_bytes = [b.strip() for b in bytes_part.split(",")]
|
||||
assert len(hex_bytes) == 32
|
||||
for b in hex_bytes:
|
||||
assert b.startswith("0x")
|
||||
int(b, 16) # raises if not valid hex
|
||||
|
||||
|
||||
def test_generate_psk_deterministic_under_same_seed() -> None:
|
||||
a = userprefs.generate_psk(seed="pytest-session-123")
|
||||
b = userprefs.generate_psk(seed="pytest-session-123")
|
||||
assert a == b, "same seed must produce same PSK"
|
||||
|
||||
|
||||
def test_generate_psk_varies_with_seed() -> None:
|
||||
seeds = ["a", "b", "pytest-1", "pytest-2", "prod-fleet-alpha"]
|
||||
psks = {userprefs.generate_psk(seed=s) for s in seeds}
|
||||
assert len(psks) == len(seeds), "seed → PSK map must be injective"
|
||||
|
||||
|
||||
def test_generate_psk_random_when_seedless() -> None:
|
||||
a = userprefs.generate_psk(seed=None)
|
||||
b = userprefs.generate_psk(seed=None)
|
||||
# Not strictly guaranteed (birthday paradox), but 256-bit randomness makes
|
||||
# a collision astronomically unlikely.
|
||||
assert a != b
|
||||
|
||||
|
||||
def test_testing_profile_contains_expected_keys() -> None:
|
||||
profile = userprefs.build_testing_profile(psk_seed="ci-run-1")
|
||||
|
||||
required = {
|
||||
"USERPREFS_CONFIG_LORA_REGION",
|
||||
"USERPREFS_LORACONFIG_MODEM_PRESET",
|
||||
"USERPREFS_LORACONFIG_CHANNEL_NUM",
|
||||
"USERPREFS_CHANNELS_TO_WRITE",
|
||||
"USERPREFS_CHANNEL_0_NAME",
|
||||
"USERPREFS_CHANNEL_0_PSK",
|
||||
"USERPREFS_CHANNEL_0_PRECISION",
|
||||
"USERPREFS_CONFIG_LORA_IGNORE_MQTT",
|
||||
"USERPREFS_MQTT_ENABLED",
|
||||
"USERPREFS_CHANNEL_0_UPLINK_ENABLED",
|
||||
"USERPREFS_CHANNEL_0_DOWNLINK_ENABLED",
|
||||
}
|
||||
assert required <= set(profile.keys())
|
||||
|
||||
# Defaults from the plan
|
||||
assert profile["USERPREFS_CONFIG_LORA_REGION"].endswith("_US")
|
||||
assert profile["USERPREFS_LORACONFIG_MODEM_PRESET"].endswith("_LONG_FAST")
|
||||
assert profile["USERPREFS_LORACONFIG_CHANNEL_NUM"] == 88
|
||||
assert profile["USERPREFS_CHANNEL_0_NAME"] == "McpTest"
|
||||
|
||||
|
||||
def test_testing_profile_rejects_unknown_region() -> None:
|
||||
with pytest.raises(ValueError, match="Unknown region"):
|
||||
userprefs.build_testing_profile(region="ATLANTIS")
|
||||
|
||||
|
||||
def test_testing_profile_rejects_unknown_modem_preset() -> None:
|
||||
with pytest.raises(ValueError, match="Unknown modem_preset"):
|
||||
userprefs.build_testing_profile(modem_preset="WARP_9")
|
||||
|
||||
|
||||
def test_testing_profile_rejects_oversized_channel_name() -> None:
|
||||
with pytest.raises(ValueError, match="11-char max"):
|
||||
userprefs.build_testing_profile(channel_name="WayTooLongChannelName")
|
||||
|
||||
|
||||
def test_testing_profile_rejects_oversized_short_name() -> None:
|
||||
with pytest.raises(ValueError, match="≤4 chars"):
|
||||
userprefs.build_testing_profile(short_name="TOOLONG")
|
||||
|
||||
|
||||
def test_disable_mqtt_false_drops_mqtt_keys() -> None:
|
||||
profile = userprefs.build_testing_profile(psk_seed="x", disable_mqtt=False)
|
||||
# When disable_mqtt is False, the MQTT-gating keys should NOT be in the
|
||||
# profile (device uses firmware defaults, whatever those are).
|
||||
assert "USERPREFS_MQTT_ENABLED" not in profile
|
||||
assert "USERPREFS_CHANNEL_0_UPLINK_ENABLED" not in profile
|
||||
|
||||
|
||||
def test_disable_position_adds_gps_disabled() -> None:
|
||||
profile = userprefs.build_testing_profile(psk_seed="x", disable_position=True)
|
||||
assert profile["USERPREFS_CONFIG_GPS_MODE"].endswith("_DISABLED")
|
||||
assert profile["USERPREFS_CONFIG_SMART_POSITION_ENABLED"] is False
|
||||
|
||||
|
||||
def test_owner_names_included_when_provided() -> None:
|
||||
profile = userprefs.build_testing_profile(
|
||||
psk_seed="x", long_name="Lab Bench 1", short_name="LB1"
|
||||
)
|
||||
assert profile["USERPREFS_CONFIG_OWNER_LONG_NAME"] == "Lab Bench 1"
|
||||
assert profile["USERPREFS_CONFIG_OWNER_SHORT_NAME"] == "LB1"
|
||||
|
||||
|
||||
def test_psk_seed_isolation_across_ci_runs() -> None:
|
||||
"""The core claim: two test labs running concurrently with different
|
||||
session seeds produce different PSKs — their meshes cannot decode each
|
||||
other's traffic."""
|
||||
lab_a = userprefs.build_testing_profile(psk_seed="lab-A-nightly")
|
||||
lab_b = userprefs.build_testing_profile(psk_seed="lab-B-nightly")
|
||||
assert lab_a["USERPREFS_CHANNEL_0_PSK"] != lab_b["USERPREFS_CHANNEL_0_PSK"]
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Unit tests for `userprefs.py`: jsonc parse, type inference, round-trip
|
||||
write, and the `temporary_overrides` context manager's byte-for-byte restore.
|
||||
|
||||
None of these require hardware. They validate the contract that the flash/
|
||||
testing-profile tools rely on — if these fail, the provisioning tier will
|
||||
produce confusing mismatches.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import userprefs
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_jsonc(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Write a minimal userPrefs.jsonc into tmp_path and point config at it."""
|
||||
content = """{
|
||||
"USERPREFS_CONFIG_LORA_REGION": "meshtastic_Config_LoRaConfig_RegionCode_US",
|
||||
"USERPREFS_LORACONFIG_CHANNEL_NUM": "88",
|
||||
// "USERPREFS_CHANNEL_0_NAME": "McpTest",
|
||||
"USERPREFS_CHANNEL_0_PSK": "{ 0x01, 0x02, 0x03 }",
|
||||
// "USERPREFS_MQTT_ENABLED": "0",
|
||||
"USERPREFS_CONFIG_LORA_IGNORE_MQTT": "true"
|
||||
}
|
||||
"""
|
||||
# Fake firmware root with a userPrefs.jsonc + platformio.ini (needed for
|
||||
# `config.firmware_root()`'s walk-up detection).
|
||||
(tmp_path / "platformio.ini").write_text("[platformio]\n", encoding="utf-8")
|
||||
jsonc = tmp_path / "userPrefs.jsonc"
|
||||
jsonc.write_text(content, encoding="utf-8")
|
||||
monkeypatch.setenv("MESHTASTIC_FIRMWARE_ROOT", str(tmp_path))
|
||||
return jsonc
|
||||
|
||||
|
||||
def test_read_state_separates_active_and_commented(sample_jsonc: Path) -> None:
|
||||
state = userprefs.read_state()
|
||||
assert set(state["active"]) == {
|
||||
"USERPREFS_CONFIG_LORA_REGION",
|
||||
"USERPREFS_LORACONFIG_CHANNEL_NUM",
|
||||
"USERPREFS_CHANNEL_0_PSK",
|
||||
"USERPREFS_CONFIG_LORA_IGNORE_MQTT",
|
||||
}
|
||||
assert set(state["commented"]) == {
|
||||
"USERPREFS_CHANNEL_0_NAME",
|
||||
"USERPREFS_MQTT_ENABLED",
|
||||
}
|
||||
|
||||
|
||||
def test_infer_type_matches_platformio_custom_py() -> None:
|
||||
# Mirrors the branch order in `bin/platformio-custom.py:222-235`.
|
||||
assert userprefs.infer_type("{ 0x01, 0x02 }") == "brace"
|
||||
assert userprefs.infer_type("88") == "number"
|
||||
assert userprefs.infer_type("-1.5") == "number"
|
||||
assert userprefs.infer_type("true") == "bool"
|
||||
assert userprefs.infer_type("false") == "bool"
|
||||
assert userprefs.infer_type("meshtastic_Config_DeviceConfig_Role_ROUTER") == "enum"
|
||||
assert userprefs.infer_type("plain string value") == "string"
|
||||
assert userprefs.infer_type(None) == "unknown"
|
||||
|
||||
|
||||
def test_temporary_overrides_restores_byte_for_byte(sample_jsonc: Path) -> None:
|
||||
"""The context manager MUST leave the file bit-identical on exit, even on
|
||||
exception — this is the safety guarantee build/flash tools rely on."""
|
||||
original = sample_jsonc.read_bytes()
|
||||
|
||||
with userprefs.temporary_overrides({"USERPREFS_CHANNEL_0_NAME": "OverrideTest"}):
|
||||
# During the context, the override is written.
|
||||
during = userprefs.read_state()
|
||||
assert "USERPREFS_CHANNEL_0_NAME" in during["active"]
|
||||
assert during["active"]["USERPREFS_CHANNEL_0_NAME"] == "OverrideTest"
|
||||
|
||||
# After: byte-identical restore.
|
||||
assert sample_jsonc.read_bytes() == original
|
||||
|
||||
|
||||
def test_temporary_overrides_restores_after_exception(sample_jsonc: Path) -> None:
|
||||
original = sample_jsonc.read_bytes()
|
||||
|
||||
with pytest.raises(RuntimeError, match="simulated"):
|
||||
with userprefs.temporary_overrides({"USERPREFS_CHANNEL_0_NAME": "Failing"}):
|
||||
raise RuntimeError("simulated mid-build failure")
|
||||
|
||||
assert sample_jsonc.read_bytes() == original
|
||||
|
||||
|
||||
def test_temporary_overrides_none_is_noop(sample_jsonc: Path) -> None:
|
||||
original = sample_jsonc.read_bytes()
|
||||
with userprefs.temporary_overrides(None) as effective:
|
||||
# No file write, and `effective` still reflects the active set.
|
||||
assert "USERPREFS_CONFIG_LORA_REGION" in effective
|
||||
assert sample_jsonc.read_bytes() == original
|
||||
|
||||
|
||||
def test_temporary_overrides_rejects_non_userprefs_keys(sample_jsonc: Path) -> None:
|
||||
with pytest.raises(ValueError, match="USERPREFS_"):
|
||||
with userprefs.temporary_overrides({"RANDOM_KEY": "value"}):
|
||||
pass
|
||||
|
||||
|
||||
def test_build_manifest_surfaces_all_keys(sample_jsonc: Path) -> None:
|
||||
"""Manifest should union the jsonc set with firmware-src consumers.
|
||||
|
||||
In the sample tmpdir there's no `src/` so `consumed_by` is empty for all
|
||||
entries; that's fine — the manifest still lists every jsonc key.
|
||||
"""
|
||||
manifest = userprefs.build_manifest()
|
||||
keys = {e["key"] for e in manifest["entries"]}
|
||||
# All 6 keys from sample_jsonc should be present.
|
||||
assert "USERPREFS_CONFIG_LORA_REGION" in keys
|
||||
assert "USERPREFS_CHANNEL_0_NAME" in keys # commented but still listed
|
||||
assert manifest["active_count"] == 4
|
||||
assert manifest["commented_count"] == 2
|
||||
Reference in New Issue
Block a user