Add MCP server for interacting with meshtastic devices and testing framework / TUI (#10194)
* Start of MCP server and test suite * Add MCP server for interacting with meshtastic devices and testing framework / TUI * Update mcp-server/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * fix mcp-server review feedback from thread Agent-Logs-Url: https://github.com/meshtastic/firmware/sessions/91dc128a-ed50-4d07-8bb2-3dc6623a05f7 Co-authored-by: thebentern <9000580+thebentern@users.noreply.github.com> * Enhance StreamAPI and PhoneAPI for improved log record handling and concurrency control * Semgrep fixes * Trunk and semgrep fixes * optimize pio streaming tee file writes Agent-Logs-Url: https://github.com/meshtastic/firmware/sessions/04e26c6b-6a2b-45be-bbeb-79ae4d0be633 Co-authored-by: thebentern <9000580+thebentern@users.noreply.github.com> * chore: remove redundant log handle assignment Agent-Logs-Url: https://github.com/meshtastic/firmware/sessions/04e26c6b-6a2b-45be-bbeb-79ae4d0be633 Co-authored-by: thebentern <9000580+thebentern@users.noreply.github.com> * Consolidate type imports and remove placeholder test files * Add tests for config persistence and more exchange messages * Refactor position test to validate on-demand request/reply behavior * Remove position request/reply test and update README for telemetry behavior * Fix transmit history file to get removed on factory reset --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
This commit is contained in:
Ben Meadors
co-authored by
GitHub
Copilot
copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
GitHub
Copilot
copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
parent
aab4cd086f
commit
c8dac10348
@@ -0,0 +1,83 @@
|
||||
"""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.skip(
|
||||
reason="test uses flash.erase_and_flash which shells to bin/device-install.sh "
|
||||
"which needs mt-esp32s3-ota.bin (not in repo). TODO: switch to "
|
||||
"esptool_erase_flash + flash.flash() like test_00_bake."
|
||||
)
|
||||
@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,60 @@
|
||||
"""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}"
|
||||
|
||||
# `modem_preset` is omitted from the protobuf→JSON dump when the
|
||||
# device is using the default enum value (LONG_FAST). If the key is
|
||||
# missing AND we expected LONG_FAST, that's a match. Otherwise compare.
|
||||
live_preset = lora.get("modem_preset")
|
||||
if live_preset is None:
|
||||
assert expected_preset_str == "LONG_FAST", (
|
||||
f"{role}: modem_preset omitted (means default LONG_FAST), "
|
||||
f"but expected {expected_preset_str!r}"
|
||||
)
|
||||
else:
|
||||
assert live_preset in (
|
||||
expected_preset_str,
|
||||
expected_preset_str.upper(),
|
||||
), f"{role}: modem_preset={live_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,108 @@
|
||||
"""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.skip(
|
||||
reason="test uses flash.erase_and_flash which shells to bin/device-install.sh "
|
||||
"which needs mt-esp32s3-ota.bin (not in repo). TODO: switch to "
|
||||
"esptool_erase_flash + flash.flash() like test_00_bake."
|
||||
)
|
||||
@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,90 @@
|
||||
"""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
|
||||
|
||||
from .._port_discovery import resolve_port_by_role
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_baked_prefs_survive_factory_reset(
|
||||
baked_single: dict[str, Any],
|
||||
test_profile: dict[str, Any],
|
||||
wait_until,
|
||||
) -> None:
|
||||
"""Runs once per connected role. Flow:
|
||||
1. Change owner name to a known-non-default value.
|
||||
2. Trigger factory_reset(full=False).
|
||||
3. Rediscover the port (macOS re-enumerates the CDC endpoint on nRF52
|
||||
factory_reset; the path can change e.g. `/dev/cu.usbmodem101` →
|
||||
`/dev/cu.usbmodem1101`).
|
||||
4. Wait for device to come back.
|
||||
5. Confirm owner is back to USERPREFS-baked default (or blank default if
|
||||
not baked), and primary channel/region/slot are still the baked values.
|
||||
"""
|
||||
role = baked_single["role"]
|
||||
port = baked_single["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)
|
||||
|
||||
# Device re-enumerates — rediscover its port before probing. nRF52's
|
||||
# CDC endpoint drops and comes back with a new `/dev/cu.usbmodem*`
|
||||
# path on macOS; ESP32-S3 usually keeps the same path but the helper
|
||||
# works either way (it just returns the current path for this role).
|
||||
# Early sleep lets the USB kernel driver settle before we start
|
||||
# polling — list_devices during a transient re-enumeration can return
|
||||
# an empty list and the helper's poll-with-backoff handles that too,
|
||||
# so the sleep is optimization not correctness.
|
||||
time.sleep(10.0)
|
||||
port = resolve_port_by_role(role, timeout_s=60.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)
|
||||
# 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)
|
||||
Reference in New Issue
Block a user