Merge pull request #199 from mkinney/unit_testing_continues

revert the stream interface change; fix tunnel tests
2025-12-28 18:37:52 -05:00 · 2021-12-31 09:40:58 -08:00
parent b92fff0da6 ab876c9efd
commit bd697864e4
5 changed files with 111 additions and 71 deletions
--- a/meshtastic/main.py
+++ b/meshtastic/main.py
@@ -498,7 +498,10 @@ def onConnected(interface):
            from . import tunnel
            # Even if others said we could close, stay open if the user asked for a tunnel
            closeNow = False
-            tunnel.Tunnel(interface, subnet=args.tunnel_net)
+            if interface.noProto:
+                logging.warning(f"Not starting Tunnel - disabled by noProto")
+            else:
+                tunnel.Tunnel(interface, subnet=args.tunnel_net)

        # if the user didn't ask for serial debugging output, we might want to exit after we've done our operation
        if (not args.seriallog) and closeNow:
--- a/meshtastic/stream_interface.py
+++ b/meshtastic/stream_interface.py
@@ -39,10 +39,7 @@ class StreamInterface(MeshInterface):
        self._wantExit = False

        # FIXME, figure out why daemon=True causes reader thread to exit too early
-        if noProto:
-            self._rxThread = None
-        else:
-            self._rxThread = threading.Thread(target=self.__reader, args=(), daemon=True)
+        self._rxThread = threading.Thread(target=self.__reader, args=(), daemon=True)

        MeshInterface.__init__(self, debugOut=debugOut, noProto=noProto)

@@ -68,8 +65,7 @@ class StreamInterface(MeshInterface):
        if not self.noProto:
            time.sleep(0.1)  # wait 100ms to give device time to start running

-        if not self.noProto:
-            self._rxThread.start()
+        self._rxThread.start()

        self._startConfig()

@@ -121,9 +117,8 @@ class StreamInterface(MeshInterface):
        # pyserial cancel_read doesn't seem to work, therefore we ask the
        # reader thread to close things for us
        self._wantExit = True
-        if not self.noProto:
-            if self._rxThread != threading.current_thread():
-                self._rxThread.join()  # wait for it to exit
+        if self._rxThread != threading.current_thread():
+            self._rxThread.join()  # wait for it to exit

    def __reader(self):
        """The reader thread that reads bytes from our stream"""
--- a/meshtastic/tests/test_main.py
+++ b/meshtastic/tests/test_main.py
@@ -1696,8 +1696,9 @@ def test_tunnel_no_args(capsys, reset_globals):


@pytest.mark.unit
+@patch('meshtastic.util.findPorts', return_value=[])
@patch('platform.system')
-def test_tunnel_tunnel_arg(mock_platform_system, capsys, reset_globals):
+def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, patched_find_ports, caplog, capsys, reset_globals):
    """Test tunnel with tunnel arg (act like we are on a linux system)"""
    a_mock = MagicMock()
    a_mock.return_value = 'Linux'
@@ -1705,18 +1706,21 @@ def test_tunnel_tunnel_arg(mock_platform_system, capsys, reset_globals):
    sys.argv = ['', '--tunnel']
    Globals.getInstance().set_args(sys.argv)
    print(f'platform.system():{platform.system()}')
-    with pytest.raises(SystemExit) as pytest_wrapped_e:
-        tunnelMain()
-    mock_platform_system.assert_called()
-    assert pytest_wrapped_e.type == SystemExit
-    assert pytest_wrapped_e.value.code == 1
-    _, err = capsys.readouterr()
-    assert not re.search(r'usage: ', err, re.MULTILINE)
+    with caplog.at_level(logging.DEBUG):
+        with pytest.raises(SystemExit) as pytest_wrapped_e:
+            tunnelMain()
+        mock_platform_system.assert_called()
+        assert pytest_wrapped_e.type == SystemExit
+        assert pytest_wrapped_e.value.code == 1
+        out, err = capsys.readouterr()
+        assert re.search(r'Warning: No Meshtastic devices detected', out, re.MULTILINE)
+        assert err == ''


