bump version to 2.7.10

feat: add named-flag/power-of-2 bitfield support to --set

meshtastic/__main__.py

@@ -66,6 +66,14 @@ from meshtastic.version import get_active_version
 
 logger = logging.getLogger(__name__)
 
+# Map dotted preference paths to the protobuf enum that defines their flags.
+# These fields are stored as uint32 bitmasks in the protobuf but have an
+# associated enum that names the individual flags.
+BITFIELD_ENUMS = {
+    "network.enabled_protocols": config_pb2.Config.NetworkConfig.ProtocolFlags,
+    "position.position_flags": config_pb2.Config.PositionConfig.PositionFlags,
+}
+
 def onReceive(packet, interface) -> None:
     """Callback invoked when a packet arrives"""
     args = mt_config.args
@@ -238,13 +246,28 @@ def setPref(config, comp_name, raw_val) -> bool:
         print("Warning: network.wifi_psk must be 8 or more characters.")
         return False
 
+    # Handle uint32 bitfields that have an associated enum of flag names.
+    bitfield_enum = None
+    if config_type.message_type is not None:
+        bitfield_path = f"{config_type.name}.{pref.name}"
+        bitfield_enum = BITFIELD_ENUMS.get(bitfield_path)
+    if bitfield_enum and isinstance(val, str):
+        # At this point fromStr() could not parse val as int/float/bool/bytes,
+        # so treat it as a comma-separated list of bitfield flag names.
+        flag_names = [name.strip() for name in val.split(",") if name.strip()]
+        try:
+            val = meshtastic.util.flags_from_list(bitfield_enum, flag_names)
+        except ValueError as e:
+            print(f"ERROR: {e}")
+            return False
+
     enumType = pref.enum_type
     # pylint: disable=C0123
     if enumType and type(val) == str:
         # We've failed so far to convert this string into an enum, try to find it by reflection
-        e = enumType.values_by_name.get(val)
+        ev = enumType.values_by_name.get(val)
-        if e:
+        if ev:
-            val = e.number
+            val = ev.number
         else:
             print(
                 f"{name[0]}.{uni_name} does not have an enum called {val}, so you can not set it."
@@ -552,6 +575,13 @@ def onConnected(interface):
             waitForAckNak = True
             interface.getNode(args.dest, False, **getNode_kwargs).resetNodeDb()
 
+        if args.add_contact:
+            closeNow = True
+            waitForAckNak = True
+            interface.getNode(args.dest, False, **getNode_kwargs).addContactURL(
+                args.add_contact
+            )
+
         if args.sendtext:
             closeNow = True
             channelIndex = mt_config.channel_index or 0
@@ -1074,6 +1104,20 @@ def onConnected(interface):
             else:
                 print("Install pyqrcode to view a QR code printed to terminal.")
 
+        if args.contact_qr:
+            closeNow = True
+            url = interface.getNode(args.dest, True, **getNode_kwargs).getContactURL(
+                args.contact_qr,
+                should_ignore=args.contact_ignore,
+                manually_verified=args.contact_verified,
+            )
+            print(f"Contact URL: {url}")
+            if pyqrcode is not None:
+                qr = pyqrcode.create(url)
+                print(qr.terminal())
+            else:
+                print("Install pyqrcode to view a QR code printed to terminal.")
+
         log_set: Optional = None  # type: ignore[annotation-unchecked]
         # we need to keep a reference to the logset so it doesn't get GCed early
 
@@ -1362,6 +1406,11 @@ def common():
             if not stripped_ham_name:
                 meshtastic.util.our_exit("ERROR: Ham radio callsign cannot be empty or contain only whitespace characters")
 
+        # Early validation for OTA firmware file before attempting device connection
+        if hasattr(args, 'ota_update') and args.ota_update is not None:
+            if not os.path.isfile(args.ota_update):
+                meshtastic.util.our_exit(f"Error: OTA firmware file not found: {args.ota_update}", 1)
+
         if have_powermon:
             create_power_meter()
 
@@ -1595,10 +1644,12 @@ def addConnectionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParse
         "--host",
         "--tcp",
         "-t",
-        help="Connect to a device using TCP, optionally passing hostname or IP address to use. (defaults to '%(const)s')",
+        help=("Connect to a device using TCP, optionally passing hostname or IP address to use. (defaults to '%(const)s'). "
+              "A port number may be specified as well, e.g. meshtastic.local:4404. The default port is 4403."),
         nargs="?",
         default=None,
         const="localhost",
+        metavar="HOST[:PORT]",
     )
 
     group.add_argument(
@@ -1858,6 +1909,24 @@ def addChannelConfigArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPa
         action="store_true",
     )
 
