diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 9ec00bede..d621fd6a2 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -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 @@ 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 - - interface.getNode(args.dest, False, **getNode_kwargs).beginSettingsTransaction() + if args.dest != BROADCAST_ADDR: + print("Configuring remote nodes is not supported.") + return - 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) + filename = args.configure[0] + fmt = getattr(args, "export_format", "auto") - 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) + 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 "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) + profile = _read_profile(filename, fmt, seed_fn=_seed_config) - 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) + closeNow = True + interface.getNode(args.dest, False, **getNode_kwargs).beginSettingsTransaction() - if "channelUrl" in configuration: - print("Setting channel url to", configuration["channelUrl"]) - interface.getNode(args.dest, **getNode_kwargs).setURL(configuration["channelUrl"]) - 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 "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 "ringtone" in configuration: - print("Setting ringtone to", configuration["ringtone"]) - interface.getNode(args.dest, **getNode_kwargs).set_ringtone(configuration["ringtone"]) - 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 "location" in configuration: - alt = 0 - lat = 0.0 - lon = 0.0 - localConfig = interface.localNode.localConfig - - 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 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 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 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 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 diff --git a/meshtastic/tests/test_main.py b/meshtastic/tests/test_main.py index 3ca0564a6..e0d65f60f 100644 --- a/meshtastic/tests/test_main.py +++ b/meshtastic/tests/test_main.py @@ -10,11 +10,13 @@ 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 @@ 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 @@ def test_main_export_config(capsys): 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 @@ def test_main_export_config(capsys): # 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(