mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2026-07-31 00:56:51 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #946 from ianmcorvidae/sim-based-testing

Browse Source 
Initial implementation of simradio-based testing with a multi-node simulated mesh
This commit is contained in:
Ian McEwen
2026-07-14 09:16:09 -07:00
committed by GitHub
parent 7a7353ca15 4339cda6bd
commit 82220ba49b
11 changed files with 1639 additions and 714 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
.github/workflows/ci.yml vendored
View File

@@ -17,6 +17,8 @@ jobs:
- "3.10"
- "3.11"
- "3.12"
- "3.13"
- "3.14"
steps:
- uses: actions/checkout@v4
- name: Install Python 3
@@ -56,12 +58,11 @@ jobs:
strategy:
matrix:
python-version:
- "3.9"
- "3.10"
- "3.11"
- "3.12"
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- name: Install Python 3
uses: actions/setup-python@v5
- name: Install meshtastic from local
@@ -70,3 +71,36 @@ jobs:
pip3 install poetry
poetry install
poetry run meshtastic --version
simradio_testing:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
channel:
- beta
- alpha
- daily
continue-on-error: ${{ matrix.channel == 'daily' }}
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- name: Install Python 3
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install meshtastic from local
run: |
python -m pip install --upgrade pip
pip3 install poetry
poetry install --all-extras --with dev
poetry run meshtastic --version
- name: Install meshtasticd (${{ matrix.channel }}) from PPA
run: |
sudo add-apt-repository -y ppa:meshtastic/${{ matrix.channel }}
sudo apt-get update
sudo apt-get install -y meshtasticd
- name: Run firmware smoke tests
run: poetry run pytest -m "smokevirt or smokemesh" -v
timeout-minutes: 15

36
.github/workflows/daily_simradio.yml vendored Normal file
View File

@@ -0,0 +1,36 @@
name: Daily SimRadio Tests
on:
schedule:
- cron: 0 6 * * *
workflow_dispatch:
permissions:
contents: read
jobs:
simradio_testing:
runs-on: ubuntu-latest
if: github.repository == 'meshtastic/meshtastic-python'
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- name: Install Python 3
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install meshtastic from local
run: |
python -m pip install --upgrade pip
pip3 install poetry
poetry install --all-extras --with dev
poetry run meshtastic --version
- name: Install meshtasticd (daily) from PPA
run: |
sudo add-apt-repository -y ppa:meshtastic/daily
sudo apt-get update
sudo apt-get install -y meshtasticd
- name: Run firmware smoke tests
run: poetry run pytest -m "smokevirt or smokemesh" -v
timeout-minutes: 15

41
meshtastic/__main__.py
View File

@@ -969,9 +969,15 @@ def onConnected(interface):
if args.ch_longslow:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_SLOW)
if args.ch_longmod:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_MODERATE)
if args.ch_longfast:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_FAST)
if args.ch_longturbo:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_TURBO)
if args.ch_medslow:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.MEDIUM_SLOW)
@@ -984,6 +990,9 @@ def onConnected(interface):
if args.ch_shortfast:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.SHORT_FAST)
if args.ch_shortturbo:
setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.SHORT_TURBO)
if args.ch_set or args.ch_enable or args.ch_disable:
closeNow = True
@@ -1960,43 +1969,61 @@ def addConfigArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
group.add_argument(
"--ch-vlongslow",
help="Change to the very long-range and slow modem preset",
help="Change to the VERY_LONG_SLOW modem preset. Deprecated since 2.5 firmware.",
action="store_true",
)
group.add_argument(
"--ch-longslow",
help="Change to the long-range and slow modem preset",
help="Change to the LONG_SLOW modem preset. Deprecated since 2.7 firmware.",
action="store_true",
)
group.add_argument(
"--ch-longmod", "--ch-longmoderate",
help="Change to the LONG_MODERATE modem preset",
action="store_true",
)
group.add_argument(
"--ch-longfast",
help="Change to the long-range and fast modem preset",
help="Change to the LONG_FAST modem preset",
action="store_true",
)
group.add_argument(
"--ch-longturbo",
help="Change to the LONG_TURBO preset",
action="store_true",
)
group.add_argument(
"--ch-medslow",
help="Change to the med-range and slow modem preset",
help="Change to the MEDIUM_SLOW modem preset",
action="store_true",
)
group.add_argument(
"--ch-medfast",
help="Change to the med-range and fast modem preset",
help="Change to the MEDIUM_FAST modem preset",
action="store_true",
)
group.add_argument(
"--ch-shortslow",
help="Change to the short-range and slow modem preset",
help="Change to the SHORT_SLOW modem preset",
action="store_true",
)
group.add_argument(
"--ch-shortfast",
help="Change to the short-range and fast modem preset",
help="Change to the SHORT_FAST modem preset",
action="store_true",
)
group.add_argument(
"--ch-shortturbo",
help="Change to the SHORT_TURBO modem preset",
action="store_true",
)

