add more unit tests

meshtastic/__init__.py

@@ -132,6 +132,7 @@ def _onTextReceive(iface, asDict):
     #
     # Usually btw this problem is caused by apps sending binary data but setting the payload type to
     # text.
+    logging.debug(f'in _onTextReceive() asDict:{asDict}')
     try:
         asBytes = asDict["decoded"]["payload"]
         asDict["decoded"]["text"] = asBytes.decode("utf-8")
@@ -142,22 +143,30 @@ def _onTextReceive(iface, asDict):
 
 def _onPositionReceive(iface, asDict):
     """Special auto parsing for received messages"""
-    p = asDict["decoded"]["position"]
-    iface._fixupPosition(p)
-    # update node DB as needed
-    iface._getOrCreateByNum(asDict["from"])["position"] = p
+    logging.debug(f'in _onPositionReceive() asDict:{asDict}')
+    if 'decoded' in asDict:
+        if 'position' in asDict['decoded'] and 'from' in asDict:
+            p = asDict["decoded"]["position"]
+            logging.debug(f'p:{p}')
+            p = iface._fixupPosition(p)
+            logging.debug(f'after fixup p:{p}')
+            # update node DB as needed
+            iface._getOrCreateByNum(asDict["from"])["position"] = p
 
 
 def _onNodeInfoReceive(iface, asDict):
     """Special auto parsing for received messages"""
-    p = asDict["decoded"]["user"]
-    # decode user protobufs and update nodedb, provide decoded version as "position" in the published msg
-    # update node DB as needed
-    n = iface._getOrCreateByNum(asDict["from"])
-    n["user"] = p
-    # We now have a node ID, make sure it is uptodate in that table
-    iface.nodes[p["id"]] = n
-    _receiveInfoUpdate(iface, asDict)
+    logging.debug(f'in _onNodeInfoReceive() asDict:{asDict}')
+    if 'decoded' in asDict:
+        if 'user' in asDict['decoded'] and 'from' in asDict:
+            p = asDict["decoded"]["user"]
+            # decode user protobufs and update nodedb, provide decoded version as "position" in the published msg
+            # update node DB as needed
+            n = iface._getOrCreateByNum(asDict["from"])
+            n["user"] = p
+            # We now have a node ID, make sure it is uptodate in that table
+            iface.nodes[p["id"]] = n
+            _receiveInfoUpdate(iface, asDict)
 
 
 def _receiveInfoUpdate(iface, asDict):

meshtastic/tests/test_init.py

@@ -0,0 +1,61 @@
+"""Meshtastic unit tests for __init__.py"""
+
+import re
+import logging
+
+from unittest.mock import MagicMock
+import pytest
+
+from meshtastic.__init__ import _onTextReceive, _onPositionReceive, _onNodeInfoReceive
+from ..serial_interface import SerialInterface
+from ..globals import Globals
+
+
+@pytest.mark.unit
+def test_init_onTextReceive_with_exception(caplog):
+    """Test _onTextReceive"""
+    args = MagicMock()
+    Globals.getInstance().set_args(args)
+    iface = MagicMock(autospec=SerialInterface)
+    packet = {}
+    with caplog.at_level(logging.DEBUG):
+        _onTextReceive(iface, packet)
+    assert re.search(r'in _onTextReceive', caplog.text, re.MULTILINE)
+    assert re.search(r'Malformatted', caplog.text, re.MULTILINE)
+
+
+@pytest.mark.unit
+def test_init_onPositionReceive(caplog):
+    """Test _onPositionReceive"""
+    args = MagicMock()
+    Globals.getInstance().set_args(args)
+    iface = MagicMock(autospec=SerialInterface)
+    packet = {
+            'from': 'foo',
+            'decoded': {
+                'position': {}
+                }
+            }
+    with caplog.at_level(logging.DEBUG):
+        _onPositionReceive(iface, packet)
+    assert re.search(r'in _onPositionReceive', caplog.text, re.MULTILINE)
+
+
+@pytest.mark.unit
+def test_init_onNodeInfoReceive(caplog, iface_with_nodes):
+    """Test _onNodeInfoReceive"""
+    args = MagicMock()
+    Globals.getInstance().set_args(args)
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    packet = {
+            'from': 'foo',
+            'decoded': {
+                'user': {
+                    'id': 'bar',
+                    },
+                }
+            }
+    with caplog.at_level(logging.DEBUG):
+        _onNodeInfoReceive(iface, packet)
+    assert re.search(r'in _onNodeInfoReceive', caplog.text, re.MULTILINE)