mirror of
https://github.com/meshtastic/python.git
synced 2026-07-31 00:56:51 -04:00
Merge pull request #944 from tylerpieper/master
add support for .cfg export/import
This commit is contained in:
Ian McEwen
@@ -5,9 +5,11 @@
|
||||
# later we can have a separate changelist to refactor main.py into smaller files
|
||||
# pylint: disable=R0917,C0302
|
||||
|
||||
from typing import List, Optional, Union
|
||||
from typing import Any, Callable, List, Optional, Union
|
||||
from types import ModuleType
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import argparse
|
||||
|
||||
argcomplete: Union[None, ModuleType] = None
|
||||
@@ -61,7 +63,7 @@ except ImportError as e:
|
||||
have_powermon = False
|
||||
powermon_exception = e
|
||||
meter = None
|
||||
from meshtastic.protobuf import admin_pb2, channel_pb2, config_pb2, portnums_pb2, mesh_pb2
|
||||
from meshtastic.protobuf import admin_pb2, channel_pb2, clientonly_pb2, config_pb2, portnums_pb2, mesh_pb2
|
||||
from meshtastic.version import get_active_version
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -742,115 +744,105 @@ def onConnected(interface):
|
||||
printConfig(node.moduleConfig)
|
||||
|
||||
if args.configure:
|
||||
with open(args.configure[0], encoding="utf8") as file:
|
||||
configuration = yaml.safe_load(file)
|
||||
closeNow = True
|
||||
if args.dest != BROADCAST_ADDR:
|
||||
print("Configuring remote nodes is not supported.")
|
||||
return
|
||||
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).beginSettingsTransaction()
|
||||
filename = args.configure[0]
|
||||
fmt = getattr(args, "export_format", "auto")
|
||||
|
||||
if "owner" in configuration:
|
||||
# Validate owner name before setting
|
||||
owner_name = str(configuration["owner"]).strip()
|
||||
if not owner_name:
|
||||
meshtastic.util.our_exit("ERROR: Long Name cannot be empty or contain only whitespace characters")
|
||||
print(f"Setting device owner to {configuration['owner']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).setOwner(configuration["owner"])
|
||||
time.sleep(0.5)
|
||||
def _seed_config(section: str, is_module: bool) -> Optional[Any]:
|
||||
"""Return the current local-node config section, if present."""
|
||||
config_obj = (
|
||||
interface.localNode.moduleConfig
|
||||
if is_module
|
||||
else interface.localNode.localConfig
|
||||
)
|
||||
return getattr(config_obj, section) if config_obj.HasField(section) else None
|
||||
|
||||
if "owner_short" in configuration:
|
||||
# Validate owner short name before setting
|
||||
owner_short_name = str(configuration["owner_short"]).strip()
|
||||
if not owner_short_name:
|
||||
meshtastic.util.our_exit("ERROR: Short Name cannot be empty or contain only whitespace characters")
|
||||
print(
|
||||
f"Setting device owner short to {configuration['owner_short']}"
|
||||
)
|
||||
waitForAckNak = True
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).setOwner(
|
||||
long_name=None, short_name=configuration["owner_short"]
|
||||
)
|
||||
time.sleep(0.5)
|
||||
profile = _read_profile(filename, fmt, seed_fn=_seed_config)
|
||||
|
||||
if "ownerShort" in configuration:
|
||||
# Validate owner short name before setting
|
||||
owner_short_name = str(configuration["ownerShort"]).strip()
|
||||
if not owner_short_name:
|
||||
meshtastic.util.our_exit("ERROR: Short Name cannot be empty or contain only whitespace characters")
|
||||
print(
|
||||
f"Setting device owner short to {configuration['ownerShort']}"
|
||||
)
|
||||
waitForAckNak = True
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).setOwner(
|
||||
long_name=None, short_name=configuration["ownerShort"]
|
||||
)
|
||||
time.sleep(0.5)
|
||||
closeNow = True
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).beginSettingsTransaction()
|
||||
|
||||
if "channel_url" in configuration:
|
||||
print("Setting channel url to", configuration["channel_url"])
|
||||
interface.getNode(args.dest, **getNode_kwargs).setURL(configuration["channel_url"])
|
||||
time.sleep(0.5)
|
||||
# Owner: combine long_name and short_name into a single setOwner call.
|
||||
# NOTE: is_licensed and is_unmessagable are not yet in DeviceProfile;
|
||||
# ref: https://github.com/meshtastic/protobufs/pull/971
|
||||
long_name = str(profile.long_name).strip() if profile.long_name else None
|
||||
short_name = str(profile.short_name).strip() if profile.short_name else None
|
||||
|
||||
if "channelUrl" in configuration:
|
||||
print("Setting channel url to", configuration["channelUrl"])
|
||||
interface.getNode(args.dest, **getNode_kwargs).setURL(configuration["channelUrl"])
|
||||
time.sleep(0.5)
|
||||
if long_name is not None and not long_name:
|
||||
meshtastic.util.our_exit("ERROR: Long Name cannot be empty or contain only whitespace characters")
|
||||
if short_name is not None and not short_name:
|
||||
meshtastic.util.our_exit("ERROR: Short Name cannot be empty or contain only whitespace characters")
|
||||
|
||||
if "canned_messages" in configuration:
|
||||
print("Setting canned message messages to", configuration["canned_messages"])
|
||||
interface.getNode(args.dest, **getNode_kwargs).set_canned_message(configuration["canned_messages"])
|
||||
time.sleep(0.5)
|
||||
if long_name or short_name:
|
||||
if long_name and short_name:
|
||||
print(f"Setting device owner to {long_name} and short name to {short_name}")
|
||||
elif long_name:
|
||||
print(f"Setting device owner to {long_name}")
|
||||
else:
|
||||
print(f"Setting device owner short to {short_name}")
|
||||
waitForAckNak = True
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).setOwner(
|
||||
long_name=long_name, short_name=short_name
|
||||
)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "ringtone" in configuration:
|
||||
print("Setting ringtone to", configuration["ringtone"])
|
||||
interface.getNode(args.dest, **getNode_kwargs).set_ringtone(configuration["ringtone"])
|
||||
time.sleep(0.5)
|
||||
if profile.channel_url:
|
||||
print(f"Setting channel url to {profile.channel_url}")
|
||||
interface.getNode(args.dest, **getNode_kwargs).setURL(profile.channel_url)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "location" in configuration:
|
||||
alt = 0
|
||||
lat = 0.0
|
||||
lon = 0.0
|
||||
localConfig = interface.localNode.localConfig
|
||||
if profile.canned_messages:
|
||||
print(f"Setting canned message messages to {profile.canned_messages}")
|
||||
interface.getNode(args.dest, **getNode_kwargs).set_canned_message(profile.canned_messages)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "alt" in configuration["location"]:
|
||||
alt = int(configuration["location"]["alt"] or 0)
|
||||
print(f"Fixing altitude at {alt} meters")
|
||||
if "lat" in configuration["location"]:
|
||||
lat = float(configuration["location"]["lat"] or 0)
|
||||
print(f"Fixing latitude at {lat} degrees")
|
||||
if "lon" in configuration["location"]:
|
||||
lon = float(configuration["location"]["lon"] or 0)
|
||||
print(f"Fixing longitude at {lon} degrees")
|
||||
if profile.ringtone:
|
||||
print(f"Setting ringtone to {profile.ringtone}")
|
||||
interface.getNode(args.dest, **getNode_kwargs).set_ringtone(profile.ringtone)
|
||||
time.sleep(0.5)
|
||||
|
||||
if profile.HasField("fixed_position"):
|
||||
# Only send the admin message when the position config
|
||||
# explicitly opts into fixed_position. The admin message
|
||||
# unconditionally enables the flag on the device, so we
|
||||
# require an explicit opt-in from the profile's config.
|
||||
if (
|
||||
profile.HasField("config")
|
||||
and profile.config.HasField("position")
|
||||
and profile.config.position.fixed_position
|
||||
):
|
||||
pos = profile.fixed_position
|
||||
lat = float(pos.latitude_i * Decimal("1e-7")) if pos.latitude_i else 0.0
|
||||
lon = float(pos.longitude_i * Decimal("1e-7")) if pos.longitude_i else 0.0
|
||||
alt = pos.altitude if pos.altitude else 0
|
||||
print(f"Fixing altitude at {alt} meters")
|
||||
print(f"Fixing latitude at {lat} degrees")
|
||||
print(f"Fixing longitude at {lon} degrees")
|
||||
print("Setting device position")
|
||||
interface.localNode.setFixedPosition(lat, lon, alt)
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).setFixedPosition(lat, lon, alt)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "config" in configuration:
|
||||
localConfig = interface.getNode(args.dest, **getNode_kwargs).localConfig
|
||||
for section in configuration["config"]:
|
||||
traverseConfig(
|
||||
section, configuration["config"][section], localConfig
|
||||
)
|
||||
interface.getNode(args.dest, **getNode_kwargs).writeConfig(
|
||||
meshtastic.util.camel_to_snake(section)
|
||||
)
|
||||
if profile.HasField("config"):
|
||||
localConfig = interface.getNode(args.dest, **getNode_kwargs).localConfig
|
||||
for field in profile.config.DESCRIPTOR.fields:
|
||||
if field.message_type is not None and profile.config.HasField(field.name):
|
||||
getattr(localConfig, field.name).CopyFrom(getattr(profile.config, field.name))
|
||||
interface.getNode(args.dest, **getNode_kwargs).writeConfig(field.name)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "module_config" in configuration:
|
||||
moduleConfig = interface.getNode(args.dest, **getNode_kwargs).moduleConfig
|
||||
for section in configuration["module_config"]:
|
||||
traverseConfig(
|
||||
section,
|
||||
configuration["module_config"][section],
|
||||
moduleConfig,
|
||||
)
|
||||
interface.getNode(args.dest, **getNode_kwargs).writeConfig(
|
||||
meshtastic.util.camel_to_snake(section)
|
||||
)
|
||||
if profile.HasField("module_config"):
|
||||
moduleConfig = interface.getNode(args.dest, **getNode_kwargs).moduleConfig
|
||||
for field in profile.module_config.DESCRIPTOR.fields:
|
||||
if field.message_type is not None and profile.module_config.HasField(field.name):
|
||||
getattr(moduleConfig, field.name).CopyFrom(getattr(profile.module_config, field.name))
|
||||
interface.getNode(args.dest, **getNode_kwargs).writeConfig(field.name)
|
||||
time.sleep(0.5)
|
||||
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).commitSettingsTransaction()
|
||||
print("Writing modified configuration to device")
|
||||
interface.getNode(args.dest, False, **getNode_kwargs).commitSettingsTransaction()
|
||||
print("Writing modified configuration to device")
|
||||
|
||||
if args.export_config:
|
||||
if args.dest != BROADCAST_ADDR:
|
||||
@@ -858,18 +850,40 @@ def onConnected(interface):
|
||||
return
|
||||
|
||||
closeNow = True
|
||||
config_txt = export_config(interface)
|
||||
|
||||
if args.export_config == "-":
|
||||
# Output to stdout (preserves legacy use of `> file.yaml`)
|
||||
print(config_txt)
|
||||
is_binary = False
|
||||
fmt = getattr(args, "export_format", "auto")
|
||||
if fmt in ("binary", "protobuf"):
|
||||
is_binary = True
|
||||
elif fmt == "yaml":
|
||||
is_binary = False
|
||||
else:
|
||||
try:
|
||||
with open(args.export_config, "w", encoding="utf-8") as f:
|
||||
f.write(config_txt)
|
||||
print(f"Exported configuration to {args.export_config}")
|
||||
except Exception as e:
|
||||
meshtastic.util.our_exit(f"ERROR: Failed to write config file: {e}")
|
||||
is_binary = args.export_config.endswith(".cfg")
|
||||
|
||||
if is_binary:
|
||||
config_bytes = export_profile(interface)
|
||||
if args.export_config == "-":
|
||||
sys.stdout.buffer.write(config_bytes)
|
||||
else:
|
||||
try:
|
||||
with open(args.export_config, "wb") as f:
|
||||
f.write(config_bytes)
|
||||
print(f"Exported profile to {args.export_config}")
|
||||
except Exception as e:
|
||||
meshtastic.util.our_exit(f"ERROR: Failed to write profile file: {e}")
|
||||
else:
|
||||
config_txt = export_config(interface)
|
||||
|
||||
if args.export_config == "-":
|
||||
# Output to stdout (preserves legacy use of `> file.yaml`)
|
||||
print(config_txt)
|
||||
else:
|
||||
try:
|
||||
with open(args.export_config, "w", encoding="utf-8") as f:
|
||||
f.write(config_txt)
|
||||
print(f"Exported configuration to {args.export_config}")
|
||||
except Exception as e:
|
||||
meshtastic.util.our_exit(f"ERROR: Failed to write config file: {e}")
|
||||
|
||||
if args.ch_set_url:
|
||||
closeNow = True
|
||||
@@ -1263,7 +1277,7 @@ def export_config(interface) -> str:
|
||||
lat = None
|
||||
lon = None
|
||||
alt = None
|
||||
if pos:
|
||||
if pos and interface.localNode.localConfig.position.fixed_position:
|
||||
lat = pos.get("latitude")
|
||||
lon = pos.get("longitude")
|
||||
alt = pos.get("altitude")
|
||||
@@ -1307,7 +1321,7 @@ def export_config(interface) -> str:
|
||||
for i in range(len(prefs[pref]['adminKey'])):
|
||||
prefs[pref]['adminKey'][i] = 'base64:' + prefs[pref]['adminKey'][i]
|
||||
if mt_config.camel_case:
|
||||
configObj["config"] = config #Identical command here and 2 lines below?
|
||||
configObj["config"] = prefs
|
||||
else:
|
||||
configObj["config"] = config
|
||||
|
||||
@@ -1332,6 +1346,162 @@ def export_config(interface) -> str:
|
||||
config_txt += yaml.dump(configObj)
|
||||
return config_txt
|
||||
|
||||
def _set_if_populated(profile, field_name, value):
|
||||
if value is None:
|
||||
return
|
||||
val = str(value).strip()
|
||||
if val:
|
||||
setattr(profile, field_name, val)
|
||||
|
||||
|
||||
def _profile_from_yaml(
|
||||
configuration: dict,
|
||||
seed_fn: Optional[Callable[[str, bool], Optional[Any]]] = None,
|
||||
) -> clientonly_pb2.DeviceProfile:
|
||||
"""Convert a YAML config dict to a DeviceProfile protobuf for uniform import.
|
||||
|
||||
If seed_fn is provided, it is called for each config/module_config section
|
||||
before applying YAML values. It should return the current protobuf message
|
||||
for that section (or None) so that unmentioned fields are preserved.
|
||||
"""
|
||||
profile = clientonly_pb2.DeviceProfile()
|
||||
if "owner" in configuration:
|
||||
_set_if_populated(profile, "long_name", configuration["owner"])
|
||||
if "owner_short" in configuration:
|
||||
_set_if_populated(profile, "short_name", configuration["owner_short"])
|
||||
elif "ownerShort" in configuration:
|
||||
_set_if_populated(profile, "short_name", configuration["ownerShort"])
|
||||
if "channel_url" in configuration:
|
||||
_set_if_populated(profile, "channel_url", configuration["channel_url"])
|
||||
elif "channelUrl" in configuration:
|
||||
_set_if_populated(profile, "channel_url", configuration["channelUrl"])
|
||||
if "canned_messages" in configuration:
|
||||
_set_if_populated(profile, "canned_messages", configuration["canned_messages"])
|
||||
if "ringtone" in configuration:
|
||||
_set_if_populated(profile, "ringtone", configuration["ringtone"])
|
||||
loc = configuration.get("location")
|
||||
if loc:
|
||||
lat = float(loc.get("lat", 0) or 0)
|
||||
lon = float(loc.get("lon", 0) or 0)
|
||||
alt = int(loc.get("alt", 0) or 0)
|
||||
if lat or lon or alt:
|
||||
profile.fixed_position.latitude_i = int(Decimal(str(lat)) * Decimal("1e7"))
|
||||
profile.fixed_position.longitude_i = int(Decimal(str(lon)) * Decimal("1e7"))
|
||||
profile.fixed_position.altitude = alt
|
||||
if "config" in configuration:
|
||||
for section in configuration["config"]:
|
||||
section_snake = meshtastic.util.camel_to_snake(section)
|
||||
if seed_fn is not None:
|
||||
seeded = seed_fn(section_snake, False)
|
||||
if seeded is not None:
|
||||
getattr(profile.config, section_snake).CopyFrom(seeded)
|
||||
traverseConfig(section, configuration["config"][section], profile.config)
|
||||
if "module_config" in configuration:
|
||||
for section in configuration["module_config"]:
|
||||
section_snake = meshtastic.util.camel_to_snake(section)
|
||||
if seed_fn is not None:
|
||||
seeded = seed_fn(section_snake, True)
|
||||
if seeded is not None:
|
||||
getattr(profile.module_config, section_snake).CopyFrom(seeded)
|
||||
traverseConfig(section, configuration["module_config"][section], profile.module_config)
|
||||
return profile
|
||||
|
||||
|
||||
def _read_profile(
|
||||
filename: str,
|
||||
fmt: str,
|
||||
seed_fn: Optional[Callable[[str, bool], Optional[Any]]] = None,
|
||||
) -> clientonly_pb2.DeviceProfile:
|
||||
"""Read a config file and return a DeviceProfile, autodetecting format by content."""
|
||||
with open(filename, "rb") as f:
|
||||
raw = f.read()
|
||||
|
||||
if fmt in ("binary", "protobuf"):
|
||||
try:
|
||||
return _parse_profile_bytes(raw)
|
||||
except Exception as e:
|
||||
meshtastic.util.our_exit(
|
||||
f"ERROR: {filename} is not a valid DeviceProfile (.cfg) file: {e}"
|
||||
)
|
||||
|
||||
if fmt == "yaml":
|
||||
try:
|
||||
configuration = yaml.safe_load(raw.decode("utf8"))
|
||||
except (UnicodeDecodeError, yaml.YAMLError) as e:
|
||||
meshtastic.util.our_exit(
|
||||
f"ERROR: {filename} is not a valid YAML config (expected UTF-8 YAML): {e}"
|
||||
)
|
||||
if not isinstance(configuration, dict):
|
||||
meshtastic.util.our_exit(
|
||||
f"ERROR: {filename} is not a valid YAML config (expected a mapping)."
|
||||
)
|
||||
return _profile_from_yaml(configuration, seed_fn=seed_fn)
|
||||
|
||||
# Auto: try YAML first (text files), fall back to protobuf (binary files).
|
||||
try:
|
||||
configuration = yaml.safe_load(raw.decode("utf8"))
|
||||
if isinstance(configuration, dict):
|
||||
return _profile_from_yaml(configuration, seed_fn=seed_fn)
|
||||
except (UnicodeDecodeError, yaml.YAMLError):
|
||||
pass
|
||||
|
||||
try:
|
||||
return _parse_profile_bytes(raw)
|
||||
except Exception as e:
|
||||
meshtastic.util.our_exit(
|
||||
f"ERROR: {filename} is not a valid YAML config or DeviceProfile (.cfg) file: {e}"
|
||||
)
|
||||
|
||||
|
||||
def _parse_profile_bytes(raw: bytes) -> clientonly_pb2.DeviceProfile:
|
||||
"""Parse raw bytes as a DeviceProfile protobuf, raising on failure."""
|
||||
profile = clientonly_pb2.DeviceProfile()
|
||||
profile.ParseFromString(raw)
|
||||
return profile
|
||||
|
||||
|
||||
def export_profile(interface) -> bytes:
|
||||
"""used in --export-config for binary .cfg files"""
|
||||
profile = clientonly_pb2.DeviceProfile()
|
||||
|
||||
owner = interface.getLongName()
|
||||
owner_short = interface.getShortName()
|
||||
channel_url = interface.localNode.getURL()
|
||||
myinfo = interface.getMyNodeInfo()
|
||||
canned_messages = interface.getCannedMessage()
|
||||
ringtone = interface.getRingtone()
|
||||
|
||||
if owner:
|
||||
profile.long_name = owner
|
||||
if owner_short:
|
||||
profile.short_name = owner_short
|
||||
if channel_url:
|
||||
profile.channel_url = channel_url
|
||||
if canned_messages:
|
||||
profile.canned_messages = canned_messages
|
||||
if ringtone:
|
||||
profile.ringtone = ringtone
|
||||
|
||||
profile.config.CopyFrom(interface.localNode.localConfig)
|
||||
profile.module_config.CopyFrom(interface.localNode.moduleConfig)
|
||||
|
||||
if interface.localNode.localConfig.position.fixed_position:
|
||||
pos = myinfo.get("position")
|
||||
if pos:
|
||||
lat = pos.get("latitude")
|
||||
lon = pos.get("longitude")
|
||||
alt = pos.get("altitude")
|
||||
|
||||
if lat or lon or alt:
|
||||
if lat:
|
||||
profile.fixed_position.latitude_i = int(Decimal(str(lat)) * Decimal("1e7"))
|
||||
if lon:
|
||||
profile.fixed_position.longitude_i = int(Decimal(str(lon)) * Decimal("1e7"))
|
||||
if alt:
|
||||
profile.fixed_position.altitude = int(alt)
|
||||
|
||||
return profile.SerializeToString()
|
||||
|
||||
|
||||
def create_power_meter():
|
||||
"""Setup the power meter."""
|
||||
@@ -1702,7 +1872,9 @@ def addImportExportArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPar
|
||||
|
||||
group.add_argument(
|
||||
"--configure",
|
||||
help="Specify a path to a yaml(.yml) file containing the desired settings for the connected device.",
|
||||
"--import-config",
|
||||
dest="configure",
|
||||
help="Specify a path to a configuration file to import. Autodetects format (yaml or binary protobuf).",
|
||||
action="append",
|
||||
)
|
||||
group.add_argument(
|
||||
@@ -1710,7 +1882,13 @@ def addImportExportArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPar
|
||||
nargs="?",
|
||||
const="-", # default to "-" if no value provided
|
||||
metavar="FILE",
|
||||
help="Export device config as YAML (to stdout if no file given)"
|
||||
help="Export device config (to stdout if no file given). Autodetects format by extension if possible."
|
||||
)
|
||||
group.add_argument(
|
||||
"--export-format",
|
||||
choices=["auto", "yaml", "binary", "protobuf"],
|
||||
default="auto",
|
||||
help="Format for export or import. 'auto' uses file extension or contents. 'binary' or 'protobuf' forces binary format. 'yaml' forces yaml."
|
||||
)
|
||||
return parser
|
||||
|
||||
|
||||
@@ -10,11 +10,13 @@ import tempfile
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import mock_open, MagicMock, patch
|
||||
|
||||
import yaml
|
||||
import pytest
|
||||
import meshtastic.__main__ as mt_main
|
||||
|
||||
from meshtastic.__main__ import (
|
||||
export_config,
|
||||
export_profile,
|
||||
initParser,
|
||||
main,
|
||||
onConnection,
|
||||
@@ -23,11 +25,14 @@ from meshtastic.__main__ import (
|
||||
setPref,
|
||||
tunnelMain,
|
||||
set_missing_flags_false,
|
||||
_profile_from_yaml,
|
||||
)
|
||||
from meshtastic import mt_config
|
||||
|
||||
from ..protobuf.channel_pb2 import Channel # pylint: disable=E0611
|
||||
from ..protobuf.config_pb2 import Config # pylint: disable=E0611
|
||||
from ..protobuf.clientonly_pb2 import DeviceProfile # pylint: disable=E0611
|
||||
from ..protobuf.localonly_pb2 import LocalConfig, LocalModuleConfig # pylint: disable=E0611
|
||||
|
||||
# from ..ble_interface import BLEInterface
|
||||
from ..mesh_interface import MeshInterface
|
||||
@@ -871,7 +876,7 @@ def test_main_sendtext_with_invalid_channel_nine(caplog, capsys):
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_sendtext_with_dest(mock_findPorts, mock_serial, mocked_open, mock_hupcl, capsys, caplog, iface_with_nodes):
|
||||
@@ -1069,7 +1074,7 @@ def test_main_seturl(capsys):
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_set_valid(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -1093,7 +1098,7 @@ def test_main_set_valid(mocked_findports, mocked_serial, mocked_open, mocked_hup
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_set_valid_wifi_psk(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -1117,7 +1122,7 @@ def test_main_set_valid_wifi_psk(mocked_findports, mocked_serial, mocked_open, m
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_set_invalid_wifi_psk(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -1144,7 +1149,7 @@ def test_main_set_invalid_wifi_psk(mocked_findports, mocked_serial, mocked_open,
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_set_valid_camel_case(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -1169,7 +1174,7 @@ def test_main_set_valid_camel_case(mocked_findports, mocked_serial, mocked_open,
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_set_with_invalid(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -1194,7 +1199,7 @@ def test_main_set_with_invalid(mocked_findports, mocked_serial, mocked_open, moc
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_configure_with_snake_case(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -1226,7 +1231,7 @@ def test_main_configure_with_snake_case(mocked_findports, mocked_serial, mocked_
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_main_configure_with_camel_case_keys(mocked_findports, mocked_serial, mocked_open, mocked_hupcl, capsys):
|
||||
@@ -2012,6 +2017,471 @@ position_flags: 35"""
|
||||
assert err == ""
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_export_profile_serializes_deviceprofile():
|
||||
"""export_profile() returns a parseable DeviceProfile protobuf"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = "foo"
|
||||
iface.getShortName.return_value = "oof"
|
||||
iface.localNode.getURL.return_value = "https://meshtastic.org/e/#test"
|
||||
iface.getCannedMessage.return_value = "Hi|Bye"
|
||||
iface.getRingtone.return_value = "24:d=32,o=5"
|
||||
iface.getMyNodeInfo.return_value = {
|
||||
"position": {"latitude": 35.88888, "longitude": -93.88888, "altitude": 304}
|
||||
}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.localConfig.position.fixed_position = True
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
raw = export_profile(iface)
|
||||
|
||||
profile = DeviceProfile()
|
||||
profile.ParseFromString(raw) # must not raise
|
||||
assert profile.long_name == "foo"
|
||||
assert profile.short_name == "oof"
|
||||
assert profile.channel_url == "https://meshtastic.org/e/#test"
|
||||
assert profile.canned_messages == "Hi|Bye"
|
||||
assert profile.ringtone == "24:d=32,o=5"
|
||||
assert profile.HasField("fixed_position")
|
||||
assert profile.fixed_position.latitude_i == 358888800
|
||||
assert profile.fixed_position.longitude_i == -938888800
|
||||
assert profile.fixed_position.altitude == 304
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_export_profile_omits_fixed_position_when_not_enabled():
|
||||
"""export_profile() omits fixed_position when position.fixed_position is False"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = "foo"
|
||||
iface.getShortName.return_value = "oof"
|
||||
iface.localNode.getURL.return_value = ""
|
||||
iface.getCannedMessage.return_value = None
|
||||
iface.getRingtone.return_value = None
|
||||
iface.getMyNodeInfo.return_value = {
|
||||
"position": {"latitude": 35.88888, "longitude": -93.88888, "altitude": 304}
|
||||
}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
raw = export_profile(iface)
|
||||
|
||||
profile = DeviceProfile()
|
||||
profile.ParseFromString(raw)
|
||||
assert not profile.HasField("fixed_position"), \
|
||||
"fixed_position should be omitted when position.fixed_position is not enabled"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_export_profile_omits_empty_fields():
|
||||
"""export_profile() should not populate protobuf oneofs for missing values"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = None
|
||||
iface.getShortName.return_value = None
|
||||
iface.localNode.getURL.return_value = ""
|
||||
iface.getCannedMessage.return_value = None
|
||||
iface.getRingtone.return_value = None
|
||||
iface.getMyNodeInfo.return_value = {}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
raw = export_profile(iface)
|
||||
|
||||
profile = DeviceProfile()
|
||||
profile.ParseFromString(raw)
|
||||
assert not profile.HasField("long_name")
|
||||
assert not profile.HasField("short_name")
|
||||
assert not profile.HasField("channel_url")
|
||||
assert not profile.HasField("canned_messages")
|
||||
assert not profile.HasField("ringtone")
|
||||
assert not profile.HasField("fixed_position")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_export_profile_converts_precisely():
|
||||
"""export_profile() uses exact arithmetic for lat/lon conversion"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = None
|
||||
iface.getShortName.return_value = None
|
||||
iface.localNode.getURL.return_value = ""
|
||||
iface.getCannedMessage.return_value = None
|
||||
iface.getRingtone.return_value = None
|
||||
iface.getMyNodeInfo.return_value = {
|
||||
"position": {"latitude": -49.82207, "longitude": 0, "altitude": 0},
|
||||
}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.localConfig.position.fixed_position = True
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
raw = export_profile(iface)
|
||||
|
||||
profile = DeviceProfile()
|
||||
profile.ParseFromString(raw)
|
||||
assert profile.fixed_position.latitude_i == -498220700
|
||||
assert profile.fixed_position.longitude_i == 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_profile_from_yaml_maps_all_fields():
|
||||
"""_profile_from_yaml() converts a YAML dict to an equivalent DeviceProfile"""
|
||||
configuration = {
|
||||
"owner": "YAML Owner",
|
||||
"owner_short": "YO",
|
||||
"channel_url": "https://meshtastic.org/e/#test",
|
||||
"canned_messages": "Hi|Bye",
|
||||
"ringtone": "24:d=32,o=5",
|
||||
"location": {"lat": 35.88888, "lon": -93.88888, "alt": 304},
|
||||
"config": {"bluetooth": {"enabled": True, "fixedPin": 123456}},
|
||||
"module_config": {"telemetry": {"deviceUpdateInterval": 900}},
|
||||
}
|
||||
profile = _profile_from_yaml(configuration)
|
||||
|
||||
assert profile.long_name == "YAML Owner"
|
||||
assert profile.short_name == "YO"
|
||||
assert profile.channel_url == "https://meshtastic.org/e/#test"
|
||||
assert profile.canned_messages == "Hi|Bye"
|
||||
assert profile.ringtone == "24:d=32,o=5"
|
||||
assert profile.HasField("fixed_position")
|
||||
assert profile.fixed_position.latitude_i == 358888800
|
||||
assert profile.fixed_position.longitude_i == -938888800
|
||||
assert profile.fixed_position.altitude == 304
|
||||
assert profile.HasField("config")
|
||||
assert profile.config.bluetooth.enabled is True
|
||||
assert profile.config.bluetooth.fixed_pin == 123456
|
||||
assert profile.HasField("module_config")
|
||||
assert profile.module_config.telemetry.device_update_interval == 900
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_profile_from_yaml_camelcase_keys():
|
||||
"""_profile_from_yaml() handles camelCase YAML keys"""
|
||||
configuration = {
|
||||
"ownerShort": "CB",
|
||||
"channelUrl": "https://meshtastic.org/e/#camel",
|
||||
}
|
||||
profile = _profile_from_yaml(configuration)
|
||||
|
||||
assert profile.short_name == "CB"
|
||||
assert profile.channel_url == "https://meshtastic.org/e/#camel"
|
||||
assert not profile.HasField("long_name")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_profile_from_yaml_converts_precisely():
|
||||
"""_profile_from_yaml() uses exact arithmetic for lat/lon conversion"""
|
||||
# -49.82207 is a value where int(float * 1e7) truncates to -498220699
|
||||
# instead of the correct -498220700
|
||||
configuration = {
|
||||
"location": {"lat": -49.82207, "lon": 0, "alt": 0},
|
||||
}
|
||||
profile = _profile_from_yaml(configuration)
|
||||
|
||||
assert profile.fixed_position.latitude_i == -498220700
|
||||
assert profile.fixed_position.longitude_i == 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_combines_setowner_into_single_call(tmp_path, capsys):
|
||||
"""Both long_name and short_name are set in a single setOwner call"""
|
||||
profile = DeviceProfile()
|
||||
profile.long_name = "Single Call"
|
||||
profile.short_name = "SC"
|
||||
cfg_path = tmp_path / "combined.cfg"
|
||||
cfg_path.write_bytes(profile.SerializeToString())
|
||||
|
||||
sys.argv = ["", "--configure", str(cfg_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
# Single combined message, not two separate ones
|
||||
assert re.search(r"Setting device owner to Single Call and short name to SC", out, re.MULTILINE)
|
||||
assert not re.search(r"Setting device owner short to SC$", out, re.MULTILINE)
|
||||
# setOwner should be called once, not twice
|
||||
iface.getNode.return_value.setOwner.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_with_binary_cfg(tmp_path, capsys):
|
||||
"""--configure with a .cfg binary file parses and applies a DeviceProfile"""
|
||||
profile = DeviceProfile()
|
||||
profile.long_name = "Binary Bob"
|
||||
profile.short_name = "BB"
|
||||
cfg_path = tmp_path / "backup.cfg"
|
||||
cfg_path.write_bytes(profile.SerializeToString())
|
||||
|
||||
sys.argv = ["", "--configure", str(cfg_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Setting device owner to Binary Bob and short name to BB", out, re.MULTILINE)
|
||||
assert re.search(r"Writing modified configuration to device", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_rejects_invalid_yaml_and_extension(tmp_path, capsys):
|
||||
"""A non-dict YAML file with a non-.cfg extension should error cleanly."""
|
||||
bad = tmp_path / "bad.yaml"
|
||||
bad.write_text("just a scalar string")
|
||||
sys.argv = ["", "--configure", str(bad)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
with pytest.raises(SystemExit):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"is not a valid YAML config or DeviceProfile", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_rejects_corrupt_cfg(tmp_path, capsys):
|
||||
"""A .cfg file that is not a valid DeviceProfile should error cleanly."""
|
||||
bad = tmp_path / "corrupt.cfg"
|
||||
bad.write_bytes(b"\x00\x01\x02 not really protobuf \xff\xfe")
|
||||
sys.argv = ["", "--configure", str(bad)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
with pytest.raises(SystemExit):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"is not a valid YAML config or DeviceProfile", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_binary_with_explicit_format(tmp_path, capsys):
|
||||
"""--configure --export-format binary forces binary parsing regardless of extension"""
|
||||
profile = DeviceProfile()
|
||||
profile.long_name = "Forced Binary"
|
||||
cfg_path = tmp_path / "notcfg.dat" # non-.cfg extension, but format forced
|
||||
cfg_path.write_bytes(profile.SerializeToString())
|
||||
|
||||
sys.argv = ["", "--configure", str(cfg_path), "--export-format", "binary"]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Setting device owner to Forced Binary", out, re.MULTILINE)
|
||||
assert re.search(r"Writing modified configuration to device", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_autodetects_binary_by_content(tmp_path, capsys):
|
||||
"""Binary DeviceProfile is detected by content even without .cfg extension"""
|
||||
profile = DeviceProfile()
|
||||
profile.long_name = "Content Detected"
|
||||
# Use .bin extension — autodetection must rely on content, not extension
|
||||
cfg_path = tmp_path / "backup.bin"
|
||||
cfg_path.write_bytes(profile.SerializeToString())
|
||||
|
||||
sys.argv = ["", "--configure", str(cfg_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Setting device owner to Content Detected", out, re.MULTILINE)
|
||||
assert re.search(r"Writing modified configuration to device", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_autodetects_yaml_by_content(tmp_path, capsys):
|
||||
"""YAML config is detected by content even with a non-standard extension"""
|
||||
yaml_path = tmp_path / "config.txt"
|
||||
yaml_path.write_text("owner: YAML Detected\n")
|
||||
|
||||
sys.argv = ["", "--configure", str(yaml_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Setting device owner to YAML Detected", out, re.MULTILINE)
|
||||
assert re.search(r"Writing modified configuration to device", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_export_config_binary_round_trip(tmp_path, capsys):
|
||||
"""Round-trip: export a .cfg via main(), then --configure it back via main()."""
|
||||
cfg_path = tmp_path / "roundtrip.cfg"
|
||||
|
||||
# Export
|
||||
sys.argv = ["", "--export-config", str(cfg_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = "Round Trip"
|
||||
iface.getShortName.return_value = "RT"
|
||||
iface.localNode.getURL.return_value = "https://meshtastic.org/e/#rt"
|
||||
iface.getCannedMessage.return_value = "Yes|No"
|
||||
iface.getRingtone.return_value = None
|
||||
iface.getMyNodeInfo.return_value = {
|
||||
"position": {"latitude": 1.0, "longitude": 2.0, "altitude": 3}
|
||||
}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.localConfig.position.fixed_position = True
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Exported profile to", out, re.MULTILINE)
|
||||
assert cfg_path.exists()
|
||||
|
||||
# Verify the exported file is a valid DeviceProfile
|
||||
profile = DeviceProfile()
|
||||
profile.ParseFromString(cfg_path.read_bytes())
|
||||
assert profile.long_name == "Round Trip"
|
||||
assert profile.short_name == "RT"
|
||||
assert profile.HasField("fixed_position")
|
||||
assert profile.fixed_position.latitude_i == int(1.0 * 1e7)
|
||||
assert profile.fixed_position.longitude_i == int(2.0 * 1e7)
|
||||
assert profile.fixed_position.altitude == 3
|
||||
|
||||
# Re-import the file we just wrote
|
||||
sys.argv = ["", "--configure", str(cfg_path)]
|
||||
mt_config.args = sys.argv
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Setting device owner to Round Trip and short name to RT", out, re.MULTILINE)
|
||||
assert re.search(r"Setting device position", out, re.MULTILINE)
|
||||
assert re.search(r"Writing modified configuration to device", out, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_does_not_set_fixed_position_for_legacy_yaml(tmp_path, capsys):
|
||||
"""Legacy YAML with location and no position config does NOT call setFixedPosition"""
|
||||
yaml_path = tmp_path / "legacy.yaml"
|
||||
yaml_path.write_text("location:\n lat: 35.0\n lon: -93.0\n alt: 100\n")
|
||||
|
||||
sys.argv = ["", "--configure", str(yaml_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert not re.search(r"Setting device position", out, re.MULTILINE)
|
||||
iface.getNode.return_value.setFixedPosition.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_skips_setfixedposition_when_config_disables_it(tmp_path, capsys):
|
||||
"""YAML with location but position.fixed_position=false skips setFixedPosition"""
|
||||
yaml_path = tmp_path / "no_fixed.yaml"
|
||||
yaml_path.write_text(
|
||||
"location:\n lat: 35.0\n lon: -93.0\n alt: 100\n"
|
||||
"config:\n position:\n fixed_position: false\n"
|
||||
)
|
||||
|
||||
sys.argv = ["", "--configure", str(yaml_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert not re.search(r"Setting device position", out, re.MULTILINE)
|
||||
iface.getNode.return_value.setFixedPosition.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_calls_setfixedposition_when_config_opt_in(tmp_path, capsys):
|
||||
"""YAML with location and position.fixed_position=true calls setFixedPosition"""
|
||||
yaml_path = tmp_path / "fixed_on.yaml"
|
||||
yaml_path.write_text(
|
||||
"location:\n lat: 35.0\n lon: -93.0\n alt: 100\n"
|
||||
"config:\n position:\n fixed_position: true\n"
|
||||
)
|
||||
|
||||
sys.argv = ["", "--configure", str(yaml_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Setting device position", out, re.MULTILINE)
|
||||
iface.getNode.return_value.setFixedPosition.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_yaml_preserves_unmentioned_fields(tmp_path, capsys):
|
||||
"""Partial YAML config does not reset unspecified fields in the same section."""
|
||||
yaml_path = tmp_path / "partial.yaml"
|
||||
yaml_path.write_text("config:\n bluetooth:\n enabled: false\n")
|
||||
|
||||
sys.argv = ["", "--configure", str(yaml_path)]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
local_config = LocalConfig()
|
||||
local_config.bluetooth.enabled = True
|
||||
local_config.bluetooth.fixed_pin = 654321
|
||||
iface.localNode.localConfig = local_config
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
# Make getNode return the same node object so apply writes back to our real config.
|
||||
iface.getNode.return_value = iface.localNode
|
||||
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Writing modified configuration to device", out, re.MULTILINE)
|
||||
assert iface.localNode.localConfig.bluetooth.enabled is False
|
||||
assert iface.localNode.localConfig.bluetooth.fixed_pin == 654321
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_main_configure_rejects_remote_node(tmp_path, capsys):
|
||||
"""--configure with --dest pointing to a remote node is rejected."""
|
||||
yaml_path = tmp_path / "remote.yaml"
|
||||
yaml_path.write_text("owner: Remote Owner\n")
|
||||
|
||||
sys.argv = ["", "--configure", str(yaml_path), "--dest", "!12345678"]
|
||||
mt_config.args = sys.argv
|
||||
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
|
||||
main()
|
||||
out, _ = capsys.readouterr()
|
||||
assert re.search(r"Configuring remote nodes is not supported", out, re.MULTILINE)
|
||||
|
||||
|
||||
# TODO
|
||||
# recursion depth exceeded error
|
||||
#@pytest.mark.unit
|
||||
@@ -2077,6 +2547,124 @@ position_flags: 35"""
|
||||
# mo.assert_called()
|
||||
|
||||
|
||||
_TRUE_DEFAULTS = [
|
||||
"bluetooth.enabled",
|
||||
"lora.sx126x_rx_boosted_gain",
|
||||
"lora.tx_enabled",
|
||||
"lora.use_preset",
|
||||
"position.position_broadcast_smart_enabled",
|
||||
"security.serial_enabled",
|
||||
]
|
||||
|
||||
_MOD_TRUE_DEFAULTS = [
|
||||
"mqtt.encryption_enabled",
|
||||
]
|
||||
|
||||
|
||||
def _set_config_bool(lc, path, value):
|
||||
section, field = path.split(".")
|
||||
setattr(getattr(lc, section), field, value)
|
||||
|
||||
|
||||
def _assert_config_bool(profile, path, expected):
|
||||
section, field = path.split(".")
|
||||
assert getattr(getattr(profile.config, section), field) is expected
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@pytest.mark.parametrize("fmt", ["binary", "yaml"])
|
||||
@pytest.mark.parametrize("value", [True, False])
|
||||
def test_round_trip_preserves_config_true_defaults(fmt, value):
|
||||
"""Export-->import preserves config_true_defaults fields for both formats"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = None
|
||||
iface.getShortName.return_value = None
|
||||
iface.localNode.getURL.return_value = ""
|
||||
iface.getCannedMessage.return_value = None
|
||||
iface.getRingtone.return_value = None
|
||||
iface.getMyNodeInfo.return_value = {}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
for path in _TRUE_DEFAULTS:
|
||||
_set_config_bool(iface.localNode.localConfig, path, value)
|
||||
for path in _MOD_TRUE_DEFAULTS:
|
||||
_set_config_bool(iface.localNode.moduleConfig, path, value)
|
||||
|
||||
if fmt == "binary":
|
||||
raw = export_profile(iface)
|
||||
profile = DeviceProfile()
|
||||
profile.ParseFromString(raw)
|
||||
else:
|
||||
yaml_str = export_config(iface)
|
||||
configuration = yaml.safe_load(yaml_str)
|
||||
profile = _profile_from_yaml(configuration)
|
||||
|
||||
assert profile.HasField("config")
|
||||
assert profile.HasField("module_config")
|
||||
for path in _TRUE_DEFAULTS:
|
||||
_assert_config_bool(profile, path, value)
|
||||
for path in _MOD_TRUE_DEFAULTS:
|
||||
section, field = path.split(".")
|
||||
assert getattr(getattr(profile.module_config, section), field) is value
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
def test_binary_and_yaml_export_consistent():
|
||||
"""Binary and YAML export paths produce equivalent DeviceProfile protos"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
iface.getLongName.return_value = "Consistency"
|
||||
iface.getShortName.return_value = "CON"
|
||||
iface.localNode.getURL.return_value = ""
|
||||
iface.getCannedMessage.return_value = None
|
||||
iface.getRingtone.return_value = None
|
||||
iface.getMyNodeInfo.return_value = {}
|
||||
iface.localNode.localConfig = LocalConfig()
|
||||
iface.localNode.moduleConfig = LocalModuleConfig()
|
||||
|
||||
lc = iface.localNode.localConfig
|
||||
lc.bluetooth.enabled = True
|
||||
lc.bluetooth.fixed_pin = 123456
|
||||
lc.lora.sx126x_rx_boosted_gain = False
|
||||
lc.lora.tx_enabled = True
|
||||
lc.lora.use_preset = True
|
||||
lc.position.position_broadcast_smart_enabled = False
|
||||
lc.security.serial_enabled = True
|
||||
|
||||
lc_mod = iface.localNode.moduleConfig
|
||||
lc_mod.mqtt.encryption_enabled = True
|
||||
|
||||
# Binary path
|
||||
raw = export_profile(iface)
|
||||
binary_profile = DeviceProfile()
|
||||
binary_profile.ParseFromString(raw)
|
||||
|
||||
# YAML path
|
||||
yaml_str = export_config(iface)
|
||||
configuration = yaml.safe_load(yaml_str)
|
||||
yaml_profile = _profile_from_yaml(configuration)
|
||||
|
||||
# Both should have config
|
||||
assert binary_profile.HasField("config") == yaml_profile.HasField("config")
|
||||
assert binary_profile.HasField("module_config") == yaml_profile.HasField("module_config")
|
||||
|
||||
# Compare individual field values
|
||||
assert binary_profile.config.bluetooth.enabled == yaml_profile.config.bluetooth.enabled
|
||||
assert binary_profile.config.bluetooth.fixed_pin == yaml_profile.config.bluetooth.fixed_pin
|
||||
assert binary_profile.config.lora.sx126x_rx_boosted_gain == yaml_profile.config.lora.sx126x_rx_boosted_gain
|
||||
assert binary_profile.config.lora.tx_enabled == yaml_profile.config.lora.tx_enabled
|
||||
assert binary_profile.config.lora.use_preset == yaml_profile.config.lora.use_preset
|
||||
assert binary_profile.config.position.position_broadcast_smart_enabled == yaml_profile.config.position.position_broadcast_smart_enabled
|
||||
assert binary_profile.config.security.serial_enabled == yaml_profile.config.security.serial_enabled
|
||||
assert binary_profile.module_config.mqtt.encryption_enabled == yaml_profile.module_config.mqtt.encryption_enabled
|
||||
|
||||
# Owner fields also match
|
||||
assert binary_profile.long_name == yaml_profile.long_name
|
||||
assert binary_profile.short_name == yaml_profile.short_name
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_set_missing_flags_false():
|
||||
"""Test set_missing_flags_false() function"""
|
||||
@@ -2906,7 +3494,7 @@ def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, caplog, capsys)
|
||||
@pytest.mark.usefixtures("reset_mt_config")
|
||||
@patch("platform.system")
|
||||
@patch("meshtastic.serial_interface.SerialInterface._set_hupcl_with_termios")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="data")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"{}")
|
||||
@patch("serial.Serial")
|
||||
@patch("meshtastic.util.findPorts", return_value=["/dev/ttyUSBfake"])
|
||||
def test_tunnel_tunnel_arg(
|
||||
|
||||
Reference in New Issue
Block a user