14
meshtastic/node.py
View File

@@ -287,12 +287,22 @@ class Node:
# for sending admin channels will also change
adminIndex = self.iface.localNode._getAdminChannelIndex()
# Snapshot serialized channel payloads from channelIndex onward so we
# can avoid writing slots whose protobuf content did not change after
# the shift. Use bytes (not message objects), because _fixupChannels()
# mutates message fields in-place.
old_channels = [
self.channels[i].SerializeToString()
for i in range(channelIndex, len(self.channels))
]
self.channels.pop(channelIndex)
self._fixupChannels() # expand back to 8 channels
index = channelIndex
while index < 8:
self.writeChannel(index, adminIndex=adminIndex)
for old_ch in old_channels:
if self.channels[index].SerializeToString() != old_ch:
self.writeChannel(index, adminIndex=adminIndex)
index += 1
# if we are updating the local node, we might end up

81
meshtastic/tests/conftest.py
View File

@@ -8,6 +8,87 @@ import pytest
from meshtastic import mt_config
from ..mesh_interface import MeshInterface
from .firmware_harness import (
CHAIN_TOPOLOGY,
DEFAULT_BASE_PORT,
SimMesh,
find_meshtasticd,
is_compatible_host,
)
from .fw_helpers import set_region
# Use a different base port for the single-node fixture so it doesn't
# conflict with the multi-node mesh fixture.
SINGLE_NODE_BASE_PORT = DEFAULT_BASE_PORT + 100
def _skip_firmware_if_unavailable() -> None:
"""Skip the test when meshtasticd can't run on this host."""
if not is_compatible_host():
pytest.skip("meshtasticd firmware tests require Linux")
if find_meshtasticd() is None:
pytest.skip(
"meshtasticd not found — set MESHTASTICD_BIN or install it on PATH"
)
@pytest.fixture(scope="function")
def firmware_node():
"""A single meshtasticd sim node for smokevirt tests.
Function-scoped so every test gets a freshly-erased node with no
state leaking from previous tests. This makes destructive commands
(``--reboot``, ``--set factory_reset true``) safe to run and lets
tests be order-independent.
Yields the SimNode instance. The node is booted with a fresh erased
config and listens on localhost at its TCP port. Region is set to US
so modem-preset tests work against firmware >= 2.8, which clamps
presets to the legal set for the current region.
"""
_skip_firmware_if_unavailable()
mesh = SimMesh(n_nodes=1, base_port=SINGLE_NODE_BASE_PORT)
mesh.start()
node = mesh.get_node(0)
set_region(node.port, "US")
# The region commit restarts the TCP listener, so reconnect the harness
# interface in case a test wants to use it directly.
if node.iface is not None:
try:
node.iface.close()
except Exception: # pylint: disable=broad-except
pass
node.connect()
yield node
mesh.stop()
@pytest.fixture(scope="function")
def firmware_mesh():
"""A 3-node chain (A-B-C) meshtasticd sim mesh for smokemesh tests.
Yields the SimMesh instance. Nodes are connected and the SIMULATOR_APP
packet bridge is running. Region is set to US for firmware >= 2.8
compatibility, interfaces are reconnected after the region change, and
node DB convergence is awaited.
"""
_skip_firmware_if_unavailable()
mesh = SimMesh(n_nodes=3, topology=CHAIN_TOPOLOGY)
mesh.start()
for node in mesh.nodes:
set_region(node.port, "US")
# The region commit restarts each node's TCP listener, so reconnect the
# harness interfaces before waiting for convergence.
for node in mesh.nodes:
if node.iface is not None:
try:
node.iface.close()
except Exception: # pylint: disable=broad-except
pass
node.connect()
mesh.wait_for_convergence(timeout=30)
yield mesh
mesh.stop()
@pytest.fixture

355
meshtastic/tests/firmware_harness.py Normal file
View File

