meshtastic/plugin/tests/e2e/test_meshtasticator_e2e.py

"""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.
        for _ in range(3):
            if _got():
                break
            peer.localNode.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")
    finally:
        pub.unsubscribe(_on_receive, "meshtastic.receive.text")
        peer.close()