@pytest.mark.unit
+@patch('meshtastic.util.findPorts', return_value=[])
@patch('platform.system')
-def test_tunnel_subnet_arg(mock_platform_system, capsys, reset_globals):
+def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, patched_find_ports, caplog, capsys, reset_globals):
    """Test tunnel with subnet arg (act like we are on a linux system)"""
    a_mock = MagicMock()
    a_mock.return_value = 'Linux'
@@ -1724,11 +1728,41 @@ def test_tunnel_subnet_arg(mock_platform_system, capsys, reset_globals):
    sys.argv = ['', '--subnet', 'foo']
    Globals.getInstance().set_args(sys.argv)
    print(f'platform.system():{platform.system()}')
-    with pytest.raises(SystemExit) as pytest_wrapped_e:
-        tunnelMain()
-    mock_platform_system.assert_called()
-    assert pytest_wrapped_e.type == SystemExit
-    assert pytest_wrapped_e.value.code == 1
-    out, err = capsys.readouterr()
-    assert not re.search(r'usage: ', err, re.MULTILINE)
-    assert re.search(r'Warning: No Meshtastic devices detected', out, re.MULTILINE)
+    with caplog.at_level(logging.DEBUG):
+        with pytest.raises(SystemExit) as pytest_wrapped_e:
+            tunnelMain()
+        mock_platform_system.assert_called()
+        assert pytest_wrapped_e.type == SystemExit
+        assert pytest_wrapped_e.value.code == 1
+        out, err = capsys.readouterr()
+        assert re.search(r'Warning: No Meshtastic devices detected', out, re.MULTILINE)
+        assert err == ''
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test tunnel with tunnel arg (act like we are on a linux system)"""
+    # Override the time.sleep so there is no loop
+    def my_sleep(amount):
+        sys.exit(3)
+
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    sys.argv = ['', '--tunnel']
+    Globals.getInstance().set_args(sys.argv)
+    print(f'platform.system():{platform.system()}')
+
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+
+    with caplog.at_level(logging.DEBUG):
+        with patch('meshtastic.serial_interface.SerialInterface', return_value=iface):
+            with patch('time.sleep', side_effect=my_sleep):
+                with pytest.raises(SystemExit) as pytest_wrapped_e:
+                    tunnelMain()
+                    mock_platform_system.assert_called()
+                assert pytest_wrapped_e.type == SystemExit
+                assert pytest_wrapped_e.value.code == 3
+                assert re.search(r'Not starting Tunnel', caplog.text, re.MULTILINE)
--- a/meshtastic/tests/test_stream_interface.py
+++ b/meshtastic/tests/test_stream_interface.py
@@ -1,7 +1,7 @@
 """Meshtastic unit tests for stream_interface.py"""

 import logging
-#import re
+import re

 from unittest.mock import MagicMock
 import pytest
@@ -37,45 +37,45 @@ def test_StreamInterface_with_noProto(caplog, reset_globals):
 ## Note: This takes a bit, so moving from unit to slow
 ## Tip: If you want to see the print output, run with '-s' flag:
 ##      pytest -s meshtastic/tests/test_stream_interface.py::test_sendToRadioImpl
-#@pytest.mark.unitslow
-#def test_sendToRadioImpl(caplog, reset_globals):
-#    """Test _sendToRadioImpl()"""
-#
-##    def add_header(b):
-##        """Add header stuffs for radio"""
-##        bufLen = len(b)
-##        header = bytes([START1, START2, (bufLen >> 8) & 0xff,  bufLen & 0xff])
-##        return header + b
-#
-#    # captured raw bytes of a Heltec2.1 radio with 2 channels (primary and a secondary channel named "gpio")
-#    raw_1_my_info = b'\x1a,\x08\xdc\x8c\xd5\xc5\x02\x18\r2\x0e1.2.49.5354c49P\x15]\xe1%\x17Eh\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01'
-#    raw_2_node_info = b'"9\x08\xdc\x8c\xd5\xc5\x02\x12(\n\t!28b5465c\x12\x0cUnknown 465c\x1a\x03?5C"\x06$o(\xb5F\\0\n\x1a\x02 1%M<\xc6a'
-#    # pylint: disable=C0301
-#    raw_3_node_info = b'"C\x08\xa4\x8c\xd5\xc5\x02\x12(\n\t!28b54624\x12\x0cUnknown 4624\x1a\x03?24"\x06$o(\xb5F$0\n\x1a\x07 5MH<\xc6a%G<\xc6a=\x00\x00\xc0@'
-#    raw_4_complete = b'@\xcf\xe5\xd1\x8c\x0e'
-#    # pylint: disable=C0301
-#    raw_5_prefs = b'Z6\r\\F\xb5(\x15\\F\xb5("\x1c\x08\x06\x12\x13*\x11\n\x0f0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#\xb8\t\x015]$\xddk5\xd5\x7f!b=M<\xc6aP\x03`F'
-#    # pylint: disable=C0301
-#    raw_6_channel0 = b'Z.\r\\F\xb5(\x15\\F\xb5("\x14\x08\x06\x12\x0b:\t\x12\x05\x18\x01"\x01\x01\x18\x015^$\xddk5\xd6\x7f!b=M<\xc6aP\x03`F'
-#    # pylint: disable=C0301
-#    raw_7_channel1 = b'ZS\r\\F\xb5(\x15\\F\xb5("9\x08\x06\x120:.\x08\x01\x12(" \xb4&\xb3\xc7\x06\xd8\xe39%\xba\xa5\xee\x8eH\x06\xf6\xf4H\xe8\xd5\xc1[ao\xb5Y\\\xb4"\xafmi*\x04gpio\x18\x025_$\xddk5\xd7\x7f!b=M<\xc6aP\x03`F'
-#    raw_8_channel2 = b'Z)\r\\F\xb5(\x15\\F\xb5("\x0f\x08\x06\x12\x06:\x04\x08\x02\x12\x005`$\xddk5\xd8\x7f!b=M<\xc6aP\x03`F'
-#    raw_blank = b''
-#
-#    test_data = b'hello'
-#    stream = MagicMock()
-#    #stream.read.return_value = add_header(test_data)
-#    stream.read.side_effect = [ raw_1_my_info, raw_2_node_info, raw_3_node_info, raw_4_complete,
-#                                raw_5_prefs, raw_6_channel0, raw_7_channel1, raw_8_channel2,
-#                                raw_blank, raw_blank]
-#    toRadio = MagicMock()
-#    toRadio.SerializeToString.return_value = test_data
-#    with caplog.at_level(logging.DEBUG):
-#        iface = StreamInterface(noProto=True, connectNow=False)
-#        iface.stream = stream
-#        iface.connect()
-#        iface._sendToRadioImpl(toRadio)
-#        assert re.search(r'Sending: ', caplog.text, re.MULTILINE)
-#        assert re.search(r'reading character', caplog.text, re.MULTILINE)
-#        assert re.search(r'In reader loop', caplog.text, re.MULTILINE)
-#        print(caplog.text)
+@pytest.mark.unitslow
+def test_sendToRadioImpl(caplog, reset_globals):
+    """Test _sendToRadioImpl()"""
+
+#    def add_header(b):
+#        """Add header stuffs for radio"""
+#        bufLen = len(b)
+#        header = bytes([START1, START2, (bufLen >> 8) & 0xff,  bufLen & 0xff])
+#        return header + b
+
+    # captured raw bytes of a Heltec2.1 radio with 2 channels (primary and a secondary channel named "gpio")
+    raw_1_my_info = b'\x1a,\x08\xdc\x8c\xd5\xc5\x02\x18\r2\x0e1.2.49.5354c49P\x15]\xe1%\x17Eh\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01'
+    raw_2_node_info = b'"9\x08\xdc\x8c\xd5\xc5\x02\x12(\n\t!28b5465c\x12\x0cUnknown 465c\x1a\x03?5C"\x06$o(\xb5F\\0\n\x1a\x02 1%M<\xc6a'
+    # pylint: disable=C0301
+    raw_3_node_info = b'"C\x08\xa4\x8c\xd5\xc5\x02\x12(\n\t!28b54624\x12\x0cUnknown 4624\x1a\x03?24"\x06$o(\xb5F$0\n\x1a\x07 5MH<\xc6a%G<\xc6a=\x00\x00\xc0@'
+    raw_4_complete = b'@\xcf\xe5\xd1\x8c\x0e'
+    # pylint: disable=C0301
+    raw_5_prefs = b'Z6\r\\F\xb5(\x15\\F\xb5("\x1c\x08\x06\x12\x13*\x11\n\x0f0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#\xb8\t\x015]$\xddk5\xd5\x7f!b=M<\xc6aP\x03`F'
+    # pylint: disable=C0301
+    raw_6_channel0 = b'Z.\r\\F\xb5(\x15\\F\xb5("\x14\x08\x06\x12\x0b:\t\x12\x05\x18\x01"\x01\x01\x18\x015^$\xddk5\xd6\x7f!b=M<\xc6aP\x03`F'
+    # pylint: disable=C0301
+    raw_7_channel1 = b'ZS\r\\F\xb5(\x15\\F\xb5("9\x08\x06\x120:.\x08\x01\x12(" \xb4&\xb3\xc7\x06\xd8\xe39%\xba\xa5\xee\x8eH\x06\xf6\xf4H\xe8\xd5\xc1[ao\xb5Y\\\xb4"\xafmi*\x04gpio\x18\x025_$\xddk5\xd7\x7f!b=M<\xc6aP\x03`F'
+    raw_8_channel2 = b'Z)\r\\F\xb5(\x15\\F\xb5("\x0f\x08\x06\x12\x06:\x04\x08\x02\x12\x005`$\xddk5\xd8\x7f!b=M<\xc6aP\x03`F'
+    raw_blank = b''
+
+    test_data = b'hello'
+    stream = MagicMock()
+    #stream.read.return_value = add_header(test_data)
+    stream.read.side_effect = [ raw_1_my_info, raw_2_node_info, raw_3_node_info, raw_4_complete,
+                                raw_5_prefs, raw_6_channel0, raw_7_channel1, raw_8_channel2,
+                                raw_blank, raw_blank]
+    toRadio = MagicMock()
+    toRadio.SerializeToString.return_value = test_data
+    with caplog.at_level(logging.DEBUG):
+        iface = StreamInterface(noProto=True, connectNow=False)
+        iface.stream = stream
+        iface.connect()
+        iface._sendToRadioImpl(toRadio)
+        assert re.search(r'Sending: ', caplog.text, re.MULTILINE)
+        assert re.search(r'reading character', caplog.text, re.MULTILINE)
+        assert re.search(r'In reader loop', caplog.text, re.MULTILINE)
+        print(caplog.text)
--- a/meshtastic/tests/test_util.py
+++ b/meshtastic/tests/test_util.py
@@ -3,13 +3,14 @@
 import re
 import logging

+from unittest.mock import patch
 import pytest

 from meshtastic.util import (fixme, stripnl, pskToString, our_exit,
                             support_info, genPSK256, fromStr, fromPSK,
                             quoteBooleans, catchAndIgnore,
                             remove_keys_from_dict, Timeout, hexstr,
-                             ipstr, readnet_u16)
+                             ipstr, readnet_u16, findPorts)


@pytest.mark.unit
@@ -211,3 +212,10 @@ def test_ipstr():
 def test_readnet_u16():
    """Test readnet_u16()"""
    assert readnet_u16(b'123456', 2) == 13108
+
+
+@pytest.mark.unit
+@patch('serial.tools.list_ports.comports', return_value=[])
+def test_findPorts_when_none_found(patch_comports):
+    """Test findPorts()"""
+    assert not findPorts()