@@ -0,0 +1,355 @@
"""Test harness for running real meshtasticd firmware instances.
Launches one or more meshtasticd processes in simulator mode (-s) and bridges
their "over-the-air" packets via the SIMULATOR_APP protocol so that multiple
instances can communicate as if via LoRa, with a configurable topology.
The harness expects meshtasticd to be available on PATH or via the
MESHTASTICD_BIN environment variable. It does not download or build the
binary itself.
"""
import logging
import os
import platform
import shutil
import signal
import socket
import subprocess
import tempfile
import time
from typing import Dict, List, Optional, Set
from pubsub import pub # type: ignore[import-untyped]
from meshtastic import BROADCAST_NUM, mesh_pb2, portnums_pb2
from meshtastic.tcp_interface import TCPInterface
logger = logging.getLogger(__name__)
HW_ID_OFFSET = 16
DEFAULT_BASE_PORT = 4404
DEFAULT_RSSI = -50
DEFAULT_SNR = 10.0
BOOT_TIMEOUT = 30
CONNECT_TIMEOUT = 30
CHAIN_TOPOLOGY: Dict[int, Set[int]] = {
0: {1},
1: {0, 2},
2: {1},
}
def find_meshtasticd() -> Optional[str]:
"""Return the path to the meshtasticd binary, or None if not found."""
env_path = os.environ.get("MESHTASTICD_BIN")
if env_path and os.path.isfile(env_path) and os.access(env_path, os.X_OK):
return env_path
return shutil.which("meshtasticd")
def is_compatible_host() -> bool:
"""True when the host can run meshtasticd natively (Linux only)."""
return platform.system() == "Linux"
def _wait_for_port(port: int, timeout: int = BOOT_TIMEOUT) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
s = socket.create_connection(("localhost", port), timeout=0.5)
s.close()
return True
except OSError:
time.sleep(0.5)
return False
class SimNode:
"""A single meshtasticd simulator instance."""
def __init__(self, node_id: int, base_port: int = DEFAULT_BASE_PORT):
self.node_id = node_id
self.hw_id = node_id + HW_ID_OFFSET
self.port = base_port + node_id
self.process: Optional[subprocess.Popen] = None
self.workdir: Optional[str] = None
self.iface: Optional[TCPInterface] = None
self._log_files: list = []
@property
def node_num(self) -> int:
"""The firmware-assigned node number (== hw_id)."""
if self.iface and self.iface.myInfo:
return self.iface.myInfo.my_node_num
return self.hw_id
def start(self, binary: str) -> None:
"""Launch the meshtasticd process in simulator mode."""
self.workdir = tempfile.mkdtemp(prefix=f"mtd_node{self.node_id}_")
vfs_dir = os.path.join(self.workdir, "vfs")
os.mkdir(vfs_dir)
# Files are closed in _kill(); keep them open for the process lifetime.
log_stdout = open( # pylint: disable=consider-using-with
os.path.join(self.workdir, "meshtasticd.log"), "wb", buffering=0
)
log_stderr = open( # pylint: disable=consider-using-with
os.path.join(self.workdir, "meshtasticd.err"), "wb", buffering=0
)
self._log_files = [log_stdout, log_stderr]
self.process = subprocess.Popen( # pylint: disable=consider-using-with
[
binary,
"-s",
"-h", str(self.hw_id),
"-p", str(self.port),
"-d", vfs_dir,
"-e",
],
stdout=log_stdout,
stderr=log_stderr,
start_new_session=True,
)
if not _wait_for_port(self.port):
self._kill()
raise RuntimeError(
f"meshtasticd node {self.node_id} did not start listening on port {self.port}"
)
def connect(self) -> None:
"""Open a TCPInterface connection to this node."""
self.iface = TCPInterface(
hostname="localhost",
portNumber=self.port,
connectNow=False,
)
self.iface.myConnect()
self.iface.connect()
def close(self) -> None:
"""Close the interface and kill the process."""
if self.iface is not None:
try:
self.iface.localNode.exitSimulator()
except Exception:
pass
try:
self.iface.close()
except Exception:
pass
self.iface = None
self._kill()
if self.workdir:
shutil.rmtree(self.workdir, ignore_errors=True)
self.workdir = None
def _kill(self) -> None:
for f in self._log_files:
try:
f.close()
except Exception:
pass
self._log_files = []
if self.process is not None:
try:
os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
except (ProcessLookupError, OSError):
pass
try:
self.process.wait(timeout=5)
except Exception:
try:
os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
except Exception:
pass
# Give OS time to release TCP port (avoid TIME_WAIT preventing
# next instance from binding the same port)
time.sleep(1.0)
self.process = None
class SimMesh:
"""Manages N meshtasticd sim instances and bridges their SIMULATOR_APP packets.
When *topology* is None every node hears every other node (full mesh).
Otherwise *topology* maps a transmitter's node index to the set of receiver
node indices that can hear it.
"""
def __init__(
self,
n_nodes: int = 1,
topology: Optional[Dict[int, Set[int]]] = None,
base_port: int = DEFAULT_BASE_PORT,
):
self.n_nodes = n_nodes
self.topology = topology
self.base_port = base_port
self.nodes: List[SimNode] = [
SimNode(i, base_port) for i in range(n_nodes)
]
self._port_to_idx: Dict[int, int] = {}
self._started = False
def start(self) -> None:
"""Launch all nodes, connect, and start the packet bridge."""
binary = find_meshtasticd()
if binary is None:
raise RuntimeError(
"meshtasticd not found. Set MESHTASTICD_BIN or install it on PATH."
)
for node in self.nodes:
node.start(binary)
for node in self.nodes:
node.connect()
self._port_to_idx[node.port] = node.node_id
pub.subscribe(self._on_sim_packet, "meshtastic.receive.simulator")
self._started = True
if self.n_nodes > 1:
self._trigger_convergence()
def _trigger_convergence(self) -> None:
"""Actively trigger NodeInfo exchange instead of waiting passively.
Sends a NODEINFO_APP packet with wantResponse from each node so the
firmware's NodeInfoModule responds with its own user info, populating
all node DBs deterministically.
"""
for node in self.nodes:
iface = node.iface
if iface is None:
continue
user = mesh_pb2.User()
user.id = f"!{node.node_num:08x}"
user.long_name = f"Node {node.node_id}"
user.short_name = f"{node.node_id:04d}"[:4]
user.hw_model = mesh_pb2.HardwareModel.PORTDUINO
try:
iface.sendData(
user,
destinationId=BROADCAST_NUM,
portNum=portnums_pb2.PortNum.NODEINFO_APP,
wantAck=False,
wantResponse=True,
)
except Exception as ex:
logger.debug("NodeInfo trigger for node %d failed: %s", node.node_id, ex)
time.sleep(5)
def stop(self) -> None:
"""Shut down all nodes and clean up."""
if not self._started:
return
try:
pub.unsubscribe(self._on_sim_packet, "meshtastic.receive.simulator")
except Exception:
pass
for node in self.nodes:
node.close()
self._started = False
def get_node(self, idx: int) -> SimNode:
"""Return the SimNode at the given index."""
return self.nodes[idx]
def get_iface(self, idx: int) -> TCPInterface:
"""Return the TCPInterface for the node at the given index."""
iface = self.nodes[idx].iface
assert iface is not None, f"node {idx} has no interface"
return iface
def wait_for_convergence(self, timeout: int = 30) -> bool:
"""Wait until every node sees all others in its node DB.
Returns True if converged, False if timed out (non-fatal — the packet
bridge forwards regardless of node DB state).
"""
if self.n_nodes <= 1:
return True
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if all(
node.iface is not None
and node.iface.nodes is not None
and len(node.iface.nodes) >= self.n_nodes
for node in self.nodes
):
return True
time.sleep(2)
logger.warning("Mesh did not fully converge within %ds", timeout)
return False
def _get_receivers(self, tx_idx: int) -> List[int]:
"""Return node indices that can hear a transmission from *tx_idx*."""
if self.topology is not None:
return sorted(self.topology.get(tx_idx, set()))
return [i for i in range(self.n_nodes) if i != tx_idx]
def _on_sim_packet(self, interface, packet) -> None:
"""Bridge callback: forward a SIMULATOR_APP packet to receiving nodes."""
tx_port = getattr(interface, "portNumber", None)
tx_idx = self._port_to_idx.get(tx_port) if tx_port else None
if tx_idx is None:
return
rx_indices = self._get_receivers(tx_idx)
if not rx_indices:
return
data = packet["decoded"]["payload"]
if hasattr(data, "SerializeToString"):
data = data.SerializeToString()
if len(data) > mesh_pb2.Constants.DATA_PAYLOAD_LEN:
logger.warning("Simulator payload too big (%d bytes), dropping", len(data))
return
mesh_packet = _build_mesh_packet(packet, data)
for rx_idx in rx_indices:
rx_iface = self.nodes[rx_idx].iface
if rx_iface is None:
continue
mesh_packet.rx_rssi = DEFAULT_RSSI
mesh_packet.rx_snr = DEFAULT_SNR
to_radio = mesh_pb2.ToRadio()
to_radio.packet.CopyFrom(mesh_packet)
try:
rx_iface._sendToRadio(to_radio)
except Exception as ex:
logger.error("Error forwarding packet to node %d: %s", rx_idx, ex)
def __enter__(self):
self.start()
return self
def __exit__(self, *exc):
self.stop()
def _build_mesh_packet(packet: dict, data: bytes) -> mesh_pb2.MeshPacket:
"""Reconstruct a MeshPacket for SIMULATOR_APP injection."""
mp = mesh_pb2.MeshPacket()
mp.decoded.payload = data
mp.decoded.portnum = portnums_pb2.PortNum.SIMULATOR_APP
mp.to = packet.get("to", BROADCAST_NUM)
setattr(mp, "from", packet.get("from", 0))
mp.id = packet.get("id", 0)
mp.want_ack = packet.get("wantAck", False)
mp.hop_limit = packet.get("hopLimit", 0)
mp.hop_start = packet.get("hopStart", 0)
mp.via_mqtt = packet.get("viaMQTT", False)
mp.relay_node = packet.get("relayNode", 0)
mp.next_hop = packet.get("nextHop", 0)
mp.channel = int(packet.get("channel", 0))
decoded = packet.get("decoded", {})
if "requestId" in decoded:
mp.decoded.request_id = decoded["requestId"]
if "wantResponse" in decoded:
mp.decoded.want_response = decoded["wantResponse"]
return mp

