working on more unit tests

meshtastic/mesh_interface.py

@@ -393,11 +393,11 @@ def getShortName(self):
             return user.get('shortName', None)
         return None
 
-    def _waitConnected(self):
+    def _waitConnected(self, timeout=15.0):
         """Block until the initial node db download is complete, or timeout
         and raise an exception"""
         if not self.noProto:
-            if not self.isConnected.wait(15.0):  # timeout after x seconds
+            if not self.isConnected.wait(timeout):  # timeout after x seconds
                 raise Exception("Timed out waiting for connection completion")
 
         # If we failed while connecting, raise the connection to the client
@@ -415,8 +415,7 @@ def _generatePacketId(self):
     def _disconnected(self):
         """Called by subclasses to tell clients this interface has disconnected"""
         self.isConnected.clear()
-        publishingThread.queueWork(lambda: pub.sendMessage(
-            "meshtastic.connection.lost", interface=self))
+        publishingThread.queueWork(lambda: pub.sendMessage("meshtastic.connection.lost", interface=self))
 
     def _startHeartbeat(self):
         """We need to send a heartbeat message to the device every X seconds"""
@@ -442,8 +441,7 @@ def _connected(self):
         if not self.isConnected.is_set():
             self.isConnected.set()
             self._startHeartbeat()
-            publishingThread.queueWork(lambda: pub.sendMessage(
-                "meshtastic.connection.established", interface=self))
+            publishingThread.queueWork(lambda: pub.sendMessage("meshtastic.connection.established", interface=self))
 
     def _startConfig(self):
         """Start device packets flowing"""

meshtastic/tests/test_mesh_interface.py

@@ -11,6 +11,7 @@
 from .. import mesh_pb2
 from ..__init__ import LOCAL_ADDR, BROADCAST_ADDR
 from ..radioconfig_pb2 import RadioConfig
+from ..util import Timeout
 
 
 @pytest.mark.unit
@@ -588,3 +589,42 @@ def test_showNodes_exclude_self(capsys, caplog, reset_globals, iface_with_nodes)
         iface.showNodes()
         iface.showNodes(includeSelf=False)
         capsys.readouterr()
+
+
+@pytest.mark.unit
+def test_waitForConfig(caplog, capsys):
+    """Test waitForConfig()"""
+    iface = MeshInterface(noProto=True)
+    # override how long to wait
+    iface._timeout = Timeout(0.01)
+    with pytest.raises(Exception) as pytest_wrapped_e:
+        iface.waitForConfig()
+        assert pytest_wrapped_e.type == Exception
+        out, err = capsys.readouterr()
+        assert re.search(r'Exception: Timed out waiting for interface config', err, re.MULTILINE)
+        assert out == ''
+
+
+@pytest.mark.unit
+def test_waitConnected_raises_an_exception(caplog, capsys):
+    """Test waitConnected()"""
+    iface = MeshInterface(noProto=True)
+    with pytest.raises(Exception) as pytest_wrapped_e:
+        iface.failure = "warn about something"
+        iface._waitConnected(0.01)
+        assert pytest_wrapped_e.type == Exception
+        out, err = capsys.readouterr()
+        assert re.search(r'warn about something', err, re.MULTILINE)
+        assert out == ''
+
+
+@pytest.mark.unit
+def test_waitConnected_isConnected_timeout(caplog, capsys):
+    """Test waitConnected()"""
+    with pytest.raises(Exception) as pytest_wrapped_e:
+        iface = MeshInterface()
+        iface._waitConnected(0.01)
+        assert pytest_wrapped_e.type == Exception
+        out, err = capsys.readouterr()
+        assert re.search(r'warn about something', err, re.MULTILINE)
+        assert out == ''

meshtastic/tests/test_node.py

@@ -888,6 +888,6 @@ def test_waitForConfig():
     anode = Node('foo', 'bar')
     radioConfig = RadioConfig()
     anode.radioConfig = radioConfig
-    anode._timeout = Timeout(0.1)
+    anode._timeout = Timeout(0.01)
     result = anode.waitForConfig()
     assert not result