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 node0(adapter):
    """Connect the adapter to simulated node 0; disconnect afterwards."""
    async def _connect():
        return await adapter.connect()

    assert asyncio.run(_connect()) is True, (
        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:
        asyncio.run(adapter.disconnect())


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):
    """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 and text.startswith("(1/"):  # our chunked framing
            received.append(text)
        elif 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}"

        asyncio.run(_send())

        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()