mirror of
https://github.com/meshtastic/python.git
synced 2026-01-16 19:58:04 -05:00
add unit tests for _handleFromRadio() and sendPosition()
This commit is contained in:
@@ -268,6 +268,7 @@ class MeshInterface:
|
||||
if timeSec == 0:
|
||||
timeSec = time.time() # returns unix timestamp in seconds
|
||||
p.time = int(timeSec)
|
||||
logging.debug(f'p.time:{p.time}')
|
||||
|
||||
return self.sendData(p, destinationId,
|
||||
portNum=portnums_pb2.PortNum.POSITION_APP,
|
||||
@@ -455,8 +456,9 @@ class MeshInterface:
|
||||
Called by subclasses."""
|
||||
fromRadio = mesh_pb2.FromRadio()
|
||||
fromRadio.ParseFromString(fromRadioBytes)
|
||||
#logging.debug(f"fromRadioBytes: {fromRadioBytes}")
|
||||
asDict = google.protobuf.json_format.MessageToDict(fromRadio)
|
||||
#logging.debug(f"Received from radio: {fromRadio}")
|
||||
logging.debug(f"Received from radio: {fromRadio}")
|
||||
if fromRadio.HasField("my_info"):
|
||||
self.myInfo = fromRadio.my_info
|
||||
self.localNode.nodeNum = self.myInfo.my_node_num
|
||||
|
||||
@@ -101,3 +101,77 @@ def test_getNode_not_local_timeout(reset_globals, capsys):
|
||||
out, err = capsys.readouterr()
|
||||
assert re.match(r'Error: Timed out waiting for node config', out)
|
||||
assert err == ''
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_sendPosition(reset_globals, caplog):
|
||||
"""Test sendPosition"""
|
||||
iface = MeshInterface(noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
iface.sendPosition()
|
||||
iface.close()
|
||||
assert re.search(r'p.time:', caplog.text, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_handleFromRadio_empty_payload(reset_globals, caplog):
|
||||
"""Test _handleFromRadio"""
|
||||
iface = MeshInterface(noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
iface._handleFromRadio(b'')
|
||||
iface.close()
|
||||
assert re.search(r'Unexpected FromRadio payload', caplog.text, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_handleFromRadio_with_my_info(reset_globals, caplog):
|
||||
"""Test _handleFromRadio with my_info"""
|
||||
# Note: I captured the '--debug --info' for the bytes below.
|
||||
# It "translates" to this:
|
||||
# my_info {
|
||||
# my_node_num: 682584012
|
||||
# num_bands: 13
|
||||
# firmware_version: "1.2.49.5354c49"
|
||||
# reboot_count: 13
|
||||
# bitrate: 17.088470458984375
|
||||
# message_timeout_msec: 300000
|
||||
# min_app_version: 20200
|
||||
# max_channels: 8
|
||||
# }
|
||||
from_radio_bytes = b'\x1a,\x08\xcc\xcf\xbd\xc5\x02\x18\r2\x0e1.2.49.5354c49P\r]0\xb5\x88Ah\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01'
|
||||
iface = MeshInterface(noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
iface._handleFromRadio(from_radio_bytes)
|
||||
iface.close()
|
||||
assert re.search(r'Received myinfo', caplog.text, re.MULTILINE)
|
||||
assert re.search(r'num_bands: 13', caplog.text, re.MULTILINE)
|
||||
assert re.search(r'max_channels: 8', caplog.text, re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_handleFromRadio_with_node_info(reset_globals, caplog):
|
||||
"""Test _handleFromRadio with node_info"""
|
||||
# Note: I captured the '--debug --info' for the bytes below.
|
||||
# It "translates" to this:
|
||||
# node_info {
|
||||
# num: 682584012
|
||||
# user {
|
||||
# id: "!28af67cc"
|
||||
# long_name: "Unknown 67cc"
|
||||
# short_name: "?CC"
|
||||
# macaddr: "$o(\257g\314"
|
||||
# hw_model: HELTEC_V2_1
|
||||
# }
|
||||
# position {
|
||||
# }
|
||||
# }
|
||||
|
||||
from_radio_bytes = b'"2\x08\xcc\xcf\xbd\xc5\x02\x12(\n\t!28af67cc\x12\x0cUnknown 67cc\x1a\x03?CC"\x06$o(\xafg\xcc0\n\x1a\x00'
|
||||
iface = MeshInterface(noProto=True)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
iface._startConfig()
|
||||
iface._handleFromRadio(from_radio_bytes)
|
||||
iface.close()
|
||||
assert re.search(r'Received nodeinfo', caplog.text, re.MULTILINE)
|
||||
assert re.search(r'682584012', caplog.text, re.MULTILINE)
|
||||
assert re.search(r'HELTEC_V2_1', caplog.text, re.MULTILINE)
|
||||
|
||||
Reference in New Issue
Block a user