374
meshtastic/tests/fw_helpers.py Normal file
View File

@@ -0,0 +1,374 @@
"""Shared helpers for meshtasticd-backed smoke tests.
Both ``test_smokevirt`` (single node) and ``test_smokemesh`` (multi-node
chain) use these helpers to drive the ``meshtastic`` CLI against real
``meshtasticd`` simulator instances and then verify the resulting firmware
state through the Python library's ``TCPInterface``.
Verifying through the library (rather than regex on CLI stdout) is the
core design choice: it makes tests robust against CLI wording changes
while still exercising both the CLI argparse path and the firmware I/O
path of every feature.
"""
from __future__ import annotations
import logging
import shlex
import socket
import subprocess
import sys
import time
from typing import Callable, List, Optional, Tuple
from pubsub import pub # type: ignore[import-untyped]
from meshtastic.tcp_interface import TCPInterface
logger = logging.getLogger(__name__)
# Pause between a CLI command finishing and a verification interface
# opening, to let the firmware flush its TCP bookkeeping. Keeps the
# simulator happy when many short-lived connections are happening.
PAUSE_AFTER_CLI = 0.2
# Pause after changing the LoRa region. The firmware may restart the TCP
# listener while committing the new radio config, so give it time to settle
# before the next CLI call or harness reconnect.
PAUSE_AFTER_REGION_CHANGE = 2.0
# ---------------------------------------------------------------------------
# CLI invocation
# ---------------------------------------------------------------------------
def resolve_cli() -> str:
"""Return a shell-invokable ``meshtastic`` command.
Prefers the in-tree module run through the current interpreter so
tests exercise the source we are editing, regardless of any
separately-installed ``meshtastic`` entry point on PATH.
"""
# The PATH binary may live outside the nono sandbox's allowed paths;
# ``python -m meshtastic`` is more portable and always available.
return f"{sys.executable} -m meshtastic"
def run_cli(
port: int,
*args: str,
timeout: int = 60,
retries: int = 2,
retry_delay: float = 1.0,
) -> Tuple[int, str]:
"""Run the ``meshtastic`` CLI against the sim node on *port*.
Returns ``(return_code, merged_stdout_stderr)``. ``--host
localhost:PORT`` is prefixed automatically. stderr is merged into
stdout so callers can match warning text such as "Warning: Need to
specify ..." regardless of which stream it lands on.
If the CLI fails to connect on the first attempt (which happens
transiently when a freshly-booted sim node needs a moment to settle
after the harness interface connects), retry up to *retries*
additional times after *retry_delay* seconds.
"""
cli = resolve_cli()
argv = [*_shlex_split(cli)]
argv.extend(["--host", f"localhost:{port}"])
argv.extend(args)
logger.debug("run_cli: %s", argv)
last_out = ""
for attempt in range(retries + 1):
try:
proc = subprocess.run(
argv,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
timeout=timeout,
check=False,
)
except subprocess.TimeoutExpired as ex:
last_out = ex.stdout.decode("utf-8", errors="replace") if ex.stdout else ""
if attempt < retries:
time.sleep(retry_delay)
continue
return 124, last_out
out = proc.stdout.decode("utf-8", errors="replace")
rc = proc.returncode
# Retry on transient connection-refused / timed-out errors that
# are common right after a sim node spins up.
transient = (
"Error connecting" in out
or "Timed out waiting for connection" in out
or "Connection reset by peer" in out
)
if rc != 0 and transient and attempt < retries:
logger.debug("run_cli: transient failure, retry %d/%d", attempt + 1, retries)
time.sleep(retry_delay)
last_out = out
continue
return rc, out
return rc, last_out
def _shlex_split(cmd: str) -> List[str]:
"""Split a shell string into argv, honoring quotes."""
return shlex.split(cmd)
# ---------------------------------------------------------------------------
# Fresh-connection state verification
# ---------------------------------------------------------------------------
def _wait_for_port(port: int, timeout: float = 30.0) -> None:
"""Wait until *port* accepts a TCP connection (firmware sim comes up
or comes back after a config-commit reboot)."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
s = socket.create_connection(("localhost", port), timeout=0.5)
s.close()
return
except OSError:
time.sleep(0.2)
raise TimeoutError(
f"port {port} did not accept connections within {timeout}s"
)
def connect_iface(
port: int,
no_nodes: bool = False,
retries: int = 4,
wait_timeout: float = 30.0,
) -> TCPInterface:
"""Open a fresh ``TCPInterface`` to *port* and block on the config exchange.
Firmware config writes (``writeChannel``, ``--seturl``, ``--factory_reset``)
can briefly restart the sim's TCP listener. We wait for the port to
come up first, then retry the connect+config-exchange a few times.
"""
last_exc: Optional[Exception] = None
for attempt in range(retries + 1):
try:
_wait_for_port(port, timeout=wait_timeout)
return TCPInterface(
hostname="localhost",
portNumber=port,
connectNow=True,
noNodes=no_nodes,
)
except Exception as ex: # pylint: disable=broad-except
last_exc = ex
if attempt < retries:
logger.debug(
"connect_iface attempt %d/%d failed: %s",
attempt + 1, retries, ex,
)
time.sleep(0.5)
continue
raise
assert last_exc is not None # type guard; unreachable
raise last_exc # pragma: no cover
def verify_state(
port: int,
verifier: Callable[[TCPInterface], None],
*,
no_nodes: bool = False,
) -> None:
"""Open a fresh interface and run *verifier(iface)*, then close.
Used after a CLI mutation to verify firmware state through the
library. Always closes the interface so the next test starts clean.
"""
iface = connect_iface(port, no_nodes=no_nodes)
try:
verifier(iface)
finally:
try:
iface.close()
except Exception: # pylint: disable=broad-except
pass
time.sleep(PAUSE_AFTER_CLI)
def cli_then_verify(
port: int,
cli_args: List[str],
verifier: Optional[Callable[[TCPInterface], None]],
*,
expect_rc: Optional[int] = 0,
no_nodes: bool = False,
cli_timeout: int = 60,
) -> str:
"""Run *cli_args* against *port*, optionally asserting *expect_rc*,
then (if *verifier* is not None) open a fresh interface and run
*verifier(iface)* against the just-mutated firmware state.
Returns the CLI stdout.
"""
rc, out = run_cli(port, *cli_args, timeout=cli_timeout)
if expect_rc is not None:
assert rc == expect_rc, f"CLI rc={rc} (expected {expect_rc}): {out}"
time.sleep(PAUSE_AFTER_CLI)
if verifier is not None:
verify_state(port, verifier, no_nodes=no_nodes)
return out
# ---------------------------------------------------------------------------
# Packet collectors (used by smokemesh receive-verification tests)
# ---------------------------------------------------------------------------
RECEIVE_TIMEOUT = 15.0
class PacketCollector:
"""Collect packets received on a specific interface via pubsub.
``Listener`` (pubsub 4.x) wraps handlers with a weak reference, so
we keep a strong reference to ``handler`` on the instance to prevent
garbage collection before the publishing thread gets to call it.
"""
def __init__(self):
self.packets: List[dict] = []
self._handler: Optional[Callable] = None
def on_receive(self, packet, interface): # pylint: disable=unused-argument
"""Append a received packet to the internal list."""
self.packets.append(packet)
def wait_for(self, count: int, timeout: float = RECEIVE_TIMEOUT) -> bool:
"""Wait until *count* packets have been collected or *timeout* expires."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if len(self.packets) >= count:
return True
time.sleep(0.2)
return len(self.packets) >= count
@property
def texts(self) -> List[str]:
"""Return text payloads from TEXT_MESSAGE_APP packets."""
return [
p.get("decoded", {}).get("text", "")
for p in self.packets
if p.get("decoded", {}).get("portnum") == "TEXT_MESSAGE_APP"
]
@property
def traceroutes(self) -> List[dict]:
"""Return TRACEROUTE_APP packets."""
return [
p for p in self.packets
if p.get("decoded", {}).get("portnum") == "TRACEROUTE_APP"
]
@property
def telemetries(self) -> List[dict]:
"""Return TELEMETRY_APP packets."""
return [
p for p in self.packets
if p.get("decoded", {}).get("portnum") == "TELEMETRY_APP"
]
@property
def positions(self) -> List[dict]:
"""Return POSITION_APP packets."""
return [
p for p in self.packets
if p.get("decoded", {}).get("portnum") == "POSITION_APP"
]
def reset(self) -> None:
"""Clear all collected packets."""
self.packets.clear()
def _subscribe_topic(
iface: TCPInterface, topic: str
) -> PacketCollector:
"""Internal: subscribe a fully-qualified *topic* and return a collector.
Filters by *interface* so multi-node tests can subscribe several
collectors concurrently without cross-talk.
"""
collector = PacketCollector()
def handler(packet, interface):
if interface is iface:
collector.on_receive(packet, interface)
pub.subscribe(handler, topic)
collector._handler = handler # strong ref; see PacketCollector docstring
return collector
def subscribe_texts(iface: TCPInterface) -> PacketCollector:
"""Subscribe to ``meshtastic.receive.text`` filtered to *iface*."""
return _subscribe_topic(iface, "meshtastic.receive.text")
def subscribe_traceroutes(iface: TCPInterface) -> PacketCollector:
"""Subscribe to ``meshtastic.receive.traceroute`` filtered to *iface*."""
return _subscribe_topic(iface, "meshtastic.receive.traceroute")
def subscribe_telemetries(iface: TCPInterface) -> PacketCollector:
"""Subscribe to ``meshtastic.receive.telemetry`` filtered to *iface*."""
return _subscribe_topic(iface, "meshtastic.receive.telemetry")
def subscribe_positions(iface: TCPInterface) -> PacketCollector:
"""Subscribe to ``meshtastic.receive.position`` filtered to *iface*."""
return _subscribe_topic(iface, "meshtastic.receive.position")
def set_region(port: int, region: str = "US") -> None:
"""Set the LoRa region on a sim node via the CLI.
The node briefly restarts its TCP listener while committing the radio
config; ``run_cli()`` handles the reconnect retry, and we sleep long
enough for the new region to take effect before callers continue.
"""
rc, out = run_cli(port, "--set", "lora.region", region)
if rc != 0:
raise RuntimeError(f"Failed to set lora.region={region}: {out}")
time.sleep(PAUSE_AFTER_REGION_CHANGE)
def unsubscribe_all(topic: str) -> None:
"""Drop every handler currently registered on *topic*.
Tests use this in a ``finally`` block to keep pubsub clean across
the function-scoped mesh fixtures.
"""
try:
pub.unsubAll(topic)
except Exception: # pylint: disable=broad-except
pass
__all__ = [
"PAUSE_AFTER_CLI",
"PAUSE_AFTER_REGION_CHANGE",
"PacketCollector",
"RECEIVE_TIMEOUT",
"cli_then_verify",
"connect_iface",
"resolve_cli",
"run_cli",
"set_region",
"subscribe_positions",
"subscribe_telemetries",
"subscribe_texts",
"subscribe_traceroutes",
"unsubscribe_all",
"verify_state",
]

