matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

feat(transport): MeshCoreTransport over pyMC companion TCP (Phase 2) (#4)

Browse source 
* feat(transport): MeshCoreTransport over pyMC companion TCP (Phase 2)

Implements MeshCoreTransport (MeshTransport impl) using the meshcore lib
over TCP to a pyMC companion frame server, bridged behind the sync
interface via a dedicated event-loop thread. Outbound channel/DM sends,
inbound message normalization into MeshMessage(transport="meshcore"),
contact/self lookups. Factory wires transport="meshcore"; supervisor is
now transport-aware (Meshtastic watchdog guarded). Dormant unless
configured; meshtastic path unchanged; full suite matches baseline.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(transport): sync loop-thread readiness before dispatch (MeshCore)

Wait on a threading.Event set from inside the event loop (via call_soon)
before dispatching the first coroutine in connect(), eliminating a startup
race where run_coroutine_threadsafe could be rejected by an is_running()
pre-check before run_forever() had begun spinning.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(transport): MeshCore broadcasts use configured channel index

The channel arg carries Meshtastic-index semantics that don't map to
MeshCore's channel table; broadcasts now always use the configured
meshcore_channel_index (also fixes explicit channel=0 being treated as
falsy). DM path unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-02 10:15:39 -06:00 • committed by GitHub
commit 316ae7351e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 865 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

9
work/meshai/config.py
View file

@ -34,8 +34,15 @@ class ConnectionConfig:
reconnect_initial_delay: float = 2.0
reconnect_max_delay: float = 60.0
reconnect_health_interval: float = 30.0
# --- transport selection (Phase 1 seam; MeshCore support is future work) ---
# --- transport selection (Phase 1 seam; MeshCore support is Phase 2) ---
transport: str = "meshtastic" # "meshtastic" | "meshcore" | "both"
# --- MeshCore transport settings (used when transport="meshcore") ---
meshcore_host: str = "100.64.0.9" # pyMC companion frame server host
meshcore_port: int = 5050 # pyMC companion frame server port
meshcore_channel_index: int = 0 # default channel index for broadcasts
meshcore_max_chars: int = 140 # placeholder max chars (Phase 3 chunker)
meshcore_auto_reconnect: bool = True # enable meshcore lib auto-reconnect
meshcore_max_reconnect_attempts: int = 5 # max reconnect attempts (0 = unlimited)
@dataclass

9
work/meshai/main.py
View file

@ -83,8 +83,13 @@ class MeshAI:
self._loop = asyncio.get_event_loop()
# --- connection supervisor (watchdog): single source of truth for link
# state + the only reconnect driver. Reconnects IN-PLACE so the container
# never needs to restart. ---
if getattr(self.config.connection, "reconnect", True):
# never needs to restart.
# Guard: watchdog is Meshtastic-specific — MeshCoreTransport manages its
# own reconnect via the meshcore lib's auto_reconnect parameter. ---
if (
getattr(self.config.connection, "reconnect", True)
and isinstance(self.connector, MeshConnector)
):
self.connector._wake = asyncio.Event()
self.connector.write_link_status("up") # we just connected ok
self._supervisor_task = asyncio.create_task(self._connection_supervisor())

11
work/meshai/transport/factory.py
View file

@ -26,10 +26,15 @@ def build_transport(config) -> MeshTransport:
from meshai.connector import MeshtasticTransport
return MeshtasticTransport(config)
if transport_name in ("meshcore", "both"):
if transport_name == "meshcore":
# Function-local import keeps the module importable without the meshcore
# lib installed (lazy import pattern mirrors the meshtastic case).
from meshai.transport.meshcore_transport import MeshCoreTransport
return MeshCoreTransport(config)
if transport_name == "both":
raise NotImplementedError(
f"Transport {transport_name!r} is not yet implemented. "
"This is a Phase 1 seam; MeshCore support lands in a later phase."
"Transport 'both' is not yet implemented (Phase 4 seam)."
)
raise ValueError(

390
work/meshai/transport/meshcore_transport.py Normal file
View file

@ -0,0 +1,390 @@
"""MeshCore transport backend (Phase 2).
Implements MeshTransport over a pyMC companion TCP frame server using the
meshcore lib. The meshcore lib is fully async; we bridge it into the sync
MeshTransport interface via a dedicated asyncio event loop running in a
daemon thread. Commands are dispatched with
``asyncio.run_coroutine_threadsafe()``.
The meshcore lib is LAZY-IMPORTED inside methods so this module can be
imported (and the test suite can run) without the lib installed.
"""
import asyncio
import logging
import threading
from typing import Callable, Optional
from .base import MeshTransport
from ..connector import MeshMessage
logger = logging.getLogger(__name__)
# Default timeout for command futures (seconds).
_COMMAND_TIMEOUT = 10.0
class MeshCoreTransport(MeshTransport):
"""MeshTransport implementation over a pyMC companion TCP frame server.
Async bridge: a dedicated asyncio event loop runs in a daemon thread
(``self._loop`` / ``self._loop_thread``). All coroutines are dispatched
via ``asyncio.run_coroutine_threadsafe(..., self._loop).result(timeout)``.
The meshcore client (``self._mc``) is created and used exclusively on that
loop.
The transport is dormant until ``connect()`` is called. When
``transport != "meshcore"`` in config the factory never instantiates this
class, so there is zero cost to existing meshtastic deployments.
"""
def __init__(self, config) -> None:
self.config = config
self._mc = None # meshcore.MeshCore instance
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._loop_thread: Optional[threading.Thread] = None
self._connected: bool = False
self._self_info: dict = {}
self._message_callback: Optional[Callable] = None
self._callback_loop: Optional[asyncio.AbstractEventLoop] = None
self._loop_ready: threading.Event = threading.Event()
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _run_coro(self, coro, timeout: float = _COMMAND_TIMEOUT):
"""Submit *coro* to the dedicated event loop and block until done.
Raises RuntimeError if the loop isn't running.
"""
if self._loop is None or not self._loop.is_running():
raise RuntimeError("MeshCoreTransport: event loop is not running")
future = asyncio.run_coroutine_threadsafe(coro, self._loop)
return future.result(timeout=timeout)
def _stop_loop(self) -> None:
"""Signal the event loop to stop and join the thread."""
if self._loop is not None:
try:
self._loop.call_soon_threadsafe(self._loop.stop)
except Exception:
pass
if self._loop_thread is not None and self._loop_thread.is_alive():
self._loop_thread.join(timeout=5.0)
self._loop = None
self._loop_thread = None
# ------------------------------------------------------------------
# Internal coroutines (run on the dedicated loop)
# ------------------------------------------------------------------
async def _do_connect(self, host: str, port: int,
auto_reconnect: bool, max_attempts: int):
"""Lazy-import meshcore and create the TCP client."""
from meshcore import MeshCore # noqa: PLC0415 (lazy import intentional)
mc = await MeshCore.create_tcp(
host, port,
auto_reconnect=auto_reconnect,
max_reconnect_attempts=max_attempts,
)
return mc
async def _setup_subscriptions(self) -> None:
"""Start auto message fetching and subscribe to inbound events."""
from meshcore import EventType # noqa: PLC0415
await self._mc.start_auto_message_fetching()
self._mc.subscribe(EventType.CONTACT_MSG_RECV, self._on_dm_event)
self._mc.subscribe(EventType.CHANNEL_MSG_RECV, self._on_channel_event)
self._mc.subscribe(EventType.DISCONNECTED, self._on_disconnect_event)
self._mc.subscribe(EventType.CONNECTED, self._on_connect_event)
async def _do_disconnect(self) -> None:
"""Stop fetching and close the meshcore connection."""
try:
await self._mc.stop_auto_message_fetching()
except Exception as exc:
logger.warning("MeshCoreTransport: stop_auto_message_fetching error: %s", exc)
try:
await self._mc.disconnect()
except Exception as exc:
logger.warning("MeshCoreTransport: mc.disconnect error: %s", exc)
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
def connect(self) -> None:
"""Connect to the pyMC companion TCP frame server."""
host = getattr(self.config, "meshcore_host", "100.64.0.9")
port = getattr(self.config, "meshcore_port", 5050)
auto_reconnect = getattr(self.config, "meshcore_auto_reconnect", True)
max_attempts = getattr(self.config, "meshcore_max_reconnect_attempts", 5)
logger.info("MeshCoreTransport: connecting to %s:%d …", host, port)
# Start the dedicated event loop in a daemon thread.
self._loop = asyncio.new_event_loop()
self._loop_ready.clear()
def _run_loop() -> None:
asyncio.set_event_loop(self._loop)
self._loop.call_soon(self._loop_ready.set)
self._loop.run_forever()
self._loop_thread = threading.Thread(
target=_run_loop,
name="meshcore-loop",
daemon=True,
)
self._loop_thread.start()
self._loop_ready.wait(timeout=5.0)
try:
mc = self._run_coro(
self._do_connect(host, port, auto_reconnect, max_attempts),
timeout=30.0,
)
except Exception as exc:
logger.error("MeshCoreTransport: connect failed: %s", exc)
self._stop_loop()
raise
if mc is None:
self._stop_loop()
raise RuntimeError(
f"MeshCore.create_tcp({host}:{port}) returned None — connection failed"
)
self._mc = mc
self._self_info = mc.self_info or {}
self._connected = True
# Subscribe to inbound events on the dedicated loop.
self._run_coro(self._setup_subscriptions())
logger.info(
"MeshCoreTransport: connected as %s (pubkey %s)",
self._self_info.get("name", "unknown"),
self._self_info.get("public_key", "?"),
)
def disconnect(self) -> None:
"""Disconnect and stop the event loop thread."""
if self._mc is not None:
try:
self._run_coro(self._do_disconnect(), timeout=10.0)
except Exception as exc:
logger.warning("MeshCoreTransport: disconnect error: %s", exc)
self._mc = None
self._connected = False
self._stop_loop()
logger.info("MeshCoreTransport: disconnected")
# ------------------------------------------------------------------
# Message I/O
# ------------------------------------------------------------------
def send_message(
self,
text: str,
destination: Optional[str] = None,
channel: int = 0,
) -> bool:
"""Send a message via MeshCore.
Args:
text: Message text (caller is responsible for length limits; see
``meshcore_max_chars`` config field for the future chunker).
destination: hex pubkey string for a DM, or None for channel send.
channel: Channel index for channel sends (0 = default).
Returns:
True if the send succeeded (not an error event).
"""
if self._mc is None:
logger.error("MeshCoreTransport: cannot send, not connected")
return False
try:
if destination:
result = self._run_coro(
self._mc.commands.send_msg(destination, text)
)
else:
# Channel-index semantics do NOT cross transports: the passed
# `channel` carries Meshtastic channel-index semantics (e.g.
# index 8) that have no relationship to MeshCore's separate
# channel table. The configured MeshCore channel is therefore
# authoritative for broadcasts, so we ignore `channel` here
# (this also avoids an explicit channel=0 being treated as
# falsy).
chan_idx = getattr(self.config, "meshcore_channel_index", 0)
result = self._run_coro(
self._mc.commands.send_chan_msg(chan_idx, text)
)
success = not result.is_error()
if not success:
logger.warning("MeshCoreTransport: send returned error event")
return success
except Exception as exc:
logger.error("MeshCoreTransport: send_message failed: %s", exc)
return False
def set_message_callback(
self,
callback: Callable,
loop: asyncio.AbstractEventLoop,
) -> None:
"""Store the meshai callback and its event loop for inbound dispatch."""
self._message_callback = callback
self._callback_loop = loop
# ------------------------------------------------------------------
# Event normalization — factored out for hermetic unit testing
# ------------------------------------------------------------------
def _normalize_dm_event(self, event) -> Optional[MeshMessage]:
"""Map a CONTACT_MSG_RECV event payload → MeshMessage.
Separated from the subscription handler so tests can call it directly
without spinning up the loop thread.
"""
try:
payload = event.payload or {}
text = payload.get("text", "")
if not text:
return None
pubkey_prefix: str = payload.get("pubkey_prefix", "")
# Best-effort contact name resolution.
sender_name = pubkey_prefix
if self._mc is not None:
try:
contact = self._mc.get_contact_by_key_prefix(pubkey_prefix)
if contact:
sender_name = contact.get("adv_name", pubkey_prefix) or pubkey_prefix
except Exception:
pass
return MeshMessage(
sender_id=pubkey_prefix,
sender_name=sender_name,
text=text,
channel=0,
is_dm=True,
packet=None,
transport="meshcore",
)
except Exception as exc:
logger.error("MeshCoreTransport: error normalizing DM event: %s", exc)
return None
def _normalize_channel_event(self, event) -> Optional[MeshMessage]:
"""Map a CHANNEL_MSG_RECV event payload → MeshMessage.
Separated from the subscription handler so tests can call it directly
without spinning up the loop thread.
"""
try:
payload = event.payload or {}
text = payload.get("text", "")
if not text:
return None
channel_idx: int = payload.get("channel_idx", 0)
# Channel messages carry no per-sender pubkey in the meshcore API.
channel_marker = f"chan:{channel_idx}"
return MeshMessage(
sender_id=channel_marker,
sender_name=channel_marker,
text=text,
channel=channel_idx,
is_dm=False,
packet=None,
transport="meshcore",
)
except Exception as exc:
logger.error("MeshCoreTransport: error normalizing channel event: %s", exc)
return None
# ------------------------------------------------------------------
# Internal event handlers (called by meshcore's event system)
# ------------------------------------------------------------------
def _on_dm_event(self, event) -> None:
"""Handle CONTACT_MSG_RECV: normalize and dispatch to meshai."""
msg = self._normalize_dm_event(event)
self._dispatch_message(msg)
def _on_channel_event(self, event) -> None:
"""Handle CHANNEL_MSG_RECV: normalize and dispatch to meshai."""
msg = self._normalize_channel_event(event)
self._dispatch_message(msg)
def _dispatch_message(self, msg: Optional[MeshMessage]) -> None:
"""Marshal a MeshMessage onto the meshai event loop (thread-safe).
Mirrors exactly what MeshtasticTransport._on_receive does:
loop.call_soon_threadsafe(lambda m=msg: asyncio.create_task(cb(m)))
"""
if msg is None or self._message_callback is None or self._callback_loop is None:
return
try:
self._callback_loop.call_soon_threadsafe(
lambda m=msg: asyncio.create_task(self._message_callback(m))
)
except Exception as exc:
logger.error("MeshCoreTransport: error dispatching message: %s", exc)
def _on_disconnect_event(self, event=None) -> None:
"""Track link state: DISCONNECTED."""
self._connected = False
logger.warning("MeshCoreTransport: DISCONNECTED event received")
def _on_connect_event(self, event=None) -> None:
"""Track link state: CONNECTED (auto-reconnect succeeded)."""
self._connected = True
logger.info("MeshCoreTransport: CONNECTED event received")
# ------------------------------------------------------------------
# Node identity / topology (MeshTransport abstract methods)
# ------------------------------------------------------------------
@property
def my_node_id(self) -> Optional[str]:
"""Our own public key (hex string), or None before connect."""
return self._self_info.get("public_key") or None
@property
def connected(self) -> bool:
"""True when the transport has an active connection."""
return self._connected and self._mc is not None
def get_node_name(self, node_id: str) -> str:
"""Resolve a pubkey prefix to a contact display name, or return node_id."""
if self._mc is None:
return node_id
try:
contact = self._mc.get_contact_by_key_prefix(node_id)
if contact:
return contact.get("adv_name", node_id) or node_id
except Exception:
pass
return node_id
def get_node_position(self, node_id: str) -> Optional[tuple]:
"""Return (adv_lat, adv_lon) for a contact, or None if not available."""
if self._mc is None:
return None
try:
contact = self._mc.get_contact_by_key_prefix(node_id)
if contact:
lat = contact.get("adv_lat")
lon = contact.get("adv_lon")
if lat is not None and lon is not None:
return (lat, lon)
except Exception:
pass
return None

1
work/pyproject.toml
View file

@ -29,6 +29,7 @@ classifiers = [
dependencies = [
"meshtastic>=2.3.0",
"meshcore>=2.3.7",
"pyyaml>=6.0",
"aiosqlite>=0.19.0",
"openai>=1.0.0",

1
work/requirements.txt
View file

@ -1,4 +1,5 @@
meshtastic>=2.3.0
meshcore>=2.3.7
pyyaml>=6.0
aiosqlite>=0.19.0
openai>=1.0.0

443
work/tests/test_meshcore_transport.py Normal file
View file

@ -0,0 +1,443 @@
"""Tests for MeshCoreTransport (Phase 2).
All tests are fully mocked — no real socket, no meshcore lib required.
The fake meshcore module is injected into sys.modules before any lazy-import
triggers, so the production code's lazy ``from meshcore import MeshCore``
gets the mock transparently.
"""
import asyncio
import sys
import threading
import types
from unittest.mock import AsyncMock, MagicMock
import pytest
# ---------------------------------------------------------------------------
# Build and register a minimal fake meshcore module
# (must happen before importing production code that lazy-imports meshcore)
# ---------------------------------------------------------------------------
def _build_fake_meshcore():
mod = types.ModuleType("meshcore")
class EventType:
CONTACT_MSG_RECV = "CONTACT_MSG_RECV"
CHANNEL_MSG_RECV = "CHANNEL_MSG_RECV"
DISCONNECTED = "DISCONNECTED"
CONNECTED = "CONNECTED"
mod.EventType = EventType
class _FakeMeshCore:
"""Minimal stand-in for meshcore.MeshCore."""
self_info = {"public_key": "aabbccdd1122", "name": "FakeNode"}
contacts = {}
async def start_auto_message_fetching(self):
pass
async def stop_auto_message_fetching(self):
pass
async def disconnect(self):
pass
def subscribe(self, event_type, callback):
pass
def get_contact_by_key_prefix(self, prefix):
return None
@classmethod
async def create_tcp(cls, host, port,
auto_reconnect=True, max_reconnect_attempts=5):
return cls()
class commands:
@staticmethod
async def send_chan_msg(chan_idx, text):
result = MagicMock()
result.is_error.return_value = False
return result
@staticmethod
async def send_msg(dst, text):
result = MagicMock()
result.is_error.return_value = False
return result
mod.MeshCore = _FakeMeshCore
return mod
# Register before production imports so lazy-import finds the mock.
sys.modules.setdefault("meshcore", _build_fake_meshcore())
# ---------------------------------------------------------------------------
# Production imports (after mock is in sys.modules)
# ---------------------------------------------------------------------------
from meshai.config import ConnectionConfig # noqa: E402
from meshai.connector import MeshMessage # noqa: E402
from meshai.transport.base import MeshTransport # noqa: E402
from meshai.transport.factory import build_transport # noqa: E402
from meshai.transport.meshcore_transport import MeshCoreTransport # noqa: E402
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _mc_config(**overrides):
"""Return a ConnectionConfig wired for meshcore."""
cfg = ConnectionConfig(
transport="meshcore",
meshcore_host="127.0.0.1",
meshcore_port=5050,
meshcore_channel_index=0,
)
for k, v in overrides.items():
setattr(cfg, k, v)
return cfg
def _transport_with_mock_mc(mc_overrides=None):
"""Create a MeshCoreTransport with _mc injected as a MagicMock.
Also starts a real dedicated loop thread so _run_coro works.
Returns (transport, mc_mock, loop).
"""
cfg = _mc_config()
t = MeshCoreTransport(cfg)
mc = MagicMock()
mc.get_contact_by_key_prefix.return_value = None
if mc_overrides:
for k, v in mc_overrides.items():
setattr(mc, k, v)
t._mc = mc
t._connected = True
loop = asyncio.new_event_loop()
t._loop = loop
thread = threading.Thread(target=loop.run_forever, daemon=True)
thread.start()
t._loop_thread = thread
return t, mc, loop
def _cleanup(t):
"""Stop the transport's dedicated event loop."""
try:
if t._loop and t._loop.is_running():
t._loop.call_soon_threadsafe(t._loop.stop)
if t._loop_thread and t._loop_thread.is_alive():
t._loop_thread.join(timeout=2.0)
except Exception:
pass
def _make_dm_event(text="hello", pubkey_prefix="aabbcc001122", **extra):
e = MagicMock()
e.payload = {"type": "PRIV", "pubkey_prefix": pubkey_prefix, "text": text, **extra}
return e
def _make_channel_event(text="chan msg", channel_idx=2, **extra):
e = MagicMock()
e.payload = {"type": "CHAN", "channel_idx": channel_idx, "text": text, **extra}
return e
# ---------------------------------------------------------------------------
# 1. Factory / subclass tests
# ---------------------------------------------------------------------------
class TestBuildTransport:
def test_returns_meshcore_transport(self):
t = build_transport(_mc_config())
assert isinstance(t, MeshCoreTransport)
def test_is_mesh_transport_subclass(self):
t = build_transport(_mc_config())
assert isinstance(t, MeshTransport)
# ---------------------------------------------------------------------------
# 2. send_message — channel (no destination)
# ---------------------------------------------------------------------------
class TestSendMessageChannel:
def test_returns_true_on_non_error_event(self):
t, mc, _ = _transport_with_mock_mc()
try:
ok = MagicMock()
ok.is_error.return_value = False
mc.commands.send_chan_msg = AsyncMock(return_value=ok)
assert t.send_message("hello") is True
mc.commands.send_chan_msg.assert_awaited_once()
finally:
_cleanup(t)
def test_returns_false_on_error_event(self):
t, mc, _ = _transport_with_mock_mc()
try:
err = MagicMock()
err.is_error.return_value = True
mc.commands.send_chan_msg = AsyncMock(return_value=err)
assert t.send_message("hello") is False
finally:
_cleanup(t)
def test_returns_false_when_not_connected(self):
cfg = _mc_config()
t = MeshCoreTransport(cfg)
# _mc is None, no loop started
assert t.send_message("test") is False
def _transport_with_configured_index(self, index):
"""Build a MeshCoreTransport whose config sets meshcore_channel_index."""
cfg = _mc_config(meshcore_channel_index=index)
t = MeshCoreTransport(cfg)
ok = MagicMock()
ok.is_error.return_value = False
loop = asyncio.new_event_loop()
mc = MagicMock()
mc.commands.send_chan_msg = AsyncMock(return_value=ok)
t._mc = mc
t._connected = True
t._loop = loop
thread = threading.Thread(target=loop.run_forever, daemon=True)
thread.start()
t._loop_thread = thread
return t, mc
def test_uses_config_channel_index_when_channel_zero(self):
t, mc = self._transport_with_configured_index(3)
try:
t.send_message("hi", channel=0)
# channel=0 must not be treated as falsy-fallthrough: broadcasts
# always use the configured meshcore_channel_index=3.
mc.commands.send_chan_msg.assert_awaited_once_with(3, "hi")
finally:
_cleanup(t)
def test_uses_config_channel_index_when_channel_default(self):
t, mc = self._transport_with_configured_index(3)
try:
t.send_message("hi") # default channel param
mc.commands.send_chan_msg.assert_awaited_once_with(3, "hi")
finally:
_cleanup(t)
def test_ignores_meshtastic_channel_index(self):
# channel=8 carries Meshtastic channel-index semantics that do NOT map
# to MeshCore's channel table; the configured index (3) is authoritative.
t, mc = self._transport_with_configured_index(3)
try:
t.send_message("hi", channel=8)
mc.commands.send_chan_msg.assert_awaited_once_with(3, "hi")
finally:
_cleanup(t)
# ---------------------------------------------------------------------------
# 3. send_message — DM (destination provided)
# ---------------------------------------------------------------------------
class TestSendMessageDM:
def test_dispatches_send_msg(self):
t, mc, _ = _transport_with_mock_mc()
try:
ok = MagicMock()
ok.is_error.return_value = False
mc.commands.send_msg = AsyncMock(return_value=ok)
result = t.send_message("hi DM", destination="aabbcc")
assert result is True
mc.commands.send_msg.assert_awaited_once_with("aabbcc", "hi DM")
finally:
_cleanup(t)
def test_send_msg_error_returns_false(self):
t, mc, _ = _transport_with_mock_mc()
try:
err = MagicMock()
err.is_error.return_value = True
mc.commands.send_msg = AsyncMock(return_value=err)
assert t.send_message("hi", destination="deadbeef") is False
finally:
_cleanup(t)
# ---------------------------------------------------------------------------
# 4. Inbound normalization — direct method calls (hermetic, no threads)
# ---------------------------------------------------------------------------
class TestNormalizeDmEvent:
def _t(self, mc=None):
t = MeshCoreTransport(_mc_config())
t._mc = mc
return t
def test_is_dm_true(self):
msg = self._t()._normalize_dm_event(_make_dm_event())
assert msg is not None
assert msg.is_dm is True
def test_transport_tag(self):
msg = self._t()._normalize_dm_event(_make_dm_event())
assert msg.transport == "meshcore"
def test_text_preserved(self):
msg = self._t()._normalize_dm_event(_make_dm_event(text="test text"))
assert msg.text == "test text"
def test_packet_is_none(self):
msg = self._t()._normalize_dm_event(_make_dm_event())
assert msg.packet is None
def test_sender_id_is_pubkey_prefix(self):
msg = self._t()._normalize_dm_event(_make_dm_event(pubkey_prefix="deadbeef"))
assert msg.sender_id == "deadbeef"
def test_sender_name_resolved_from_contact(self):
mc = MagicMock()
mc.get_contact_by_key_prefix.return_value = {"adv_name": "Alice"}
msg = self._t(mc)._normalize_dm_event(_make_dm_event(pubkey_prefix="aabbcc"))
assert msg.sender_name == "Alice"
def test_sender_name_falls_back_to_prefix_when_no_contact(self):
mc = MagicMock()
mc.get_contact_by_key_prefix.return_value = None
msg = self._t(mc)._normalize_dm_event(_make_dm_event(pubkey_prefix="ffff00"))
assert msg.sender_name == "ffff00"
def test_empty_text_returns_none(self):
assert self._t()._normalize_dm_event(_make_dm_event(text="")) is None
class TestNormalizeChannelEvent:
def _t(self):
t = MeshCoreTransport(_mc_config())
return t
def test_is_dm_false(self):
msg = self._t()._normalize_channel_event(_make_channel_event())
assert msg is not None
assert msg.is_dm is False
def test_transport_tag(self):
msg = self._t()._normalize_channel_event(_make_channel_event())
assert msg.transport == "meshcore"
def test_channel_idx_set(self):
msg = self._t()._normalize_channel_event(_make_channel_event(channel_idx=5))
assert msg.channel == 5
def test_text_preserved(self):
msg = self._t()._normalize_channel_event(_make_channel_event(text="RF traffic"))
assert msg.text == "RF traffic"
def test_empty_text_returns_none(self):
assert self._t()._normalize_channel_event(_make_channel_event(text="")) is None
# ---------------------------------------------------------------------------
# 5. Inbound dispatch → callback delivery (via _dispatch_message)
# ---------------------------------------------------------------------------
class TestInboundDispatch:
"""Verify that _dispatch_message delivers MeshMessage to registered callback."""
def _run_drain(self, loop, seconds=0.1):
async def _drain():
await asyncio.sleep(seconds)
loop.run_until_complete(_drain())
def test_dm_event_delivered_to_callback(self):
loop = asyncio.new_event_loop()
received = []
async def cb(msg):
received.append(msg)
t = MeshCoreTransport(_mc_config())
t.set_message_callback(cb, loop)
event = _make_dm_event(text="callback test")
msg = t._normalize_dm_event(event)
t._dispatch_message(msg)
self._run_drain(loop)
loop.close()
assert len(received) == 1
assert received[0].is_dm is True
assert received[0].text == "callback test"
assert received[0].transport == "meshcore"
def test_channel_event_delivered_to_callback(self):
loop = asyncio.new_event_loop()
received = []
async def cb(msg):
received.append(msg)
t = MeshCoreTransport(_mc_config())
t.set_message_callback(cb, loop)
event = _make_channel_event(text="chan callback", channel_idx=1)
msg = t._normalize_channel_event(event)
t._dispatch_message(msg)
self._run_drain(loop)
loop.close()
assert len(received) == 1
assert received[0].is_dm is False
assert received[0].channel == 1
assert received[0].transport == "meshcore"
def test_none_msg_not_dispatched(self):
loop = asyncio.new_event_loop()
received = []
async def cb(msg):
received.append(msg)
t = MeshCoreTransport(_mc_config())
t.set_message_callback(cb, loop)
t._dispatch_message(None)
self._run_drain(loop)
loop.close()
assert received == []
# ---------------------------------------------------------------------------
# 6. my_node_id property
# ---------------------------------------------------------------------------
class TestMyNodeId:
def test_returns_public_key_from_self_info(self):
t = MeshCoreTransport(_mc_config())
t._self_info = {"public_key": "deadbeef1234", "name": "TestNode"}
assert t.my_node_id == "deadbeef1234"
def test_returns_none_before_connect(self):
t = MeshCoreTransport(_mc_config())
assert t.my_node_id is None
def test_returns_none_when_self_info_empty(self):
t = MeshCoreTransport(_mc_config())
t._self_info = {}
assert t.my_node_id is None

10
work/tests/test_transport_abstraction.py
View file

@ -70,10 +70,14 @@ class TestBuildTransport:
transport = build_transport(cfg)
assert isinstance(transport, MeshtasticTransport)
def test_meshcore_raises_not_implemented(self):
def test_meshcore_returns_meshcore_transport(self):
# Phase 2: meshcore is now implemented; build_transport returns a
# MeshCoreTransport instance (MeshTransport subclass).
from meshai.transport.meshcore_transport import MeshCoreTransport
cfg = self._config_with("meshcore")
with pytest.raises(NotImplementedError):
build_transport(cfg)
transport = build_transport(cfg)
assert isinstance(transport, MeshCoreTransport)
assert isinstance(transport, MeshTransport)
def test_both_raises_not_implemented(self):
cfg = self._config_with("both")