feat: migrate openHop modem transport names

This commit is contained in:
yellowcooln committed 2026-08-20 09:40:09 -04:00
1 parent 2a55dd6034
commit 9511a8477f
13 files changed
+699 -190

No files matched your search

+2 -2
View File
@@ -65,8 +65,8 @@ openHop Repeater supports these radio backends:
- **SX1262 over Linux SPI**: set `radio_type: sx1262`
- **SX1262 over CH341 USB-to-SPI**: set `radio_type: sx1262_ch341`
- **pyMC TCP modem**: set `radio_type: pymc_tcp`
- **pyMC USB-CDC modem**: set `radio_type: pymc_usb`
- **openHop Modem over Wi-Fi/Ethernet**: set `radio_type: modem_tcp`
- **openHop Modem over USB-CDC**: set `radio_type: modem_usb`
- **KISS serial modem**: set `radio_type: kiss`
- **No radio hardware**: set `radio_type: null` for setup, testing, or API-only work
+10 -10
View File
@@ -320,10 +320,10 @@ sensors:
# bus_number: 1 # I2C bus number (1 for Raspberry Pi default)
# low_cell_mv: 3150 # Per-cell warning threshold in mV
# Example pyMC modem sensor. This reads a networked pyMC modem's
# Example openHop Modem sensor. This reads a networked openHop Modem's
# HTTP /api/stats JSON API for modem diagnostics, including GPS fields
# when the modem exposes them. If password is set to the same value as
# pymc_tcp.token, the repeater can use one shared modem secret.
# modem_tcp.token, the repeater can use one shared modem secret.
# - type: pymc_modem
# name: modem
# enabled: true
@@ -332,7 +332,7 @@ sensors:
# port: 80
# endpoint: "/api/stats"
# username: "admin"
# password: "" # modem HTTP password; may match pymc_tcp.token
# password: "" # modem HTTP password; may match modem_tcp.token
# poll_interval_seconds: 60.0 # avoid contending with modem traffic
# timeout_seconds: 2.0
@@ -486,8 +486,8 @@ identities:
# - sx1262 (Linux spidev + system GPIO)
# - sx1262_ch341 (CH341 USB-to-SPI + CH341 GPIO 0-7)
# - kiss (KISS-modem over a serial port; alias: kiss-modem)
# - pymc_tcp (pymc_usb firmware modem over Wi-Fi/TCP)
# - pymc_usb (pymc_usb firmware modem over USB-CDC)
# - modem_tcp (openHop Modem over Wi-Fi/Ethernet TCP)
# - modem_usb (openHop Modem over USB-CDC)
# - null/none (disable radio hardware; daemon starts without RF I/O)
radio_type: null
@@ -544,19 +544,19 @@ radio:
# # kiss_txtail_ms: 0 # tail after TX (rarely needed)
# # kiss_full_duplex: false # disable carrier-sense/CSMA entirely (not recommended)
# pymc_usb firmware modem over Wi-Fi/TCP (when radio_type: pymc_tcp).
# openHop Modem over Wi-Fi/Ethernet TCP (when radio_type: modem_tcp).
# Requires openhop-core with the TCPLoRaRadio driver
# pymc_tcp:
# host: "pymc-3e2834.local" # modem hostname / mDNS name / LAN IP
# modem_tcp:
# host: "openhop-modem.local" # modem hostname / mDNS name / LAN IP
# port: 5055 # firmware default
# token: "" # empty = no auth; set to match firmware NVS token
# connect_timeout: 5.0 # seconds; deferred-connect retries with backoff on failure
# lbt_enabled: true # Listen-Before-Talk via firmware CAD
# lbt_max_attempts: 5
# pymc_usb firmware modem over USB-CDC (when radio_type: pymc_usb).
# openHop Modem over USB-CDC (when radio_type: modem_usb).
# Requires openhop-core with the USBLoRaRadio driver
# pymc_usb:
# modem_usb:
# port: "/dev/ttyACM0" # USB-CDC device; udev rule may symlink to /dev/lora-modem
# baudrate: 921600 # must match firmware monitor_speed
# lbt_enabled: true
+8 -8
View File
@@ -440,19 +440,19 @@
"dio3_tcxo_voltage": 1.8,
"preamble_length": 32
},
"pymc_usb": {
"name": "pymc_usb modem (USB-CDC)",
"description": "ESP32-S3 / nRF52 board running pymc_usb firmware as a USB-CDC LoRa modem. Pick this if the modem is plugged into the host's USB port (e.g. /dev/ttyACM0). Edit the 'pymc_usb' section after first-run to point at the right serial device.",
"modem_usb": {
"name": "openHop Modem (USB-CDC)",
"description": "ESP32-S3 / nRF52 openHop Modem connected over USB-CDC. Pick this if the modem is plugged into the host's USB port (e.g. /dev/ttyACM0). Edit the 'modem_usb' section after first-run to point at the right serial device.",
"connection_type": "usb",
"radio_type": "pymc_usb",
"radio_type": "modem_usb",
"tx_power": 22,
"preamble_length": 32
},
"pymc_tcp": {
"name": "pymc_tcp modem (Wi-Fi / Ethernet)",
"description": "ESP32 board running pymc_usb firmware exposed as a TCP server over Wi-Fi or Ethernet. After first-run, set 'pymc_tcp.host' to the modem's LAN address or mDNS name (e.g. pymc-3e2834.local).",
"modem_tcp": {
"name": "openHop Modem (Wi-Fi / Ethernet)",
"description": "openHop Modem exposed as a TCP server over Wi-Fi or Ethernet. After first-run, set 'modem_tcp.host' to the modem's LAN address or mDNS name (e.g. openhop-modem.local).",
"connection_type": "network",
"radio_type": "pymc_tcp",
"radio_type": "modem_tcp",
"tx_power": 22,
"preamble_length": 32
},
+19 -15
View File
@@ -7,6 +7,7 @@ from typing import Any, Dict, Optional, overload
import yaml
from repeater.exceptions import ConfigurationError
from repeater.modem_config import normalize_modem_config
from repeater.policy_engine import default_policy_engine_config
logger = logging.getLogger("Config")
@@ -214,6 +215,7 @@ def load_config(config_path: Optional[str] = None) -> Dict[str, Any]:
try:
with open(config_path) as f:
config = yaml.safe_load(f) or {}
config = normalize_modem_config(config)
logger.info(f"Loaded config from {config_path}")
except Exception as e:
raise ConfigurationError(f"Failed to load configuration from {config_path}: {e}") from e
@@ -341,6 +343,7 @@ def save_config(config_data: Dict[str, Any], config_path: Optional[str] = None)
)
try:
config_data = normalize_modem_config(config_data)
# Create backup of existing config
config_file = Path(config_path)
if config_file.exists():
@@ -449,6 +452,7 @@ def _load_or_create_identity_key(path: Optional[str] = None) -> bytes:
def get_radio_for_board(board_config: dict):
board_config = normalize_modem_config(board_config)
@overload
def _parse_int(value, *, default: None = None) -> Optional[int]: ...
@@ -661,25 +665,25 @@ def get_radio_for_board(board_config: dict):
return radio
elif radio_type == "pymc_tcp":
elif radio_type == "modem_tcp":
try:
from openhop_core.hardware.tcp_radio import TCPLoRaRadio
except ImportError:
raise RuntimeError(
"pymc_tcp radio requires openhop-core >= the release that includes "
"modem_tcp radio requires openhop-core >= the release that includes "
"the openhop-core release with TCP modem support. "
"Reinstall the [hardware] extra to pick it up."
) from None
tcp_cfg = board_config.get("pymc_tcp")
tcp_cfg = board_config.get("modem_tcp")
if not tcp_cfg:
raise ValueError(
"Missing 'pymc_tcp' section in configuration file for radio_type: pymc_tcp"
"Missing 'modem_tcp' section in configuration file for radio_type: modem_tcp"
)
host = tcp_cfg.get("host")
if not host:
raise ValueError("Missing 'host' in 'pymc_tcp' section (modem hostname or LAN IP)")
raise ValueError("Missing 'host' in 'modem_tcp' section (modem hostname or LAN IP)")
radio_cfg = board_config.get("radio") or {}
radio = TCPLoRaRadio(
@@ -701,29 +705,29 @@ def get_radio_for_board(board_config: dict):
try:
radio.begin()
except Exception as e:
raise RuntimeError(f"Failed to initialize pymc_tcp radio: {e}") from e
raise RuntimeError(f"Failed to initialize modem_tcp radio: {e}") from e
return BaselineCrcCounterRadio(radio)
elif radio_type == "pymc_usb":
elif radio_type == "modem_usb":
try:
from openhop_core.hardware.usb_radio import USBLoRaRadio
except ImportError:
raise RuntimeError(
"pymc_usb radio requires openhop-core >= the release that includes "
"modem_usb radio requires openhop-core >= the release that includes "
"the openhop-core release with TCP modem support. "
"Reinstall the [hardware] extra to pick it up."
) from None
usb_cfg = board_config.get("pymc_usb")
usb_cfg = board_config.get("modem_usb")
if not usb_cfg:
raise ValueError(
"Missing 'pymc_usb' section in configuration file for radio_type: pymc_usb"
"Missing 'modem_usb' section in configuration file for radio_type: modem_usb"
)
port = usb_cfg.get("port")
if not port:
raise ValueError("Missing 'port' in 'pymc_usb' section (e.g. /dev/ttyACM0)")
raise ValueError("Missing 'port' in 'modem_usb' section (e.g. /dev/ttyACM0)")
radio_cfg = board_config.get("radio") or {}
radio = USBLoRaRadio(
@@ -743,13 +747,13 @@ def get_radio_for_board(board_config: dict):
try:
radio.begin()
except Exception as e:
raise RuntimeError(f"Failed to initialize pymc_usb radio: {e}") from e
raise RuntimeError(f"Failed to initialize modem_usb radio: {e}") from e
return BaselineCrcCounterRadio(radio)
raise RuntimeError(
f"Unknown radio type: {radio_type}. "
"Supported: sx1262, sx1262_ch341, kiss (or kiss-modem), pymc_tcp, pymc_usb"
"Supported: sx1262, sx1262_ch341, kiss (or kiss-modem), modem_tcp, modem_usb"
)
@@ -773,8 +777,8 @@ def _merge_radio_entry(global_config: dict, entry: dict) -> dict:
"sx1262",
"ch341",
"kiss",
"pymc_tcp",
"pymc_usb",
"modem_tcp",
"modem_usb",
):
if key in entry:
merged[key] = entry[key]
+2
View File
@@ -5,6 +5,7 @@ from typing import Any, Dict, List, Optional
import yaml
from repeater.logging_utils import normalize_log_level
from repeater.modem_config import normalize_modem_config_in_place
logger = logging.getLogger("ConfigManager")
@@ -242,6 +243,7 @@ class ConfigManager:
True if successful, False otherwise
"""
try:
normalize_modem_config_in_place(self.config)
os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
with open(self.config_path, "w") as f:
# Use safe_dump with explicit width to prevent line wrapping
+6 -4
View File
@@ -33,6 +33,7 @@ from openhop_core.protocol.packet_utils import (
from repeater.airtime import AirtimeManager
from repeater.data_acquisition import StorageCollector
from repeater.modem_config import normalize_modem_config
from repeater.neighbour_links import NeighbourLinkTracker
from repeater.policy_engine import PolicyDecision, PolicyEngine
@@ -1659,6 +1660,7 @@ class RepeaterHandler(BaseHandler):
return self._cached_noise_floor
def get_stats(self) -> dict:
runtime_config = normalize_modem_config(self.config, warn=False)
uptime_seconds = time.time() - self.start_time
@@ -1737,14 +1739,14 @@ class RepeaterHandler(BaseHandler):
"radio": self.config.get(
"radio", {}
), # Read from live config, not cached radio_config
"radios": self.config.get("radios") or [],
"radios": runtime_config.get("radios") or [],
"fabric": self.config.get("fabric") or {},
"radio_type": self.config.get("radio_type"),
"radio_type": runtime_config.get("radio_type"),
"sx1262": self.config.get("sx1262", {}),
"ch341": self.config.get("ch341", {}),
"kiss": self.config.get("kiss", {}),
"pymc_usb": self.config.get("pymc_usb", {}),
"pymc_tcp": self.config.get("pymc_tcp", {}),
"modem_usb": runtime_config.get("modem_usb", {}),
"modem_tcp": runtime_config.get("modem_tcp", {}),
"duty_cycle": {
"max_airtime_percent": max_duty_cycle_percent,
"enforcement_enabled": duty_cycle_config.get("enforcement_enabled", True),
+60
View File
@@ -0,0 +1,60 @@
"""Compatibility boundary for openHop Modem transport configuration names."""
from __future__ import annotations
import copy
import logging
from typing import Any
logger = logging.getLogger(__name__)
LEGACY_MODEM_RADIO_TYPES = {
"pymc_tcp": "modem_tcp",
"pymc_usb": "modem_usb",
}
def _normalize_board_config(config: dict[str, Any], *, warn: bool) -> None:
radio_type = config.get("radio_type")
if isinstance(radio_type, str):
normalized_type = radio_type.lower().strip()
if normalized_type in LEGACY_MODEM_RADIO_TYPES:
config["radio_type"] = LEGACY_MODEM_RADIO_TYPES[normalized_type]
for legacy_key, canonical_key in LEGACY_MODEM_RADIO_TYPES.items():
if legacy_key not in config:
continue
if canonical_key in config:
if warn:
logger.warning(
"Both legacy configuration key '%s' and canonical key '%s' are present; canonical key wins",
legacy_key,
canonical_key,
)
else:
config[canonical_key] = config[legacy_key]
del config[legacy_key]
def normalize_modem_config_in_place(config: dict, *, warn: bool = True) -> dict:
"""Canonicalize modem transport keys in place without replacing unrelated objects."""
if not isinstance(config, dict):
return config
_normalize_board_config(config, warn=warn)
radios = config.get("radios")
if isinstance(radios, list):
for entry in radios:
if isinstance(entry, dict):
_normalize_board_config(entry, warn=warn)
return config
def normalize_modem_config(config: dict, *, warn: bool = True) -> dict:
"""Return a deep-copied canonical runtime config; canonical values win conflicts."""
if not isinstance(config, dict):
return config
normalized = copy.deepcopy(config)
normalize_modem_config_in_place(normalized, warn=warn)
return normalized
+161 -68
View File
@@ -26,6 +26,7 @@ from repeater.companion.utils import (
validate_companion_config_capacity,
)
from repeater.config import resolve_storage_dir
from repeater.modem_config import LEGACY_MODEM_RADIO_TYPES, normalize_modem_config
from repeater.policy_engine import PolicyEngine
from repeater.service_utils import get_buildroot_image_info
from repeater.utils_packet import create_scoped_advert_packet
@@ -1431,17 +1432,40 @@ class APIEndpoints:
for dev in glob.glob(pattern):
devices.append({"device": str(dev), "description": str(dev)})
# De-duplicate by device path while preserving the first description.
# Docker deployments may mount the host device namespace separately.
import glob
if os.path.isdir("/host/dev"):
for pattern in (
"/host/dev/ttyACM*",
"/host/dev/ttyUSB*",
"/host/dev/serial/by-id/*",
):
try:
for dev in glob.glob(pattern):
devices.append({"device": str(dev), "description": str(dev)})
except (OSError, PermissionError):
continue
# De-duplicate by resolved target. Prefer stable host by-id aliases.
dedup = {}
for item in devices:
dev = item.get("device")
if dev and dev not in dedup:
if not dev:
continue
try:
target = os.path.realpath(dev)
except (OSError, PermissionError):
continue
preferred = str(dev).startswith("/host/dev/serial/by-id/")
existing = dedup.get(target)
if existing is None or preferred:
# Docker reads `devices:` entries as host:container[:perms] and
# cannot escape a colon inside the path, so a /dev/serial/by-id/
# name carrying an ESP32-S3 MAC serial is unusable in a
# container. Flag it so the UI can offer a symlink instead.
item["docker_safe"] = ":" not in dev
dedup[dev] = item
dedup[target] = item
sorted_devices = sorted(dedup.values(), key=lambda x: x["device"])
return self._success(sorted_devices)
@@ -1483,6 +1507,7 @@ class APIEndpoints:
return {"success": False, "error": "Node name too long (max 31 bytes in UTF-8)"}
hardware_key = data.get("hardware_key", "").strip()
hardware_key = LEGACY_MODEM_RADIO_TYPES.get(hardware_key, hardware_key)
if not hardware_key:
return {"success": False, "error": "Hardware selection is required"}
@@ -1564,47 +1589,58 @@ class APIEndpoints:
)
if "preamble_length" not in config_yaml["radio"]:
config_yaml["radio"]["preamble_length"] = 32
elif hardware_key == "pymc_usb":
# pymc_usb modem: external SX1262 board over USB-CDC.
# Accept pymc_usb_port / pymc_usb_baudrate from the request body
elif hardware_key == "modem_usb":
# modem_usb modem: external SX1262 board over USB-CDC.
# Accept modem_usb_port / modem_usb_baudrate from the request body
# (mirrors the KISS pattern) so a future SPA can expose inputs;
# fall back to /dev/ttyACM0 at 921600 baud, which matches the
# firmware default and the typical USB-CDC modem device on Linux.
config_yaml["radio_type"] = "pymc_usb"
usb_port = (data.get("pymc_usb_port") or "").strip() or "/dev/ttyACM0"
usb_baud = int(data.get("pymc_usb_baudrate", data.get("pymc_usb_baud", 921600)))
pymc_usb_section = config_yaml.setdefault("pymc_usb", {})
pymc_usb_section["port"] = usb_port
pymc_usb_section["baudrate"] = usb_baud
pymc_usb_section.setdefault("lbt_enabled", True)
pymc_usb_section.setdefault("lbt_max_attempts", 5)
config_yaml["radio_type"] = "modem_usb"
usb_port = (
data.get("modem_usb_port") or data.get("pymc_usb_port") or ""
).strip() or "/dev/ttyACM0"
usb_baud = int(
data.get(
"modem_usb_baudrate",
data.get("pymc_usb_baudrate", data.get("pymc_usb_baud", 921600)),
)
)
modem_usb_section = config_yaml.setdefault("modem_usb", {})
modem_usb_section["port"] = usb_port
modem_usb_section["baudrate"] = usb_baud
modem_usb_section.setdefault("lbt_enabled", True)
modem_usb_section.setdefault("lbt_max_attempts", 5)
if tx_power_preset is not None:
config_yaml["radio"]["tx_power"] = tx_power_preset
elif "tx_power" in hw_config:
config_yaml["radio"]["tx_power"] = hw_config.get("tx_power", 22)
if "preamble_length" in hw_config:
config_yaml["radio"]["preamble_length"] = hw_config.get("preamble_length", 32)
elif hardware_key == "pymc_tcp":
# pymc_tcp modem: external SX1262 board exposed as TCP over Wi-Fi/Ethernet.
elif hardware_key == "modem_tcp":
# modem_tcp modem: external SX1262 board exposed as TCP over Wi-Fi/Ethernet.
# 'host' has no sensible default — must be the modem's LAN address or
# mDNS name. Accept it from the request body if the SPA provides it,
# otherwise write a clearly-placeholder hostname so the file is valid
# YAML and the user gets a startup error pointing them at the right
# section to edit (see config.py: ValueError 'Missing host …').
config_yaml["radio_type"] = "pymc_tcp"
tcp_host = (data.get("pymc_tcp_host") or "").strip() or "REPLACE_WITH_MODEM_HOST"
tcp_port = int(data.get("pymc_tcp_port", 5055))
pymc_tcp_section = config_yaml.setdefault("pymc_tcp", {})
pymc_tcp_section["host"] = tcp_host
pymc_tcp_section["port"] = tcp_port
tcp_token = data.get("pymc_tcp_token")
config_yaml["radio_type"] = "modem_tcp"
tcp_host = (
data.get("modem_tcp_host") or data.get("pymc_tcp_host") or ""
).strip() or "REPLACE_WITH_MODEM_HOST"
tcp_port = int(data.get("modem_tcp_port", data.get("pymc_tcp_port", 5055)))
modem_tcp_section = config_yaml.setdefault("modem_tcp", {})
modem_tcp_section["host"] = tcp_host
modem_tcp_section["port"] = tcp_port
tcp_token = data.get("modem_tcp_token")
if tcp_token is None:
tcp_token = data.get("pymc_tcp_token")
if tcp_token is not None:
pymc_tcp_section["token"] = str(tcp_token)
modem_tcp_section["token"] = str(tcp_token)
else:
pymc_tcp_section.setdefault("token", "")
pymc_tcp_section.setdefault("connect_timeout", 5.0)
pymc_tcp_section.setdefault("lbt_enabled", True)
pymc_tcp_section.setdefault("lbt_max_attempts", 5)
modem_tcp_section.setdefault("token", "")
modem_tcp_section.setdefault("connect_timeout", 5.0)
modem_tcp_section.setdefault("lbt_enabled", True)
modem_tcp_section.setdefault("lbt_max_attempts", 5)
if tx_power_preset is not None:
config_yaml["radio"]["tx_power"] = tx_power_preset
elif "tx_power" in hw_config:
@@ -1680,7 +1716,8 @@ class APIEndpoints:
config_yaml["sx1262"]["use_gpiod_backend"] = hw_config.get(
"use_gpiod_backend", False
)
# Write updated config
# Explicit setup writes canonical names only.
config_yaml = normalize_modem_config(config_yaml)
with open(self._config_path, "w") as f:
yaml.dump(config_yaml, f, default_flow_style=False, sort_keys=False)
@@ -1718,14 +1755,14 @@ class APIEndpoints:
if hardware_key == "kiss":
result_config["kiss_port"] = config_yaml.get("kiss", {}).get("port")
result_config["kiss_baud_rate"] = config_yaml.get("kiss", {}).get("baud_rate")
elif hardware_key == "pymc_usb":
pymc_usb_cfg = config_yaml.get("pymc_usb", {})
result_config["pymc_usb_port"] = pymc_usb_cfg.get("port")
result_config["pymc_usb_baudrate"] = pymc_usb_cfg.get("baudrate")
elif hardware_key == "pymc_tcp":
pymc_tcp_cfg = config_yaml.get("pymc_tcp", {})
result_config["pymc_tcp_host"] = pymc_tcp_cfg.get("host")
result_config["pymc_tcp_port"] = pymc_tcp_cfg.get("port")
elif hardware_key == "modem_usb":
modem_usb_cfg = config_yaml.get("modem_usb", {})
result_config["modem_usb_port"] = modem_usb_cfg.get("port")
result_config["modem_usb_baudrate"] = modem_usb_cfg.get("baudrate")
elif hardware_key == "modem_tcp":
modem_tcp_cfg = config_yaml.get("modem_tcp", {})
result_config["modem_tcp_host"] = modem_tcp_cfg.get("host")
result_config["modem_tcp_port"] = modem_tcp_cfg.get("port")
# token deliberately omitted from response (sensitive)
return {
"success": True,
@@ -1748,15 +1785,16 @@ class APIEndpoints:
def stats(self):
try:
stats = self.stats_getter() if self.stats_getter else {}
runtime_config = normalize_modem_config(self.config, warn=False)
# Include active radio configuration in stats so UI can hydrate
# directly from this endpoint without additional config fetches.
stats["radio_type"] = self.config.get("radio_type")
stats["sx1262"] = self.config.get("sx1262", {})
stats["ch341"] = self.config.get("ch341", {})
stats["kiss"] = self.config.get("kiss", {})
stats["pymc_usb"] = self.config.get("pymc_usb", {})
stats["pymc_tcp"] = self.config.get("pymc_tcp", {})
stats["radios"] = self.config.get("radios") or []
stats["radio_type"] = runtime_config.get("radio_type")
stats["sx1262"] = runtime_config.get("sx1262", {})
stats["ch341"] = runtime_config.get("ch341", {})
stats["kiss"] = runtime_config.get("kiss", {})
stats["modem_usb"] = runtime_config.get("modem_usb", {})
stats["modem_tcp"] = runtime_config.get("modem_tcp", {})
stats["radios"] = runtime_config.get("radios") or []
stats["fabric"] = self.config.get("fabric") or {}
meta = {}
daemon = getattr(self, "daemon_instance", None)
@@ -2982,6 +3020,7 @@ class APIEndpoints:
config_yaml = None
if isinstance(config_yaml, dict):
config_yaml = normalize_modem_config(config_yaml)
repeater = config_yaml.get("repeater")
if not isinstance(repeater, dict):
add_error("repeater", "Missing required section 'repeater'")
@@ -3014,8 +3053,8 @@ class APIEndpoints:
"sx1262",
"sx1262_ch341",
"kiss",
"pymc_tcp",
"pymc_usb",
"modem_tcp",
"modem_usb",
"none",
"null",
"disabled",
@@ -3026,7 +3065,7 @@ class APIEndpoints:
if radio_type not in known_radio_types:
add_error(
"radio_type",
"Unsupported radio_type. Supported: sx1262, sx1262_ch341, kiss, pymc_tcp, pymc_usb, none/null",
"Unsupported radio_type. Supported: sx1262, sx1262_ch341, kiss, modem_tcp, modem_usb, none/null",
)
radio_disabled = radio_type in ("", "none", "null", "disabled", "off", "no_radio")
@@ -3159,45 +3198,45 @@ class APIEndpoints:
elif baud <= 0:
add_error("kiss.baud_rate", "KISS baud_rate must be greater than zero")
if radio_type == "pymc_usb":
usb_cfg = config_yaml.get("pymc_usb")
if radio_type == "modem_usb":
usb_cfg = config_yaml.get("modem_usb")
if not isinstance(usb_cfg, dict):
add_error(
"pymc_usb",
"Missing required section 'pymc_usb' for radio_type pymc_usb",
"modem_usb",
"Missing required section 'modem_usb' for radio_type modem_usb",
)
usb_cfg = {}
port = (usb_cfg.get("port") if isinstance(usb_cfg, dict) else "") or ""
if not str(port).strip():
add_error("pymc_usb.port", "pymc_usb.port is required")
baud = as_int((usb_cfg or {}).get("baudrate"), "pymc_usb.baudrate")
add_error("modem_usb.port", "modem_usb.port is required")
baud = as_int((usb_cfg or {}).get("baudrate"), "modem_usb.baudrate")
if baud is not None and baud <= 0:
add_error(
"pymc_usb.baudrate", "pymc_usb.baudrate must be greater than zero"
"modem_usb.baudrate", "modem_usb.baudrate must be greater than zero"
)
if radio_type == "pymc_tcp":
tcp_cfg = config_yaml.get("pymc_tcp")
if radio_type == "modem_tcp":
tcp_cfg = config_yaml.get("modem_tcp")
if not isinstance(tcp_cfg, dict):
add_error(
"pymc_tcp",
"Missing required section 'pymc_tcp' for radio_type pymc_tcp",
"modem_tcp",
"Missing required section 'modem_tcp' for radio_type modem_tcp",
)
tcp_cfg = {}
host = (tcp_cfg.get("host") if isinstance(tcp_cfg, dict) else "") or ""
host_str = str(host).strip()
if not host_str:
add_error("pymc_tcp.host", "pymc_tcp.host is required")
add_error("modem_tcp.host", "modem_tcp.host is required")
elif host_str == "REPLACE_WITH_MODEM_HOST":
add_error(
"pymc_tcp.host",
"modem_tcp.host",
"Replace placeholder host with your modem hostname or IP",
)
port = as_int((tcp_cfg or {}).get("port"), "pymc_tcp.port")
port = as_int((tcp_cfg or {}).get("port"), "modem_tcp.port")
if port is None:
add_error("pymc_tcp.port", "pymc_tcp.port is required")
add_error("modem_tcp.port", "modem_tcp.port is required")
elif port < 1 or port > 65535:
add_error("pymc_tcp.port", "pymc_tcp.port must be 1-65535")
add_error("modem_tcp.port", "modem_tcp.port must be 1-65535")
if radio_disabled:
add_warning("radio_type", "Radio is disabled (radio_type none/null/off)")
@@ -7756,7 +7795,7 @@ class APIEndpoints:
import copy
full_backup = str(include_secrets).lower() in ("true", "1", "yes")
exported = copy.deepcopy(self.config)
exported = normalize_modem_config(copy.deepcopy(self.config))
if full_backup:
# Convert binary identity key to hex for JSON serialisation
@@ -7782,6 +7821,16 @@ class APIEndpoints:
if "identity_key" in rep:
del rep["identity_key"]
tcp_cfg = exported.get("modem_tcp")
if isinstance(tcp_cfg, dict) and "token" in tcp_cfg:
tcp_cfg["token"] = "*** REDACTED ***"
for radio_entry in exported.get("radios") or []:
if not isinstance(radio_entry, dict):
continue
nested_tcp_cfg = radio_entry.get("modem_tcp")
if isinstance(nested_tcp_cfg, dict) and "token" in nested_tcp_cfg:
nested_tcp_cfg["token"] = "*** REDACTED ***"
# Redact identity keys in companion / room_server configs
for section in ("room_servers", "companions"):
entries = exported.get("identities", {}).get(section, []) or []
@@ -7874,6 +7923,50 @@ class APIEndpoints:
if not imported_config or not isinstance(imported_config, dict):
return self._error("Missing or invalid 'config' object in request body")
imported_config = normalize_modem_config(imported_config)
# Redacted exports are intentionally safe to re-import. Preserve the
# currently configured modem token instead of saving the sentinel.
current_config = normalize_modem_config(self.config, warn=False)
imported_tcp = imported_config.get("modem_tcp")
current_tcp = current_config.get("modem_tcp")
if isinstance(imported_tcp, dict) and imported_tcp.get("token") == "*** REDACTED ***":
if isinstance(current_tcp, dict) and "token" in current_tcp:
imported_tcp["token"] = current_tcp["token"]
else:
imported_tcp.pop("token", None)
current_radios = current_config.get("radios")
if not isinstance(current_radios, list):
current_radios = []
current_radios_by_id = {
entry.get("id"): entry
for entry in current_radios
if isinstance(entry, dict) and entry.get("id") is not None
}
imported_radios = imported_config.get("radios")
if isinstance(imported_radios, list):
for index, entry in enumerate(imported_radios):
if not isinstance(entry, dict):
continue
imported_radio_tcp = entry.get("modem_tcp")
if not (
isinstance(imported_radio_tcp, dict)
and imported_radio_tcp.get("token") == "*** REDACTED ***"
):
continue
current_entry = current_radios_by_id.get(entry.get("id"))
if current_entry is None and index < len(current_radios):
candidate = current_radios[index]
current_entry = candidate if isinstance(candidate, dict) else None
current_radio_tcp = (
current_entry.get("modem_tcp") if isinstance(current_entry, dict) else None
)
if isinstance(current_radio_tcp, dict) and "token" in current_radio_tcp:
imported_radio_tcp["token"] = current_radio_tcp["token"]
else:
imported_radio_tcp.pop("token", None)
# Sections we allow to be imported
ALLOWED_SECTIONS = {
"repeater",
@@ -7884,8 +7977,8 @@ class APIEndpoints:
"sx1262",
"ch341",
"kiss",
"pymc_usb",
"pymc_tcp",
"modem_usb",
"modem_tcp",
"mqtt_brokers",
"mqtt",
"identities",
@@ -7948,8 +8041,8 @@ class APIEndpoints:
"sx1262",
"ch341",
"kiss",
"pymc_usb",
"pymc_tcp",
"modem_usb",
"modem_tcp",
"radio_type",
"mqtt_brokers",
"letsmesh",
+139 -7
View File
@@ -430,7 +430,7 @@ def test_hardware_options_loads_installed_file(tmp_path):
hardware_file = tmp_path / "radio-settings.json"
hardware_file.write_text(
'{"hardware":{"pymc_usb":{"name":"USB","description":"desc","radio_type":"pymc_usb"}}}',
'{"hardware":{"modem_usb":{"name":"openHop Modem (USB-CDC)","description":"desc","radio_type":"modem_usb"}}}',
encoding="utf-8",
)
@@ -438,8 +438,22 @@ def test_hardware_options_loads_installed_file(tmp_path):
result = api.hardware_options()
assert len(result["hardware"]) == 1
assert result["hardware"][0]["key"] == "pymc_usb"
assert result["hardware"][0]["name"] == "USB"
assert result["hardware"][0]["key"] == "modem_usb"
assert result["hardware"][0]["name"] == "openHop Modem (USB-CDC)"
def test_repository_hardware_options_use_canonical_modem_names():
import json
settings = json.loads(
(Path(__file__).parents[1] / "radio-settings.json").read_text(encoding="utf-8")
)["hardware"]
assert "modem_usb" in settings and "modem_tcp" in settings
assert "pymc_usb" not in settings and "pymc_tcp" not in settings
assert settings["modem_usb"]["radio_type"] == "modem_usb"
assert settings["modem_usb"]["name"] == "openHop Modem (USB-CDC)"
assert settings["modem_tcp"]["name"] == "openHop Modem (Wi-Fi / Ethernet)"
def test_radio_presets_returns_error_when_file_missing(tmp_path):
@@ -568,6 +582,48 @@ def test_serial_ports_marks_ordinary_paths_docker_safe(cherrypy_ctx):
assert result["data"][0]["docker_safe"] is True
def test_serial_ports_scans_host_dev_and_prefers_by_id_alias(cherrypy_ctx):
del cherrypy_ctx
api = _make_api()
numbered = "/host/dev/ttyACM0"
by_id = "/host/dev/serial/by-id/openhop-modem"
def fake_glob(pattern):
return {
"/host/dev/ttyACM*": [numbered],
"/host/dev/ttyUSB*": [],
"/host/dev/serial/by-id/*": [by_id],
}.get(pattern, [])
with (
patch("serial.tools.list_ports.comports", return_value=[]),
patch("os.path.isdir", side_effect=lambda path: path == "/host/dev"),
patch("glob.glob", side_effect=fake_glob),
patch("os.path.realpath", side_effect=lambda path: "/host/dev/ttyACM0"),
):
result = api.serial_ports()
assert result["success"] is True
assert [item["device"] for item in result["data"]] == [by_id]
def test_serial_ports_tolerates_broken_host_alias(cherrypy_ctx):
del cherrypy_ctx
api = _make_api()
broken = "/host/dev/serial/by-id/broken"
with (
patch("serial.tools.list_ports.comports", return_value=[]),
patch("os.path.isdir", return_value=True),
patch("glob.glob", side_effect=lambda pattern: [broken] if "by-id" in pattern else []),
patch("os.path.realpath", side_effect=OSError("inaccessible")),
):
result = api.serial_ports()
assert result["success"] is True
assert result["data"] == []
def test_config_export_redacts_secrets_and_identity_keys(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "GET"
@@ -628,6 +684,37 @@ def test_config_export_full_backup_includes_hex_keys(cherrypy_ctx):
assert result["data"]["meta"]["includes_secrets"] is True
def test_config_export_canonicalizes_modem_keys_and_redacts_token(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "GET"
api = _make_api(
{
"radio_type": "pymc_tcp",
"pymc_tcp": {"host": "legacy.local", "token": "tcp-secret"},
"radios": [
{"id": "usb", "radio_type": "pymc_usb", "pymc_usb": {"port": "/dev/x"}},
{
"id": "tcp",
"radio_type": "modem_tcp",
"modem_tcp": {"host": "nested.local", "token": "nested-secret"},
},
],
}
)
redacted = api.config_export()["data"]["config"]
full = api.config_export(include_secrets="true")["data"]["config"]
assert redacted["radio_type"] == "modem_tcp"
assert redacted["modem_tcp"]["token"] == "*** REDACTED ***"
assert "pymc_tcp" not in redacted
assert redacted["radios"][0]["radio_type"] == "modem_usb"
assert "pymc_usb" not in redacted["radios"][0]
assert redacted["radios"][1]["modem_tcp"]["token"] == "*** REDACTED ***"
assert full["modem_tcp"]["token"] == "tcp-secret"
assert full["radios"][1]["modem_tcp"]["token"] == "nested-secret"
def test_config_import_rejects_missing_config_object(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "POST"
@@ -658,6 +745,18 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
{"name": "c1", "identity_key": bytes.fromhex("C0FFEE")},
]
},
"modem_tcp": {"host": "current.local", "port": 5055, "token": "keep-token"},
"radios": [
{
"id": "tcp-link",
"radio_type": "modem_tcp",
"modem_tcp": {
"host": "current-nested.local",
"port": 5055,
"token": "keep-nested-token",
},
}
],
}
)
@@ -679,6 +778,22 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
},
"radio": {"frequency": 915000000},
"radio_type": "pymc_usb",
"pymc_tcp": {
"host": "restored.local",
"port": 6000,
"token": "*** REDACTED ***",
},
"radios": [
{
"id": "tcp-link",
"radio_type": "pymc_tcp",
"pymc_tcp": {
"host": "restored-nested.local",
"port": 6001,
"token": "*** REDACTED ***",
},
}
],
"unknown": {"x": 1},
}
}
@@ -690,7 +805,14 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
assert result["success"] is True
assert result["restart_required"] is True
assert set(result["sections_updated"]) == {"repeater", "identities", "radio", "radio_type"}
assert set(result["sections_updated"]) == {
"repeater",
"identities",
"radio",
"radios",
"radio_type",
"modem_tcp",
}
sec = api.config["repeater"]["security"]
assert sec["admin_password"] == "keep-admin"
@@ -699,6 +821,16 @@ def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx):
assert api.config["repeater"]["identity_key"] == bytes.fromhex("AABBCC")
assert "identity_file" not in api.config["repeater"]
assert api.config["identities"]["companions"][0]["identity_key"] == bytes.fromhex("C0FFEE")
assert api.config["modem_tcp"] == {
"host": "restored.local",
"port": 6000,
"token": "keep-token",
}
assert api.config["radios"][0]["modem_tcp"] == {
"host": "restored-nested.local",
"port": 6001,
"token": "keep-nested-token",
}
def test_openapi_success_sets_content_type(cherrypy_ctx):
@@ -1282,7 +1414,7 @@ radio_type: weird_radio
assert "radio_type" in paths
def test_validate_config_pymc_tcp_placeholder_and_bad_port(cherrypy_ctx, tmp_path):
def test_validate_config_legacy_tcp_reports_canonical_error_paths(cherrypy_ctx, tmp_path):
request, _ = cherrypy_ctx
request.method = "GET"
api = _make_api()
@@ -1313,8 +1445,8 @@ pymc_tcp:
assert result["success"] is True
assert result["data"]["valid"] is False
paths = {e["path"] for e in result["data"]["errors"]}
assert "pymc_tcp.host" in paths
assert "pymc_tcp.port" in paths
assert "modem_tcp.host" in paths
assert "modem_tcp.port" in paths
def test_validate_config_sx1262_ch341_missing_sections(cherrypy_ctx, tmp_path):
+154
View File
@@ -0,0 +1,154 @@
import copy
import logging
import yaml
from repeater.modem_config import normalize_modem_config
def test_canonical_modem_config_is_unchanged_and_input_is_not_mutated():
original = {
"radio_type": "modem_tcp",
"modem_tcp": {"host": "modem.local", "token": "secret"},
"modem_usb": {"port": "/dev/ttyACM0"},
}
before = copy.deepcopy(original)
normalized = normalize_modem_config(original)
assert normalized == before
assert original == before
assert normalized is not original
def test_legacy_top_level_type_and_section_are_canonicalized():
normalized = normalize_modem_config(
{"radio_type": "pymc_tcp", "pymc_tcp": {"host": "modem.local"}}
)
assert normalized == {
"radio_type": "modem_tcp",
"modem_tcp": {"host": "modem.local"},
}
def test_canonical_section_wins_conflict_without_logging_values(caplog):
config = {
"radio_type": "pymc_tcp",
"pymc_tcp": {"host": "legacy.example", "token": "legacy-secret"},
"modem_tcp": {"host": "canonical.example", "token": "canonical-secret"},
}
with caplog.at_level(logging.WARNING):
normalized = normalize_modem_config(config)
assert normalized["radio_type"] == "modem_tcp"
assert normalized["modem_tcp"]["host"] == "canonical.example"
assert "pymc_tcp" not in normalized
assert "pymc_tcp" in caplog.text and "modem_tcp" in caplog.text
assert "legacy.example" not in caplog.text
assert "canonical.example" not in caplog.text
assert "legacy-secret" not in caplog.text
assert "canonical-secret" not in caplog.text
def test_each_multi_radio_entry_is_normalized_independently():
normalized = normalize_modem_config(
{
"radios": [
{"id": "tcp", "radio_type": "pymc_tcp", "pymc_tcp": {"host": "x"}},
{
"id": "usb",
"radio_type": "modem_usb",
"pymc_usb": {"port": "old"},
"modem_usb": {"port": "new"},
},
"malformed",
]
}
)
assert normalized["radios"][0] == {
"id": "tcp",
"radio_type": "modem_tcp",
"modem_tcp": {"host": "x"},
}
assert normalized["radios"][1]["radio_type"] == "modem_usb"
assert normalized["radios"][1]["modem_usb"] == {"port": "new"}
assert "pymc_usb" not in normalized["radios"][1]
assert normalized["radios"][2] == "malformed"
def test_unrelated_and_malformed_values_are_preserved():
config = {
"radio_type": "kiss",
"pymc_tcp": "not-a-mapping",
"modem_tcp": ["also", "not", "a", "mapping"],
"radios": None,
}
normalized = normalize_modem_config(config)
assert normalized["radio_type"] == "kiss"
assert normalized["modem_tcp"] == ["also", "not", "a", "mapping"]
assert "pymc_tcp" not in normalized
assert normalized["radios"] is None
def test_load_config_normalizes_in_memory_without_rewriting_file(tmp_path):
from repeater.config import load_config
path = tmp_path / "config.yaml"
persisted = {
"radio_type": "pymc_usb",
"pymc_usb": {"port": "/dev/ttyACM0"},
"storage": {"storage_dir": str(tmp_path / "data")},
"repeater": {"security": {}},
}
path.write_text(yaml.safe_dump(persisted), encoding="utf-8")
before = path.read_text(encoding="utf-8")
loaded = load_config(str(path))
assert loaded["radio_type"] == "modem_usb"
assert loaded["modem_usb"]["port"] == "/dev/ttyACM0"
assert "pymc_usb" not in loaded
assert path.read_text(encoding="utf-8") == before
def test_config_manager_save_lazily_canonicalizes_in_memory_and_yaml(tmp_path):
from repeater.config_manager import ConfigManager
path = tmp_path / "config.yaml"
config = {
"radio_type": "pymc_tcp",
"pymc_tcp": {"host": "legacy.local", "token": "secret"},
"radios": [{"id": "usb", "radio_type": "pymc_usb", "pymc_usb": {"port": "/dev/x"}}],
}
assert ConfigManager(str(path), config).save_to_file() is True
written = yaml.safe_load(path.read_text(encoding="utf-8"))
assert config == written
assert written["radio_type"] == "modem_tcp"
assert "pymc_tcp" not in written
assert written["radios"][0]["radio_type"] == "modem_usb"
assert "pymc_usb" not in written["radios"][0]
def test_save_config_serializes_only_canonical_modem_keys(tmp_path):
from repeater.config import save_config
path = tmp_path / "config.yaml"
config = {
"radio_type": "pymc_usb",
"pymc_usb": {"port": "/dev/ttyACM0"},
}
assert save_config(config, str(path)) is True
written = yaml.safe_load(path.read_text(encoding="utf-8"))
assert written == {
"radio_type": "modem_usb",
"modem_usb": {"port": "/dev/ttyACM0"},
}
+4 -4
View File
@@ -61,15 +61,15 @@ def test_merge_radio_entry_inherits_and_overrides():
}
entry = {
"id": "link",
"radio_type": "pymc_usb",
"radio_type": "modem_usb",
"radio": {"frequency": 2, "tx_power": 22},
"pymc_usb": {"port": "/dev/ttyACM0"},
"modem_usb": {"port": "/dev/ttyACM0"},
}
merged = _merge_radio_entry(global_cfg, entry)
assert merged["_radio_id"] == "link"
assert merged["radio_type"] == "pymc_usb"
assert merged["radio_type"] == "modem_usb"
assert merged["radio"]["frequency"] == 2
assert merged["pymc_usb"]["port"] == "/dev/ttyACM0"
assert merged["modem_usb"]["port"] == "/dev/ttyACM0"
assert "sx1262" in merged # inherited leftover ok; factory uses radio_type
+58 -22
View File
@@ -62,10 +62,10 @@ def test_get_radio_for_board_missing_radio_type_returns_null_radio():
assert type(radio).__name__ == "NullRadio"
# ─── pymc_tcp / pymc_usb branches ────────────────────────────────────
# ─── modem_tcp / modem_usb branches ────────────────────────────────────
def _pymc_radio_cfg():
def _modem_radio_cfg():
"""Common radio params for the pymc_* tests."""
return {
"frequency": 869618000,
@@ -78,7 +78,7 @@ def _pymc_radio_cfg():
}
def test_get_radio_for_board_pymc_tcp(monkeypatch):
def test_get_radio_for_board_modem_tcp(monkeypatch):
pytest.importorskip("openhop_core.hardware.tcp_radio")
captured = {}
@@ -92,8 +92,8 @@ def test_get_radio_for_board_pymc_tcp(monkeypatch):
)
board_config = {
"radio_type": "pymc_tcp",
"pymc_tcp": {
"radio_type": "modem_tcp",
"modem_tcp": {
"host": "pymc-3e2834.local",
"port": 5055,
"token": "shared-secret",
@@ -101,7 +101,7 @@ def test_get_radio_for_board_pymc_tcp(monkeypatch):
"lbt_enabled": False,
"lbt_max_attempts": 3,
},
"radio": _pymc_radio_cfg(),
"radio": _modem_radio_cfg(),
}
get_radio_for_board(board_config)
@@ -116,7 +116,7 @@ def test_get_radio_for_board_pymc_tcp(monkeypatch):
assert captured["lbt_max_attempts"] == 3
def test_get_radio_for_board_pymc_tcp_requires_host(monkeypatch):
def test_get_radio_for_board_modem_tcp_requires_host(monkeypatch):
pytest.importorskip("openhop_core.hardware.tcp_radio")
monkeypatch.setattr(
@@ -125,16 +125,16 @@ def test_get_radio_for_board_pymc_tcp_requires_host(monkeypatch):
)
board_config = {
"radio_type": "pymc_tcp",
"pymc_tcp": {"port": 5055},
"radio": _pymc_radio_cfg(),
"radio_type": "modem_tcp",
"modem_tcp": {"port": 5055},
"radio": _modem_radio_cfg(),
}
with pytest.raises(ValueError, match="Missing 'host'"):
get_radio_for_board(board_config)
def test_get_radio_for_board_pymc_usb(monkeypatch):
def test_get_radio_for_board_modem_usb(monkeypatch):
pytest.importorskip("openhop_core.hardware.usb_radio")
captured = {}
@@ -148,12 +148,12 @@ def test_get_radio_for_board_pymc_usb(monkeypatch):
)
board_config = {
"radio_type": "pymc_usb",
"pymc_usb": {
"radio_type": "modem_usb",
"modem_usb": {
"port": "/dev/ttyACM0",
"baudrate": 921600,
},
"radio": _pymc_radio_cfg(),
"radio": _modem_radio_cfg(),
}
get_radio_for_board(board_config)
@@ -162,12 +162,12 @@ def test_get_radio_for_board_pymc_usb(monkeypatch):
assert captured["baudrate"] == 921600
assert captured["frequency"] == 869618000
assert captured["sync_word"] == 0x12
# LBT defaults preserved when omitted from pymc_usb section.
# LBT defaults preserved when omitted from modem_usb section.
assert captured["lbt_enabled"] is True
assert captured["lbt_max_attempts"] == 5
def test_get_radio_for_board_pymc_usb_requires_port(monkeypatch):
def test_get_radio_for_board_modem_usb_requires_port(monkeypatch):
pytest.importorskip("openhop_core.hardware.usb_radio")
monkeypatch.setattr(
@@ -176,18 +176,54 @@ def test_get_radio_for_board_pymc_usb_requires_port(monkeypatch):
)
board_config = {
"radio_type": "pymc_usb",
"radio_type": "modem_usb",
# Section present (baudrate set) but `port` deliberately omitted to
# exercise the inner "Missing 'port'" guard rather than the outer
# "Missing 'pymc_usb' section" one.
"pymc_usb": {"baudrate": 921600},
"radio": _pymc_radio_cfg(),
# "Missing 'modem_usb' section" one.
"modem_usb": {"baudrate": 921600},
"radio": _modem_radio_cfg(),
}
with pytest.raises(ValueError, match="Missing 'port'"):
get_radio_for_board(board_config)
@pytest.mark.parametrize(
("legacy_type", "legacy_section", "module_name", "class_name", "settings"),
[
(
"pymc_tcp",
"pymc_tcp",
"openhop_core.hardware.tcp_radio",
"TCPLoRaRadio",
{"host": "legacy.local"},
),
(
"pymc_usb",
"pymc_usb",
"openhop_core.hardware.usb_radio",
"USBLoRaRadio",
{"port": "/dev/ttyACM0"},
),
],
)
def test_legacy_modem_aliases_still_construct_radio(
monkeypatch, legacy_type, legacy_section, module_name, class_name, settings
):
pytest.importorskip(module_name)
monkeypatch.setattr(f"{module_name}.{class_name}", lambda **kwargs: _DummyRadio())
radio = get_radio_for_board(
{
"radio_type": legacy_type,
legacy_section: settings,
"radio": _modem_radio_cfg(),
}
)
assert type(radio).__name__ == "BaselineCrcCounterRadio"
# ─── kiss branch: optional CSMA / key-up tuning forwarding ────────────
@@ -220,7 +256,7 @@ def test_get_radio_for_board_kiss_forwards_csma_tuning(monkeypatch):
"tx_delay_ms": 50,
"kiss_full_duplex": True,
},
"radio": _pymc_radio_cfg(),
"radio": _modem_radio_cfg(),
}
get_radio_for_board(board_config)
@@ -238,7 +274,7 @@ def test_get_radio_for_board_kiss_omits_unset_tuning(monkeypatch):
board_config = {
"radio_type": "kiss",
"kiss": {"port": "/dev/ttyACM0", "baud_rate": 115200},
"radio": _pymc_radio_cfg(),
"radio": _modem_radio_cfg(),
}
get_radio_for_board(board_config)
@@ -1,4 +1,4 @@
"""Tests for setup_wizard pymc_usb / pymc_tcp branches.
"""Tests for setup_wizard modem_usb / modem_tcp branches.
These verify that when the first-run /setup wizard is finished with one of
the two pymc_* hardware tiles selected, api_endpoints.setup_wizard() writes
@@ -44,15 +44,15 @@ def wizard_env(tmp_path, monkeypatch):
radio_settings = {
"hardware": {
"pymc_usb": {
"name": "pymc_usb modem (USB-CDC)",
"radio_type": "pymc_usb",
"modem_usb": {
"name": "openHop Modem (USB-CDC)",
"radio_type": "modem_usb",
"tx_power": 22,
"preamble_length": 16,
},
"pymc_tcp": {
"name": "pymc_tcp modem (Wi-Fi / Ethernet)",
"radio_type": "pymc_tcp",
"modem_tcp": {
"name": "openHop Modem (Wi-Fi / Ethernet)",
"radio_type": "modem_tcp",
"tx_power": 22,
"preamble_length": 16,
},
@@ -91,42 +91,42 @@ def _read_yaml(path):
return yaml.safe_load(f)
# ─── pymc_usb ─────────────────────────────────────────────────────────
# ─── modem_usb ─────────────────────────────────────────────────────────
def test_wizard_pymc_usb_defaults(wizard_env):
def test_wizard_modem_usb_defaults(wizard_env):
tmp_path, config_path, endpoints, set_request = wizard_env
body = dict(_BASE_REQUEST, hardware_key="pymc_usb")
body = dict(_BASE_REQUEST, hardware_key="modem_usb")
set_request(body)
result = endpoints.setup_wizard()
assert result["success"] is True
assert result["config"]["radio_type"] == "pymc_usb"
assert result["config"]["pymc_usb_port"] == "/dev/ttyACM0"
assert result["config"]["pymc_usb_baudrate"] == 921600
assert result["config"]["radio_type"] == "modem_usb"
assert result["config"]["modem_usb_port"] == "/dev/ttyACM0"
assert result["config"]["modem_usb_baudrate"] == 921600
written = _read_yaml(config_path)
assert written["radio_type"] == "pymc_usb"
assert written["pymc_usb"]["port"] == "/dev/ttyACM0"
assert written["pymc_usb"]["baudrate"] == 921600
assert written["pymc_usb"]["lbt_enabled"] is True
assert written["pymc_usb"]["lbt_max_attempts"] == 5
assert written["radio_type"] == "modem_usb"
assert written["modem_usb"]["port"] == "/dev/ttyACM0"
assert written["modem_usb"]["baudrate"] == 921600
assert written["modem_usb"]["lbt_enabled"] is True
assert written["modem_usb"]["lbt_max_attempts"] == 5
assert written["radio"]["tx_power"] == 22
assert written["radio"]["preamble_length"] == 16
# config.py rejects pymc_usb if 'sx1262' / 'ch341' keys leak in — none here.
# config.py rejects modem_usb if 'sx1262' / 'ch341' keys leak in — none here.
assert "sx1262" not in written
def test_wizard_pymc_usb_overrides_from_request(wizard_env):
def test_wizard_modem_usb_overrides_from_request(wizard_env):
tmp_path, config_path, endpoints, set_request = wizard_env
body = dict(
_BASE_REQUEST,
hardware_key="pymc_usb",
pymc_usb_port="/dev/ttyUSB0",
pymc_usb_baudrate=115200,
hardware_key="modem_usb",
modem_usb_port="/dev/ttyUSB0",
modem_usb_baudrate=115200,
)
set_request(body)
@@ -134,48 +134,48 @@ def test_wizard_pymc_usb_overrides_from_request(wizard_env):
assert result["success"] is True
written = _read_yaml(config_path)
assert written["pymc_usb"]["port"] == "/dev/ttyUSB0"
assert written["pymc_usb"]["baudrate"] == 115200
assert written["modem_usb"]["port"] == "/dev/ttyUSB0"
assert written["modem_usb"]["baudrate"] == 115200
# ─── pymc_tcp ─────────────────────────────────────────────────────────
# ─── modem_tcp ─────────────────────────────────────────────────────────
def test_wizard_pymc_tcp_placeholder(wizard_env):
def test_wizard_modem_tcp_placeholder(wizard_env):
"""No host in request → wizard writes a sentinel placeholder. config.py
will then refuse to start with a clear error pointing at pymc_tcp.host."""
will then refuse to start with a clear error pointing at modem_tcp.host."""
tmp_path, config_path, endpoints, set_request = wizard_env
body = dict(_BASE_REQUEST, hardware_key="pymc_tcp")
body = dict(_BASE_REQUEST, hardware_key="modem_tcp")
set_request(body)
result = endpoints.setup_wizard()
assert result["success"] is True
assert result["config"]["radio_type"] == "pymc_tcp"
assert result["config"]["pymc_tcp_host"] == "REPLACE_WITH_MODEM_HOST"
assert result["config"]["pymc_tcp_port"] == 5055
assert result["config"]["radio_type"] == "modem_tcp"
assert result["config"]["modem_tcp_host"] == "REPLACE_WITH_MODEM_HOST"
assert result["config"]["modem_tcp_port"] == 5055
written = _read_yaml(config_path)
assert written["radio_type"] == "pymc_tcp"
assert written["pymc_tcp"]["host"] == "REPLACE_WITH_MODEM_HOST"
assert written["pymc_tcp"]["port"] == 5055
assert written["pymc_tcp"]["token"] == ""
assert written["pymc_tcp"]["connect_timeout"] == 5.0
assert written["pymc_tcp"]["lbt_enabled"] is True
assert written["radio_type"] == "modem_tcp"
assert written["modem_tcp"]["host"] == "REPLACE_WITH_MODEM_HOST"
assert written["modem_tcp"]["port"] == 5055
assert written["modem_tcp"]["token"] == ""
assert written["modem_tcp"]["connect_timeout"] == 5.0
assert written["modem_tcp"]["lbt_enabled"] is True
# token deliberately stripped from response.
assert "pymc_tcp_token" not in result["config"]
assert "modem_tcp_token" not in result["config"]
def test_wizard_pymc_tcp_full_fields(wizard_env):
def test_wizard_modem_tcp_full_fields(wizard_env):
tmp_path, config_path, endpoints, set_request = wizard_env
body = dict(
_BASE_REQUEST,
hardware_key="pymc_tcp",
pymc_tcp_host="pymc-3e2834.local",
pymc_tcp_port=6000,
pymc_tcp_token="hunter2",
hardware_key="modem_tcp",
modem_tcp_host="pymc-3e2834.local",
modem_tcp_port=6000,
modem_tcp_token="hunter2",
)
set_request(body)
@@ -183,9 +183,35 @@ def test_wizard_pymc_tcp_full_fields(wizard_env):
assert result["success"] is True
written = _read_yaml(config_path)
assert written["pymc_tcp"]["host"] == "pymc-3e2834.local"
assert written["pymc_tcp"]["port"] == 6000
assert written["pymc_tcp"]["token"] == "hunter2"
assert written["modem_tcp"]["host"] == "pymc-3e2834.local"
assert written["modem_tcp"]["port"] == 6000
assert written["modem_tcp"]["token"] == "hunter2"
assert "modem_tcp_token" not in result["config"]
assert "hunter2" not in str(result)
def test_wizard_accepts_legacy_tcp_fields_but_emits_canonical_config(wizard_env):
_tmp_path, config_path, endpoints, set_request = wizard_env
set_request(
dict(
_BASE_REQUEST,
hardware_key="pymc_tcp",
pymc_tcp_host="legacy.local",
pymc_tcp_port=6001,
pymc_tcp_token="legacy-secret",
)
)
result = endpoints.setup_wizard()
written = _read_yaml(config_path)
assert result["success"] is True
assert result["config"]["hardware"] == "modem_tcp"
assert result["config"]["radio_type"] == "modem_tcp"
assert result["config"]["modem_tcp_host"] == "legacy.local"
assert "pymc_tcp" not in written
assert written["modem_tcp"]["token"] == "legacy-secret"
assert "legacy-secret" not in str(result)
# ─── KISS regression guard ────────────────────────────────────────────
@@ -213,7 +239,7 @@ def test_wizard_rejected_after_setup_complete(wizard_env):
configured = {
"repeater": {"node_name": "already-set", "security": {"admin_password": "verysecret"}},
"radio_type": "pymc_tcp",
"radio_type": "modem_tcp",
"radio": {
"frequency": 869618000,
"spreading_factor": 8,
@@ -224,7 +250,7 @@ def test_wizard_rejected_after_setup_complete(wizard_env):
with open(config_path, "w") as f:
yaml.safe_dump(configured, f)
body = dict(_BASE_REQUEST, hardware_key="pymc_tcp", pymc_tcp_host="modem.local")
body = dict(_BASE_REQUEST, hardware_key="modem_tcp", modem_tcp_host="modem.local")
set_request(body)
result = endpoints.setup_wizard()
@@ -236,7 +262,7 @@ def test_wizard_rejected_after_setup_complete(wizard_env):
def test_wizard_rejects_short_admin_password(wizard_env):
_tmp_path, _config_path, endpoints, set_request = wizard_env
body = dict(_BASE_REQUEST, hardware_key="pymc_tcp", admin_password="short7")
body = dict(_BASE_REQUEST, hardware_key="modem_tcp", admin_password="short7")
set_request(body)
result = endpoints.setup_wizard()