Half-close TCP connections so the last write is not lost

close() went straight to shutdown(SHUT_RDWR) + close(). Doing that while either
side still has unread data makes the stack send RST rather than FIN, and the
device's FromRadio stream means there is essentially always unread data.

Winsock discards data that was received but not yet read when an RST arrives, so
an admin message written moments earlier is thrown away before the device reads
it. Linux delivers the buffered bytes before reporting ECONNRESET, which is why
this only surfaced on Windows: every one-shot TCP write there (--set, --seturl,
--sendtext) reported success and did nothing.

writeConfig() does not wait for an ack, so close() runs microseconds after
sendall(). The device cannot compensate; it polls every 5ms and the RST beats its
next read.

Send FIN first with shutdown(SHUT_WR), which lets the device consume what we
wrote, wait briefly for the reader thread to drain and exit, then tear down
exactly as before. The full shutdown is kept so a reader still blocked in recv()
is unblocked for the join in StreamInterface.close(). The wait is bounded: the
device is not obliged to close just because we half-closed.

Fixes #957

.coderabbit.yaml

@@ -0,0 +1,16 @@
+language: en-US
+reviews:
+  profile: chill
+  high_level_summary: true
+  poem: false
+  auto_review:
+    enabled: true
+    drafts: false
+    ignore_usernames:
+      - renovate
+      - renovate[bot]
+  abort_on_close: true
+
+  path_filters:
+    - "!/protobufs/**"
+    - "!/meshtastic/protobuf/**"

.github/workflows/ci.yml

@@ -17,6 +17,8 @@ jobs:
           - "3.10"
           - "3.11"
           - "3.12"
+          - "3.13"
+          - "3.14"
     steps:
       - uses: actions/checkout@v4
       - name: Install Python 3
@@ -56,12 +58,11 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - "3.9"
-          - "3.10"
-          - "3.11"
           - "3.12"
     steps:
       - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
       - name: Install Python 3
         uses: actions/setup-python@v5
       - name: Install meshtastic from local
@@ -70,3 +71,36 @@ jobs:
           pip3 install poetry
           poetry install
           poetry run meshtastic --version
+
+  simradio_testing:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        channel:
+          - beta
+          - alpha
+          - daily
+    continue-on-error: ${{ matrix.channel == 'daily' }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: Install Python 3
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install meshtastic from local
+        run: |
+          python -m pip install --upgrade pip
+          pip3 install poetry
+          poetry install --all-extras --with dev
+          poetry run meshtastic --version
+      - name: Install meshtasticd (${{ matrix.channel }}) from PPA
+        run: |
+          sudo add-apt-repository -y ppa:meshtastic/${{ matrix.channel }}
+          sudo apt-get update
+          sudo apt-get install -y meshtasticd
+      - name: Run firmware smoke tests
+        run: poetry run pytest -m "smokevirt or smokemesh" -v
+        timeout-minutes: 15

.github/workflows/daily_simradio.yml

@@ -0,0 +1,36 @@
+name: Daily SimRadio Tests
+
+on:
+  schedule:
+    - cron: 0 6 * * *
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  simradio_testing:
+    runs-on: ubuntu-latest
+    if: github.repository == 'meshtastic/meshtastic-python'
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: Install Python 3
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install meshtastic from local
+        run: |
+          python -m pip install --upgrade pip
+          pip3 install poetry
+          poetry install --all-extras --with dev
+          poetry run meshtastic --version
+      - name: Install meshtasticd (daily) from PPA
+        run: |
+          sudo add-apt-repository -y ppa:meshtastic/daily
+          sudo apt-get update
+          sudo apt-get install -y meshtasticd
+      - name: Run firmware smoke tests
+        run: poetry run pytest -m "smokevirt or smokemesh" -v
+        timeout-minutes: 15

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 @@ 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:
+            if args.dest != BROADCAST_ADDR:
-                configuration = yaml.safe_load(file)
+                print("Configuring remote nodes is not supported.")
-                closeNow = True
+                return
 
-                interface.getNode(args.dest, False, **getNode_kwargs).beginSettingsTransaction()
+            filename = args.configure[0]
+            fmt = getattr(args, "export_format", "auto")
 
-                if "owner" in configuration:
+            def _seed_config(section: str, is_module: bool) -> Optional[Any]:
-                    # Validate owner name before setting
+                """Return the current local-node config section, if present."""
-                    owner_name = str(configuration["owner"]).strip()
+                config_obj = (
-                    if not owner_name:
+                    interface.localNode.moduleConfig
-                        meshtastic.util.our_exit("ERROR: Long Name cannot be empty or contain only whitespace characters")
+                    if is_module
-                    print(f"Setting device owner to {configuration['owner']}")
+                    else interface.localNode.localConfig
-                    waitForAckNak = True
+                )
-                    interface.getNode(args.dest, False, **getNode_kwargs).setOwner(configuration["owner"])
+                return getattr(config_obj, section) if config_obj.HasField(section) else None
-                    time.sleep(0.5)
 
-                if "owner_short" in configuration:
+            profile = _read_profile(filename, fmt, seed_fn=_seed_config)
-                    # 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)
 
-                if "ownerShort" in configuration:
+            closeNow = True
-                    # Validate owner short name before setting
+            interface.getNode(args.dest, False, **getNode_kwargs).beginSettingsTransaction()
-                    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)
 
-                if "channel_url" in configuration:
+            # Owner: combine long_name and short_name into a single setOwner call.
-                    print("Setting channel url to", configuration["channel_url"])
+            # NOTE: is_licensed and is_unmessagable are not yet in DeviceProfile;
-                    interface.getNode(args.dest, **getNode_kwargs).setURL(configuration["channel_url"])
+            # ref: https://github.com/meshtastic/protobufs/pull/971
-                    time.sleep(0.5)
+            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:
+            if long_name is not None and not long_name:
-                    print("Setting channel url to", configuration["channelUrl"])
+                meshtastic.util.our_exit("ERROR: Long Name cannot be empty or contain only whitespace characters")
-                    interface.getNode(args.dest, **getNode_kwargs).setURL(configuration["channelUrl"])
+            if short_name is not None and not short_name:
-                    time.sleep(0.5)
+                meshtastic.util.our_exit("ERROR: Short Name cannot be empty or contain only whitespace characters")
 
-                if "canned_messages" in configuration:
+            if long_name or short_name:
-                    print("Setting canned message messages to", configuration["canned_messages"])
+                if long_name and short_name:
-                    interface.getNode(args.dest, **getNode_kwargs).set_canned_message(configuration["canned_messages"])
+                    print(f"Setting device owner to {long_name} and short name to {short_name}")
-                    time.sleep(0.5)
+                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:
+            if profile.channel_url:
-                    print("Setting ringtone to", configuration["ringtone"])
+                print(f"Setting channel url to {profile.channel_url}")
-                    interface.getNode(args.dest, **getNode_kwargs).set_ringtone(configuration["ringtone"])
+                interface.getNode(args.dest, **getNode_kwargs).setURL(profile.channel_url)
-                    time.sleep(0.5)
+                time.sleep(0.5)
 
