eric
6125130e6a
Run 10538 booted the full simulator and ran pytest — both failures were test-side: the installed meshtastic client moved sendText from Node to the Interface, and the outbound chunked message never arrived at the peer (cause still unknown). Fix the sendText call and add in-run diagnostics so the next failure is self-explaining: - e2e: outbound-timeout failure now reports what arrived at the peer and the simulator log tail; sim now boots with -v (meshtastic debug logging). - run_e2e.sh: on pytest failure in docker mode, dump per-node meshtasticd logs (meshtasticator-nodes.log) before cleanup removes the container.
297 lines
10 KiB
Python
297 lines
10 KiB
Python
"""End-to-end tests: hermes-meshtastic adapter against live Meshtasticator nodes.
|
|
|
|
What these tests validate (the "radio-level" layer no contract test can):
|
|
* the adapter's real TCP/protobuf handshake against MeshtasticD (the actual
|
|
device software, not a mock) as run by the Meshtasticator interactive
|
|
simulator (https://meshtastic.org/docs/software/meshtasticator/),
|
|
* inbound text reception across the simulated LoRa mesh,
|
|
* outbound chunked sends arriving at another simulated node,
|
|
* the whole path: pubsub callback -> MessageEvent construction (real gateway
|
|
types) -> handle_message dispatch.
|
|
|
|
What they do NOT validate (out of scope even for Meshtasticator): true RF
|
|
(silicon-level LoRa, real propagation/interference, regulatory duty cycle).
|
|
|
|
Running
|
|
-------
|
|
1. On a Linux host (or a machine with Docker): clone meshtastic/meshtasticator
|
|
and boot the interactive simulator (see scripts/meshtasticator_e2e/).
|
|
2. Point the plugin's own contract venv at the booted nodes and run:
|
|
cd plugin && MESHTASTICATOR_E2E=1 E2E_PORTS=4404,4405 \
|
|
E2E_CHANNEL_INDEX=0 .venv-contract/bin/python -m pytest tests/e2e -q
|
|
|
|
Environment
|
|
-----------
|
|
MESHTASTICATOR_E2E = "1" enables these tests (required; without it the
|
|
module skips so ordinary unit/CI runs stay fast).
|
|
E2E_PORTS = comma-separated TCP API ports of the simulated nodes
|
|
(first = the node the adapter connects to, default 4404).
|
|
E2E_HOST = host where the simulator exposes the ports (default 127.0.0.1).
|
|
E2E_CHANNEL_INDEX = Meshtastic channel index the adapter listens/sends on
|
|
(default 0 = primary channel, matching a stock sim node).
|
|
|
|
NOTE: this module is an initial scaffold. Structural validation (imports,
|
|
fixtures, flow) passes, but a full green run requires a Linux/Docker host with
|
|
a MeshtasticD build — tracked as the remaining verification step.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import pytest
|
|
|
|
if os.environ.get("MESHTASTICATOR_E2E") != "1":
|
|
pytest.skip(
|
|
"set MESHTASTICATOR_E2E=1 to run against Meshtasticator nodes "
|
|
"(see scripts/meshtasticator_e2e/README.md)",
|
|
allow_module_level=True,
|
|
)
|
|
|
|
# Real Hermes gateway + real Meshtastic client library are required.
|
|
pytest.importorskip("gateway.platforms.base")
|
|
pytest.importorskip("meshtastic.tcp_interface")
|
|
|
|
from gateway.config import PlatformConfig # noqa: E402
|
|
from gateway.platform_registry import ( # noqa: E402
|
|
PlatformEntry,
|
|
platform_registry,
|
|
)
|
|
|
|
import hermes_meshtastic # noqa: E402
|
|
from hermes_meshtastic import adapter as plugin_adapter # noqa: E402
|
|
|
|
E2E_HOST = os.environ.get("E2E_HOST", "127.0.0.1")
|
|
E2E_PORTS = [int(p) for p in os.environ.get("E2E_PORTS", "4404,4405").split(",") if p]
|
|
E2E_CHANNEL_INDEX = int(os.environ.get("E2E_CHANNEL_INDEX", "0"))
|
|
assert len(E2E_PORTS) >= 2, "E2E_PORTS needs at least two node ports (adapter + peer)"
|
|
|
|
NODE0_PORT = E2E_PORTS[0] # the node our adapter connects to
|
|
NODE1_PORT = E2E_PORTS[1] # the peer node that sends / observes
|
|
CONNECT_TIMEOUT_S = 30
|
|
RADIO_SETTLE_S = 2
|
|
|
|
|
|
def _register_meshtastic_platform():
|
|
"""Register the plugin (mirrors discovery) so Platform('meshtastic') resolves."""
|
|
if platform_registry.is_registered("meshtastic"):
|
|
return
|
|
kwargs = {}
|
|
|
|
class _Ctx:
|
|
def register_platform(self, **kw):
|
|
kwargs.update(kw)
|
|
|
|
hermes_meshtastic.register(_Ctx())
|
|
platform_registry.register(
|
|
PlatformEntry(
|
|
name=kwargs["name"],
|
|
label=kwargs["label"],
|
|
adapter_factory=kwargs["adapter_factory"],
|
|
check_fn=kwargs["check_fn"],
|
|
required_env=[],
|
|
max_message_length=kwargs.get("max_message_length", 0),
|
|
emoji=kwargs.get("emoji", "🔌"),
|
|
platform_hint=kwargs.get("platform_hint", ""),
|
|
)
|
|
)
|
|
|
|
|
|
class _ProbeAdapter(plugin_adapter.get_adapter_class()):
|
|
"""MeshtasticAdapter that records inbound MessageEvents instead of
|
|
dispatching to a Hermes agent handler (no agent runs in these tests)."""
|
|
|
|
def __init__(self, config):
|
|
super().__init__(config)
|
|
self.received_events = []
|
|
|
|
async def handle_message(self, event):
|
|
self.received_events.append(event)
|
|
# Intentionally do not call super(): the real base would forward to a
|
|
# message handler that only exists inside a running Hermes gateway.
|
|
return None
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def adapter():
|
|
_register_meshtastic_platform()
|
|
cfg = PlatformConfig(
|
|
enabled=True,
|
|
extra={
|
|
"host": E2E_HOST,
|
|
"port": NODE0_PORT,
|
|
"channel_index": E2E_CHANNEL_INDEX,
|
|
},
|
|
)
|
|
return _ProbeAdapter(cfg)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def loop():
|
|
"""Long-lived event loop in a background thread.
|
|
|
|
The adapter mirrors the Hermes gateway contract: connect()/watchdog/
|
|
handle_message() assume a *running* loop (the gateway's). A per-call
|
|
asyncio.run() would close the loop right after connect() and strand both
|
|
the watchdog and inbound dispatch, so the e2e module keeps one loop alive
|
|
and drives coroutines into it with run_coroutine_threadsafe.
|
|
"""
|
|
import threading
|
|
|
|
_loop = asyncio.new_event_loop()
|
|
thread = threading.Thread(target=_loop.run_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield _loop
|
|
finally:
|
|
# Cancel leftover tasks (e.g. a watchdog started by a failed connect
|
|
# whose disconnect teardown never ran) before closing the loop.
|
|
async def _shutdown():
|
|
for task in asyncio.all_tasks(_loop):
|
|
task.cancel()
|
|
|
|
try:
|
|
asyncio.run_coroutine_threadsafe(_shutdown(), _loop).result(timeout=5)
|
|
except Exception:
|
|
pass
|
|
_loop.call_soon_threadsafe(_loop.stop)
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def _run_in_loop(loop, coro, timeout_s=CONNECT_TIMEOUT_S):
|
|
import concurrent.futures
|
|
|
|
try:
|
|
return asyncio.run_coroutine_threadsafe(coro, loop).result(timeout=timeout_s)
|
|
except concurrent.futures.TimeoutError:
|
|
pytest.fail(f"coroutine did not finish within {timeout_s}s")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def node0(adapter, loop):
|
|
"""Connect the adapter to simulated node 0 (retrying while the node's TCP
|
|
API comes up); disconnect afterwards."""
|
|
import time
|
|
|
|
deadline = time.monotonic() + CONNECT_TIMEOUT_S
|
|
last_ok = False
|
|
while time.monotonic() < deadline:
|
|
last_ok = bool(_run_in_loop(loop, adapter.connect()))
|
|
if last_ok:
|
|
break
|
|
time.sleep(3)
|
|
assert last_ok, (
|
|
f"adapter failed to connect to simulated node at {E2E_HOST}:{NODE0_PORT} "
|
|
"(is the simulator running and the node reachable?)"
|
|
)
|
|
try:
|
|
yield adapter
|
|
finally:
|
|
try:
|
|
_run_in_loop(loop, adapter.disconnect(), timeout_s=10)
|
|
except Exception: # loop teardown must not mask test failures
|
|
pass
|
|
|
|
|
|
def _wait_for(predicate, timeout_s=15.0, interval_s=0.5, what="condition"):
|
|
"""Poll until *predicate* (callable -> bool) is true, else pytest.fail."""
|
|
import time
|
|
|
|
deadline = time.monotonic() + timeout_s
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return
|
|
time.sleep(interval_s)
|
|
pytest.fail(f"timed out waiting for {what}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Radio-level flows
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_adapter_receives_broadcast_from_peer(node0):
|
|
"""Peer node 1 broadcasts text over the simulated mesh; the adapter must
|
|
receive it as a real gateway MessageEvent."""
|
|
import time
|
|
|
|
from meshtastic import tcp_interface
|
|
|
|
sent = f"e2e-inbound-{int(time.time())}"
|
|
|
|
peer = tcp_interface.TCPInterface(hostname=E2E_HOST, portNumber=NODE1_PORT)
|
|
try:
|
|
time.sleep(RADIO_SETTLE_S)
|
|
|
|
def _got():
|
|
return any(getattr(e, "text", None) == sent for e in node0.received_events)
|
|
|
|
# Radio delivery is not guaranteed per attempt; retry a few times.
|
|
# (meshtastic client >=2.8: sendText lives on the Interface, not Node.)
|
|
for _ in range(3):
|
|
if _got():
|
|
break
|
|
peer.sendText(
|
|
text=sent, destinationId="^all", channelIndex=E2E_CHANNEL_INDEX
|
|
)
|
|
time.sleep(1)
|
|
_wait_for(_got, timeout_s=25, what="inbound broadcast from peer node")
|
|
finally:
|
|
peer.close()
|
|
|
|
|
|
def test_adapter_chunked_send_reaches_peer(node0, loop):
|
|
"""A long outbound message must be chunked and arrive at the peer node."""
|
|
import time
|
|
|
|
from meshtastic import tcp_interface
|
|
from pubsub import pub
|
|
|
|
content = " ".join(f"word{n}" for n in range(60)) # well over 180 chars -> chunks
|
|
received = []
|
|
|
|
def _on_receive(packet, interface):
|
|
decoded = packet.get("decoded", {}) or {}
|
|
text = decoded.get("text", "")
|
|
if text:
|
|
received.append(text)
|
|
|
|
peer = tcp_interface.TCPInterface(hostname=E2E_HOST, portNumber=NODE1_PORT)
|
|
try:
|
|
time.sleep(RADIO_SETTLE_S)
|
|
pub.subscribe(_on_receive, "meshtastic.receive.text")
|
|
|
|
async def _send():
|
|
result = await node0.send("channel_e2e", content)
|
|
assert result.success, f"adapter.send failed: {result.error}"
|
|
|
|
_run_in_loop(loop, _send(), timeout_s=60)
|
|
|
|
def _all_chunks_arrived():
|
|
# Adapter frames multi-part sends as "(i/n) <chunk>"; the final
|
|
# chunk carries the message tail.
|
|
return any(str(r).rstrip().endswith("word59") for r in received)
|
|
|
|
_wait_for(_all_chunks_arrived, timeout_s=45, what="chunked outbound send at peer")
|
|
|
|
except Exception as exc:
|
|
# Fast-fail diagnostics: what arrived at the peer, and what the
|
|
# simulator itself logged (RF forwarding/routing noise).
|
|
tail = ""
|
|
try:
|
|
sim_log = os.path.join(os.getcwd(), "..", "..", "meshtasticator-sim.log")
|
|
if not os.path.exists(sim_log):
|
|
sim_log = os.path.join(os.getcwd(), "meshtasticator-sim.log")
|
|
if os.path.exists(sim_log):
|
|
tail = "\n".join(
|
|
open(sim_log, errors="replace").read().splitlines()[-40:]
|
|
)
|
|
except OSError:
|
|
pass
|
|
raise AssertionError(
|
|
f"{exc}\nreceived at peer so far: {received[-10:]!r}\n"
|
|
f"--- meshtasticator-sim.log tail ---\n{tail}"
|
|
) from exc
|
|
finally:
|
|
pub.unsubscribe(_on_receive, "meshtastic.receive.text")
|
|
peer.close()
|