mirror of
https://github.com/meshtastic/python.git
synced 2026-07-31 09:06:28 -04:00
Compare commits
12 Commits
2.7.9
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4f2ce3dfc2 | ||
|
|
307d8d81e7 | ||
|
|
405a6bbc5d | ||
|
|
db746a0981 | ||
|
|
13b8cdcb04 | ||
|
|
9d445098f4 | ||
|
|
8c84074c1d | ||
|
|
1cffae6add | ||
|
|
e4f0fb222b | ||
|
|
953d01206b | ||
|
|
c71d3df332 | ||
|
|
0413d00825 |
@@ -552,6 +552,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 +1081,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
|
||||
|
||||
@@ -1858,6 +1879,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.",
|
||||
@@ -2095,6 +2134,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.",
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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}")
|
||||
|
||||
@@ -2998,6 +2998,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):
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -37,6 +37,7 @@ from meshtastic.util import (
|
||||
stripnl,
|
||||
support_info,
|
||||
message_to_json,
|
||||
to_node_num,
|
||||
Acknowledgment
|
||||
)
|
||||
|
||||
@@ -715,3 +716,69 @@ 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
|
||||
|
||||
@@ -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:
|
||||
|
||||
90
poetry.lock
generated
90
poetry.lock
generated
@@ -1,4 +1,4 @@
|
||||
# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
|
||||
# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand.
|
||||
|
||||
[[package]]
|
||||
name = "altgraph"
|
||||
@@ -3552,7 +3552,7 @@ description = "Type annotations for pandas"
|
||||
optional = true
|
||||
python-versions = ">=3.9"
|
||||
groups = ["main"]
|
||||
markers = "python_version < \"3.11\" and extra == \"analysis\""
|
||||
markers = "extra == \"analysis\""
|
||||
files = [
|
||||
{file = "pandas_stubs-2.2.2.240807-py3-none-any.whl", hash = "sha256:893919ad82be4275f0d07bb47a95d08bae580d3fdea308a7acfcb3f02e76186e"},
|
||||
{file = "pandas_stubs-2.2.2.240807.tar.gz", hash = "sha256:64a559725a57a449f46225fbafc422520b7410bff9252b661a225b5559192a93"},
|
||||
@@ -3562,23 +3562,6 @@ files = [
|
||||
numpy = ">=1.23.5"
|
||||
types-pytz = ">=2022.1.1"
|
||||
|
||||
[[package]]
|
||||
name = "pandas-stubs"
|
||||
version = "2.3.2.250926"
|
||||
description = "Type annotations for pandas"
|
||||
optional = true
|
||||
python-versions = ">=3.10"
|
||||
groups = ["main"]
|
||||
markers = "extra == \"analysis\" and python_version >= \"3.11\""
|
||||
files = [
|
||||
{file = "pandas_stubs-2.3.2.250926-py3-none-any.whl", hash = "sha256:81121818453dcfe00f45c852f4dceee043640b813830f6e7bd084a4ef7ff7270"},
|
||||
{file = "pandas_stubs-2.3.2.250926.tar.gz", hash = "sha256:c64b9932760ceefb96a3222b953e6a251321a9832a28548be6506df473a66406"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
numpy = ">=1.23.5"
|
||||
types-pytz = ">=2022.1.1"
|
||||
|
||||
[[package]]
|
||||
name = "pandocfilters"
|
||||
version = "1.5.1"
|
||||
@@ -4054,7 +4037,6 @@ description = ""
|
||||
optional = false
|
||||
python-versions = ">=3.9"
|
||||
groups = ["main", "dev"]
|
||||
markers = "python_version < \"3.11\""
|
||||
files = [
|
||||
{file = "protobuf-6.33.6-cp310-abi3-win32.whl", hash = "sha256:7d29d9b65f8afef196f8334e80d6bc1d5d4adedb449971fefd3723824e6e77d3"},
|
||||
{file = "protobuf-6.33.6-cp310-abi3-win_amd64.whl", hash = "sha256:0cd27b587afca21b7cfa59a74dcbd48a50f0a6400cfb59391340ad729d91d326"},
|
||||
@@ -4068,25 +4050,6 @@ files = [
|
||||
{file = "protobuf-6.33.6.tar.gz", hash = "sha256:a6768d25248312c297558af96a9f9c929e8c4cee0659cb07e780731095f38135"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "protobuf"
|
||||
version = "7.35.0"
|
||||
description = ""
|
||||
optional = false
|
||||
python-versions = ">=3.10"
|
||||
groups = ["main", "dev"]
|
||||
markers = "python_version >= \"3.11\""
|
||||
files = [
|
||||
{file = "protobuf-7.35.0-cp310-abi3-macosx_10_9_universal2.whl", hash = "sha256:66be6c513931c794fa92c080ffee41671390da3d79da219cf9c0c0907f035dda"},
|
||||
{file = "protobuf-7.35.0-cp310-abi3-manylinux2014_aarch64.whl", hash = "sha256:fcbe42a4ac09d3ec9c987ddfcd956afd0b15f1ff613bd8371bde9405ffd5c8e5"},
|
||||
{file = "protobuf-7.35.0-cp310-abi3-manylinux2014_s390x.whl", hash = "sha256:4cbf5cc286130e06a6c9bbefac442431173906dfcc979712183d4adcc01b37ee"},
|
||||
{file = "protobuf-7.35.0-cp310-abi3-manylinux2014_x86_64.whl", hash = "sha256:6c0f98f10c8a05ea30f8993dfef2de093d27b490fdae78bb60c8343795d55011"},
|
||||
{file = "protobuf-7.35.0-cp310-abi3-win32.whl", hash = "sha256:4c4617b83ade0e279d1d2bfe04025a1adb87f9ed657de038620dc0ff959357f6"},
|
||||
{file = "protobuf-7.35.0-cp310-abi3-win_amd64.whl", hash = "sha256:f05bcadf9a2a6b8dda047007075135fb7d08c73d9177aabc067e1be46881a201"},
|
||||
{file = "protobuf-7.35.0-py3-none-any.whl", hash = "sha256:c13f325cf242bad135c350629eeb5d54b24228eb472fb3e2e9ebbd4c5dc20ca0"},
|
||||
{file = "protobuf-7.35.0.tar.gz", hash = "sha256:a2efd84605f41e559f1881b0912b44099d0a2ac9bf46b3474823f10fb393b0e6"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "psutil"
|
||||
version = "7.1.3"
|
||||
@@ -4443,6 +4406,7 @@ python-versions = ">=3.3, <4"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "Pypubsub-4.0.3-py3-none-any.whl", hash = "sha256:7f716bae9388afe01ff82b264ba8a96a8ae78b42bb1f114f2716ca8f9e404e2a"},
|
||||
{file = "pypubsub-4.0.3.tar.gz", hash = "sha256:32d662de3ade0fb0880da92df209c62a4803684de5ccb8d19421c92747a258c7"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4874,7 +4838,6 @@ description = "Python HTTP for Humans."
|
||||
optional = false
|
||||
python-versions = ">=3.9"
|
||||
groups = ["main", "analysis"]
|
||||
markers = "python_version < \"3.11\""
|
||||
files = [
|
||||
{file = "requests-2.32.5-py3-none-any.whl", hash = "sha256:2462f94637a34fd532264295e186976db0f5d453d1cdd31473c85a6a161affb6"},
|
||||
{file = "requests-2.32.5.tar.gz", hash = "sha256:dbba0bac56e100853db0ea71b82b4dfd5fe2bf6d3754a8893c3af500cec7d7cf"},
|
||||
@@ -4890,29 +4853,6 @@ urllib3 = ">=1.21.1,<3"
|
||||
socks = ["PySocks (>=1.5.6,!=1.5.7)"]
|
||||
use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
|
||||
|
||||
[[package]]
|
||||
name = "requests"
|
||||
version = "2.34.2"
|
||||
description = "Python HTTP for Humans."
|
||||
optional = false
|
||||
python-versions = ">=3.10"
|
||||
groups = ["main", "analysis"]
|
||||
markers = "python_version >= \"3.11\""
|
||||
files = [
|
||||
{file = "requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0"},
|
||||
{file = "requests-2.34.2.tar.gz", hash = "sha256:f288924cae4e29463698d6d60bc6a4da69c89185ad1e0bcc4104f584e960b9ed"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
certifi = ">=2023.5.7"
|
||||
charset_normalizer = ">=2,<4"
|
||||
idna = ">=2.5,<4"
|
||||
urllib3 = ">=1.26,<3"
|
||||
|
||||
[package.extras]
|
||||
socks = ["PySocks (>=1.5.6,!=1.5.7)"]
|
||||
use-chardet-on-py3 = ["chardet (>=3.0.2,<8)"]
|
||||
|
||||
[[package]]
|
||||
name = "retrying"
|
||||
version = "1.4.2"
|
||||
@@ -5515,22 +5455,22 @@ files = [
|
||||
|
||||
[[package]]
|
||||
name = "tornado"
|
||||
version = "6.5.6"
|
||||
version = "6.5.7"
|
||||
description = "Tornado is a Python web framework and asynchronous networking library, originally developed at FriendFeed."
|
||||
optional = false
|
||||
python-versions = ">=3.9"
|
||||
groups = ["analysis"]
|
||||
groups = ["analysis", "dev"]
|
||||
files = [
|
||||
{file = "tornado-6.5.6-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:65fcfaafb079435c2c19dc9e07c0f1cf0fa9051759ed0a7d0a3ba7ea7f64919c"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:38bc01b4acacded2de63ae78023548e41ebe6fbed3ec05a796d7ae3ad893887e"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:b942e6a137fda31ff54bf8e6e2c8d1c37f1f50583f3ed53fb840b53b9601d104"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8666946e70171b8c3f1fc9b7876fac492e84822c4c7f3746f4e8f8bc9ac92a79"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:1c34cfab7ad6d104f052f55de06d39bbafc5885cfeb4da688803308dbcfa90b7"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:385f35e4e22fb52551dfcda4cdc8c30c61c2c001aef5ddad99cdfe116952efd3"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-win32.whl", hash = "sha256:db475f1b67b2809b10bb16264829087724ca8d24fe4ed47f7b8675cae453ef86"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-win_amd64.whl", hash = "sha256:6739bf1e8eb09230f1280ddbd3236f0309db70f2c551a8dbc40f62babdf82f79"},
|
||||
{file = "tornado-6.5.6-cp39-abi3-win_arm64.whl", hash = "sha256:2543597b24a695d72338a9a77818362d72387c03ae173f1f169eadc5c91466ac"},
|
||||
{file = "tornado-6.5.6.tar.gz", hash = "sha256:9a365179fe8ff6b8766f602c0f67c185d778193e9bdd828b19f0b6ed7764177d"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:148b2eb15c2c765a50796172c1e499649b35f30d2e3c3d3e15913cfa56bfb163"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:9da38de27f1da3b78a966f0dae12b5a1ea9afe72ca805d84ff06508272ddf100"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:8d759e71906ee783f8867b93bf26a265743da4c1e2f4a018464c1ba019862972"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8a46347a18f23fb92b396beebe0fb78f61dda0cc302445202c16203d8a18848b"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:7778b30bef919231265e91c69963ce0f49a1e9c07ac900bbe75b19ce2575ba92"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:e726f0c75da7726eec023aa62751ff8878bd2737e34fbdd33b1ae5897d2200f5"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-win32.whl", hash = "sha256:f8de3bf12d3efdd0cbe7c8887868198f8a91415e3f29fcf258d9b8eb7b1d9ae4"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-win_amd64.whl", hash = "sha256:de942f843533a039ef9fa3d9c88c7cd8a7c94553fb5ad0154270989b3d99a2c4"},
|
||||
{file = "tornado-6.5.7-cp39-abi3-win_arm64.whl", hash = "sha256:ff934fce95643af5f11efdae618eaa73d469dc588641e5c8d19295a0c65c4796"},
|
||||
{file = "tornado-6.5.7.tar.gz", hash = "sha256:66c513a76cda70d53907bc27cf1447557699c2e95aa48ba27a442ff61c3ddfc2"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
Reference in New Issue
Block a user