-                if "location" in configuration:
+            if profile.canned_messages:
-                    alt = 0
+                print(f"Setting canned message messages to {profile.canned_messages}")
-                    lat = 0.0
+                interface.getNode(args.dest, **getNode_kwargs).set_canned_message(profile.canned_messages)
-                    lon = 0.0
+                time.sleep(0.5)
-                    localConfig = interface.localNode.localConfig
 
-                    if "alt" in configuration["location"]:
+            if profile.ringtone:
-                        alt = int(configuration["location"]["alt"] or 0)
+                print(f"Setting ringtone to {profile.ringtone}")
-                        print(f"Fixing altitude at {alt} meters")
+                interface.getNode(args.dest, **getNode_kwargs).set_ringtone(profile.ringtone)
-                    if "lat" in configuration["location"]:
+                time.sleep(0.5)
-                        lat = float(configuration["location"]["lat"] or 0)
+
-                        print(f"Fixing latitude at {lat} degrees")
+            if profile.HasField("fixed_position"):
-                    if "lon" in configuration["location"]:
+                # Only send the admin message when the position config
-                        lon = float(configuration["location"]["lon"] or 0)
+                # explicitly opts into fixed_position.  The admin message
-                        print(f"Fixing longitude at {lon} degrees")
+                # 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:
+            if profile.HasField("config"):
-                    localConfig = interface.getNode(args.dest, **getNode_kwargs).localConfig
+                localConfig = interface.getNode(args.dest, **getNode_kwargs).localConfig
-                    for section in configuration["config"]:
+                for field in profile.config.DESCRIPTOR.fields:
-                        traverseConfig(
+                    if field.message_type is not None and profile.config.HasField(field.name):
-                            section, configuration["config"][section], localConfig
+                        getattr(localConfig, field.name).CopyFrom(getattr(profile.config, field.name))
-                        )
+                        interface.getNode(args.dest, **getNode_kwargs).writeConfig(field.name)
-                        interface.getNode(args.dest, **getNode_kwargs).writeConfig(
-                            meshtastic.util.camel_to_snake(section)
-                        )
                         time.sleep(0.5)
 
-                if "module_config" in configuration:
+            if profile.HasField("module_config"):
-                    moduleConfig = interface.getNode(args.dest, **getNode_kwargs).moduleConfig
+                moduleConfig = interface.getNode(args.dest, **getNode_kwargs).moduleConfig
-                    for section in configuration["module_config"]:
+                for field in profile.module_config.DESCRIPTOR.fields:
-                        traverseConfig(
+                    if field.message_type is not None and profile.module_config.HasField(field.name):
-                            section,
+                        getattr(moduleConfig, field.name).CopyFrom(getattr(profile.module_config, field.name))
-                            configuration["module_config"][section],
+                        interface.getNode(args.dest, **getNode_kwargs).writeConfig(field.name)
-                            moduleConfig,
-                        )
-                        interface.getNode(args.dest, **getNode_kwargs).writeConfig(
-                            meshtastic.util.camel_to_snake(section)
-                        )
                         time.sleep(0.5)
 
-                interface.getNode(args.dest, False, **getNode_kwargs).commitSettingsTransaction()
+            interface.getNode(args.dest, False, **getNode_kwargs).commitSettingsTransaction()
-                print("Writing modified configuration to device")
+            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 == "-":
+            is_binary = False
-                # Output to stdout (preserves legacy use of `> file.yaml`)
+            fmt = getattr(args, "export_format", "auto")
-                print(config_txt)
+            if fmt in ("binary", "protobuf"):
+                is_binary = True
+            elif fmt == "yaml":
+                is_binary = False
             else:
-                try:
+                is_binary = args.export_config.endswith(".cfg")
-                    with open(args.export_config, "w", encoding="utf-8") as f:
+
-                        f.write(config_txt)
+            if is_binary:
-                    print(f"Exported configuration to {args.export_config}")
+                config_bytes = export_profile(interface)
-                except Exception as e:
+                if args.export_config == "-":
-                    meshtastic.util.our_exit(f"ERROR: Failed to write config file: {e}")
+                    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
@@ -955,9 +969,15 @@ def onConnected(interface):
         if args.ch_longslow:
             setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_SLOW)
 
+        if args.ch_longmod:
+            setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_MODERATE)
+
         if args.ch_longfast:
             setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_FAST)
 
+        if args.ch_longturbo:
+            setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.LONG_TURBO)
+
         if args.ch_medslow:
             setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.MEDIUM_SLOW)
 
@@ -970,6 +990,9 @@ def onConnected(interface):
         if args.ch_shortfast:
             setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.SHORT_FAST)
 
+        if args.ch_shortturbo:
+            setSimpleConfig(config_pb2.Config.LoRaConfig.ModemPreset.SHORT_TURBO)
+
         if args.ch_set or args.ch_enable or args.ch_disable:
             closeNow = True
 
@@ -1263,7 +1286,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 +1330,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 +1355,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 +1881,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 +1891,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
 
@@ -1782,43 +1969,61 @@ def addConfigArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
 
     group.add_argument(
         "--ch-vlongslow",
-        help="Change to the very long-range and slow modem preset",
+        help="Change to the VERY_LONG_SLOW modem preset. Deprecated since 2.5 firmware.",
         action="store_true",
     )
 
     group.add_argument(
         "--ch-longslow",
-        help="Change to the long-range and slow modem preset",
+        help="Change to the LONG_SLOW modem preset. Deprecated since 2.7 firmware.",
+        action="store_true",
+    )
+
+    group.add_argument(
+        "--ch-longmod", "--ch-longmoderate",
+        help="Change to the LONG_MODERATE modem preset",
         action="store_true",
     )
 
     group.add_argument(
         "--ch-longfast",
-        help="Change to the long-range and fast modem preset",
+        help="Change to the LONG_FAST modem preset",
+        action="store_true",
+    )
+
+    group.add_argument(
+        "--ch-longturbo",
+        help="Change to the LONG_TURBO preset",
         action="store_true",
     )
 
     group.add_argument(
         "--ch-medslow",
-        help="Change to the med-range and slow modem preset",
+        help="Change to the MEDIUM_SLOW modem preset",
         action="store_true",
     )
 
     group.add_argument(
         "--ch-medfast",
-        help="Change to the med-range and fast modem preset",
+        help="Change to the MEDIUM_FAST modem preset",
         action="store_true",
     )
 
     group.add_argument(
         "--ch-shortslow",
-        help="Change to the short-range and slow modem preset",
+        help="Change to the SHORT_SLOW modem preset",
         action="store_true",
     )
 
     group.add_argument(
         "--ch-shortfast",
-        help="Change to the short-range and fast modem preset",
+        help="Change to the SHORT_FAST modem preset",
+        action="store_true",
+    )
+
+    group.add_argument(
+        "--ch-shortturbo",
+        help="Change to the SHORT_TURBO modem preset",
         action="store_true",
     )
 

meshtastic/node.py