+    group.add_argument(
+        "--contact-qr",
+        help="Display a QR code for a node's contact data. "
+        "Use the node ID with a '!' or '0x' prefix or the node number. "
+        "Also shows the shareable contact URL.",
+        metavar="!xxxxxxxx",
+    )
+    group.add_argument(
+        "--contact-verified",
+        help="Set the IS_KEY_MANUALLY_VERIFIED bit in the generated contact URL",
+        action="store_true",
+    )
+    group.add_argument(
+        "--contact-ignore",
+        help="Mark this contact as blocked/ignored in the generated contact URL",
+        action="store_true",
+    )
+
     group.add_argument(
         "--ch-enable",
         help="Enable the specified channel. Use --ch-add instead whenever possible.",
@@ -1910,7 +1979,8 @@ def addPositionConfigArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentP
 
     group.add_argument(
         "--pos-fields",
-        help="Specify fields to send when sending a position. Use no argument for a list of valid values. "
+        help="Deprecated: use '--set position.position_flags FLAG1,FLAG2' instead. "
+        "Specify fields to send when sending a position. Use no argument for a list of valid values. "
         "Can pass multiple values as a space separated list like "
         "this: '--pos-fields ALTITUDE HEADING SPEED'",
         nargs="*",
@@ -2095,6 +2165,13 @@ def addRemoteAdminArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPars
         action="store_true",
     )
 
+    group.add_argument(
+        "--add-contact",
+        help="Add a contact (User) to the NodeDB from a shareable URL. "
+        "Example: https://meshtastic.org/v/#<base64>",
+        metavar="URL",
+    )
+
     group.add_argument(
         "--set-time",
         help="Set the time to the provided unix epoch timestamp, or the system's current time if omitted or 0.",

meshtastic/mesh_interface.py

@@ -687,6 +687,13 @@ class MeshInterface:  # pylint: disable=R0902
 
     def onResponseTraceRoute(self, p: dict):
         """on response for trace route"""
+        if p["decoded"]["portnum"] == "ROUTING_APP":
+            error = p["decoded"]["routing"]["errorReason"]
+            if error != "NONE":
+                print(f"Traceroute failed: {error}")
+            self._acknowledgment.receivedTraceRoute = True
+            return
+
         UNK_SNR = -128 # Value representing unknown SNR
 
         routeDiscovery = mesh_pb2.RouteDiscovery()

meshtastic/node.py

@@ -380,6 +380,46 @@ class Node:
         s = s.replace("=", "").replace("+", "-").replace("/", "_")
         return f"https://meshtastic.org/e/#{s}"
 
+    def getContactURL(self, node_id: Union[int, str], should_ignore: bool = False, manually_verified: bool = False):
+        """Generate a shareable contact URL for the specified node"""
+        nodeNum = to_node_num(node_id)
+
+        node = self.iface.nodesByNum.get(nodeNum)
+        if not node or not node.get("user"):
+            our_exit(f"Warning: Node {node_id} not found in NodeDB")
+
+        contact = admin_pb2.SharedContact()
+        contact.node_num = nodeNum
+
+        u = node["user"]
+        if u.get("id"):
+            contact.user.id = u["id"]
+        if u.get("macaddr"):
+            contact.user.macaddr = base64.b64decode(u["macaddr"])
+        if u.get("longName"):
+            contact.user.long_name = u["longName"]
+        if u.get("shortName"):
+            contact.user.short_name = u["shortName"]
+        if u.get("hwModel") and u["hwModel"] != "UNSET":
+            contact.user.hw_model = mesh_pb2.HardwareModel.Value(u["hwModel"])
+        if u.get("role"):
+            contact.user.role = config_pb2.Config.DeviceConfig.Role.Value(u["role"])
+        if u.get("publicKey"):
+            contact.user.public_key = base64.b64decode(u["publicKey"])
+        if u.get("isLicensed"):
+            contact.user.is_licensed = u["isLicensed"]
+        if u.get("isUnmessagable") is not None:
+            contact.user.is_unmessagable = u["isUnmessagable"]
+        if should_ignore:
+            contact.should_ignore = True
+        if manually_verified:
+            contact.manually_verified = True
+
+        data = contact.SerializeToString()
+        s = base64.urlsafe_b64encode(data).decode("ascii")
+        s = s.replace("=", "").replace("+", "-").replace("/", "_")
+        return f"https://meshtastic.org/v/#{s}"
+
     def setURL(self, url: str, addOnly: bool = False):
         """Set mesh network URL"""
         if self.localConfig is None or self.channels is None:
@@ -445,6 +485,32 @@ class Node:
         self.ensureSessionKey()
         self._sendAdmin(p)
 
+    def addContactURL(self, url: str):
+        """Add a contact (User) to the NodeDB from a shareable URL"""
+        self.ensureSessionKey()
+
+        splitURL = url.split("/#")
+        if len(splitURL) == 1:
+            our_exit(f"Warning: Invalid URL '{url}'")
+        b64 = splitURL[-1]
+
+        missing_padding = len(b64) % 4
+        if missing_padding:
+            b64 += "=" * (4 - missing_padding)
+
+        decodedURL = base64.urlsafe_b64decode(b64)
+        contact = admin_pb2.SharedContact()
+        contact.ParseFromString(decodedURL)
+
+        p = admin_pb2.AdminMessage()
+        p.add_contact.CopyFrom(contact)
+
+        if self == self.iface.localNode:
+            onResponse = None
+        else:
+            onResponse = self.onAckNak
+        return self._sendAdmin(p, onResponse=onResponse)
+
     def onResponseRequestRingtone(self, p):
         """Handle the response packet for requesting ringtone part 1"""
         logger.debug(f"onResponseRequestRingtone() p:{p}")

meshtastic/protobuf/mesh_pb2.pyi

@@ -594,6 +594,22 @@ class _HardwareModelEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._
     """
     Lilygo T-Echo Card
     """
+    SEEED_WIO_TRACKER_L2: _HardwareModel.ValueType  # 137
+    """
+    Seeed Tracker L2
+    """
+    CROWPANEL_P4: _HardwareModel.ValueType  # 138
+    """
+    Elecrow CrowPanel Advance P4 models, ESP32-P4 and TFT with SX1262 radio plugin
+    """
+    HELTEC_MESH_TOWER_V2: _HardwareModel.ValueType  # 139
+    """
+    Heltec Mesh Tower V2
+    """
+    MESHNOLOGY_W10: _HardwareModel.ValueType  # 140
+    """
+    Meshnology W10
+    """
     PRIVATE_HW: _HardwareModel.ValueType  # 255
     """
     ------------------------------------------------------------------------------------------------------------------------------------------
@@ -1171,6 +1187,22 @@ T_ECHO_CARD: HardwareModel.ValueType  # 136
 """
 Lilygo T-Echo Card
 """
+SEEED_WIO_TRACKER_L2: HardwareModel.ValueType  # 137
+"""
+Seeed Tracker L2
+"""
+CROWPANEL_P4: HardwareModel.ValueType  # 138
+"""
+Elecrow CrowPanel Advance P4 models, ESP32-P4 and TFT with SX1262 radio plugin
+"""
+HELTEC_MESH_TOWER_V2: HardwareModel.ValueType  # 139
+"""
+Heltec Mesh Tower V2
+"""
+MESHNOLOGY_W10: HardwareModel.ValueType  # 140
+"""
+Meshnology W10
+"""
 PRIVATE_HW: HardwareModel.ValueType  # 255
 """
 ------------------------------------------------------------------------------------------------------------------------------------------

meshtastic/tests/test_main.py

@@ -20,12 +20,14 @@ from meshtastic.__main__ import (
     onConnection,
     onNode,
     onReceive,
+    setPref,
     tunnelMain,
     set_missing_flags_false,
 )
 from meshtastic import mt_config
 
 from ..protobuf.channel_pb2 import Channel # pylint: disable=E0611
+from ..protobuf.config_pb2 import Config # pylint: disable=E0611
 
 # from ..ble_interface import BLEInterface
 from ..mesh_interface import MeshInterface
@@ -2998,6 +3000,56 @@ def test_remove_ignored_node():
         main()
 
     mocked_node.removeIgnored.assert_called_once_with("!12345678")
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_add_contact_url():
+    """Test --add-contact with a shareable URL"""
+    url = "https://meshtastic.org/v/#CKqkvZgIElEKCSE4MzBmNTIyYRIQUm9hZHJ1bm5lciBSaWRnZRoEUktTTiIGAAAAAAAAKAk4AkIgRxo_Fw_ergQIhRqBbrHasLYy3gU-Ay8hrhu4OVnIPQc=" # pylint: disable=line-too-long
+    sys.argv = ["", "--add-contact", url]
+    mt_config.args = sys.argv
+    mocked_node = MagicMock(autospec=Node)
+    iface = MagicMock(autospec=SerialInterface)
+    iface.getNode.return_value = mocked_node
+    with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
+        main()
+
+    mocked_node.addContactURL.assert_called_once_with(url)
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_contact_qr():
+    """Test --contact-qr with a node ID"""
+    sys.argv = ["", "--contact-qr", "!830f522a"]
+    mt_config.args = sys.argv
+    mocked_node = MagicMock(autospec=Node)
+    iface = MagicMock(autospec=SerialInterface)
+    iface.getNode.return_value = mocked_node
+    mocked_node.iface = iface
+    with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
+        main()
+
+    mocked_node.getContactURL.assert_called_once_with("!830f522a", should_ignore=False, manually_verified=False)
+    mocked_node.getContactURL.reset_mock()
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_contact_qr_with_flags():
+    """Test --contact-qr with --contact-verified and --contact-ignore"""
+    sys.argv = ["", "--contact-qr", "!830f522a", "--contact-verified", "--contact-ignore"]
+    mt_config.args = sys.argv
+    mocked_node = MagicMock(autospec=Node)
+    iface = MagicMock(autospec=SerialInterface)
+    iface.getNode.return_value = mocked_node
+    mocked_node.iface = iface
+    with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):
+        main()
+
+    mocked_node.getContactURL.assert_called_once_with("!830f522a", should_ignore=True, manually_verified=True)
+
+
 @pytest.mark.unit
 @pytest.mark.usefixtures("reset_mt_config")
 def test_main_set_owner_whitespace_only(capsys):
@@ -3107,6 +3159,9 @@ def test_main_ota_update_file_not_found(capsys):
 
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
+    out, _ = capsys.readouterr()
+    assert "OTA firmware file not found" in out
+    assert "/nonexistent/firmware.bin" in out
 
 
 @pytest.mark.unit
@@ -3151,3 +3206,48 @@ def test_main_ota_update_retries(mock_our_exit, mock_ota_class, capsys):
 
     finally:
         os.unlink(firmware_file)
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_main_setPref_network_enabled_protocols_by_name(capsys):
+    """Test setPref() accepts bitfield flag names for network.enabled_protocols."""
+    config = Config()
+    assert setPref(config, "network.enabled_protocols", "UDP_BROADCAST") is True
+    assert config.network.enabled_protocols == 1
+    out, _ = capsys.readouterr()
+    assert "Set network.enabled_protocols to UDP_BROADCAST" in out
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_main_setPref_position_flags_multiple(capsys):
+    """Test setPref() accepts comma-separated bitfield flag names."""
+    config = Config()
+    assert setPref(config, "position.position_flags", "ALTITUDE,SPEED") is True
+    assert config.position.position_flags == 513
+    out, _ = capsys.readouterr()
+    assert "Set position.position_flags to ALTITUDE,SPEED" in out
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_main_setPref_bitfield_raw_integer(capsys):
+    """Test setPref() still accepts raw integers for bitfields."""
+    config = Config()
+    assert setPref(config, "network.enabled_protocols", "0") is True
+    assert config.network.enabled_protocols == 0
+    out, _ = capsys.readouterr()
+    assert "Set network.enabled_protocols to 0" in out
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_main_setPref_bitfield_invalid_name(capsys):
+    """Test setPref() rejects unknown bitfield flag names."""
+    config = Config()
+    assert setPref(config, "network.enabled_protocols", "TCP") is False
+    out, _ = capsys.readouterr()
+    assert "Unknown flag 'TCP'" in out
+    assert "NO_BROADCAST" in out
+    assert "UDP_BROADCAST" in out

meshtastic/tests/test_mesh_interface.py

@@ -757,3 +757,43 @@ def test_timeago_fuzz(seconds):
     """Fuzz _timeago to ensure it works with any integer"""
     val = _timeago(seconds)
     assert re.match(r"(now|\d+ (secs?|mins?|hours?|days?|months?|years?))", val)
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_onResponseTraceRoute_routing_error(capsys):
+    """Test that onResponseTraceRoute handles ROUTING_APP error packets correctly."""
+    iface = MeshInterface(noProto=True)
+
+    packet = {
+        "decoded": {
+            "portnum": "ROUTING_APP",
+            "routing": {"errorReason": "MAX_RETRANSMIT"},
+        }
+    }
+
+    iface.onResponseTraceRoute(packet)
+
+    assert iface._acknowledgment.receivedTraceRoute is True
+    out, _ = capsys.readouterr()
+    assert "Traceroute failed: MAX_RETRANSMIT" in out
+
+
+@pytest.mark.unit
+@pytest.mark.usefixtures("reset_mt_config")
+def test_onResponseTraceRoute_routing_none(capsys):
+    """Test that onResponseTraceRoute does not print error for ROUTING_APP with NONE errorReason."""
+    iface = MeshInterface(noProto=True)
+
+    packet = {
+        "decoded": {
+            "portnum": "ROUTING_APP",
+            "routing": {"errorReason": "NONE"},
+        }
+    }
+
+    iface.onResponseTraceRoute(packet)
+
+    assert iface._acknowledgment.receivedTraceRoute is True
+    out, _ = capsys.readouterr()
+    assert "Traceroute failed" not in out

meshtastic/tests/test_node.py

@@ -1,17 +1,20 @@
 """Meshtastic unit tests for node.py"""
 # pylint: disable=C0302
 
+import base64
 import logging
 import re
 from unittest.mock import MagicMock, patch
 
 import pytest
+from hypothesis import given, strategies as st
 
-from ..protobuf import admin_pb2, localonly_pb2, config_pb2
+from ..protobuf import admin_pb2, localonly_pb2, config_pb2, mesh_pb2, nanopb_pb2
 from ..protobuf.channel_pb2 import Channel # pylint: disable=E0611
 from ..node import Node
 from ..serial_interface import SerialInterface
 from ..mesh_interface import MeshInterface
+from ..util import to_node_num
 
 # from ..config_pb2 import Config
 # from ..cannedmessages_pb2 import (CannedMessagePluginMessagePart1, CannedMessagePluginMessagePart2,
@@ -19,6 +22,11 @@ from ..mesh_interface import MeshInterface
 #                                  CannedMessagePluginMessagePart5)
 # from ..util import Timeout
 
+# Extract nanopb max_size constraints from the User protobuf descriptor
+_USER_NANOPB = {
+    field.name: field.GetOptions().Extensions[nanopb_pb2.nanopb]
+    for field in mesh_pb2.User.DESCRIPTOR.fields
+}
 
 @pytest.mark.unit
 def test_node(capsys):
@@ -339,6 +347,248 @@ def test_setURL_valid_URL_but_no_settings(capsys):
     assert err == ""
 
 
+@pytest.mark.unit
+@pytest.mark.parametrize("node_id,node_data,should_ignore,manually_verified", [
+    pytest.param(
+        "!830f522a",
+        {
+            "num": 2198819370,
+            "user": {
+                "id": "!830f522a",
+                "longName": "Roadrunner Ridge",
+                "shortName": "RKSN",
+                "macaddr": "AAAAAAAAAAA=",
+                "hwModel": "RAK4631",
+                "role": "ROUTER",
+                "publicKey": "Rx8XD96uBAiFGoFusdqwti3eBT4DLyGuG7g5Wcg9Bw==",
+                "isLicensed": True,
+                "isUnmessagable": False,
+            },
+        },
+        True,
+        True,
+        id="all_fields_all_flags",
+    ),
+    pytest.param(
+        "!12345678",
+        {
+            "num": 305419896,
+            "user": {
+                "id": "!12345678",
+                "longName": "Test Node",
+                "shortName": "TN",
+                "macaddr": "QkVTVEVWRVI=",
+                "hwModel": "TBEAM",
+            },
+        },
+        False,
+        False,
+        id="minimal_fields_no_flags",
+    ),
+    pytest.param(
+        305419896,
+        {
+            "num": 305419896,
+            "user": {
+                "id": "!12345678",
+                "longName": "Another Node",
+                "shortName": "AN",
+                "macaddr": "QkVTVEVWRVI=",
+                "hwModel": "HELTEC_V3",
+                "role": "CLIENT",
+                "publicKey": "AAAAAAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=",
+                "isLicensed": False,
+            },
+        },
+        True,
+        False,
+        id="int_node_id_should_ignore_only",
+    ),
+    pytest.param(
+        "!deadbeef",
+        {
+            "num": 3735928559,
+            "user": {
+                "id": "!deadbeef",
+                "longName": "Minimal Contact",
+                "shortName": "MC",
+                "macaddr": "BQYHCAkKCw==",
+                "hwModel": "UNSET",
+                "role": "CLIENT_MUTE",
+            },
+        },
+        False,
+        True,
+        id="unset_hw_model_verified_only",
+    ),
+    pytest.param(
+        "!1a2b3c4d",
+        {
+            "num": 439041101,
+            "user": {
+                "id": "!1a2b3c4d",
+                "longName": "Licensed Node",
+                "shortName": "LN",
+                "macaddr": "DA0ODxAREg==",
+                "hwModel": "NANO_G1",
+                "isLicensed": True,
+                "isUnmessagable": True,
+            },
+        },
+        False,
+        False,
+        id="licensed_unmessagable_no_flags",
+    ),
+])
+def test_contact_url_roundtrip(node_id, node_data, should_ignore, manually_verified):
+    """Verify that contact URL generation via getContactURL() and parsing via addContactURL() is fully reversible"""
+    iface = MagicMock(autospec=MeshInterface)
+    node_num = to_node_num(node_id)
+    iface.nodesByNum = {node_num: node_data}
+    iface.localNode = None
+
+    anode = Node(iface, node_num, noProto=True)
+
+    sent_admin = []
+    def capture_send(p, *_args, **_kwargs):
+        sent_admin.append(p)
+
+    with patch.object(anode, "_sendAdmin", side_effect=capture_send):
+        url = anode.getContactURL(node_id, should_ignore=should_ignore, manually_verified=manually_verified)
+        assert url.startswith("https://meshtastic.org/v/#")
+
+        anode.addContactURL(url)
+
+    assert len(sent_admin) == 1
+    contact = sent_admin[0].add_contact
+    u = node_data["user"]
+
+    assert contact.node_num == node_num
+    assert contact.user.id == u["id"]
+    assert contact.user.long_name == u["longName"]
+    assert contact.user.short_name == u["shortName"]
+    assert contact.user.macaddr == base64.b64decode(u["macaddr"])
+
+    if u.get("hwModel") and u["hwModel"] != "UNSET":
+        assert contact.user.hw_model == mesh_pb2.HardwareModel.Value(u["hwModel"])
+    if u.get("role"):
+        assert contact.user.role == config_pb2.Config.DeviceConfig.Role.Value(u["role"])
+    if u.get("publicKey"):
+        assert contact.user.public_key == base64.b64decode(u["publicKey"])
+    if u.get("isLicensed"):
+        assert contact.user.is_licensed is True
+    if u.get("isUnmessagable") is not None:
+        assert contact.user.is_unmessagable == u["isUnmessagable"]
+
+    assert contact.should_ignore == should_ignore
+    assert contact.manually_verified == manually_verified
+
+
+@st.composite
+def contact_url_roundtrip_params(draw):
+    """Hypothesis strategy: generate a full node config and roundtrip flags"""
+    should_ignore = draw(st.booleans())
+    manually_verified = draw(st.booleans())
+
+    node_num = draw(st.integers(min_value=6, max_value=2**32 - 2))
+    node_id = f"!{node_num:08x}"
+
+    hw_model = draw(st.sampled_from(list(mesh_pb2.HardwareModel.keys())))
+    role = draw(st.one_of(
+        st.none(),
+        st.sampled_from(list(config_pb2.Config.DeviceConfig.Role.keys())),
+    ))
+
+    long_name = draw(st.text(
+        min_size=1, max_size=_USER_NANOPB['long_name'].max_size
+    ))
+    short_name = draw(st.text(
+        min_size=1, max_size=_USER_NANOPB['short_name'].max_size
+    ))
+
+    macaddr_bytes = draw(st.binary(
+        min_size=_USER_NANOPB['macaddr'].max_size,
+        max_size=_USER_NANOPB['macaddr'].max_size,
+    ))
+    macaddr_b64 = base64.b64encode(macaddr_bytes).decode("ascii")
+
+    has_public_key = draw(st.booleans())
+    public_key_b64 = None
+    if has_public_key:
+        pk_bytes = draw(st.binary(
+            min_size=_USER_NANOPB['public_key'].max_size,
+            max_size=_USER_NANOPB['public_key'].max_size,
+        ))
+        public_key_b64 = base64.b64encode(pk_bytes).decode("ascii")
+
+    is_licensed = draw(st.booleans())
+    is_unmessagable = draw(st.booleans())
+
+    node_data = {
+        "num": node_num,
+        "user": {
+            "id": node_id,
+            "longName": long_name,
+            "shortName": short_name,
+            "macaddr": macaddr_b64,
+            "hwModel": hw_model,
+            "isLicensed": is_licensed,
+            "isUnmessagable": is_unmessagable,
+        },
+    }
+    if role is not None:
+        node_data["user"]["role"] = role
+    if public_key_b64 is not None:
+        node_data["user"]["publicKey"] = public_key_b64
+
+    return node_num, node_data, should_ignore, manually_verified
+
+
+@pytest.mark.unit
+@given(contact_url_roundtrip_params())
+def test_contact_url_roundtrip_hypothesis(params):
+    """Property: roundtrip preserves data across random field configurations"""
+    node_num, node_data, should_ignore, manually_verified = params
+
+    iface = MagicMock(autospec=MeshInterface)
+    iface.nodesByNum = {node_num: node_data}
+    iface.localNode = None
+
+    anode = Node(iface, node_num, noProto=True)
+
+    sent_admin = []
+    def capture_send(p, *_args, **_kwargs):
+        sent_admin.append(p)
+
+    with patch.object(anode, "_sendAdmin", side_effect=capture_send):
+        url = anode.getContactURL(
+            node_num,
+            should_ignore=should_ignore,
+            manually_verified=manually_verified,
+        )
+        anode.addContactURL(url)
+
+    assert len(sent_admin) == 1
+    contact = sent_admin[0].add_contact
+    u = node_data["user"]
+
+    assert contact.node_num == node_num
+    assert contact.user.id == u["id"]
+    assert contact.user.long_name == u["longName"]
+    assert contact.user.short_name == u["shortName"]
+    assert contact.user.macaddr == base64.b64decode(u["macaddr"])
+    assert contact.user.hw_model == mesh_pb2.HardwareModel.Value(u["hwModel"])
+
+    if "role" in u:
+        assert contact.user.role == config_pb2.Config.DeviceConfig.Role.Value(u["role"])
+    if "publicKey" in u:
+        assert contact.user.public_key == base64.b64decode(u["publicKey"])
+    assert contact.user.is_licensed == u["isLicensed"]
+    assert contact.user.is_unmessagable == u["isUnmessagable"]
+    assert contact.should_ignore == should_ignore
+    assert contact.manually_verified == manually_verified
+
+
 # TODO
 # @pytest.mark.unit
 # def test_showChannels(capsys):

meshtastic/tests/test_showNodes_favorite.py

@@ -153,25 +153,6 @@ def test_showNodes_favorite_asterisk_display(capsys, _iface_with_favorite_nodes)
     assert err == ""
 
 
-@pytest.mark.unit
-def test_showNodes_favorite_field_formatting():
-    """Test the formatting logic for isFavorite field"""
-    # Test favorite node
-    raw_value = True
-    formatted_value = "*" if raw_value else ""
-    assert formatted_value == "*"
-
-    # Test non-favorite node
-    raw_value = False
-    formatted_value = "*" if raw_value else ""
-    assert formatted_value == ""
-
-    # Test None/missing value
-    raw_value = None
-    formatted_value = "*" if raw_value else ""
-    assert formatted_value == ""
-
-
 @pytest.mark.unit
 def test_showNodes_with_custom_fields_including_favorite(capsys, _iface_with_favorite_nodes):
     """Test that isFavorite can be specified in custom showFields"""

meshtastic/tests/test_util.py

@@ -3,13 +3,15 @@
 import json
 import logging
 import re
+import time
+from types import SimpleNamespace
 from unittest.mock import patch
 
 import pytest
 from hypothesis import given, strategies as st
 
 from meshtastic.supported_device import SupportedDevice
-from meshtastic.protobuf import mesh_pb2
+from meshtastic.protobuf import mesh_pb2, config_pb2
 from meshtastic.util import (
     DEFAULT_KEY,
     Timeout,
@@ -20,6 +22,8 @@ from meshtastic.util import (
     convert_mac_addr,
     eliminate_duplicate_port,
     findPorts,
+    flags_from_list,
+    flags_to_list,
     fixme,
     fromPSK,
     fromStr,
@@ -37,6 +41,7 @@ from meshtastic.util import (
     stripnl,
     support_info,
     message_to_json,
+    to_node_num,
     Acknowledgment
 )
 
@@ -234,20 +239,111 @@ def test_remove_keys_from_dict_nested():
     assert remove_keys_from_dict(keys, adict) == exp
 
 
-@pytest.mark.unitslow
+@pytest.mark.unit
-def test_Timeout_not_found():
+def test_Timeout_waitForSet_found():
-    """Test Timeout()"""
+    """waitForSet returns True when all required attrs are truthy on target."""
-    to = Timeout(0.2)
+    to = Timeout(0.01)
-    attrs = "foo"
+    target = SimpleNamespace(foo=1, bar="ok")
-    to.waitForSet("bar", attrs)
+    assert to.waitForSet(target, ("foo", "bar")) is True
 
 
-@pytest.mark.unitslow
+@pytest.mark.unit
-def test_Timeout_found():
+@patch("meshtastic.util.time.sleep")
-    """Test Timeout()"""
+def test_Timeout_waitForSet_not_found(mock_sleep):  # pylint: disable=unused-argument
-    to = Timeout(0.2)
+    """waitForSet returns False when attrs remain falsy until timeout."""
-    attrs = ()
+    to = Timeout(0.01)
-    to.waitForSet("bar", attrs)
+    target = SimpleNamespace(foo=None, bar=0)
+    assert to.waitForSet(target, ("foo", "bar")) is False
+
+
+@pytest.mark.unit
+def test_Timeout_waitForSet_empty_attrs():
+    """waitForSet returns True immediately when attrs is empty (vacuous truth)."""
+    to = Timeout(0.01)
+    assert to.waitForSet(object(), ()) is True
+
+
+@pytest.mark.unit
+def test_Timeout_reset():
+    """reset() sets expireTime to now + expireTimeout."""
+    to = Timeout(maxSecs=5)
+    before = time.time()
+    to.reset()
+    assert to.expireTime == pytest.approx(before + 5, abs=0.1)
+
+
+@pytest.mark.unit
+def test_Timeout_reset_custom():
+    """reset(expireTimeout) overrides the stored expireTimeout."""
+    to = Timeout(maxSecs=5)
+    before = time.time()
+    to.reset(expireTimeout=10)
+    assert to.expireTime == pytest.approx(before + 10, abs=0.1)
+
+
+@pytest.mark.unit
+def test_Timeout_waitForAckNak_found():
+    """waitForAckNak returns True when any acknowledgment attr is set, and clears it."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    ack.receivedAck = True
+    assert to.waitForAckNak(ack) is True
+    assert ack.receivedAck is False  # cleared after detection
+
+
+@pytest.mark.unit
+@patch("meshtastic.util.time.sleep")
+def test_Timeout_waitForAckNak_not_found(mock_sleep):  # pylint: disable=unused-argument
+    """waitForAckNak returns False when no acknowledgment attr is set."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    assert to.waitForAckNak(ack) is False
+
+
+@pytest.mark.unit
+def test_Timeout_waitForTraceRoute_found():
+    """waitForTraceRoute returns True on receivedTraceRoute, and clears it."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    ack.receivedTraceRoute = True
+    assert to.waitForTraceRoute(3.0, ack) is True
+    assert ack.receivedTraceRoute is False
+
+
+@pytest.mark.unit
+@patch("meshtastic.util.time.sleep")
+def test_Timeout_waitForTraceRoute_not_found(mock_sleep):  # pylint: disable=unused-argument
+    """waitForTraceRoute returns False on timeout."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    assert to.waitForTraceRoute(3.0, ack) is False
+
+
+@pytest.mark.unit
+def test_Timeout_waitForTelemetry_found():
+    """waitForTelemetry returns True when receivedTelemetry is set."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    ack.receivedTelemetry = True
+    assert to.waitForTelemetry(ack) is True
+
+
+@pytest.mark.unit
+def test_Timeout_waitForPosition_found():
+    """waitForPosition returns True when receivedPosition is set."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    ack.receivedPosition = True
+    assert to.waitForPosition(ack) is True
+
+
+@pytest.mark.unit
+def test_Timeout_waitForWaypoint_found():
+    """waitForWaypoint returns True when receivedWaypoint is set."""
+    to = Timeout(0.01)
+    ack = Acknowledgment()
+    ack.receivedWaypoint = True
+    assert to.waitForWaypoint(ack) is True
 
 
 @pytest.mark.unitslow
@@ -715,3 +811,162 @@ def test_generate_channel_hash_fuzz_aes256(channel_name, key_bytes):
     "Test generate_channel_hash with fuzzed channel names and 256-bit keys, ensuring it produces single-byte values"
     hashed = generate_channel_hash(channel_name, key_bytes)
     assert 0 <= hashed <= 0xFF
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("input_val,expected", [
+    # int passthrough
+    (0, 0),
+    (1, 1),
+    (6, 6),
+    (502009325, 502009325),
+    (2198819370, 2198819370),
+    (0xFFFFFFFF, 0xFFFFFFFF),
+    # !hex format (always treated as hex)
+    ("!00000000", 0x00000000),
+    ("!00000001", 0x00000001),
+    ("!00000010", 0x00000010),
+    ("!000000ff", 0x000000FF),
+    ("!830f522a", 0x830F522A),
+    ("!1dec0ded", 0x1DEC0DED),
+    ("!ffffffff", 0xFFFFFFFF),
+    ("!FFFFFFFF", 0xFFFFFFFF),
+    # 0xhex format
+    ("0x00000000", 0x00000000),
+    ("0x00000010", 0x00000010),
+    ("0x830f522a", 0x830F522A),
+    ("0x1dec0ded", 0x1DEC0DED),
+    ("0xFFFFFFFF", 0xFFFFFFFF),
+    # Unprefixed hex string (falls back to hex when decimal fails)
+    ("830f522a", 0x830F522A),
+    ("1dec0ded", 0x1DEC0DED),
+    # Decimal string
+    ("42", 42),
+    ("12345678", 12345678),
+    ("0", 0),
+    ("1", 1),
+    # With whitespace
+    ("  !830f522a  ", 2198819370),
+    ("  !00000010  ", 16),
+    ("  0x830f522a  ", 2198819370),
+])
+def test_to_node_num(input_val, expected):
+    """Test to_node_num with various valid inputs"""
+    assert to_node_num(input_val) == expected
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("input_val", [
+    "",
+    "!",
+    "!!",
+    "!0x10",
+    "!xyz",
+])
+def test_to_node_num_invalid(input_val):
+    """Test to_node_num raises ValueError for invalid inputs"""
+    with pytest.raises(ValueError):
+        to_node_num(input_val)
+
+
+@pytest.mark.unit
+@given(st.integers(min_value=0, max_value=2**32 - 1))
+def test_to_node_num_hypothesis_roundtrip(n):
+    """Property: all supported input formats roundtrip for any valid node number"""
+    assert to_node_num(n) == n
+    assert to_node_num(f"!{n:08x}") == n
+    assert to_node_num(f"0x{n:x}") == n
+    assert to_node_num(str(n)) == n
+
+
+_EXCLUDED_MODULES = mesh_pb2.ExcludedModules
+_POSITION_FLAGS = config_pb2.Config.PositionConfig.PositionFlags
+_NETWORK_PROTOCOLS = config_pb2.Config.NetworkConfig.ProtocolFlags
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("flag_type,flags,expected", [
+    (_EXCLUDED_MODULES, 0, []),
+    (_EXCLUDED_MODULES, 1, ["MQTT_CONFIG"]),
+    (_EXCLUDED_MODULES, 3, ["MQTT_CONFIG", "SERIAL_CONFIG"]),
+    (_EXCLUDED_MODULES, 0x7FFF, [
+        "MQTT_CONFIG", "SERIAL_CONFIG", "EXTNOTIF_CONFIG", "STOREFORWARD_CONFIG",
+        "RANGETEST_CONFIG", "TELEMETRY_CONFIG", "CANNEDMSG_CONFIG", "AUDIO_CONFIG",
+        "REMOTEHARDWARE_CONFIG", "NEIGHBORINFO_CONFIG", "AMBIENTLIGHTING_CONFIG",
+        "DETECTIONSENSOR_CONFIG", "PAXCOUNTER_CONFIG", "BLUETOOTH_CONFIG",
+        "NETWORK_CONFIG",
+    ]),
+    (_EXCLUDED_MODULES, 0x8000, ["UNKNOWN_ADDITIONAL_FLAGS(32768)"]),
+    (_EXCLUDED_MODULES, 0x8001, ["MQTT_CONFIG", "UNKNOWN_ADDITIONAL_FLAGS(32768)"]),
+    (_POSITION_FLAGS, 0, []),
+    (_POSITION_FLAGS, 0x09, ["ALTITUDE", "DOP"]),
+    (_POSITION_FLAGS, 0x1FF, [
+        "ALTITUDE", "ALTITUDE_MSL", "GEOIDAL_SEPARATION", "DOP", "HVDOP",
+        "SATINVIEW", "SEQ_NO", "TIMESTAMP", "HEADING",
+    ]),
+])
+def test_flags_to_list(flag_type, flags, expected):
+    """Test flags_to_list decodes set bits in enum order and reports unknown remainders."""
+    assert flags_to_list(flag_type, flags) == expected
+
+
+@pytest.mark.unit
+@given(st.integers(min_value=0, max_value=0xFFFFF))
+def test_flags_to_list_conservation(flags):
+    """Property: flags_to_list partitions `flags` into known names plus an exact unknown remainder.
+
+    Every known bit that is set must appear as a name, and the leftover reported in
+    UNKNOWN_ADDITIONAL_FLAGS(...) must together with the named bits reconstruct the input.
+    """
+    for flag_type in (_EXCLUDED_MODULES, _POSITION_FLAGS):
+        known_union = 0
+        for key in flag_type.keys():
+            value = flag_type.Value(key)
+            if key != "EXCLUDED_NONE" and value:
+                known_union |= value
+
+        result = flags_to_list(flag_type, flags)
+
+        accounted = 0
+        leftover = 0
+        for name in result:
+            if name.startswith("UNKNOWN_ADDITIONAL_FLAGS("):
+                leftover = int(name[len("UNKNOWN_ADDITIONAL_FLAGS("):-1])
+            else:
+                accounted |= flag_type.Value(name)
+
+        assert accounted == (flags & known_union)
+        assert (accounted | leftover) == flags
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("flag_type, flags, expected", [
+    (_NETWORK_PROTOCOLS, ["UDP_BROADCAST"], 1),
+    (_NETWORK_PROTOCOLS, ["NO_BROADCAST"], 0),
+    (_NETWORK_PROTOCOLS, [], 0),
+    (_POSITION_FLAGS, ["ALTITUDE"], 1),
+    (_POSITION_FLAGS, ["ALTITUDE", "SPEED"], 513),
+    (_POSITION_FLAGS, ["ALTITUDE", " SPEED "], 513),
+])
+def test_flags_from_list(flag_type, flags, expected):
+    """Test flags_from_list combines named flags into the expected bitmask."""
+    assert flags_from_list(flag_type, flags) == expected
+
+
+@pytest.mark.unit
+def test_flags_from_list_unknown_flag():
+    """Test flags_from_list raises ValueError for an unknown flag name."""
+    with pytest.raises(ValueError, match="Unknown flag 'TCP'"):
+        flags_from_list(_NETWORK_PROTOCOLS, ["UDP_BROADCAST", "TCP"])
+
+
+@pytest.mark.unit
+@given(st.lists(st.sampled_from(list(_POSITION_FLAGS.keys())), unique=True))
+def test_flags_from_list_roundtrip(flags):
+    """Property: flags_from_list and flags_to_list are inverses for known position flags."""
+    combined = flags_from_list(_POSITION_FLAGS, flags)
+    decoded = flags_to_list(_POSITION_FLAGS, combined)
+    # flags_to_list drops zero-value flags and may report unknown remainders,
+    # but for combinations of known non-zero flags it should return the same set of names.
+    nonzero_flags = {f for f in flags if _POSITION_FLAGS.Value(f)}
+    assert set(decoded) == nonzero_flags

meshtastic/util.py

@@ -728,7 +728,7 @@ def to_node_num(node_id: Union[int, str]) -> int:
         return node_id
     s = str(node_id).strip()
     if s.startswith("!"):
-        s = s[1:]
+        s = "0x" + s[1:]
     if s.lower().startswith("0x"):
         return int(s, 16)
     try:
@@ -737,14 +737,41 @@ def to_node_num(node_id: Union[int, str]) -> int:
         return int(s, 16)
 
 def flags_to_list(flag_type, flags: int) -> List[str]:
-    """Given a flag_type that's a protobuf EnumTypeWrapper, and a flag int, give a list of flags enabled."""
+    """Given a flag_type that's a protobuf EnumTypeWrapper, and a flag int, give a list of flags enabled.
+
+    Zero-valued members (e.g. EXCLUDED_NONE, UNSET, NO_BROADCAST) never appear in the
+    result: they hold no bit, so `flags & value` is always False for them, and a flags
+    value of 0 therefore decodes to an empty list rather than a named "no flags" entry.
+    Any leftover bits not corresponding to a known member are reported via
+    `UNKNOWN_ADDITIONAL_FLAGS(<leftover>)`.
+    """
     ret = []
     for key in flag_type.keys():
-        if key == "EXCLUDED_NONE":
-            continue
         if flags & flag_type.Value(key):
             ret.append(key)
             flags = flags - flag_type.Value(key)
     if flags > 0:
         ret.append(f"UNKNOWN_ADDITIONAL_FLAGS({flags})")
     return ret
+
+
+def flags_from_list(flag_type, flags: List[str]) -> int:
+    """Given a flag_type that's a protobuf EnumTypeWrapper, and a list of flag names, return the combined bitmask.
+
+    Zero-valued members (e.g. EXCLUDED_NONE, UNSET, NO_BROADCAST) are accepted but are
+    no-ops: they OR in 0 and thus set nothing. A list consisting solely of such a member
+    (or an empty list) yields 0, which round-trips back through flags_to_list as an
+    empty list rather than the original member name -- see flags_to_list's docstring.
+    """
+    result = 0
+    valid_names = list(flag_type.keys())
+    for flag_name in flags:
+        flag_name = flag_name.strip()
+        if not flag_name:
+            continue
+        if flag_name not in valid_names:
+            raise ValueError(
+                f"Unknown flag '{flag_name}'. Valid choices: {', '.join(sorted(valid_names))}"
+            )
+        result |= flag_type.Value(flag_name)
+    return result

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "meshtastic"
-version = "2.7.9"
+version = "2.7.10"
 description = "Python API & client shell for talking to Meshtastic devices"
 authors = ["Meshtastic Developers <contact@meshtastic.org>"]
 license = "GPL-3.0-only"