mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2025-12-31 11:57:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

wip extracting onResponse() closures to class methods so they can be tested

Browse Source
This commit is contained in:
Mike Kinney
2021-12-22 11:11:17 -08:00
parent 2eeffdcd49
commit 586a3eb463
2 changed files with 179 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
meshtastic/node.py
View File

@@ -8,6 +8,8 @@ from . import portnums_pb2, apponly_pb2, admin_pb2, channel_pb2
from .util import pskToString, stripnl, Timeout, our_exit, fromPSK
class Node:
"""A model of a (local or remote) node in the mesh
@@ -222,26 +224,29 @@ class Node:
self.writeChannel(ch.index)
i = i + 1
def onResponseRequestSettings(self, p):
"""Handle the response packet for requesting settings _requestSettings()"""
logging.debug(f'onResponseRequestSetting() p:{p}')
errorFound = False
if 'routing' in p["decoded"]:
if p["decoded"]["routing"]["errorReason"] != "NONE":
errorFound = True
print(f'Error on response: {p["decoded"]["routing"]["errorReason"]}')
if errorFound is False:
self.radioConfig = p["decoded"]["admin"]["raw"].get_radio_response
logging.debug(f'self.radioConfig:{self.radioConfig}')
logging.debug("Received radio config, now fetching channels...")
self._timeout.reset() # We made foreward progress
self._requestChannel(0) # now start fetching channels
def _requestSettings(self):
"""Done with initial config messages, now send regular
MeshPackets to ask for settings."""
p = admin_pb2.AdminMessage()
p.get_radio_request = True
def onResponse(p):
"""A closure to handle the response packet"""
errorFound = False
if 'routing' in p["decoded"]:
if p["decoded"]["routing"]["errorReason"] != "NONE":
errorFound = True
print(f'Error on response: {p["decoded"]["routing"]["errorReason"]}')
if errorFound is False:
self.radioConfig = p["decoded"]["admin"]["raw"].get_radio_response
logging.debug(f'self.radioConfig:{self.radioConfig}')
logging.debug("Received radio config, now fetching channels...")
self._timeout.reset() # We made foreward progress
self._requestChannel(0) # now start fetching channels
# Show progress message for super slow operations
if self != self.iface.localNode:
print("Requesting preferences from remote node.")
@@ -252,7 +257,7 @@ class Node:
print(" 4. All devices have been rebooted after all of the above. (optional, but recommended)")
print("Note: This could take a while (it requests remote channel configs, then writes config)")
return self._sendAdmin(p, wantResponse=True, onResponse=onResponse)
return self._sendAdmin(p, wantResponse=True, onResponse=self.onResponseRequestSettings)
def exitSimulator(self):
"""Tell a simulator node to exit (this message
@@ -293,6 +298,34 @@ class Node:
self.channels.append(ch)
index += 1
def onResponseRequestChannel(self, p):
"""Handle the response packet for requesting a channel _requestChannel()"""
logging.debug(f'onResponseRequestChannel() p:{p}')
c = p["decoded"]["admin"]["raw"].get_channel_response
self.partialChannels.append(c)
self._timeout.reset() # We made foreward progress
logging.debug(f"Received channel {stripnl(c)}")
index = c.index
# for stress testing, we can always download all channels
fastChannelDownload = True
# Once we see a response that has NO settings, assume
# we are at the end of channels and stop fetching
quitEarly = (c.role == channel_pb2.Channel.Role.DISABLED) and fastChannelDownload
if quitEarly or index >= self.iface.myInfo.max_channels - 1:
logging.debug("Finished downloading channels")
self.channels = self.partialChannels
self._fixupChannels()
# FIXME, the following should only be called after we have settings and channels
self.iface._connected() # Tell everyone else we are ready to go
else:
self._requestChannel(index + 1)
def _requestChannel(self, channelNum: int):
"""Done with initial config messages, now send regular
MeshPackets to ask for settings"""
@@ -306,33 +339,7 @@ class Node:
else:
logging.debug(f"Requesting channel {channelNum}")
def onResponse(p):
"""A closure to handle the response packet for requesting a channel"""
c = p["decoded"]["admin"]["raw"].get_channel_response
self.partialChannels.append(c)
self._timeout.reset() # We made foreward progress
logging.debug(f"Received channel {stripnl(c)}")
index = c.index
# for stress testing, we can always download all channels
fastChannelDownload = True
# Once we see a response that has NO settings, assume
# we are at the end of channels and stop fetching
quitEarly = (c.role == channel_pb2.Channel.Role.DISABLED) and fastChannelDownload
if quitEarly or index >= self.iface.myInfo.max_channels - 1:
logging.debug("Finished downloading channels")
self.channels = self.partialChannels
self._fixupChannels()
# FIXME, the following should only be called after we have settings and channels
self.iface._connected() # Tell everyone else we are ready to go
else:
self._requestChannel(index + 1)
return self._sendAdmin(p, wantResponse=True, onResponse=onResponse)
return self._sendAdmin(p, wantResponse=True, onResponse=self.onResponseRequestChannel)
# pylint: disable=R1710

130
meshtastic/tests/test_node.py
View File

@@ -623,3 +623,133 @@ def test_requestChannel_localNode(caplog):
anode._requestChannel(0)
assert re.search(r'Requesting channel 0', caplog.text, re.MULTILINE)
assert not re.search(r'from remote node', caplog.text, re.MULTILINE)
# TODO:
#@pytest.mark.unit
#def test_onResponseRequestChannel(caplog):
# """Test onResponseRequestChannel()"""
# anode = Node('foo', 'bar')
# radioConfig = RadioConfig()
# anode.radioConfig = radioConfig
@pytest.mark.unit
def test_onResponseRequestSetting(caplog):
"""Test onResponseRequestSetting()"""
# Note: Split out the get_radio_response to a MagicMock
# so it could be "returned" (not really sure how to do that
# in a python dict.
amsg = MagicMock(autospec=AdminMessage)
amsg.get_radio_response = """{
preferences {
phone_timeout_secs: 900
ls_secs: 300
position_broadcast_smart: true
position_flags: 35
}
}"""
# TODO: not sure if this tmpraw is formatted correctly
tmpraw = """from: 2475227164
to: 2475227164
decoded {
portnum: ADMIN_APP
payload: "*\016\n\0140\204\007P\254\002\210\001\001\260\t#"
request_id: 3145147848
}
id: 365963704
rx_time: 1640195197
hop_limit: 3
priority: RELIABLE"""
packet = {
'from': 2475227164,
'to': 2475227164,
'decoded': {
'portnum': 'ADMIN_APP',
'payload': b'*\x0e\n\x0c0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#',
'requestId': 3145147848,
'admin': {
'getRadioResponse': {
'preferences': {
'phoneTimeoutSecs': 900,
'lsSecs': 300,
'positionBroadcastSmart': True,
'positionFlags': 35
}
},
'raw': amsg
},
'id': 365963704,
'rxTime': 1640195197,
'hopLimit': 3,
'priority': 'RELIABLE',
'raw': tmpraw,
'fromId': '!9388f81c',
'toId': '!9388f81c'
}
}
iface = MagicMock(autospec=SerialInterface)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
mo.localNode.getChannelByName.return_value = None
mo.myInfo.max_channels = 8
anode = Node(mo, 'bar', noProto=True)
radioConfig = RadioConfig()
anode.radioConfig = radioConfig
# Note: Have to do this next line because every call to MagicMock object/method returns a new magic mock
mo.localNode = anode
with caplog.at_level(logging.DEBUG):
anode.onResponseRequestSettings(packet)
assert re.search(r'Received radio config, now fetching channels..', caplog.text, re.MULTILINE)
@pytest.mark.unit
def test_onResponseRequestSetting_with_error(capsys):
"""Test onResponseRequestSetting() with an error"""
packet = {
'from': 2475227164,
'to': 2475227164,
'decoded': {
'portnum': 'ADMIN_APP',
'payload': b'*\x0e\n\x0c0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#',
'requestId': 3145147848,
'routing': {
'errorReason': 'some made up error',
},
'admin': {
'getRadioResponse': {
'preferences': {
'phoneTimeoutSecs': 900,
'lsSecs': 300,
'positionBroadcastSmart': True,
'positionFlags': 35
}
},
},
'id': 365963704,
'rxTime': 1640195197,
'hopLimit': 3,
'priority': 'RELIABLE',
'fromId': '!9388f81c',
'toId': '!9388f81c'
}
}
iface = MagicMock(autospec=SerialInterface)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
mo.localNode.getChannelByName.return_value = None
mo.myInfo.max_channels = 8
anode = Node(mo, 'bar', noProto=True)
radioConfig = RadioConfig()
anode.radioConfig = radioConfig
# Note: Have to do this next line because every call to MagicMock object/method returns a new magic mock
mo.localNode = anode
anode.onResponseRequestSettings(packet)
out, err = capsys.readouterr()
assert re.search(r'Error on response', out)
assert err == ''