@@ -287,12 +287,22 @@ class Node:
         # for sending admin channels will also change
         adminIndex = self.iface.localNode._getAdminChannelIndex()
 
+        # Snapshot serialized channel payloads from channelIndex onward so we
+        # can avoid writing slots whose protobuf content did not change after
+        # the shift. Use bytes (not message objects), because _fixupChannels()
+        # mutates message fields in-place.
+        old_channels = [
+            self.channels[i].SerializeToString()
+            for i in range(channelIndex, len(self.channels))
+        ]
+
         self.channels.pop(channelIndex)
         self._fixupChannels()  # expand back to 8 channels
 
         index = channelIndex
-        while index < 8:
+        for old_ch in old_channels:
-            self.writeChannel(index, adminIndex=adminIndex)
+            if self.channels[index].SerializeToString() != old_ch:
+                self.writeChannel(index, adminIndex=adminIndex)
             index += 1
 
             # if we are updating the local node, we might end up

meshtastic/tcp_interface.py

@@ -11,6 +11,11 @@ from typing import Optional
 from meshtastic.stream_interface import StreamInterface
 
 DEFAULT_TCP_PORT = 4403
+
+# How long close() gives the device to consume what we last wrote before forcing
+# the connection down. See close() for why this exists.
+GRACEFUL_CLOSE_TIMEOUT = 0.25
+
 logger = logging.getLogger(__name__)
 
 
@@ -80,6 +85,18 @@ class TCPInterface(StreamInterface):
         server_address = (self.hostname, self.portNumber)
         self.socket = socket.create_connection(server_address)
 
+    def _wait_for_reader_exit(self, timeout: float) -> None:
+        """Wait briefly for the reader thread to drain and exit after a half-close.
+
+        Returns as soon as it exits, or after timeout: the device is not obliged
+        to close just because we did.
+        """
+        rx = getattr(self, "_rxThread", None)
+        if rx is None or rx is threading.current_thread():
+            return
+        with contextlib.suppress(Exception):
+            rx.join(timeout)
+
     def close(self) -> None:
         """Close a connection to the device."""
         logger.debug("Closing TCP stream")
@@ -87,6 +104,20 @@ class TCPInterface(StreamInterface):
         # Therefore force a shutdown first to unblock reader thread reads.
         self._wantExit = True
         if self.socket is not None:
+            # Half-close first. shutdown(SHUT_WR) sends FIN, which tells the
+            # device we are done writing and lets it consume what we last wrote
+            # before the connection goes away -- typically the admin message a
+            # one-shot command such as `--set` just sent.
+            #
+            # Going straight to shutdown(SHUT_RDWR) + close() while either side
+            # still has unread data makes the stack send RST instead. Winsock
+            # then discards data the peer had already received but not yet read,
+            # so the write is lost; Linux delivers it before reporting
+            # ECONNRESET, which is why this only bites on Windows.
+            with contextlib.suppress(Exception):
+                self.socket.shutdown(socket.SHUT_WR)
+            self._wait_for_reader_exit(GRACEFUL_CLOSE_TIMEOUT)
+
             with contextlib.suppress(
                 Exception
             ):  # Ignore errors in shutdown, because we might have a race with the server

meshtastic/tests/conftest.py

@@ -8,6 +8,87 @@ import pytest
 from meshtastic import mt_config
 
 from ..mesh_interface import MeshInterface
+from .firmware_harness import (
+    CHAIN_TOPOLOGY,
+    DEFAULT_BASE_PORT,
+    SimMesh,
+    find_meshtasticd,
+    is_compatible_host,
+)
+from .fw_helpers import set_region
+
+# Use a different base port for the single-node fixture so it doesn't
+# conflict with the multi-node mesh fixture.
+SINGLE_NODE_BASE_PORT = DEFAULT_BASE_PORT + 100
+
+
+def _skip_firmware_if_unavailable() -> None:
+    """Skip the test when meshtasticd can't run on this host."""
+    if not is_compatible_host():
+        pytest.skip("meshtasticd firmware tests require Linux")
+    if find_meshtasticd() is None:
+        pytest.skip(
+            "meshtasticd not found — set MESHTASTICD_BIN or install it on PATH"
+        )
+
+
+@pytest.fixture(scope="function")
+def firmware_node():
+    """A single meshtasticd sim node for smokevirt tests.
+
+    Function-scoped so every test gets a freshly-erased node with no
+    state leaking from previous tests. This makes destructive commands
+    (``--reboot``, ``--set factory_reset true``) safe to run and lets
+    tests be order-independent.
+
+    Yields the SimNode instance.  The node is booted with a fresh erased
+    config and listens on localhost at its TCP port.  Region is set to US
+    so modem-preset tests work against firmware >= 2.8, which clamps
+    presets to the legal set for the current region.
+    """
+    _skip_firmware_if_unavailable()
+    mesh = SimMesh(n_nodes=1, base_port=SINGLE_NODE_BASE_PORT)
+    mesh.start()
+    node = mesh.get_node(0)
+    set_region(node.port, "US")
+    # The region commit restarts the TCP listener, so reconnect the harness
+    # interface in case a test wants to use it directly.
+    if node.iface is not None:
+        try:
+            node.iface.close()
+        except Exception:  # pylint: disable=broad-except
+            pass
+    node.connect()
+    yield node
+    mesh.stop()
+
+
+@pytest.fixture(scope="function")
+def firmware_mesh():
+    """A 3-node chain (A-B-C) meshtasticd sim mesh for smokemesh tests.
+
+    Yields the SimMesh instance.  Nodes are connected and the SIMULATOR_APP
+    packet bridge is running.  Region is set to US for firmware >= 2.8
+    compatibility, interfaces are reconnected after the region change, and
+    node DB convergence is awaited.
+    """
+    _skip_firmware_if_unavailable()
+    mesh = SimMesh(n_nodes=3, topology=CHAIN_TOPOLOGY)
+    mesh.start()
+    for node in mesh.nodes:
+        set_region(node.port, "US")
+    # The region commit restarts each node's TCP listener, so reconnect the
+    # harness interfaces before waiting for convergence.
+    for node in mesh.nodes:
+        if node.iface is not None:
+            try:
+                node.iface.close()
+            except Exception:  # pylint: disable=broad-except
+                pass
+        node.connect()
+    mesh.wait_for_convergence(timeout=30)
+    yield mesh
+    mesh.stop()
 
 
 @pytest.fixture

meshtastic/tests/firmware_harness.py