72
meshtastic/tests/test_node.py
View File

@@ -669,6 +669,78 @@ def test_getChannelByChannelIndex():
assert anode.getChannelByChannelIndex(9) is None
def _build_channels(highest_secondary_index: int):
"""Build an 8-slot channel table with contiguous active channels.
Slot 0 is PRIMARY. Slots 1..highest_secondary_index are SECONDARY.
Remaining slots are DISABLED.
"""
channels = []
for idx in range(8):
ch = Channel()
ch.index = idx
if idx == 0:
ch.role = Channel.Role.PRIMARY
ch.settings.name = "primary"
elif idx <= highest_secondary_index:
ch.role = Channel.Role.SECONDARY
ch.settings.name = f"ch{idx}"
else:
ch.role = Channel.Role.DISABLED
channels.append(ch)
return channels
@pytest.mark.unit
@pytest.mark.parametrize(
"highest_secondary_index,delete_index,expected_writes",
[
pytest.param(1, 1, [1], id="active-0-1-del-1"),
pytest.param(2, 1, [1, 2], id="active-0-2-del-1"),
pytest.param(3, 1, [1, 2, 3], id="active-0-3-del-1"),
pytest.param(3, 2, [2, 3], id="active-0-3-del-2"),
],
)
def test_delete_channel_writes_only_changed_suffix(
highest_secondary_index, delete_index, expected_writes
):
"""deleteChannel should only write slots whose payload changed."""
iface = MagicMock()
anode = Node(iface, "bar", noProto=True)
iface.localNode = anode
anode.channels = _build_channels(highest_secondary_index)
writes = []
def fake_write(channel_index, adminIndex=0):
writes.append((channel_index, adminIndex))
anode.writeChannel = fake_write
anode.deleteChannel(delete_index)
written_indices = [idx for idx, _ in writes]
assert written_indices == expected_writes
assert all(admin_idx == 0 for _, admin_idx in writes)
assert 0 not in written_indices
assert all(idx < 4 for idx in written_indices)
@pytest.mark.unit
def test_delete_channel_rejects_primary():
"""deleteChannel should refuse deleting PRIMARY slot 0."""
iface = MagicMock()
anode = Node(iface, "bar", noProto=True)
iface.localNode = anode
anode.channels = _build_channels(3)
with pytest.raises(SystemExit) as pytest_wrapped_e:
anode.deleteChannel(0)
assert pytest_wrapped_e.type is SystemExit
assert pytest_wrapped_e.value.code == 1
# TODO
# @pytest.mark.unit
# def test_deleteChannel_try_to_delete_primary_channel(capsys):

