diff --git a/work/dashboard-frontend/src/components/SerialPortPicker.tsx b/work/dashboard-frontend/src/components/SerialPortPicker.tsx new file mode 100644 index 0000000..5c1afeb --- /dev/null +++ b/work/dashboard-frontend/src/components/SerialPortPicker.tsx @@ -0,0 +1,117 @@ +import { useState } from 'react' +import { RefreshCw, Radio, Check } from 'lucide-react' +import { getSerialPorts, type SerialPort } from '@/lib/api' + +// Reusable USB serial-port picker. Manual text entry is always available; the +// "Detect USB devices" button lists ports from GET /api/serial-ports and, when a +// port is picked, sets the STABLE by-id path (port.stable_path) — that path +// survives ttyACM* enumeration hops, which is the whole point. +export default function SerialPortPicker({ + value, + onChange, + label = 'Serial Port', + helper = 'Device path for your USB radio — click Detect to auto-fill a stable by-id path', +}: { + value: string + onChange: (path: string) => void + label?: string + helper?: string +}) { + const [ports, setPorts] = useState(null) + const [note, setNote] = useState('') + const [loading, setLoading] = useState(false) + const [error, setError] = useState(null) + + const detect = async () => { + setLoading(true) + setError(null) + try { + const res = await getSerialPorts() + setPorts(res.ports) + setNote(res.note || '') + } catch (err) { + setError(err instanceof Error ? err.message : 'Failed to list serial ports') + setPorts([]) + } finally { + setLoading(false) + } + } + + return ( +
+
+ +
+ onChange(e.target.value)} + placeholder="/dev/serial/by-id/usb-... (or /dev/ttyACM0)" + className="flex-1 px-3 py-2 bg-[#0a0e17] border border-[#1e2a3a] rounded text-sm text-slate-200 font-mono focus:outline-none focus:border-accent placeholder-slate-600" + /> + +
+ {helper &&

{helper}

} +
+ + {error && ( +
{error}
+ )} + + {ports !== null && !error && ( + ports.length === 0 ? ( +
+ No USB serial devices found — is the device passed through to the container? +
+ ) : ( +
+ {ports.map((p) => { + const selected = value === p.stable_path + const title = p.product || p.description || p.device + return ( + + ) + })} +
+ ) + )} + + {note &&

{note}

} +
+ ) +} diff --git a/work/dashboard-frontend/src/lib/api.ts b/work/dashboard-frontend/src/lib/api.ts index 3e47345..6f4bf7e 100644 --- a/work/dashboard-frontend/src/lib/api.ts +++ b/work/dashboard-frontend/src/lib/api.ts @@ -239,6 +239,29 @@ async function fetchJson(url: string): Promise { return response.json() } +export interface SerialPort { + device: string + by_id?: string | null + by_path?: string | null + stable_path: string + description?: string + vid?: number | null + pid?: number | null + serial_number?: string | null + manufacturer?: string | null + product?: string | null + likely_radio: boolean +} + +export interface SerialPortsResponse { + ports: SerialPort[] + note: string +} + +export async function getSerialPorts(): Promise { + return fetchJson('/api/serial-ports') +} + export async function fetchStatus(): Promise { return fetchJson('/api/status') } diff --git a/work/dashboard-frontend/src/pages/Config.tsx b/work/dashboard-frontend/src/pages/Config.tsx index 7390734..d9e2254 100644 --- a/work/dashboard-frontend/src/pages/Config.tsx +++ b/work/dashboard-frontend/src/pages/Config.tsx @@ -5,6 +5,7 @@ import { ManagedSecret } from '@/components/ManagedSecret' import { useDirty } from '@/context/DirtyContext' import NodePicker from '@/components/NodePicker' import ChannelPicker from '@/components/ChannelPicker' +import SerialPortPicker from '@/components/SerialPortPicker' import { Settings, Bot, MessageSquare, Database, Brain, Eye, Terminal, Cpu, Cloud, BookOpen, Activity, @@ -739,13 +740,11 @@ export function ConnectionSection({ data, onChange }: { data: ConnectionConfig; info="Serial: direct USB connection to a Meshtastic radio. TCP: connect over the network to a radio's IP or to meshtasticd running on another machine." /> {data.type === 'serial' ? ( - onChange({ ...data, serial_port: v })} - placeholder="/dev/ttyUSB0" - helper="Device path for your USB radio" - info="Usually /dev/ttyUSB0 on Linux or /dev/ttyACM0. Check with 'ls /dev/tty*' after plugging in your radio." + helper="Device path for your USB radio — Detect fills a stable by-id path" /> ) : (
diff --git a/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx b/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx index 5e2d4f7..3c27d2f 100644 --- a/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx +++ b/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx @@ -1,7 +1,8 @@ import { useState, useEffect, useCallback } from 'react' import { Link } from 'react-router-dom' import { Save, RotateCcw, RefreshCw, Check, ChevronRight } from 'lucide-react' -import { TextInput, NumberInput, Toggle, ListInput } from './Config' +import { TextInput, NumberInput, Toggle, ListInput, SelectInput } from './Config' +import SerialPortPicker from '@/components/SerialPortPicker' import { notifyRestartRequired } from '@/components/RestartBanner' import { fetchConfig as apiFetchConfig, updateConfig as apiUpdateConfig, getMeshcoreChannels, sendTestMessage } from '@/lib/api' import { useDirty } from '@/context/DirtyContext' @@ -18,6 +19,10 @@ interface ConnectionConfig { meshcore_port?: number meshcore_auto_reconnect?: boolean meshcore_max_reconnect_attempts?: number + meshcore_conn_type?: string + meshcore_serial_port?: string + meshcore_baud?: number + meshcore_ble_address?: string [key: string]: unknown } @@ -236,27 +241,67 @@ export default function MeshCoreConnection() {
MeshCore Connection

- Set the host and port to enable MeshCore; leave host blank to disable. - Meshtastic is always active. + Choose how MeshAI reaches your MeshCore node. TCP talks to a companion + frame server; Serial connects to a USB-attached node; BLE pairs over + Bluetooth. Meshtastic is always active.

-
+ upd({ meshcore_conn_type: v })} + options={[ + { value: 'tcp', label: 'TCP (companion)' }, + { value: 'serial', label: 'Serial (USB)' }, + { value: 'ble', label: 'BLE' }, + ]} + helper="TCP for a companion frame server, Serial for a USB node, BLE for Bluetooth" + /> + {(config.meshcore_conn_type ?? 'tcp') === 'tcp' && ( +
+ upd({ meshcore_host: v })} + placeholder="192.168.1.100" + helper="IP or hostname of the companion frame server" + info="The MeshCore companion (frame server) host. Active when non-empty in TCP mode." + /> + upd({ meshcore_port: v })} + min={1} + max={65535} + helper="MeshCore TCP port (default 5525)" + /> +
+ )} + {(config.meshcore_conn_type ?? 'tcp') === 'serial' && ( + <> + upd({ meshcore_serial_port: v })} + helper="USB-attached MeshCore node — Detect fills a stable by-id path" + /> + upd({ meshcore_baud: v })} + min={1200} + helper="Serial baud rate (default 115200)" + /> + + )} + {(config.meshcore_conn_type ?? 'tcp') === 'ble' && ( upd({ meshcore_host: v })} - placeholder="192.168.1.100" - helper="IP or hostname — leave blank to disable MeshCore" - info="MeshCore is active when this field is non-empty." + label="BLE Address" + value={config.meshcore_ble_address ?? ''} + onChange={(v) => upd({ meshcore_ble_address: v })} + placeholder="AA:BB:CC:DD:EE:FF" + helper="Leave blank to scan/pair the first available device" /> - upd({ meshcore_port: v })} - min={1} - max={65535} - helper="MeshCore TCP port (default 5525)" - /> -
+ )}
FastAPI: from .api.alert_routes import router as alert_router from .api.notification_routes import router as notification_router from .api.debug_routes import router as debug_router + from .api.serial_ports_routes import router as serial_ports_router app.include_router(system_router, prefix="/api") + app.include_router(serial_ports_router, prefix="/api") app.include_router(adapter_config_router, prefix="/api") app.include_router(curation_router, prefix="/api") app.include_router(config_router, prefix="/api") diff --git a/work/meshai/serial_ports.py b/work/meshai/serial_ports.py new file mode 100644 index 0000000..ae741ab --- /dev/null +++ b/work/meshai/serial_ports.py @@ -0,0 +1,122 @@ +"""USB serial port scanner for MeshCore connection setup. + +Enumerates available serial ports, resolves stable by-id/by-path symlinks, +and flags likely radio devices by USB VID. +""" + +import os +import re + +from serial.tools.list_ports import comports + +# Module-level dir constants — tests monkeypatch these to tmp_path locations. +BY_ID_DIR = "/dev/serial/by-id" +BY_PATH_DIR = "/dev/serial/by-path" + +# USB VIDs for known mesh-radio hardware: +# 0x239A Adafruit/RAK nRF52840 +# 0x1915 Nordic Semiconductor +# 0x10C4 Silicon Labs CP210x (ESP32 bridge) +# 0x1A86 WCH CH340/CH9102 +# 0x55D4 WCH CH9102 (alternate VID) +RADIO_VIDS: frozenset[int] = frozenset({0x239A, 0x1915, 0x10C4, 0x1A86, 0x55D4}) + +# Pattern for ACM/USB tty devices (not legacy ttyS*) +_ACMUSB_RE = re.compile(r"tty(ACM|USB)\d") +# Pattern for legacy ttyS ports to exclude +_TTYS_RE = re.compile(r"ttyS\d") + + +def serial_by_id_available() -> bool: + """Return True when /dev/serial/by-id (or the module constant) exists.""" + return os.path.isdir(BY_ID_DIR) + + +def _build_symlink_map(dirpath: str) -> dict[str, str]: + """Return a mapping of realpath -> symlink path for all entries in dirpath. + + Returns an empty dict if the directory doesn't exist or can't be read. + """ + result: dict[str, str] = {} + try: + entries = os.listdir(dirpath) + except OSError: + return result + for entry in entries: + link = os.path.join(dirpath, entry) + try: + real = os.path.realpath(link) + result[real] = link + except OSError: + pass + return result + + +def list_serial_ports() -> list[dict]: + """Enumerate and return usable serial ports. + + Includes a port if it has a USB VID/PID OR its device name matches + /dev/tty(ACM|USB). Excludes /dev/ttyS* legacy ports (unless they carry + a vid/pid — edge case handled by checking vid presence first). + + Each entry dict: + device, by_id, by_path, description, hwid, vid, pid, + serial_number, manufacturer, product, likely_radio, stable_path + + Never raises — returns [] on unexpected error. + """ + try: + return _scan_ports() + except Exception: + return [] + + +def _scan_ports() -> list[dict]: + by_id_map = _build_symlink_map(BY_ID_DIR) + by_path_map = _build_symlink_map(BY_PATH_DIR) + + ports = [] + for p in comports(): + device: str = p.device or "" + vid: int | None = p.vid + pid: int | None = p.pid + basename = os.path.basename(device) + + # Exclude legacy ttyS* UNLESS it has a VID (unlikely but handled correctly) + if _TTYS_RE.match(basename) and vid is None: + continue + + # Include: has VID/PID or looks like ACM/USB tty + if vid is None and pid is None and not _ACMUSB_RE.match(basename): + continue + + # Resolve stable path via realpath comparison. + real_device = os.path.realpath(device) if device else device + by_id_link: str | None = by_id_map.get(real_device) + by_path_link: str | None = by_path_map.get(real_device) + + if by_id_link: + stable_path = by_id_link + elif by_path_link: + stable_path = by_path_link + else: + stable_path = device + + likely_radio: bool = vid in RADIO_VIDS if vid is not None else False + + ports.append({ + "device": device, + "by_id": by_id_link, + "by_path": by_path_link, + "description": getattr(p, "description", "") or "", + "hwid": getattr(p, "hwid", "") or "", + "vid": vid, + "pid": pid, + "serial_number": getattr(p, "serial_number", None), + "manufacturer": getattr(p, "manufacturer", None), + "product": getattr(p, "product", None), + "likely_radio": likely_radio, + "stable_path": stable_path, + }) + + return ports diff --git a/work/meshai/transport/factory.py b/work/meshai/transport/factory.py index b054e40..01132bf 100644 --- a/work/meshai/transport/factory.py +++ b/work/meshai/transport/factory.py @@ -3,6 +3,27 @@ from .base import MeshTransport +def meshcore_enabled(config) -> bool: + """Return True when the selected MeshCore connection mode is configured. + + Back-compat: an existing config with ``meshcore_host`` set and the default + ``meshcore_conn_type="tcp"`` correctly returns True (tcp branch). + + Mode rules: + - tcp → meshcore_host is non-empty + - serial → meshcore_serial_port is non-empty + - ble → meshcore_ble_address is non-empty (address optional for ble scan, + but we require it to be set to prevent accidental activation) + """ + conn_type = (getattr(config, "meshcore_conn_type", "tcp") or "tcp").strip() + if conn_type == "serial": + return bool((getattr(config, "meshcore_serial_port", "") or "").strip()) + if conn_type == "ble": + return bool((getattr(config, "meshcore_ble_address", "") or "").strip()) + # tcp (default) + return bool((getattr(config, "meshcore_host", "") or "").strip()) + + def build_transport(config, meshcore_context=None) -> MeshTransport: """Instantiate and return the active MeshTransport derived from config. @@ -10,9 +31,10 @@ def build_transport(config, meshcore_context=None) -> MeshTransport: separate mode flag: - Meshtastic is always the base transport. - - MeshCore is active when ``config.meshcore_host`` is a non-empty string. + - MeshCore is active when the selected mode's field is configured + (see ``meshcore_enabled``). - Both configured → CompositeTransport wrapping both. - - MeshCore host blank (default) → Meshtastic only. + - MeshCore not configured → Meshtastic only. Args: config: A ConnectionConfig (or duck-compatible object). @@ -25,8 +47,7 @@ def build_transport(config, meshcore_context=None) -> MeshTransport: from meshai.connector import MeshtasticTransport meshtastic = MeshtasticTransport(config) - meshcore_host = getattr(config, "meshcore_host", "") or "" - if meshcore_host.strip(): + if meshcore_enabled(config): from meshai.transport.meshcore_transport import MeshCoreTransport from meshai.transport.composite_transport import CompositeTransport return CompositeTransport( diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py index 1909c44..bf1e402 100644 --- a/work/meshai/transport/meshcore_transport.py +++ b/work/meshai/transport/meshcore_transport.py @@ -391,15 +391,36 @@ class MeshCoreTransport(MeshTransport): # 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.""" + async def _do_connect(self, conn_type: str, target: str, + auto_reconnect: bool, max_attempts: int, + host: str = "", port: int = 5050, + serial_port: str = "", baud: int = 115200, + ble_address: str = ""): + """Lazy-import meshcore and create the appropriate client. + + ``conn_type`` selects tcp | serial | ble. ``target`` is a human-readable + string used only for logging (built in connect()). + """ 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, - ) + if conn_type == "serial": + mc = await MeshCore.create_serial( + serial_port, + baudrate=baud, + auto_reconnect=auto_reconnect, + max_reconnect_attempts=max_attempts, + ) + elif conn_type == "ble": + mc = await MeshCore.create_ble( + address=(ble_address or None), + auto_reconnect=auto_reconnect, + max_reconnect_attempts=max_attempts, + ) + else: # tcp (default) + mc = await MeshCore.create_tcp( + host, port, + auto_reconnect=auto_reconnect, + max_reconnect_attempts=max_attempts, + ) return mc async def _setup_subscriptions(self) -> None: @@ -437,13 +458,25 @@ class MeshCoreTransport(MeshTransport): # ------------------------------------------------------------------ def connect(self) -> None: - """Connect to the pyMC companion TCP frame server.""" - host = getattr(self.config, "meshcore_host", "100.64.0.9") + """Connect to the MeshCore companion (tcp, serial, or ble).""" + conn_type = getattr(self.config, "meshcore_conn_type", "tcp") or "tcp" + host = getattr(self.config, "meshcore_host", "") port = getattr(self.config, "meshcore_port", 5050) + serial_port = getattr(self.config, "meshcore_serial_port", "") + baud = getattr(self.config, "meshcore_baud", 115200) + ble_address = getattr(self.config, "meshcore_ble_address", "") 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) + # Build a human-readable target string for logging. + if conn_type == "serial": + target = f"serial:{serial_port}@{baud}" + elif conn_type == "ble": + target = f"ble:{ble_address or 'auto'}" + else: + target = f"{host}:{port}" + + logger.info("MeshCoreTransport: connecting to %s …", target) # Start the dedicated event loop in a daemon thread. self._loop = asyncio.new_event_loop() @@ -464,7 +497,13 @@ class MeshCoreTransport(MeshTransport): try: mc = self._run_coro( - self._do_connect(host, port, auto_reconnect, max_attempts), + self._do_connect( + conn_type, target, + auto_reconnect, max_attempts, + host=host, port=port, + serial_port=serial_port, baud=baud, + ble_address=ble_address, + ), timeout=30.0, ) except Exception as exc: @@ -475,7 +514,7 @@ class MeshCoreTransport(MeshTransport): if mc is None: self._stop_loop() raise RuntimeError( - f"MeshCore.create_tcp({host}:{port}) returned None — connection failed" + f"MeshCore connect ({target}) returned None — connection failed" ) self._mc = mc diff --git a/work/tests/test_meshcore_conn_type.py b/work/tests/test_meshcore_conn_type.py new file mode 100644 index 0000000..e429dee --- /dev/null +++ b/work/tests/test_meshcore_conn_type.py @@ -0,0 +1,327 @@ +"""Tests for MeshCore multi-transport connection type support. + +Covers: + - ConnectionConfig validation of meshcore_conn_type + - meshcore_enabled() factory helper + - _do_connect() selects the right MeshCore.create_* per mode +""" + +import asyncio +import sys +import types +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +# --------------------------------------------------------------------------- +# Build a recording fake meshcore module +# --------------------------------------------------------------------------- + +def _build_recording_meshcore(): + """Build a fake meshcore module that records which create_* was called.""" + mod = types.ModuleType("meshcore") + + class EventType: + CONTACT_MSG_RECV = "CONTACT_MSG_RECV" + CHANNEL_MSG_RECV = "CHANNEL_MSG_RECV" + DISCONNECTED = "DISCONNECTED" + CONNECTED = "CONNECTED" + ACK = "ACK" + + mod.EventType = EventType + + class _FakeMeshCore: + # Shared call log: list of (method_name, args, kwargs) + _calls: list = [] + + 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 + + async def ensure_contacts(self, follow=False): + return True + + 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, + debug=False, only_error=False, default_timeout=None): + cls._calls.append(("create_tcp", (host, port), { + "auto_reconnect": auto_reconnect, + "max_reconnect_attempts": max_reconnect_attempts, + })) + return cls() + + @classmethod + async def create_serial(cls, port, baudrate=115200, + auto_reconnect=True, max_reconnect_attempts=5, + debug=False, only_error=False, default_timeout=None, + cx_dly=0.1): + cls._calls.append(("create_serial", (port,), { + "baudrate": baudrate, + "auto_reconnect": auto_reconnect, + "max_reconnect_attempts": max_reconnect_attempts, + })) + return cls() + + @classmethod + async def create_ble(cls, address=None, client=None, device=None, pin=None, + debug=False, only_error=False, default_timeout=None, + auto_reconnect=True, max_reconnect_attempts=5): + cls._calls.append(("create_ble", (), { + "address": address, + "auto_reconnect": auto_reconnect, + "max_reconnect_attempts": max_reconnect_attempts, + })) + 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 + + @staticmethod + async def send_advert(flood=False): + pass + + mod.MeshCore = _FakeMeshCore + return mod + + +# Register before production imports so lazy-import finds the mock. +# Use a unique key so we don't clobber test_meshcore_transport.py's module. +_fake_mc_mod = _build_recording_meshcore() +sys.modules.setdefault("meshcore", _fake_mc_mod) + + +# --------------------------------------------------------------------------- +# Production imports (after mock is registered) +# --------------------------------------------------------------------------- + +from meshai.config import ConnectionConfig # noqa: E402 +from meshai.transport.factory import build_transport, meshcore_enabled # noqa: E402 +from meshai.transport.meshcore_transport import MeshCoreTransport # noqa: E402 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _get_fake_mc(): + """Return the _FakeMeshCore class (reachable via sys.modules).""" + return sys.modules["meshcore"].MeshCore + + +def _clear_calls(): + _get_fake_mc()._calls.clear() + + +def _run(coro): + """Run a coroutine in a fresh event loop.""" + return asyncio.run(coro) + + +# --------------------------------------------------------------------------- +# Part 1: ConnectionConfig validation +# --------------------------------------------------------------------------- + +class TestConnectionConfigValidation: + def test_invalid_conn_type_raises(self): + with pytest.raises(ValueError, match="meshcore_conn_type"): + ConnectionConfig(meshcore_conn_type="bogus") + + def test_tcp_accepted(self): + cfg = ConnectionConfig(meshcore_conn_type="tcp") + assert cfg.meshcore_conn_type == "tcp" + + def test_serial_accepted(self): + cfg = ConnectionConfig(meshcore_conn_type="serial") + assert cfg.meshcore_conn_type == "serial" + + def test_ble_accepted(self): + cfg = ConnectionConfig(meshcore_conn_type="ble") + assert cfg.meshcore_conn_type == "ble" + + def test_default_is_tcp(self): + cfg = ConnectionConfig() + assert cfg.meshcore_conn_type == "tcp" + + def test_new_fields_have_defaults(self): + cfg = ConnectionConfig() + assert cfg.meshcore_serial_port == "" + assert cfg.meshcore_baud == 115200 + assert cfg.meshcore_ble_address == "" + + +# --------------------------------------------------------------------------- +# Part 2: meshcore_enabled() factory helper +# --------------------------------------------------------------------------- + +class TestMeshcoreEnabled: + def test_tcp_with_host_true(self): + cfg = ConnectionConfig(meshcore_conn_type="tcp", meshcore_host="127.0.0.1") + assert meshcore_enabled(cfg) is True + + def test_tcp_without_host_false(self): + cfg = ConnectionConfig(meshcore_conn_type="tcp", meshcore_host="") + assert meshcore_enabled(cfg) is False + + def test_serial_with_port_true(self): + cfg = ConnectionConfig(meshcore_conn_type="serial", meshcore_serial_port="/dev/ttyACM0") + assert meshcore_enabled(cfg) is True + + def test_serial_without_port_false(self): + cfg = ConnectionConfig(meshcore_conn_type="serial", meshcore_serial_port="") + assert meshcore_enabled(cfg) is False + + def test_serial_ignores_host(self): + """serial mode: having meshcore_host set does NOT enable MeshCore.""" + cfg = ConnectionConfig( + meshcore_conn_type="serial", + meshcore_host="127.0.0.1", + meshcore_serial_port="", + ) + assert meshcore_enabled(cfg) is False + + def test_ble_with_address_true(self): + cfg = ConnectionConfig(meshcore_conn_type="ble", meshcore_ble_address="AA:BB:CC:DD:EE:FF") + assert meshcore_enabled(cfg) is True + + def test_ble_without_address_false(self): + cfg = ConnectionConfig(meshcore_conn_type="ble", meshcore_ble_address="") + assert meshcore_enabled(cfg) is False + + def test_back_compat_tcp_default(self): + """Existing config: meshcore_host set, default conn_type='tcp' → True.""" + cfg = ConnectionConfig(meshcore_host="100.64.0.9") + # conn_type defaults to "tcp" + assert cfg.meshcore_conn_type == "tcp" + assert meshcore_enabled(cfg) is True + + def test_whitespace_only_host_false(self): + cfg = ConnectionConfig(meshcore_conn_type="tcp", meshcore_host=" ") + assert meshcore_enabled(cfg) is False + + +# --------------------------------------------------------------------------- +# Part 3: _do_connect selects correct create_* method +# --------------------------------------------------------------------------- + +class TestDoConnectSelection: + """Call _do_connect directly (it's just an async method) via asyncio.run.""" + + def setup_method(self): + _clear_calls() + + def _make_transport(self, **cfg_kwargs): + cfg = ConnectionConfig(**cfg_kwargs) + return MeshCoreTransport(cfg), cfg + + def test_tcp_calls_create_tcp(self): + t, cfg = self._make_transport( + meshcore_conn_type="tcp", + meshcore_host="127.0.0.1", + meshcore_port=5050, + meshcore_auto_reconnect=True, + meshcore_max_reconnect_attempts=3, + ) + _run(t._do_connect( + "tcp", "127.0.0.1:5050", + True, 3, + host="127.0.0.1", port=5050, + serial_port="", baud=115200, + ble_address="", + )) + calls = _get_fake_mc()._calls + assert len(calls) == 1 + name, args, kwargs = calls[0] + assert name == "create_tcp" + assert args == ("127.0.0.1", 5050) + assert kwargs["auto_reconnect"] is True + assert kwargs["max_reconnect_attempts"] == 3 + + def test_serial_calls_create_serial(self): + t, cfg = self._make_transport( + meshcore_conn_type="serial", + meshcore_serial_port="/dev/ttyACM0", + meshcore_baud=115200, + meshcore_auto_reconnect=True, + meshcore_max_reconnect_attempts=5, + ) + _run(t._do_connect( + "serial", "serial:/dev/ttyACM0@115200", + True, 5, + serial_port="/dev/ttyACM0", baud=115200, + )) + calls = _get_fake_mc()._calls + assert len(calls) == 1 + name, args, kwargs = calls[0] + assert name == "create_serial" + assert args == ("/dev/ttyACM0",) + assert kwargs["baudrate"] == 115200 + assert kwargs["auto_reconnect"] is True + assert kwargs["max_reconnect_attempts"] == 5 + + def test_serial_does_not_pass_cx_dly(self): + """create_serial should use default cx_dly (not passed explicitly).""" + t, _ = self._make_transport(meshcore_conn_type="serial") + _run(t._do_connect( + "serial", "serial:/dev/ttyACM0@115200", + True, 5, + serial_port="/dev/ttyACM0", baud=115200, + )) + calls = _get_fake_mc()._calls + assert "cx_dly" not in calls[0][2] + + def test_ble_calls_create_ble(self): + t, _ = self._make_transport( + meshcore_conn_type="ble", + meshcore_ble_address="AA:BB:CC:DD:EE:FF", + meshcore_auto_reconnect=False, + meshcore_max_reconnect_attempts=2, + ) + _run(t._do_connect( + "ble", "ble:AA:BB:CC:DD:EE:FF", + False, 2, + ble_address="AA:BB:CC:DD:EE:FF", + )) + calls = _get_fake_mc()._calls + assert len(calls) == 1 + name, args, kwargs = calls[0] + assert name == "create_ble" + assert kwargs["address"] == "AA:BB:CC:DD:EE:FF" + assert kwargs["auto_reconnect"] is False + assert kwargs["max_reconnect_attempts"] == 2 + + def test_ble_empty_address_passes_none(self): + """Empty ble_address → address=None (scan for any device).""" + t, _ = self._make_transport(meshcore_conn_type="ble") + _run(t._do_connect( + "ble", "ble:auto", + True, 5, + ble_address="", + )) + calls = _get_fake_mc()._calls + assert calls[0][2]["address"] is None diff --git a/work/tests/test_serial_ports.py b/work/tests/test_serial_ports.py new file mode 100644 index 0000000..eae6c87 --- /dev/null +++ b/work/tests/test_serial_ports.py @@ -0,0 +1,253 @@ +"""Tests for meshai.serial_ports — USB serial port scanner. + +All tests are hermetic: no real /dev access, no real pyserial comports call. +Fake comport objects use types.SimpleNamespace. +""" + +import os +import types + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _fake_port( + device, + vid=None, + pid=None, + description="Test Device", + hwid="USB", + serial_number="SN001", + manufacturer="Acme", + product="Widget", +): + return types.SimpleNamespace( + device=device, + vid=vid, + pid=pid, + description=description, + hwid=hwid, + serial_number=serial_number, + manufacturer=manufacturer, + product=product, + ) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def reset_module_cache(): + """Ensure meshai.serial_ports is re-imported fresh for each test when + BY_ID_DIR/BY_PATH_DIR are monkeypatched.""" + yield + + +# --------------------------------------------------------------------------- +# (a) stable_path resolves to by-id link when present +# --------------------------------------------------------------------------- + +def test_stable_path_by_id(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + # Create a fake device file and a by-id symlink pointing to it. + fake_device = tmp_path / "ttyACM0" + fake_device.write_text("") + + by_id_dir = tmp_path / "by-id" + by_id_dir.mkdir() + link = by_id_dir / "usb-RAK-nRF52840_ABC123-if00" + link.symlink_to(fake_device) + + monkeypatch.setattr(sp, "BY_ID_DIR", str(by_id_dir)) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "by-path-nonexistent")) + + port = _fake_port(str(fake_device), vid=0x239A, pid=0x0001) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["by_id"] == str(link) + assert result[0]["stable_path"] == str(link) + assert result[0]["by_path"] is None + + +# --------------------------------------------------------------------------- +# (b) falls back to by-path when only by-path matches +# --------------------------------------------------------------------------- + +def test_stable_path_by_path_fallback(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + fake_device = tmp_path / "ttyACM0" + fake_device.write_text("") + + by_path_dir = tmp_path / "by-path" + by_path_dir.mkdir() + link = by_path_dir / "platform-fd500000.pcie-pci-0000:01:00.0-usb-0:1.1:1.0" + link.symlink_to(fake_device) + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "by-id-nonexistent")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(by_path_dir)) + + port = _fake_port(str(fake_device), vid=0x239A, pid=0x0001) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["by_id"] is None + assert result[0]["by_path"] == str(link) + assert result[0]["stable_path"] == str(link) + + +# --------------------------------------------------------------------------- +# (c) falls back to raw device when neither dir has a match +# --------------------------------------------------------------------------- + +def test_stable_path_raw_device_fallback(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + fake_device = tmp_path / "ttyACM0" + fake_device.write_text("") + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "by-id-missing")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "by-path-missing")) + + port = _fake_port(str(fake_device), vid=0x239A, pid=0x0001) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["by_id"] is None + assert result[0]["by_path"] is None + assert result[0]["stable_path"] == str(fake_device) + + +# --------------------------------------------------------------------------- +# (d) likely_radio True for known VID, False for unknown +# --------------------------------------------------------------------------- + +def test_likely_radio_known_vid(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + fake_device = tmp_path / "ttyACM0" + fake_device.write_text("") + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "noid")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "nopath")) + + port = _fake_port(str(fake_device), vid=0x239A, pid=0x0001) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["likely_radio"] is True + + +def test_likely_radio_unknown_vid(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + fake_device = tmp_path / "ttyACM1" + fake_device.write_text("") + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "noid")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "nopath")) + + port = _fake_port(str(fake_device), vid=0x1234, pid=0x5678) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["likely_radio"] is False + + +# --------------------------------------------------------------------------- +# (e) ttyS* legacy port is excluded (no vid/pid) +# --------------------------------------------------------------------------- + +def test_ttys_legacy_excluded(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "noid")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "nopath")) + + legacy = _fake_port("/dev/ttyS0", vid=None, pid=None) + monkeypatch.setattr(sp, "comports", lambda: [legacy]) + + result = sp.list_serial_ports() + assert result == [] + + +def test_ttys_with_vid_included(tmp_path, monkeypatch): + """Edge case: a ttyS* port WITH a vid/pid should still be included.""" + import meshai.serial_ports as sp + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "noid")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "nopath")) + + # ttyS* + vid → NOT excluded by the legacy rule (the rule only applies when vid is None) + port = _fake_port("/dev/ttyS0", vid=0x10C4, pid=0xEA60) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + + +# --------------------------------------------------------------------------- +# (f) missing by-id / by-path dirs → no raise, stable_path == device +# --------------------------------------------------------------------------- + +def test_missing_serial_dirs_no_raise(tmp_path, monkeypatch): + import meshai.serial_ports as sp + + # Both dirs are nonexistent paths — should not raise. + monkeypatch.setattr(sp, "BY_ID_DIR", "/does/not/exist/by-id") + monkeypatch.setattr(sp, "BY_PATH_DIR", "/does/not/exist/by-path") + + port = _fake_port("/dev/ttyACM0", vid=0x1915, pid=0x520F) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["by_id"] is None + assert result[0]["by_path"] is None + assert result[0]["stable_path"] == "/dev/ttyACM0" + + +# --------------------------------------------------------------------------- +# Extra: multiple VIDs from RADIO_VIDS set +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("vid", [0x239A, 0x1915, 0x10C4, 0x1A86, 0x55D4]) +def test_all_radio_vids_flagged(tmp_path, monkeypatch, vid): + import meshai.serial_ports as sp + + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "noid")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "nopath")) + + port = _fake_port("/dev/ttyACM0", vid=vid, pid=0x0001) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["likely_radio"] is True + + +# --------------------------------------------------------------------------- +# serial_by_id_available +# --------------------------------------------------------------------------- + +def test_serial_by_id_available_true(tmp_path, monkeypatch): + import meshai.serial_ports as sp + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path)) + assert sp.serial_by_id_available() is True + + +def test_serial_by_id_available_false(monkeypatch): + import meshai.serial_ports as sp + monkeypatch.setattr(sp, "BY_ID_DIR", "/does/not/exist/by-id") + assert sp.serial_by_id_available() is False