@@ -0,0 +1,355 @@
+"""Test harness for running real meshtasticd firmware instances.
+
+Launches one or more meshtasticd processes in simulator mode (-s) and bridges
+their "over-the-air" packets via the SIMULATOR_APP protocol so that multiple
+instances can communicate as if via LoRa, with a configurable topology.
+
+The harness expects meshtasticd to be available on PATH or via the
+MESHTASTICD_BIN environment variable.  It does not download or build the
+binary itself.
+"""
+import logging
+import os
+import platform
+import shutil
+import signal
+import socket
+import subprocess
+import tempfile
+import time
+from typing import Dict, List, Optional, Set
+
+from pubsub import pub  # type: ignore[import-untyped]
+
+from meshtastic import BROADCAST_NUM, mesh_pb2, portnums_pb2
+from meshtastic.tcp_interface import TCPInterface
+
+logger = logging.getLogger(__name__)
+
+HW_ID_OFFSET = 16
+DEFAULT_BASE_PORT = 4404
+DEFAULT_RSSI = -50
+DEFAULT_SNR = 10.0
+BOOT_TIMEOUT = 30
+CONNECT_TIMEOUT = 30
+
+CHAIN_TOPOLOGY: Dict[int, Set[int]] = {
+    0: {1},
+    1: {0, 2},
+    2: {1},
+}
+
+
+def find_meshtasticd() -> Optional[str]:
+    """Return the path to the meshtasticd binary, or None if not found."""
+    env_path = os.environ.get("MESHTASTICD_BIN")
+    if env_path and os.path.isfile(env_path) and os.access(env_path, os.X_OK):
+        return env_path
+    return shutil.which("meshtasticd")
+
+
+def is_compatible_host() -> bool:
+    """True when the host can run meshtasticd natively (Linux only)."""
+    return platform.system() == "Linux"
+
+
+def _wait_for_port(port: int, timeout: int = BOOT_TIMEOUT) -> bool:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            s = socket.create_connection(("localhost", port), timeout=0.5)
+            s.close()
+            return True
+        except OSError:
+            time.sleep(0.5)
+    return False
+
+
+class SimNode:
+    """A single meshtasticd simulator instance."""
+
+    def __init__(self, node_id: int, base_port: int = DEFAULT_BASE_PORT):
+        self.node_id = node_id
+        self.hw_id = node_id + HW_ID_OFFSET
+        self.port = base_port + node_id
+        self.process: Optional[subprocess.Popen] = None
+        self.workdir: Optional[str] = None
+        self.iface: Optional[TCPInterface] = None
+        self._log_files: list = []
+
+    @property
+    def node_num(self) -> int:
+        """The firmware-assigned node number (== hw_id)."""
+        if self.iface and self.iface.myInfo:
+            return self.iface.myInfo.my_node_num
+        return self.hw_id
+
+    def start(self, binary: str) -> None:
+        """Launch the meshtasticd process in simulator mode."""
+        self.workdir = tempfile.mkdtemp(prefix=f"mtd_node{self.node_id}_")
+        vfs_dir = os.path.join(self.workdir, "vfs")
+        os.mkdir(vfs_dir)
+        # Files are closed in _kill(); keep them open for the process lifetime.
+        log_stdout = open(  # pylint: disable=consider-using-with
+            os.path.join(self.workdir, "meshtasticd.log"), "wb", buffering=0
+        )
+        log_stderr = open(  # pylint: disable=consider-using-with
+            os.path.join(self.workdir, "meshtasticd.err"), "wb", buffering=0
+        )
+        self._log_files = [log_stdout, log_stderr]
+        self.process = subprocess.Popen(  # pylint: disable=consider-using-with
+            [
+                binary,
+                "-s",
+                "-h", str(self.hw_id),
+                "-p", str(self.port),
+                "-d", vfs_dir,
+                "-e",
+            ],
+            stdout=log_stdout,
+            stderr=log_stderr,
+            start_new_session=True,
+        )
+        if not _wait_for_port(self.port):
+            self._kill()
+            raise RuntimeError(
+                f"meshtasticd node {self.node_id} did not start listening on port {self.port}"
+            )
+    def connect(self) -> None:
+        """Open a TCPInterface connection to this node."""
+        self.iface = TCPInterface(
+            hostname="localhost",
+            portNumber=self.port,
+            connectNow=False,
+        )
+        self.iface.myConnect()
+        self.iface.connect()
+
+    def close(self) -> None:
+        """Close the interface and kill the process."""
+        if self.iface is not None:
+            try:
+                self.iface.localNode.exitSimulator()
+            except Exception:
+                pass
+            try:
+                self.iface.close()
+            except Exception:
+                pass
+            self.iface = None
+        self._kill()
+        if self.workdir:
+            shutil.rmtree(self.workdir, ignore_errors=True)
+            self.workdir = None
+
+    def _kill(self) -> None:
+        for f in self._log_files:
+            try:
+                f.close()
+            except Exception:
+                pass
+        self._log_files = []
+        if self.process is not None:
+            try:
+                os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
+            except (ProcessLookupError, OSError):
+                pass
+            try:
+                self.process.wait(timeout=5)
+            except Exception:
+                try:
+                    os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
+                except Exception:
+                    pass
+            # Give OS time to release TCP port (avoid TIME_WAIT preventing
+            # next instance from binding the same port)
+            time.sleep(1.0)
+            self.process = None
+
+
+class SimMesh:
+    """Manages N meshtasticd sim instances and bridges their SIMULATOR_APP packets.
+
+    When *topology* is None every node hears every other node (full mesh).
+    Otherwise *topology* maps a transmitter's node index to the set of receiver
+    node indices that can hear it.
+    """
+
+    def __init__(
+        self,
+        n_nodes: int = 1,
+        topology: Optional[Dict[int, Set[int]]] = None,
+        base_port: int = DEFAULT_BASE_PORT,
+    ):
+        self.n_nodes = n_nodes
+        self.topology = topology
+        self.base_port = base_port
+        self.nodes: List[SimNode] = [
+            SimNode(i, base_port) for i in range(n_nodes)
+        ]
+        self._port_to_idx: Dict[int, int] = {}
+        self._started = False
+
+    def start(self) -> None:
+        """Launch all nodes, connect, and start the packet bridge."""
+        binary = find_meshtasticd()
+        if binary is None:
+            raise RuntimeError(
+                "meshtasticd not found. Set MESHTASTICD_BIN or install it on PATH."
+            )
+
+        for node in self.nodes:
+            node.start(binary)
+
+        for node in self.nodes:
+            node.connect()
+            self._port_to_idx[node.port] = node.node_id
+
+        pub.subscribe(self._on_sim_packet, "meshtastic.receive.simulator")
+        self._started = True
+
+        if self.n_nodes > 1:
+            self._trigger_convergence()
+
+    def _trigger_convergence(self) -> None:
+        """Actively trigger NodeInfo exchange instead of waiting passively.
+
+        Sends a NODEINFO_APP packet with wantResponse from each node so the
+        firmware's NodeInfoModule responds with its own user info, populating
+        all node DBs deterministically.
+        """
+        for node in self.nodes:
+            iface = node.iface
+            if iface is None:
+                continue
+            user = mesh_pb2.User()
+            user.id = f"!{node.node_num:08x}"
+            user.long_name = f"Node {node.node_id}"
+            user.short_name = f"{node.node_id:04d}"[:4]
+            user.hw_model = mesh_pb2.HardwareModel.PORTDUINO
+            try:
+                iface.sendData(
+                    user,
+                    destinationId=BROADCAST_NUM,
+                    portNum=portnums_pb2.PortNum.NODEINFO_APP,
+                    wantAck=False,
+                    wantResponse=True,
+                )
+            except Exception as ex:
+                logger.debug("NodeInfo trigger for node %d failed: %s", node.node_id, ex)
+        time.sleep(5)
+
+    def stop(self) -> None:
+        """Shut down all nodes and clean up."""
+        if not self._started:
+            return
+        try:
+            pub.unsubscribe(self._on_sim_packet, "meshtastic.receive.simulator")
+        except Exception:
+            pass
+        for node in self.nodes:
+            node.close()
+        self._started = False
+
+    def get_node(self, idx: int) -> SimNode:
+        """Return the SimNode at the given index."""
+        return self.nodes[idx]
+
+    def get_iface(self, idx: int) -> TCPInterface:
+        """Return the TCPInterface for the node at the given index."""
+        iface = self.nodes[idx].iface
+        assert iface is not None, f"node {idx} has no interface"
+        return iface
+
+    def wait_for_convergence(self, timeout: int = 30) -> bool:
+        """Wait until every node sees all others in its node DB.
+
+        Returns True if converged, False if timed out (non-fatal — the packet
+        bridge forwards regardless of node DB state).
+        """
+        if self.n_nodes <= 1:
+            return True
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if all(
+                node.iface is not None
+                and node.iface.nodes is not None
+                and len(node.iface.nodes) >= self.n_nodes
+                for node in self.nodes
+            ):
+                return True
+            time.sleep(2)
+        logger.warning("Mesh did not fully converge within %ds", timeout)
+        return False
+
+    def _get_receivers(self, tx_idx: int) -> List[int]:
+        """Return node indices that can hear a transmission from *tx_idx*."""
+        if self.topology is not None:
+            return sorted(self.topology.get(tx_idx, set()))
+        return [i for i in range(self.n_nodes) if i != tx_idx]
+
+    def _on_sim_packet(self, interface, packet) -> None:
+        """Bridge callback: forward a SIMULATOR_APP packet to receiving nodes."""
+        tx_port = getattr(interface, "portNumber", None)
+        tx_idx = self._port_to_idx.get(tx_port) if tx_port else None
+        if tx_idx is None:
+            return
+
+        rx_indices = self._get_receivers(tx_idx)
+        if not rx_indices:
+            return
+
+        data = packet["decoded"]["payload"]
+        if hasattr(data, "SerializeToString"):
+            data = data.SerializeToString()
+
+        if len(data) > mesh_pb2.Constants.DATA_PAYLOAD_LEN:
+            logger.warning("Simulator payload too big (%d bytes), dropping", len(data))
+            return
+
+        mesh_packet = _build_mesh_packet(packet, data)
+
+        for rx_idx in rx_indices:
+            rx_iface = self.nodes[rx_idx].iface
+            if rx_iface is None:
+                continue
+            mesh_packet.rx_rssi = DEFAULT_RSSI
+            mesh_packet.rx_snr = DEFAULT_SNR
+            to_radio = mesh_pb2.ToRadio()
+            to_radio.packet.CopyFrom(mesh_packet)
+            try:
+                rx_iface._sendToRadio(to_radio)
+            except Exception as ex:
+                logger.error("Error forwarding packet to node %d: %s", rx_idx, ex)
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, *exc):
+        self.stop()
+
+
+def _build_mesh_packet(packet: dict, data: bytes) -> mesh_pb2.MeshPacket:
+    """Reconstruct a MeshPacket for SIMULATOR_APP injection."""
+    mp = mesh_pb2.MeshPacket()
+    mp.decoded.payload = data
+    mp.decoded.portnum = portnums_pb2.PortNum.SIMULATOR_APP
+    mp.to = packet.get("to", BROADCAST_NUM)
+    setattr(mp, "from", packet.get("from", 0))
+    mp.id = packet.get("id", 0)
+    mp.want_ack = packet.get("wantAck", False)
+    mp.hop_limit = packet.get("hopLimit", 0)
+    mp.hop_start = packet.get("hopStart", 0)
+    mp.via_mqtt = packet.get("viaMQTT", False)
+    mp.relay_node = packet.get("relayNode", 0)
+    mp.next_hop = packet.get("nextHop", 0)
+    mp.channel = int(packet.get("channel", 0))
+
+    decoded = packet.get("decoded", {})
+    if "requestId" in decoded:
+        mp.decoded.request_id = decoded["requestId"]
+    if "wantResponse" in decoded:
+        mp.decoded.want_response = decoded["wantResponse"]
+
+    return mp

