mirror of
https://github.com/meshtastic/python.git
synced 2026-01-18 20:57:57 -05:00
appease pylint, tests, mypy
This commit is contained in:
@@ -203,7 +203,7 @@ def _onAdminReceive(iface, asDict):
|
||||
"""Special auto parsing for received messages"""
|
||||
logging.debug(f"in _onAdminReceive() asDict:{asDict}")
|
||||
if "decoded" in asDict and "from" in asDict and "admin" in asDict["decoded"]:
|
||||
adminMessage: admin_pb2.AdminMessage = asDict["decoded"]["admin"]["raw"]
|
||||
adminMessage = asDict["decoded"]["admin"]["raw"]
|
||||
iface._getOrCreateByNum(asDict["from"])["adminSessionPassKey"] = adminMessage.session_passkey
|
||||
|
||||
"""Well known message payloads can register decoders for automatic protobuf parsing"""
|
||||
|
||||
@@ -854,9 +854,9 @@ class Node:
|
||||
logging.debug(f"adminIndex:{adminIndex}")
|
||||
if isinstance(self.nodeNum, int):
|
||||
nodeid = self.nodeNum
|
||||
elif self.nodeNum.startswith("!"):
|
||||
else: # assume string starting with !
|
||||
nodeid = int(self.nodeNum[1:],16)
|
||||
if ("adminSessionPassKey" in self.iface._getOrCreateByNum(nodeid)):
|
||||
if "adminSessionPassKey" in self.iface._getOrCreateByNum(nodeid):
|
||||
p.session_passkey = self.iface._getOrCreateByNum(nodeid).get("adminSessionPassKey")
|
||||
return self.iface.sendData(
|
||||
p,
|
||||
@@ -869,9 +869,15 @@ class Node:
|
||||
)
|
||||
|
||||
def ensureSessionKey(self):
|
||||
if isinstance(self.nodeNum, int):
|
||||
nodeid = self.nodeNum
|
||||
elif self.nodeNum.startswith("!"):
|
||||
nodeid = int(self.nodeNum[1:],16)
|
||||
if self.iface._getOrCreateByNum(nodeid).get("adminSessionPassKey") is None:
|
||||
self.requestConfig(admin_pb2.AdminMessage.SESSIONKEY_CONFIG)
|
||||
"""If our entry in iface.nodesByNum doesn't already have an adminSessionPassKey, make a request to get one"""
|
||||
if self.noProto:
|
||||
logging.warning(
|
||||
f"Not ensuring session key, because protocol use is disabled by noProto"
|
||||
)
|
||||
else:
|
||||
if isinstance(self.nodeNum, int):
|
||||
nodeid = self.nodeNum
|
||||
else: # assume string starting with !
|
||||
nodeid = int(self.nodeNum[1:],16)
|
||||
if self.iface._getOrCreateByNum(nodeid).get("adminSessionPassKey") is None:
|
||||
self.requestConfig(admin_pb2.AdminMessage.SESSIONKEY_CONFIG)
|
||||
|
||||
@@ -231,7 +231,9 @@ def test_node(capsys):
|
||||
@pytest.mark.unit
|
||||
def test_exitSimulator(caplog):
|
||||
"""Test exitSimulator"""
|
||||
anode = Node("foo", "bar", noProto=True)
|
||||
interface = MeshInterface()
|
||||
interface.nodesByNum = {}
|
||||
anode = Node(interface, "!ba400000", noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
anode.exitSimulator()
|
||||
assert re.search(r"in exitSimulator", caplog.text, re.MULTILINE)
|
||||
@@ -240,7 +242,9 @@ def test_exitSimulator(caplog):
|
||||
@pytest.mark.unit
|
||||
def test_reboot(caplog):
|
||||
"""Test reboot"""
|
||||
anode = Node(MeshInterface(), 1234567890, noProto=True)
|
||||
interface = MeshInterface()
|
||||
interface.nodesByNum = {}
|
||||
anode = Node(interface, 1234567890, noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
anode.reboot()
|
||||
assert re.search(r"Telling node to reboot", caplog.text, re.MULTILINE)
|
||||
@@ -249,7 +253,9 @@ def test_reboot(caplog):
|
||||
@pytest.mark.unit
|
||||
def test_shutdown(caplog):
|
||||
"""Test shutdown"""
|
||||
anode = Node(MeshInterface(), 1234567890, noProto=True)
|
||||
interface = MeshInterface()
|
||||
interface.nodesByNum = {}
|
||||
anode = Node(interface, 1234567890, noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
anode.shutdown()
|
||||
assert re.search(r"Telling node to shutdown", caplog.text, re.MULTILINE)
|
||||
|
||||
Reference in New Issue
Block a user