128
meshtastic/tests/test_smokemesh.py Normal file
View File

@@ -0,0 +1,128 @@
"""Meshtastic smoke tests with multiple meshtasticd sim instances.
Uses the ``firmware_mesh`` session fixture which provides a 3-node chain
topology (A-B-C) where A hears B, B hears A and C, and C hears B.
The SIMULATOR_APP packet bridge forwards transmissions between nodes
according to this topology.
"""
import time
import pytest
from .fw_helpers import (
RECEIVE_TIMEOUT,
subscribe_texts,
subscribe_traceroutes,
unsubscribe_all,
)
@pytest.mark.smokemesh
def test_smokemesh_node_db_convergence(firmware_mesh):
"""Each node should see all 3 nodes in its node DB after convergence."""
counts = [len(n.iface.nodes) for n in firmware_mesh.nodes if n.iface]
if any(c < 3 for c in counts):
pytest.skip(f"Mesh did not converge (counts={counts})")
for i, node in enumerate(firmware_mesh.nodes):
iface = node.iface
assert iface is not None
assert len(iface.nodes) >= 3, f"node {i} only sees {len(iface.nodes)} nodes"
@pytest.mark.smokemesh
def test_smokemesh_broadcast_text(firmware_mesh):
"""A broadcast from node A should arrive on node B."""
collector = subscribe_texts(firmware_mesh.get_iface(1))
try:
firmware_mesh.get_iface(0).sendText("hello mesh", wantAck=False)
assert collector.wait_for(1)
assert "hello mesh" in collector.texts
finally:
unsubscribe_all("meshtastic.receive.text")
@pytest.mark.smokemesh
def test_smokemesh_dm(firmware_mesh):
"""A DM from node A to node B should arrive on B."""
dest = firmware_mesh.get_node(1).node_num
collector = subscribe_texts(firmware_mesh.get_iface(1))
try:
firmware_mesh.get_iface(0).sendText(
"hey B", destinationId=dest, wantAck=False
)
assert collector.wait_for(1)
assert "hey B" in collector.texts
finally:
unsubscribe_all("meshtastic.receive.text")
@pytest.mark.smokemesh
def test_smokemesh_dm_across_relay(firmware_mesh):
"""A DM from node A to node C must relay through B (chain topology)."""
dest = firmware_mesh.get_node(2).node_num
collector = subscribe_texts(firmware_mesh.get_iface(2))
try:
firmware_mesh.get_iface(0).sendText(
"relay test", destinationId=dest, wantAck=False
)
assert collector.wait_for(1), "node C did not receive the DM within timeout"
assert "relay test" in collector.texts
finally:
unsubscribe_all("meshtastic.receive.text")
@pytest.mark.smokemesh
def test_smokemesh_hop_limit_prevents_relay(firmware_mesh):
"""A broadcast with hopLimit=0 from A reaches B but B does not relay to C."""
col_b = subscribe_texts(firmware_mesh.get_iface(1))
col_c = subscribe_texts(firmware_mesh.get_iface(2))
try:
firmware_mesh.get_iface(0).sendText(
"hop0", wantAck=False, hopLimit=0
)
assert col_b.wait_for(1), "B should receive A's broadcast"
assert "hop0" in col_b.texts, "B got wrong text"
time.sleep(RECEIVE_TIMEOUT)
assert "hop0" not in col_c.texts, (
"C should NOT receive — B must not relay hopLimit=0"
)
finally:
unsubscribe_all("meshtastic.receive.text")
@pytest.mark.smokemesh
def test_smokemesh_show_nodes(firmware_mesh):
"""showNodes should report the other nodes in the mesh."""
for i in range(3):
iface = firmware_mesh.get_iface(i)
iface.showNodes()
@pytest.mark.smokemesh
def test_smokemesh_traceroute_across_relay(firmware_mesh):
"""Traceroute from A to C should show route via B in both directions."""
col_a = subscribe_traceroutes(firmware_mesh.get_iface(0))
col_c = subscribe_traceroutes(firmware_mesh.get_iface(2))
try:
src_a = firmware_mesh.get_node(0).node_num
dest_c = firmware_mesh.get_node(2).node_num
node_b = firmware_mesh.get_node(1).node_num
firmware_mesh.get_iface(0).sendTraceRoute(dest=dest_c, hopLimit=3)
time.sleep(2)
assert len(col_a.traceroutes) >= 1, "A did not receive traceroute response"
a_resp = col_a.traceroutes[0]
assert a_resp["from"] == dest_c, "response source should be C"
route = a_resp["decoded"]["traceroute"]
assert route.get("route") == [node_b], "forward route should be A→B→C"
assert route.get("routeBack") == [node_b], "return route should be C→B→A"
assert len(col_c.traceroutes) >= 1, "C did not receive traceroute request"
c_req = col_c.traceroutes[0]
assert c_req["from"] == src_a, "request source should be A"
finally:
unsubscribe_all("meshtastic.receive.traceroute")

1209
meshtastic/tests/test_smokevirt.py
View File

File diff suppressed because it is too large Load Diff

3
pytest.ini
View File

@@ -1,6 +1,6 @@
[pytest]
addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples and not smokevirt"
addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples and not smokevirt and not smokemesh"
filterwarnings =
ignore::DeprecationWarning
@@ -13,4 +13,5 @@ markers =
smoke1: runs smoke tests on a single device connected via USB
smoke2: runs smoke tests on a two devices connected via USB
smokewifi: runs smoke test on an esp32 device setup with wifi
smokemesh: runs smoke tests against multiple meshtasticd sim instances
examples: runs the examples tests which validates the library