meshtastic/tests/fw_helpers.py

@@ -0,0 +1,374 @@
+"""Shared helpers for meshtasticd-backed smoke tests.
+
+Both ``test_smokevirt`` (single node) and ``test_smokemesh`` (multi-node
+chain) use these helpers to drive the ``meshtastic`` CLI against real
+``meshtasticd`` simulator instances and then verify the resulting firmware
+state through the Python library's ``TCPInterface``.
+
+Verifying through the library (rather than regex on CLI stdout) is the
+core design choice: it makes tests robust against CLI wording changes
+while still exercising both the CLI argparse path and the firmware I/O
+path of every feature.
+"""
+from __future__ import annotations
+
+import logging
+import shlex
+import socket
+import subprocess
+import sys
+import time
+from typing import Callable, List, Optional, Tuple
+
+from pubsub import pub  # type: ignore[import-untyped]
+
+from meshtastic.tcp_interface import TCPInterface
+
+logger = logging.getLogger(__name__)
+
+# Pause between a CLI command finishing and a verification interface
+# opening, to let the firmware flush its TCP bookkeeping. Keeps the
+# simulator happy when many short-lived connections are happening.
+PAUSE_AFTER_CLI = 0.2
+
+# Pause after changing the LoRa region. The firmware may restart the TCP
+# listener while committing the new radio config, so give it time to settle
+# before the next CLI call or harness reconnect.
+PAUSE_AFTER_REGION_CHANGE = 2.0
+
+
+# ---------------------------------------------------------------------------
+# CLI invocation
+# ---------------------------------------------------------------------------
+
+def resolve_cli() -> str:
+    """Return a shell-invokable ``meshtastic`` command.
+
+    Prefers the in-tree module run through the current interpreter so
+    tests exercise the source we are editing, regardless of any
+    separately-installed ``meshtastic`` entry point on PATH.
+    """
+    # The PATH binary may live outside the nono sandbox's allowed paths;
+    # ``python -m meshtastic`` is more portable and always available.
+    return f"{sys.executable} -m meshtastic"
+
+
+def run_cli(
+    port: int,
+    *args: str,
+    timeout: int = 60,
+    retries: int = 2,
+    retry_delay: float = 1.0,
+) -> Tuple[int, str]:
+    """Run the ``meshtastic`` CLI against the sim node on *port*.
+
+    Returns ``(return_code, merged_stdout_stderr)``. ``--host
+    localhost:PORT`` is prefixed automatically. stderr is merged into
+    stdout so callers can match warning text such as "Warning: Need to
+    specify ..." regardless of which stream it lands on.
+
+    If the CLI fails to connect on the first attempt (which happens
+    transiently when a freshly-booted sim node needs a moment to settle
+    after the harness interface connects), retry up to *retries*
+    additional times after *retry_delay* seconds.
+    """
+    cli = resolve_cli()
+    argv = [*_shlex_split(cli)]
+    argv.extend(["--host", f"localhost:{port}"])
+    argv.extend(args)
+    logger.debug("run_cli: %s", argv)
+
+    last_out = ""
+    for attempt in range(retries + 1):
+        try:
+            proc = subprocess.run(
+                argv,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                timeout=timeout,
+                check=False,
+            )
+        except subprocess.TimeoutExpired as ex:
+            last_out = ex.stdout.decode("utf-8", errors="replace") if ex.stdout else ""
+            if attempt < retries:
+                time.sleep(retry_delay)
+                continue
+            return 124, last_out
+
+        out = proc.stdout.decode("utf-8", errors="replace")
+        rc = proc.returncode
+
+        # Retry on transient connection-refused / timed-out errors that
+        # are common right after a sim node spins up.
+        transient = (
+            "Error connecting" in out
+            or "Timed out waiting for connection" in out
+            or "Connection reset by peer" in out
+        )
+        if rc != 0 and transient and attempt < retries:
+            logger.debug("run_cli: transient failure, retry %d/%d", attempt + 1, retries)
+            time.sleep(retry_delay)
+            last_out = out
+            continue
+        return rc, out
+    return rc, last_out
+
+
+def _shlex_split(cmd: str) -> List[str]:
+    """Split a shell string into argv, honoring quotes."""
+    return shlex.split(cmd)
+
+
+# ---------------------------------------------------------------------------
+# Fresh-connection state verification
+# ---------------------------------------------------------------------------
+
+def _wait_for_port(port: int, timeout: float = 30.0) -> None:
+    """Wait until *port* accepts a TCP connection (firmware sim comes up
+    or comes back after a config-commit reboot)."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            s = socket.create_connection(("localhost", port), timeout=0.5)
+            s.close()
+            return
+        except OSError:
+            time.sleep(0.2)
+    raise TimeoutError(
+        f"port {port} did not accept connections within {timeout}s"
+    )
+
+
+def connect_iface(
+    port: int,
+    no_nodes: bool = False,
+    retries: int = 4,
+    wait_timeout: float = 30.0,
+) -> TCPInterface:
+    """Open a fresh ``TCPInterface`` to *port* and block on the config exchange.
+
+    Firmware config writes (``writeChannel``, ``--seturl``, ``--factory_reset``)
+    can briefly restart the sim's TCP listener. We wait for the port to
+    come up first, then retry the connect+config-exchange a few times.
+    """
+    last_exc: Optional[Exception] = None
+    for attempt in range(retries + 1):
+        try:
+            _wait_for_port(port, timeout=wait_timeout)
+            return TCPInterface(
+                hostname="localhost",
+                portNumber=port,
+                connectNow=True,
+                noNodes=no_nodes,
+            )
+        except Exception as ex:  # pylint: disable=broad-except
+            last_exc = ex
+            if attempt < retries:
+                logger.debug(
+                    "connect_iface attempt %d/%d failed: %s",
+                    attempt + 1, retries, ex,
+                )
+                time.sleep(0.5)
+                continue
+            raise
+    assert last_exc is not None  # type guard; unreachable
+    raise last_exc  # pragma: no cover
+
+
+def verify_state(
+    port: int,
+    verifier: Callable[[TCPInterface], None],
+    *,
+    no_nodes: bool = False,
+) -> None:
+    """Open a fresh interface and run *verifier(iface)*, then close.
+
+    Used after a CLI mutation to verify firmware state through the
+    library. Always closes the interface so the next test starts clean.
+    """
+    iface = connect_iface(port, no_nodes=no_nodes)
+    try:
+        verifier(iface)
+    finally:
+        try:
+            iface.close()
+        except Exception:  # pylint: disable=broad-except
+            pass
+        time.sleep(PAUSE_AFTER_CLI)
+
+
+def cli_then_verify(
+    port: int,
+    cli_args: List[str],
+    verifier: Optional[Callable[[TCPInterface], None]],
+    *,
+    expect_rc: Optional[int] = 0,
+    no_nodes: bool = False,
+    cli_timeout: int = 60,
+) -> str:
+    """Run *cli_args* against *port*, optionally asserting *expect_rc*,
+    then (if *verifier* is not None) open a fresh interface and run
+    *verifier(iface)* against the just-mutated firmware state.
+
+    Returns the CLI stdout.
+    """
+    rc, out = run_cli(port, *cli_args, timeout=cli_timeout)
+    if expect_rc is not None:
+        assert rc == expect_rc, f"CLI rc={rc} (expected {expect_rc}): {out}"
+    time.sleep(PAUSE_AFTER_CLI)
+    if verifier is not None:
+        verify_state(port, verifier, no_nodes=no_nodes)
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Packet collectors (used by smokemesh receive-verification tests)
+# ---------------------------------------------------------------------------
+
+RECEIVE_TIMEOUT = 15.0
+
+
+class PacketCollector:
+    """Collect packets received on a specific interface via pubsub.
+
+    ``Listener`` (pubsub 4.x) wraps handlers with a weak reference, so
+    we keep a strong reference to ``handler`` on the instance to prevent
+    garbage collection before the publishing thread gets to call it.
+    """
+
+    def __init__(self):
+        self.packets: List[dict] = []
+        self._handler: Optional[Callable] = None
+
+    def on_receive(self, packet, interface):  # pylint: disable=unused-argument
+        """Append a received packet to the internal list."""
+        self.packets.append(packet)
+
+    def wait_for(self, count: int, timeout: float = RECEIVE_TIMEOUT) -> bool:
+        """Wait until *count* packets have been collected or *timeout* expires."""
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if len(self.packets) >= count:
+                return True
+            time.sleep(0.2)
+        return len(self.packets) >= count
+
+    @property
+    def texts(self) -> List[str]:
+        """Return text payloads from TEXT_MESSAGE_APP packets."""
+        return [
+            p.get("decoded", {}).get("text", "")
+            for p in self.packets
+            if p.get("decoded", {}).get("portnum") == "TEXT_MESSAGE_APP"
+        ]
+
+    @property
+    def traceroutes(self) -> List[dict]:
+        """Return TRACEROUTE_APP packets."""
+        return [
+            p for p in self.packets
+            if p.get("decoded", {}).get("portnum") == "TRACEROUTE_APP"
+        ]
+
+    @property
+    def telemetries(self) -> List[dict]:
+        """Return TELEMETRY_APP packets."""
+        return [
+            p for p in self.packets
+            if p.get("decoded", {}).get("portnum") == "TELEMETRY_APP"
+        ]
+
+    @property
+    def positions(self) -> List[dict]:
+        """Return POSITION_APP packets."""
+        return [
+            p for p in self.packets
+            if p.get("decoded", {}).get("portnum") == "POSITION_APP"
+        ]
+
+    def reset(self) -> None:
+        """Clear all collected packets."""
+        self.packets.clear()
+
+
+def _subscribe_topic(
+    iface: TCPInterface, topic: str
+) -> PacketCollector:
+    """Internal: subscribe a fully-qualified *topic* and return a collector.
+
+    Filters by *interface* so multi-node tests can subscribe several
+    collectors concurrently without cross-talk.
+    """
+    collector = PacketCollector()
+
+    def handler(packet, interface):
+        if interface is iface:
+            collector.on_receive(packet, interface)
+
+    pub.subscribe(handler, topic)
+    collector._handler = handler  # strong ref; see PacketCollector docstring
+    return collector
+
+
+def subscribe_texts(iface: TCPInterface) -> PacketCollector:
+    """Subscribe to ``meshtastic.receive.text`` filtered to *iface*."""
+    return _subscribe_topic(iface, "meshtastic.receive.text")
+
+
+def subscribe_traceroutes(iface: TCPInterface) -> PacketCollector:
+    """Subscribe to ``meshtastic.receive.traceroute`` filtered to *iface*."""
+    return _subscribe_topic(iface, "meshtastic.receive.traceroute")
+
+
+def subscribe_telemetries(iface: TCPInterface) -> PacketCollector:
+    """Subscribe to ``meshtastic.receive.telemetry`` filtered to *iface*."""
+    return _subscribe_topic(iface, "meshtastic.receive.telemetry")
+
+
+def subscribe_positions(iface: TCPInterface) -> PacketCollector:
+    """Subscribe to ``meshtastic.receive.position`` filtered to *iface*."""
+    return _subscribe_topic(iface, "meshtastic.receive.position")
+
+
+def set_region(port: int, region: str = "US") -> None:
+    """Set the LoRa region on a sim node via the CLI.
+
+    The node briefly restarts its TCP listener while committing the radio
+    config; ``run_cli()`` handles the reconnect retry, and we sleep long
+    enough for the new region to take effect before callers continue.
+    """
+    rc, out = run_cli(port, "--set", "lora.region", region)
+    if rc != 0:
+        raise RuntimeError(f"Failed to set lora.region={region}: {out}")
+    time.sleep(PAUSE_AFTER_REGION_CHANGE)
+
+
+def unsubscribe_all(topic: str) -> None:
+    """Drop every handler currently registered on *topic*.
+
+    Tests use this in a ``finally`` block to keep pubsub clean across
+    the function-scoped mesh fixtures.
+    """
+    try:
+        pub.unsubAll(topic)
+    except Exception:  # pylint: disable=broad-except
+        pass
+
+
+__all__ = [
+    "PAUSE_AFTER_CLI",
+    "PAUSE_AFTER_REGION_CHANGE",
+    "PacketCollector",
+    "RECEIVE_TIMEOUT",
+    "cli_then_verify",
+    "connect_iface",
+    "resolve_cli",
+    "run_cli",
+    "set_region",
+    "subscribe_positions",
+    "subscribe_telemetries",
+    "subscribe_texts",
+    "subscribe_traceroutes",
+    "unsubscribe_all",
+    "verify_state",
+]

meshtastic/tests/test_main.py

@@ -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(

meshtastic/tests/test_node.py

@@ -669,6 +669,78 @@ def test_getChannelByChannelIndex():
     assert anode.getChannelByChannelIndex(9) is None
 
 
+def _build_channels(highest_secondary_index: int):
+    """Build an 8-slot channel table with contiguous active channels.
+
+    Slot 0 is PRIMARY. Slots 1..highest_secondary_index are SECONDARY.
+    Remaining slots are DISABLED.
+    """
+    channels = []
+    for idx in range(8):
+        ch = Channel()
+        ch.index = idx
+        if idx == 0:
+            ch.role = Channel.Role.PRIMARY
+            ch.settings.name = "primary"
+        elif idx <= highest_secondary_index:
+            ch.role = Channel.Role.SECONDARY
+            ch.settings.name = f"ch{idx}"
+        else:
+            ch.role = Channel.Role.DISABLED
+        channels.append(ch)
+    return channels
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    "highest_secondary_index,delete_index,expected_writes",
+    [
+        pytest.param(1, 1, [1], id="active-0-1-del-1"),
+        pytest.param(2, 1, [1, 2], id="active-0-2-del-1"),
+        pytest.param(3, 1, [1, 2, 3], id="active-0-3-del-1"),
+        pytest.param(3, 2, [2, 3], id="active-0-3-del-2"),
+    ],
+)
+def test_delete_channel_writes_only_changed_suffix(
+    highest_secondary_index, delete_index, expected_writes
+):
+    """deleteChannel should only write slots whose payload changed."""
+    iface = MagicMock()
+    anode = Node(iface, "bar", noProto=True)
+    iface.localNode = anode
+    anode.channels = _build_channels(highest_secondary_index)
+
+    writes = []
+
+    def fake_write(channel_index, adminIndex=0):
+        writes.append((channel_index, adminIndex))
+
+    anode.writeChannel = fake_write
+
+    anode.deleteChannel(delete_index)
+
+    written_indices = [idx for idx, _ in writes]
+    assert written_indices == expected_writes
+    assert all(admin_idx == 0 for _, admin_idx in writes)
+    assert 0 not in written_indices
+    assert all(idx < 4 for idx in written_indices)
+
+
+@pytest.mark.unit
+def test_delete_channel_rejects_primary():
+    """deleteChannel should refuse deleting PRIMARY slot 0."""
+    iface = MagicMock()
+    anode = Node(iface, "bar", noProto=True)
+    iface.localNode = anode
+    anode.channels = _build_channels(3)
+
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        anode.deleteChannel(0)
+
+    assert pytest_wrapped_e.type is SystemExit
+    assert pytest_wrapped_e.value.code == 1
+
+
 # TODO
 # @pytest.mark.unit
 # def test_deleteChannel_try_to_delete_primary_channel(capsys):

meshtastic/tests/test_smokemesh.py

@@ -0,0 +1,128 @@
+"""Meshtastic smoke tests with multiple meshtasticd sim instances.
+
+Uses the ``firmware_mesh`` session fixture which provides a 3-node chain
+topology (A-B-C) where A hears B, B hears A and C, and C hears B.
+The SIMULATOR_APP packet bridge forwards transmissions between nodes
+according to this topology.
+"""
+import time
+
+import pytest
+
+from .fw_helpers import (
+    RECEIVE_TIMEOUT,
+    subscribe_texts,
+    subscribe_traceroutes,
+    unsubscribe_all,
+)
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_node_db_convergence(firmware_mesh):
+    """Each node should see all 3 nodes in its node DB after convergence."""
+    counts = [len(n.iface.nodes) for n in firmware_mesh.nodes if n.iface]
+    if any(c < 3 for c in counts):
+        pytest.skip(f"Mesh did not converge (counts={counts})")
+    for i, node in enumerate(firmware_mesh.nodes):
+        iface = node.iface
+        assert iface is not None
+        assert len(iface.nodes) >= 3, f"node {i} only sees {len(iface.nodes)} nodes"
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_broadcast_text(firmware_mesh):
+    """A broadcast from node A should arrive on node B."""
+    collector = subscribe_texts(firmware_mesh.get_iface(1))
+    try:
+        firmware_mesh.get_iface(0).sendText("hello mesh", wantAck=False)
+        assert collector.wait_for(1)
+        assert "hello mesh" in collector.texts
+    finally:
+        unsubscribe_all("meshtastic.receive.text")
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_dm(firmware_mesh):
+    """A DM from node A to node B should arrive on B."""
+    dest = firmware_mesh.get_node(1).node_num
+    collector = subscribe_texts(firmware_mesh.get_iface(1))
+    try:
+        firmware_mesh.get_iface(0).sendText(
+            "hey B", destinationId=dest, wantAck=False
+        )
+        assert collector.wait_for(1)
+        assert "hey B" in collector.texts
+    finally:
+        unsubscribe_all("meshtastic.receive.text")
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_dm_across_relay(firmware_mesh):
+    """A DM from node A to node C must relay through B (chain topology)."""
+    dest = firmware_mesh.get_node(2).node_num
+    collector = subscribe_texts(firmware_mesh.get_iface(2))
+    try:
+        firmware_mesh.get_iface(0).sendText(
+            "relay test", destinationId=dest, wantAck=False
+        )
+        assert collector.wait_for(1), "node C did not receive the DM within timeout"
+        assert "relay test" in collector.texts
+    finally:
+        unsubscribe_all("meshtastic.receive.text")
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_hop_limit_prevents_relay(firmware_mesh):
+    """A broadcast with hopLimit=0 from A reaches B but B does not relay to C."""
+    col_b = subscribe_texts(firmware_mesh.get_iface(1))
+    col_c = subscribe_texts(firmware_mesh.get_iface(2))
+    try:
+        firmware_mesh.get_iface(0).sendText(
+            "hop0", wantAck=False, hopLimit=0
+        )
+        assert col_b.wait_for(1), "B should receive A's broadcast"
+        assert "hop0" in col_b.texts, "B got wrong text"
+
+        time.sleep(RECEIVE_TIMEOUT)
+        assert "hop0" not in col_c.texts, (
+            "C should NOT receive — B must not relay hopLimit=0"
+        )
+    finally:
+        unsubscribe_all("meshtastic.receive.text")
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_show_nodes(firmware_mesh):
+    """showNodes should report the other nodes in the mesh."""
+    for i in range(3):
+        iface = firmware_mesh.get_iface(i)
+        iface.showNodes()
+
+
+@pytest.mark.smokemesh
+def test_smokemesh_traceroute_across_relay(firmware_mesh):
+    """Traceroute from A to C should show route via B in both directions."""
+    col_a = subscribe_traceroutes(firmware_mesh.get_iface(0))
+    col_c = subscribe_traceroutes(firmware_mesh.get_iface(2))
+    try:
+        src_a = firmware_mesh.get_node(0).node_num
+        dest_c = firmware_mesh.get_node(2).node_num
+        node_b = firmware_mesh.get_node(1).node_num
+
+        firmware_mesh.get_iface(0).sendTraceRoute(dest=dest_c, hopLimit=3)
+
+        time.sleep(2)
+
+        assert len(col_a.traceroutes) >= 1, "A did not receive traceroute response"
+        a_resp = col_a.traceroutes[0]
+        assert a_resp["from"] == dest_c, "response source should be C"
+
+        route = a_resp["decoded"]["traceroute"]
+        assert route.get("route") == [node_b], "forward route should be A→B→C"
+        assert route.get("routeBack") == [node_b], "return route should be C→B→A"
+
+        assert len(col_c.traceroutes) >= 1, "C did not receive traceroute request"
+        c_req = col_c.traceroutes[0]
+        assert c_req["from"] == src_a, "request source should be A"
+    finally:
+        unsubscribe_all("meshtastic.receive.traceroute")

meshtastic/tests/test_tcp_interface.py

@@ -1,6 +1,7 @@
 """Meshtastic unit tests for tcp_interface.py"""
 
 import re
+import socket
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -78,6 +79,56 @@ def test_TCPInterface_close_shutdowns_socket_before_super_close():
     assert iface.socket is None
 
 
+@pytest.mark.unit
+def test_TCPInterface_close_half_closes_before_shutdown():
+    """Close should send FIN and let the device drain before forcing the link down.
+
+    Going straight to shutdown(SHUT_RDWR)/close() with data still unread makes the
+    stack send RST, and Winsock then discards data the device had received but not
+    yet read, silently losing writes such as `--set`.
+    """
+    iface = TCPInterface(hostname="localhost", noProto=True, connectNow=False)
+    sock = MagicMock()
+    iface.socket = sock
+    call_order = []
+
+    with patch.object(TCPInterface, "_socket_shutdown", autospec=True) as mock_shutdown:
+        with patch.object(
+            TCPInterface, "_wait_for_reader_exit", autospec=True
+        ) as mock_wait:
+            with patch(
+                "meshtastic.stream_interface.StreamInterface.close", autospec=True
+            ) as mock_super_close:
+                sock.shutdown.side_effect = lambda how: call_order.append(f"shutdown_{how}")
+                mock_wait.side_effect = lambda _self, _t: call_order.append("wait")
+                mock_shutdown.side_effect = lambda _self: call_order.append("full_shutdown")
+                mock_super_close.side_effect = lambda _self: call_order.append("super_close")
+
+                iface.close()
+
+    assert call_order == [
+        f"shutdown_{socket.SHUT_WR}",
+        "wait",
+        "full_shutdown",
+        "super_close",
+    ]
+
+
+@pytest.mark.unit
+def test_TCPInterface_close_survives_half_close_failure():
+    """A peer that already vanished must not break close()."""
+    iface = TCPInterface(hostname="localhost", noProto=True, connectNow=False)
+    sock = MagicMock()
+    sock.shutdown.side_effect = OSError("already gone")
+    iface.socket = sock
+
+    with patch("meshtastic.stream_interface.StreamInterface.close", autospec=True):
+        iface.close()  # must not raise
+
+    sock.close.assert_called_once()
+    assert iface.socket is None
+
+
 @pytest.mark.unit
 def test_TCPInterface_reconnect():
     """Test that _reconnect correctly reconnects"""

pytest.ini

@@ -1,6 +1,6 @@
 [pytest]
 
-addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples and not smokevirt"
+addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples and not smokevirt and not smokemesh"
 
 filterwarnings =
     ignore::DeprecationWarning
@@ -13,4 +13,5 @@ markers =
     smoke1: runs smoke tests on a single device connected via USB
     smoke2: runs smoke tests on a two devices connected via USB
     smokewifi: runs smoke test on an esp32 device setup with wifi
+    smokemesh: runs smoke tests against multiple meshtasticd sim instances
     examples: runs the examples tests which validates the library