Merge pull request #203 from mkinney/minor_changes

do not print line for export; comment out ble test; do not send decoded

.coveragerc

@@ -1,2 +1,6 @@
 [run]
 omit = meshtastic/*_pb2.py,meshtastic/tests/*.py,meshtastic/test.py
+
+[report]
+exclude_lines =
+    if __name__ == .__main__.:

.github/workflows/ci.yml

@@ -32,6 +32,18 @@ jobs:
         run: pylint meshtastic
       - name: Run tests with pytest
         run: pytest --cov=meshtastic
+      - name: Generate coverage report
+        run: |
+          pytest --cov=meshtastic --cov-report=xml
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v1
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          file: ./coverage.xml
+          flags: unittests
+          name: codecov-umbrella
+          yml: ./codecov.yml
+          fail_ci_if_error: true
   validate:
     runs-on: ubuntu-latest
     steps:

Makefile

@@ -16,7 +16,7 @@ lint:
 
 # show the slowest unit tests
 slow:
-	pytest --durations=0
+	pytest --durations=5
 
 # run the coverage report and open results in a browser
 cov:

README.md

@@ -2,6 +2,7 @@
 
 [![Open in Visual Studio Code](https://open.vscode.dev/badges/open-in-vscode.svg)](https://open.vscode.dev/meshtastic/Meshtastic-python)
 ![Unit Tests](https://github.com/meshtastic/Meshtastic-python/actions/workflows/ci.yml/badge.svg)
+[![codecov](https://codecov.io/gh/meshtastic/Meshtastic-python/branch/master/graph/badge.svg?token=TIWPJL73KV)](https://codecov.io/gh/meshtastic/Meshtastic-python)
 
 A python client for using [Meshtastic](https://www.meshtastic.org) devices. This small library (and example application) provides an easy API for sending and receiving messages over mesh radios. It also provides access to any of the operations/data available in the device user interface or the Android application. Events are delivered using a publish-subscribe model, and you can subscribe to only the message types you are interested in.
 

meshtastic/__main__.py

@@ -14,13 +14,11 @@ import pkg_resources
 import meshtastic.util
 import meshtastic.test
 from . import remote_hardware
+from .ble_interface import BLEInterface
 from . import portnums_pb2, channel_pb2, radioconfig_pb2
 from .globals import Globals
 
 
-have_tunnel = platform.system() == 'Linux'
-"""We only import the tunnel code if we are on a platform that can run it. """
-
 def onReceive(packet, interface):
     """Callback invoked when a packet arrives"""
     our_globals = Globals.getInstance()
@@ -45,7 +43,7 @@ def onReceive(packet, interface):
                 interface.sendText(reply)
 
     except Exception as ex:
-        print(ex)
+        print(f'Warning: There is no field {ex} in the packet.')
 
 
 def onConnection(interface, topic=pub.AUTO_TOPIC):
@@ -137,7 +135,9 @@ def onConnected(interface):
         our_globals = Globals.getInstance()
         args = our_globals.get_args()
 
-        print("Connected to radio")
+        # do not print this line if we are exporting the config
+        if not args.export_config:
+            print("Connected to radio")
 
         def getNode():
             """This operation could be expensive, so we try to cache the results"""
@@ -495,12 +495,16 @@ def onConnected(interface):
             qr = pyqrcode.create(url)
             print(qr.terminal())
 
+        have_tunnel = platform.system() == 'Linux'
         if have_tunnel and args.tunnel:
             # pylint: disable=C0415
             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:
@@ -627,19 +631,18 @@ def common():
 
             subscribe()
             if args.ble:
-                client = meshtastic.ble_interface.BLEInterface(args.ble, debugOut=logfile, noProto=args.noproto)
+                client = BLEInterface(args.ble, debugOut=logfile, noProto=args.noproto)
             elif args.host:
-                client = meshtastic.tcp_interface.TCPInterface(
-                    args.host, debugOut=logfile, noProto=args.noproto)
+                client = meshtastic.tcp_interface.TCPInterface(args.host, debugOut=logfile, noProto=args.noproto)
             else:
-                client = meshtastic.serial_interface.SerialInterface(
-                    args.port, debugOut=logfile, noProto=args.noproto)
+                client = meshtastic.serial_interface.SerialInterface(args.port, debugOut=logfile, noProto=args.noproto)
 
             # We assume client is fully connected now
             onConnected(client)
             #if logfile:
                 #logfile.close()
 
+            have_tunnel = platform.system() == 'Linux'
             if args.noproto or args.reply or (have_tunnel and args.tunnel):  # loop until someone presses ctrlc
                 while True:
                     time.sleep(1000)
@@ -805,11 +808,13 @@ def initParser():
     parser.add_argument('--unset-router', dest='deprecated',
                         action='store_false', help='Deprecated, use "--set is_router false" instead')
 
+    have_tunnel = platform.system() == 'Linux'
     if have_tunnel:
-        parser.add_argument('--tunnel',
-                            action='store_true', help="Create a TUN tunnel device for forwarding IP packets over the mesh")
-        parser.add_argument(
-            "--subnet", dest='tunnel_net', help="Sets the local-end subnet address for the TUN IP bridge", default=None)
+        parser.add_argument('--tunnel', action='store_true',
+                            help="Create a TUN tunnel device for forwarding IP packets over the mesh")
+        parser.add_argument("--subnet", dest='tunnel_net',
+                            help="Sets the local-end subnet address for the TUN IP bridge. (ex: 10.115' which is the default)",
+                            default=None)
 
     parser.set_defaults(deprecated=None)
 

meshtastic/globals.py

@@ -30,6 +30,7 @@ class Globals:
         self.target_node = None
         self.channel_index = None
         self.logfile = None
+        self.tunnelInstance = None
 
     def reset(self):
         """Reset all of our globals. If you add a member, add it to this method, too."""
@@ -37,6 +38,8 @@ class Globals:
         self.parser = None
         self.target_node = None
         self.channel_index = None
+        self.logfile = None
+        self.tunnelInstance = None
 
     # setters
     def set_args(self, args):
@@ -59,6 +62,10 @@ class Globals:
         """Set the logfile"""
         self.logfile = logfile
 
+    def set_tunnelInstance(self, tunnelInstance):
+        """Set the tunnelInstance"""
+        self.tunnelInstance = tunnelInstance
+
     # getters
     def get_args(self):
         """Get args"""
@@ -79,3 +86,7 @@ class Globals:
     def get_logfile(self):
         """Get logfile"""
         return self.logfile
+
+    def get_tunnelInstance(self):
+        """Get tunnelInstance"""
+        return self.tunnelInstance

meshtastic/mesh_interface.py

@@ -18,7 +18,7 @@ from google.protobuf.json_format import MessageToJson
 
 import meshtastic.node
 from . import portnums_pb2, mesh_pb2
-from .util import stripnl, Timeout, our_exit, remove_keys_from_dict
+from .util import stripnl, Timeout, our_exit, remove_keys_from_dict, convert_mac_addr
 from .__init__ import LOCAL_ADDR, BROADCAST_NUM, BROADCAST_ADDR, ResponseHandler, publishingThread, OUR_APP_VERSION, protocols
 
 class MeshInterface:
@@ -87,6 +87,15 @@ class MeshInterface:
                 # when the TBeam is first booted, it sometimes shows the 'raw' data
                 # so, we will just remove any raw keys
                 n2 = remove_keys_from_dict('raw', n)
+                n2 = remove_keys_from_dict('decode', n2)
+
+                # if we have 'macaddr', re-format it
+                if 'macaddr' in n2['user']:
+                    val = n2['user']['macaddr']
+                    # decode the base64 value
+                    addr = convert_mac_addr(val)
+                    n2['user']['macaddr'] = addr
+
                 nodes = nodes + f"  {stripnl(n2)}"
         infos = owner + myinfo + mesh + nodes
         print(infos)
@@ -507,7 +516,8 @@ class MeshInterface:
         elif fromRadio.HasField("node_info"):
             node = asDict["nodeInfo"]
             try:
-                self._fixupPosition(node["position"])
+                newpos = self._fixupPosition(node["position"])
+                node["position"] = newpos
             except:
                 logging.debug("Node without position")
 
@@ -539,12 +549,14 @@ class MeshInterface:
         """Convert integer lat/lon into floats
 
         Arguments:
-            position {Position dictionary} -- object ot fix up
+            position {Position dictionary} -- object to fix up
+        Returns the position with the updated keys
         """
         if "latitudeI" in position:
             position["latitude"] = position["latitudeI"] * 1e-7
         if "longitudeI" in position:
             position["longitude"] = position["longitudeI"] * 1e-7
+        return position
 
     def _nodeNumToId(self, num):
         """Map a node node number to a node ID

meshtastic/tcp_interface.py

@@ -17,8 +17,6 @@ class TCPInterface(StreamInterface):
             hostname {string} -- Hostname/IP address of the device to connect to
         """
 
-        # Instead of wrapping as a stream, we use the native socket API
-        # self.stream = sock.makefile('rw')
         self.stream = None
 
         self.hostname = hostname

meshtastic/tests/test_int.py

@@ -6,13 +6,21 @@ import pytest
 
 
 @pytest.mark.int
-def test_int_no_args():
-    """Test without any args"""
+def test_int_meshtastic_no_args():
+    """Test meshtastic without any args"""
     return_value, out = subprocess.getstatusoutput('meshtastic')
     assert re.match(r'usage: meshtastic', out)
     assert return_value == 1
 
 
+@pytest.mark.int
+def test_int_mesh_tunnel_no_args():
+    """Test mesh-tunnel without any args"""
+    return_value, out = subprocess.getstatusoutput('mesh-tunnel')
+    assert re.match(r'usage: mesh-tunnel', out)
+    assert return_value == 1
+
+
 @pytest.mark.int
 def test_int_version():
     """Test '--version'."""

meshtastic/tests/test_main.py

@@ -5,16 +5,17 @@ import sys
 import os
 import re
 import logging
+import platform
 
 from unittest.mock import patch, MagicMock
 import pytest
 
-from meshtastic.__main__ import initParser, main, Globals, onReceive, onConnection, export_config, getPref, setPref
+from meshtastic.__main__ import initParser, main, Globals, onReceive, onConnection, export_config, getPref, setPref, onNode, tunnelMain
 #from ..radioconfig_pb2 import UserPreferences
 import meshtastic.radioconfig_pb2
 from ..serial_interface import SerialInterface
 from ..tcp_interface import TCPInterface
-from ..ble_interface import BLEInterface
+#from ..ble_interface import BLEInterface
 from ..node import Node
 from ..channel_pb2 import Channel
 from ..remote_hardware import onGPIOreceive
@@ -62,7 +63,7 @@ def test_main_main_version(capsys, reset_globals):
 
 
 @pytest.mark.unit
-def test_main_main_no_args(reset_globals):
+def test_main_main_no_args(reset_globals, capsys):
     """Test with no args"""
     sys.argv = ['']
     Globals.getInstance().set_args(sys.argv)
@@ -71,6 +72,8 @@ def test_main_main_no_args(reset_globals):
         main()
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
+    _, err = capsys.readouterr()
+    assert re.search(r'usage:', err, re.MULTILINE)
 
 
 @pytest.mark.unit
@@ -111,7 +114,7 @@ def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals):
 
 @pytest.mark.unit
 @patch('meshtastic.util.findPorts', return_value=[])
-def test_main_test_no_ports(patched_find_ports, reset_globals):
+def test_main_test_no_ports(patched_find_ports, reset_globals, capsys):
     """Test --test with no hardware"""
     sys.argv = ['', '--test']
     Globals.getInstance().set_args(sys.argv)
@@ -122,11 +125,14 @@ def test_main_test_no_ports(patched_find_ports, reset_globals):
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
     patched_find_ports.assert_called()
+    out, err = capsys.readouterr()
+    assert re.search(r'Warning: Must have at least two devices connected to USB', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
 @patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1'])
-def test_main_test_one_port(patched_find_ports, reset_globals):
+def test_main_test_one_port(patched_find_ports, reset_globals, capsys):
     """Test --test with one fake port"""
     sys.argv = ['', '--test']
     Globals.getInstance().set_args(sys.argv)
@@ -137,12 +143,14 @@ def test_main_test_one_port(patched_find_ports, reset_globals):
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
     patched_find_ports.assert_called()
+    out, err = capsys.readouterr()
+    assert re.search(r'Warning: Must have at least two devices connected to USB', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
 @patch('meshtastic.test.testAll', return_value=True)
-@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1', '/dev/ttyFake2'])
-def test_main_test_two_ports_success(patched_find_ports, patched_test_all, reset_globals):
+def test_main_test_two_ports_success(patched_test_all, reset_globals, capsys):
     """Test --test two fake ports and testAll() is a simulated success"""
     sys.argv = ['', '--test']
     Globals.getInstance().set_args(sys.argv)
@@ -151,14 +159,15 @@ def test_main_test_two_ports_success(patched_find_ports, patched_test_all, reset
         main()
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 0
-    # TODO: why does this fail? patched_find_ports.assert_called()
     patched_test_all.assert_called()
+    out, err = capsys.readouterr()
+    assert re.search(r'Test was a success.', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
 @patch('meshtastic.test.testAll', return_value=False)
-@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1', '/dev/ttyFake2'])
-def test_main_test_two_ports_fails(patched_find_ports, patched_test_all, reset_globals):
+def test_main_test_two_ports_fails(patched_test_all, reset_globals, capsys):
     """Test --test two fake ports and testAll() is a simulated failure"""
     sys.argv = ['', '--test']
     Globals.getInstance().set_args(sys.argv)
@@ -167,8 +176,10 @@ def test_main_test_two_ports_fails(patched_find_ports, patched_test_all, reset_g
         main()
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
-    # TODO: why does this fail? patched_find_ports.assert_called()
     patched_test_all.assert_called()
+    out, err = capsys.readouterr()
+    assert re.search(r'Test was not successful.', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
@@ -209,23 +220,24 @@ def test_main_info_with_tcp_interface(capsys, reset_globals):
         mo.assert_called()
 
 
-@pytest.mark.unit
-def test_main_info_with_ble_interface(capsys, reset_globals):
-    """Test --info"""
-    sys.argv = ['', '--info', '--ble', 'foo']
-    Globals.getInstance().set_args(sys.argv)
-
-    iface = MagicMock(autospec=BLEInterface)
-    def mock_showInfo():
-        print('inside mocked showInfo')
-    iface.showInfo.side_effect = mock_showInfo
-    with patch('meshtastic.ble_interface.BLEInterface', return_value=iface) as mo:
-        main()
-        out, err = capsys.readouterr()
-        assert re.search(r'Connected to radio', out, re.MULTILINE)
-        assert re.search(r'inside mocked showInfo', out, re.MULTILINE)
-        assert err == ''
-        mo.assert_called()
+# TODO: comment out ble (for now)
+#@pytest.mark.unit
+#def test_main_info_with_ble_interface(capsys, reset_globals):
+#    """Test --info"""
+#    sys.argv = ['', '--info', '--ble', 'foo']
+#    Globals.getInstance().set_args(sys.argv)
+#
+#    iface = MagicMock(autospec=BLEInterface)
+#    def mock_showInfo():
+#        print('inside mocked showInfo')
+#    iface.showInfo.side_effect = mock_showInfo
+#    with patch('meshtastic.ble_interface.BLEInterface', return_value=iface) as mo:
+#        main()
+#        out, err = capsys.readouterr()
+#        assert re.search(r'Connected to radio', out, re.MULTILINE)
+#        assert re.search(r'inside mocked showInfo', out, re.MULTILINE)
+#        assert err == ''
+#        mo.assert_called()
 
 
 @pytest.mark.unit
@@ -1232,18 +1244,23 @@ def test_main_setchan(capsys, reset_globals):
             main()
         assert pytest_wrapped_e.type == SystemExit
         assert pytest_wrapped_e.value.code == 1
+        _, err = capsys.readouterr()
+        assert re.search(r'usage:', err, re.MULTILINE)
 
 
 @pytest.mark.unit
-def test_main_onReceive_empty(caplog, reset_globals):
+def test_main_onReceive_empty(caplog, reset_globals, capsys):
     """Test onReceive"""
-    sys.argv = ['']
-    Globals.getInstance().set_args(sys.argv)
+    args = MagicMock()
+    Globals.getInstance().set_args(args)
     iface = MagicMock(autospec=SerialInterface)
-    packet = {'decoded': 'foo'}
+    packet = {}
     with caplog.at_level(logging.DEBUG):
         onReceive(packet, iface)
     assert re.search(r'in onReceive', caplog.text, re.MULTILINE)
+    out, err = capsys.readouterr()
+    assert re.search(r"Warning: There is no field 'to' in the packet.", out, re.MULTILINE)
+    assert err == ''
 
 
 #    TODO: use this captured position app message (might want/need in the future)
@@ -1258,7 +1275,7 @@ def test_main_onReceive_empty(caplog, reset_globals):
 #            }
 
 @pytest.mark.unit
-def test_main_onReceive_with_sendtext(caplog, reset_globals):
+def test_main_onReceive_with_sendtext(caplog, capsys, reset_globals):
     """Test onReceive with sendtext
        The entire point of this test is to make sure the interface.close() call
        is made in onReceive().
@@ -1287,63 +1304,48 @@ def test_main_onReceive_with_sendtext(caplog, reset_globals):
             onReceive(packet, iface)
         assert re.search(r'in onReceive', caplog.text, re.MULTILINE)
         mo.assert_called()
+        out, err = capsys.readouterr()
+        assert re.search(r'Sending text message hello to', out, re.MULTILINE)
+        assert err == ''
 
 
-# TODO: re-write this with updated
-# temp disable this test as we need to have a separate thread going with '--reply'
-#@pytest.mark.unit
-#def test_main_onReceive_with_reply(caplog, capsys, reset_globals):
-#    """Test onReceive with a reply
-#       To capture: on one device run '--sendtext aaa --reply' and on another
-#       device run '--sendtext bbb --reply', then back to the first device and
-#       run '--sendtext aaa2 --reply'. You should now see a "Sending reply" message.
-#    """
-#    sys.argv = ['', '--sendtext', 'hello', '--reply']
-#    Globals.getInstance().set_args(sys.argv)
-#
-#    # Note: 'TEXT_MESSAGE_APP' value is 1
-#
-#    send_packet = {
-#            'to': 4294967295,
-#            'decoded': {
-#                'portnum': 1,
-#                'payload': "hello"
-#                },
-#            'id': 334776977,
-#            'hop_limit': 3,
-#            'want_ack': True
-#            }
-#
-#    reply_packet = {
-#            'from': 682968668,
-#            'to': 4294967295,
-#            'decoded': {
-#                'portnum': 'TEXT_MESSAGE_APP',
-#                'payload': b'bbb',
-#                'text': 'bbb'
-#                },
-#            'id': 1709936182,
-#            'rxTime': 1640381999,
-#            'rxSnr': 6.0,
-#            'hopLimit': 3,
-#            'raw': 'faked',
-#            'fromId': '!28b5465c',
-#            'toId': '^all'
-#            }
-#
-#    iface = MagicMock(autospec=SerialInterface)
-#    iface.myInfo.my_node_num = 4294967295
-#
-#    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
-#        with caplog.at_level(logging.DEBUG):
-#            main()
-#            onReceive(send_packet, iface)
-#            onReceive(reply_packet, iface)
-#        assert re.search(r'in onReceive', caplog.text, re.MULTILINE)
-#        out, err = capsys.readouterr()
-#        assert re.search(r'got msg ', out, re.MULTILINE)
-#        assert err == ''
-#        mo.assert_called()
+@pytest.mark.unit
+def test_main_onReceive_with_text(caplog, capsys, reset_globals):
+    """Test onReceive with text
+    """
+    args = MagicMock()
+    args.sendtext.return_value = 'foo'
+    Globals.getInstance().set_args(args)
+
+    # Note: 'TEXT_MESSAGE_APP' value is 1
+    # Note: Some of this is faked below.
+    packet = {
+            'to': 4294967295,
+            'decoded': {
+                'portnum': 1,
+                'payload': "hello",
+                'text': "faked"
+                },
+            'id': 334776977,
+            'hop_limit': 3,
+            'want_ack': True,
+            'rxSnr': 6.0,
+            'hopLimit': 3,
+            'raw': 'faked',
+            'fromId': '!28b5465c',
+            'toId': '^all'
+            }
+
+    iface = MagicMock(autospec=SerialInterface)
+    iface.myInfo.my_node_num = 4294967295
+
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface):
+        with caplog.at_level(logging.DEBUG):
+            onReceive(packet, iface)
+        assert re.search(r'in onReceive', caplog.text, re.MULTILINE)
+        out, err = capsys.readouterr()
+        assert re.search(r'Sending reply', out, re.MULTILINE)
+        assert err == ''
 
 
 @pytest.mark.unit
@@ -1381,6 +1383,10 @@ fixed_position: true
 position_flags: 35"""
         export_config(mo)
     out, err = capsys.readouterr()
+
+    # ensure we do not output this line
+    assert not re.search(r'Connected to radio', out, re.MULTILINE)
+
     assert re.search(r'owner: foo', out, re.MULTILINE)
     assert re.search(r'channel_url: bar', out, re.MULTILINE)
     assert re.search(r'location:', out, re.MULTILINE)
@@ -1406,7 +1412,7 @@ def test_main_export_config_called_from_main(capsys, reset_globals):
     with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
         main()
         out, err = capsys.readouterr()
-        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert not re.search(r'Connected to radio', out, re.MULTILINE)
         assert re.search(r'# start of Meshtastic configure yaml', out, re.MULTILINE)
         assert err == ''
         mo.assert_called()
@@ -1524,6 +1530,36 @@ def test_main_getPref_valid_field(capsys, reset_globals):
     assert err == ''
 
 
+@pytest.mark.unit
+def test_main_getPref_valid_field_string(capsys, reset_globals):
+    """Test getPref() with a valid field and value as a string"""
+    prefs = MagicMock()
+    prefs.DESCRIPTOR.fields_by_name.get.return_value = 'wifi_ssid'
+    prefs.wifi_ssid = 'foo'
+    prefs.ls_secs = 300
+    prefs.fixed_position = False
+
+    getPref(prefs, 'wifi_ssid')
+    out, err = capsys.readouterr()
+    assert re.search(r'wifi_ssid: foo', out, re.MULTILINE)
+    assert err == ''
+
+
+@pytest.mark.unit
+def test_main_getPref_valid_field_bool(capsys, reset_globals):
+    """Test getPref() with a valid field and value as a bool"""
+    prefs = MagicMock()
+    prefs.DESCRIPTOR.fields_by_name.get.return_value = 'fixed_position'
+    prefs.wifi_ssid = 'foo'
+    prefs.ls_secs = 300
+    prefs.fixed_position = False
+
+    getPref(prefs, 'fixed_position')
+    out, err = capsys.readouterr()
+    assert re.search(r'fixed_position: False', out, re.MULTILINE)
+    assert err == ''
+
+
 @pytest.mark.unit
 def test_main_getPref_invalid_field(capsys, reset_globals):
     """Test getPref() with an invalid field"""
@@ -1556,7 +1592,7 @@ def test_main_getPref_invalid_field(capsys, reset_globals):
 
 
 @pytest.mark.unit
-def test_main_setPref_valid_field(capsys, reset_globals):
+def test_main_setPref_valid_field_int(capsys, reset_globals):
     """Test setPref() with a valid field"""
 
     class Field:
@@ -1606,3 +1642,152 @@ def test_main_setPref_invalid_field(capsys, reset_globals):
     # ensure they are sorted
     assert re.search(r'fixed_position\s+is_router\s+ls_secs', out, re.MULTILINE)
     assert err == ''
+
+
+@pytest.mark.unit
+def test_main_ch_set_psk_no_ch_index(capsys, reset_globals):
+    """Test --ch-set psk """
+    sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local']
+    Globals.getInstance().set_args(sys.argv)
+
+    iface = MagicMock(autospec=TCPInterface)
+    with patch('meshtastic.tcp_interface.TCPInterface', return_value=iface) as mo:
+        with pytest.raises(SystemExit) as pytest_wrapped_e:
+            main()
+        out, err = capsys.readouterr()
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r"Warning: Need to specify '--ch-index'", out, re.MULTILINE)
+        assert err == ''
+        assert pytest_wrapped_e.type == SystemExit
+        assert pytest_wrapped_e.value.code == 1
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_ch_set_psk_with_ch_index(capsys, reset_globals):
+    """Test --ch-set psk """
+    sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local', '--ch-index', '0']
+    Globals.getInstance().set_args(sys.argv)
+
+    iface = MagicMock(autospec=TCPInterface)
+    with patch('meshtastic.tcp_interface.TCPInterface', return_value=iface) as mo:
+        main()
+    out, err = capsys.readouterr()
+    assert re.search(r'Connected to radio', out, re.MULTILINE)
+    assert re.search(r"Writing modified channels to device", out, re.MULTILINE)
+    assert err == ''
+    mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_ch_set_name_with_ch_index(capsys, reset_globals):
+    """Test --ch-set setting other than psk"""
+    sys.argv = ['', '--ch-set', 'name', 'foo', '--host', 'meshtastic.local', '--ch-index', '0']
+    Globals.getInstance().set_args(sys.argv)
+
+    iface = MagicMock(autospec=TCPInterface)
+    with patch('meshtastic.tcp_interface.TCPInterface', return_value=iface) as mo:
+        main()
+    out, err = capsys.readouterr()
+    assert re.search(r'Connected to radio', out, re.MULTILINE)
+    assert re.search(r'Set name to foo', out, re.MULTILINE)
+    assert re.search(r"Writing modified channels to device", out, re.MULTILINE)
+    assert err == ''
+    mo.assert_called()
+
+
+@pytest.mark.unit
+def test_onNode(capsys, reset_globals):
+    """Test onNode"""
+    onNode('foo')
+    out, err = capsys.readouterr()
+    assert re.search(r'Node changed', out, re.MULTILINE)
+    assert err == ''
+
+
+@pytest.mark.unit
+def test_tunnel_no_args(capsys, reset_globals):
+    """Test tunnel no arguments"""
+    sys.argv = ['']
+    Globals.getInstance().set_args(sys.argv)
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        tunnelMain()
+    assert pytest_wrapped_e.type == SystemExit
+    assert pytest_wrapped_e.value.code == 1
+    _, err = capsys.readouterr()
+    assert re.search(r'usage: ', err, re.MULTILINE)
+
+
+@pytest.mark.unit
+@patch('meshtastic.util.findPorts', return_value=[])
+@patch('platform.system')
+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'
+    mock_platform_system.side_effect = a_mock
+    sys.argv = ['', '--tunnel']
+    Globals.getInstance().set_args(sys.argv)
+    print(f'platform.system():{platform.system()}')
+    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_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'
+    mock_platform_system.side_effect = a_mock
+    sys.argv = ['', '--subnet', 'foo']
+    Globals.getInstance().set_args(sys.argv)
+    print(f'platform.system():{platform.system()}')
+    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, capsys):
+    """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)
+
+    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)
+        out, err = capsys.readouterr()
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert err == ''

meshtastic/tests/test_mesh_interface.py

@@ -57,10 +57,8 @@ def test_MeshInterface(capsys, reset_globals):
 def test_getMyUser(reset_globals, iface_with_nodes):
     """Test getMyUser()"""
     iface = iface_with_nodes
-
     iface.myInfo.my_node_num = 2475227164
     myuser = iface.getMyUser()
-    print(f'myuser:{myuser}')
     assert myuser is not None
     assert myuser["id"] == '!9388f81c'
 
@@ -460,3 +458,88 @@ def test_generatePacketId(capsys, reset_globals):
         assert re.search(r'Not connected yet, can not generate packet', out, re.MULTILINE)
         assert err == ''
     assert pytest_wrapped_e.type == Exception
+
+
+@pytest.mark.unit
+def test_fixupPosition_empty_pos(capsys, reset_globals):
+    """Test _fixupPosition()"""
+    iface = MeshInterface(noProto=True)
+    pos = {}
+    newpos = iface._fixupPosition(pos)
+    assert newpos == pos
+
+
+@pytest.mark.unit
+def test_fixupPosition_no_changes_needed(capsys, reset_globals):
+    """Test _fixupPosition()"""
+    iface = MeshInterface(noProto=True)
+    pos = {"latitude": 101, "longitude": 102}
+    newpos = iface._fixupPosition(pos)
+    assert newpos == pos
+
+
+@pytest.mark.unit
+def test_fixupPosition(capsys, reset_globals):
+    """Test _fixupPosition()"""
+    iface = MeshInterface(noProto=True)
+    pos = {"latitudeI": 1010000000, "longitudeI": 1020000000}
+    newpos = iface._fixupPosition(pos)
+    assert newpos == {"latitude": 101.0,
+                      "latitudeI": 1010000000,
+                      "longitude": 102.0,
+                      "longitudeI": 1020000000}
+
+
+@pytest.mark.unit
+def test_nodeNumToId(capsys, reset_globals, iface_with_nodes):
+    """Test _nodeNumToId()"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    someid = iface._nodeNumToId(2475227164)
+    assert someid == '!9388f81c'
+
+
+@pytest.mark.unit
+def test_nodeNumToId_not_found(capsys, reset_globals, iface_with_nodes):
+    """Test _nodeNumToId()"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    someid = iface._nodeNumToId(123)
+    assert someid is None
+
+
+@pytest.mark.unit
+def test_nodeNumToId_to_all(capsys, reset_globals, iface_with_nodes):
+    """Test _nodeNumToId()"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    someid = iface._nodeNumToId(0xffffffff)
+    assert someid == '^all'
+
+
+@pytest.mark.unit
+def test_getOrCreateByNum_minimal(capsys, reset_globals, iface_with_nodes):
+    """Test _getOrCreateByNum()"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    tmp = iface._getOrCreateByNum(123)
+    assert tmp == {'num': 123}
+
+
+@pytest.mark.unit
+def test_getOrCreateByNum_not_found(capsys, reset_globals, iface_with_nodes):
+    """Test _getOrCreateByNum()"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    with pytest.raises(Exception) as pytest_wrapped_e:
+        iface._getOrCreateByNum(0xffffffff)
+    assert pytest_wrapped_e.type == Exception
+
+
+@pytest.mark.unit
+def test_getOrCreateByNum(capsys, reset_globals, iface_with_nodes):
+    """Test _getOrCreateByNum()"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    tmp = iface._getOrCreateByNum(2475227164)
+    assert tmp['num'] == 2475227164

meshtastic/tests/test_node.py

@@ -29,7 +29,7 @@ def test_node(capsys):
 
 
 @pytest.mark.unit
-def test_node_reqquestConfig():
+def test_node_requestConfig(capsys):
     """Test run requestConfig"""
     iface = MagicMock(autospec=SerialInterface)
     amesg = MagicMock(autospec=AdminMessage)
@@ -37,6 +37,9 @@ def test_node_reqquestConfig():
         with patch('meshtastic.admin_pb2.AdminMessage', return_value=amesg):
             anode = Node(mo, 'bar')
             anode.requestConfig()
+    out, err = capsys.readouterr()
+    assert re.search(r'Requesting preferences from remote node', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
@@ -106,13 +109,16 @@ def test_reboot(caplog):
 
 
 @pytest.mark.unit
-def test_setURL_empty_url():
+def test_setURL_empty_url(capsys):
     """Test reboot"""
     anode = Node('foo', 'bar', noProto=True)
     with pytest.raises(SystemExit) as pytest_wrapped_e:
         anode.setURL('')
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
+    out, err = capsys.readouterr()
+    assert re.search(r'Warning: No RadioConfig has been read', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
@@ -133,7 +139,7 @@ def test_setURL_valid_URL(caplog):
 
 
 @pytest.mark.unit
-def test_setURL_valid_URL_but_no_settings(caplog):
+def test_setURL_valid_URL_but_no_settings(caplog, capsys):
     """Test setURL"""
     iface = MagicMock(autospec=SerialInterface)
     url = "https://www.meshtastic.org/d/#"
@@ -143,6 +149,9 @@ def test_setURL_valid_URL_but_no_settings(caplog):
         anode.setURL(url)
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
+    out, err = capsys.readouterr()
+    assert re.search(r'Warning: There were no settings', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit
@@ -623,7 +632,7 @@ def test_writeConfig(caplog):
 
 
 @pytest.mark.unit
-def test_requestChannel_not_localNode(caplog):
+def test_requestChannel_not_localNode(caplog, capsys):
     """Test _requestChannel()"""
     iface = MagicMock(autospec=SerialInterface)
     with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -633,6 +642,9 @@ def test_requestChannel_not_localNode(caplog):
         with caplog.at_level(logging.DEBUG):
             anode._requestChannel(0)
             assert re.search(r'Requesting channel 0 info from remote node', caplog.text, re.MULTILINE)
+        out, err = capsys.readouterr()
+        assert re.search(r'Requesting channel 0 info', out, re.MULTILINE)
+        assert err == ''
 
 
 @pytest.mark.unit

meshtastic/tests/test_remote_hardware.py

@@ -79,7 +79,7 @@ def test_watchGPIOs(caplog):
 
 
 @pytest.mark.unit
-def test_sendHardware_no_nodeid():
+def test_sendHardware_no_nodeid(capsys):
     """Test sending no nodeid to _sendHardware()"""
     iface = MagicMock(autospec=SerialInterface)
     with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -87,3 +87,6 @@ def test_sendHardware_no_nodeid():
             rhw = RemoteHardwareClient(mo)
             rhw._sendHardware(None, None)
         assert pytest_wrapped_e.type == SystemExit
+    out, err = capsys.readouterr()
+    assert re.search(r'Warning: Must use a destination node ID', out)
+    assert err == ''

meshtastic/tests/test_serial_interface.py

@@ -11,7 +11,7 @@ from ..serial_interface import SerialInterface
 @pytest.mark.unit
 @patch('serial.Serial')
 @patch('meshtastic.util.findPorts', return_value=['/dev/ttyUSBfake'])
-def test_SerialInterface_single_port(mocked_findPorts, mocked_serial):
+def test_SerialInterface_single_port(mocked_findPorts, mocked_serial, capsys):
     """Test that we can instantiate a SerialInterface with a single port"""
     iface = SerialInterface(noProto=True)
     iface.showInfo()
@@ -19,6 +19,12 @@ def test_SerialInterface_single_port(mocked_findPorts, mocked_serial):
     iface.close()
     mocked_findPorts.assert_called()
     mocked_serial.assert_called()
+    out, err = capsys.readouterr()
+    assert re.search(r'Nodes in mesh', out, re.MULTILINE)
+    assert re.search(r'Preferences', out, re.MULTILINE)
+    assert re.search(r'Channels', out, re.MULTILINE)
+    assert re.search(r'Primary channel', out, re.MULTILINE)
+    assert err == ''
 
 
 @pytest.mark.unit

meshtastic/tests/test_stream_interface.py

@@ -34,9 +34,9 @@ def test_StreamInterface_with_noProto(caplog, reset_globals):
         assert data == test_data
 
 
-# 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
+## 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()"""
@@ -78,4 +78,3 @@ def test_sendToRadioImpl(caplog, reset_globals):
         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)

meshtastic/tests/test_tcp_interface.py

@@ -13,6 +13,7 @@ def test_TCPInterface(capsys):
     """Test that we can instantiate a TCPInterface"""
     with patch('socket.socket') as mock_socket:
         iface = TCPInterface(hostname='localhost', noProto=True)
+        iface.myConnect()
         iface.showInfo()
         iface.localNode.showInfo()
         out, err = capsys.readouterr()
@@ -24,3 +25,11 @@ def test_TCPInterface(capsys):
         assert err == ''
         assert mock_socket.called
         iface.close()
+
+
+@pytest.mark.unit
+def test_TCPInterface_without_connecting(capsys):
+    """Test that we can instantiate a TCPInterface with connectNow as false"""
+    with patch('socket.socket'):
+        iface = TCPInterface(hostname='localhost', noProto=True, connectNow=False)
+        assert iface.socket is None

meshtastic/tests/test_tunnel.py

@@ -0,0 +1,266 @@
+"""Meshtastic unit tests for tunnel.py"""
+
+import re
+import sys
+import logging
+
+from unittest.mock import patch, MagicMock
+import pytest
+
+from ..tcp_interface import TCPInterface
+from ..tunnel import Tunnel, onTunnelReceive
+from ..globals import Globals
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_Tunnel_on_non_linux_system(mock_platform_system, reset_globals):
+    """Test that we cannot instantiate a Tunnel on a non Linux system"""
+    a_mock = MagicMock()
+    a_mock.return_value = 'notLinux'
+    mock_platform_system.side_effect = a_mock
+    with patch('socket.socket') as mock_socket:
+        with pytest.raises(Exception) as pytest_wrapped_e:
+            iface = TCPInterface(hostname='localhost', noProto=True)
+            Tunnel(iface)
+        assert pytest_wrapped_e.type == Exception
+        assert mock_socket.called
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_Tunnel_without_interface(mock_platform_system, reset_globals):
+    """Test that we can not instantiate a Tunnel without a valid interface"""
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with pytest.raises(Exception) as pytest_wrapped_e:
+        Tunnel(None)
+    assert pytest_wrapped_e.type == Exception
+
+
+@pytest.mark.unitslow
+@patch('platform.system')
+def test_Tunnel_with_interface(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test that we can not instantiate a Tunnel without a valid interface"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.WARNING):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            assert tun == Globals.getInstance().get_tunnelInstance()
+            iface.close()
+    assert re.search(r'Not creating a TapDevice()', caplog.text, re.MULTILINE)
+    assert re.search(r'Not starting TUN reader', caplog.text, re.MULTILINE)
+    assert re.search(r'Not sending packet', caplog.text, re.MULTILINE)
+
+
+@pytest.mark.unitslow
+@patch('platform.system')
+def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test onTunnelReceive"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    sys.argv = ['']
+    Globals.getInstance().set_args(sys.argv)
+    packet = {'decoded': { 'payload': 'foo'}, 'from': 2475227164}
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            Globals.getInstance().set_tunnelInstance(tun)
+            onTunnelReceive(packet, iface)
+    assert re.search(r'in onTunnelReceive', caplog.text, re.MULTILINE)
+    assert re.search(r'Ignoring message we sent', caplog.text, re.MULTILINE)
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_onTunnelReceive_from_someone_else(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test onTunnelReceive"""
+    iface = iface_with_nodes
+    iface.myInfo.my_node_num = 2475227164
+    sys.argv = ['']
+    Globals.getInstance().set_args(sys.argv)
+    packet = {'decoded': { 'payload': 'foo'}, 'from': 123}
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            Globals.getInstance().set_tunnelInstance(tun)
+            onTunnelReceive(packet, iface)
+    assert re.search(r'in onTunnelReceive', caplog.text, re.MULTILINE)
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_random(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # random packet
+    packet = b'1234567890123456789012345678901234567890'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert not ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # faked IGMP
+    packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_icmp(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # faked ICMP
+    packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert re.search(r'forwarding ICMP message', caplog.text, re.MULTILINE)
+            assert not ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_udp(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # faked UDP
+    packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert re.search(r'forwarding udp', caplog.text, re.MULTILINE)
+            assert not ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # faked UDP
+    packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07\x6c\x07\x6c\x00\x00\x00'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    # Note: custom logging level
+    LOG_TRACE = 5
+    with caplog.at_level(LOG_TRACE):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert re.search(r'ignoring blacklisted UDP', caplog.text, re.MULTILINE)
+            assert ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_tcp(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # faked TCP
+    packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert re.search(r'forwarding tcp', caplog.text, re.MULTILINE)
+            assert not ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _shouldFilterPacket()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    # faked TCP
+    packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x17\x0c\x17\x0c\x00\x00\x00'
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    # Note: custom logging level
+    LOG_TRACE = 5
+    with caplog.at_level(LOG_TRACE):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            ignore = tun._shouldFilterPacket(packet)
+            assert re.search(r'ignoring blacklisted TCP', caplog.text, re.MULTILINE)
+            assert ignore
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_ipToNodeId_none(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _ipToNodeId()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            nodeid = tun._ipToNodeId('something not useful')
+            assert nodeid is None
+
+
+@pytest.mark.unit
+@patch('platform.system')
+def test_ipToNodeId_all(mock_platform_system, caplog, reset_globals, iface_with_nodes):
+    """Test _ipToNodeId()"""
+    iface = iface_with_nodes
+    iface.noProto = True
+    a_mock = MagicMock()
+    a_mock.return_value = 'Linux'
+    mock_platform_system.side_effect = a_mock
+    with caplog.at_level(logging.DEBUG):
+        with patch('socket.socket'):
+            tun = Tunnel(iface)
+            nodeid = tun._ipToNodeId(b'\x00\x00\xff\xff')
+            assert nodeid == '^all'

meshtastic/tests/test_util.py

@@ -3,12 +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)
+                             remove_keys_from_dict, Timeout, hexstr,
+                             ipstr, readnet_u16, findPorts, convert_mac_addr)
 
 
 @pytest.mark.unit
@@ -39,7 +41,7 @@ def test_fromStr():
     assert fromStr('abc') == 'abc'
 
 
-@pytest.mark.unit
+@pytest.mark.unitslow
 def test_quoteBooleans():
     """Test quoteBooleans"""
     assert quoteBooleans('') == ''
@@ -92,7 +94,7 @@ def test_pskToString_one_byte_non_zero_value():
     assert pskToString(bytes([0x01])) == 'default'
 
 
-@pytest.mark.unit
+@pytest.mark.unitslow
 def test_pskToString_many_bytes():
     """Test pskToString many bytes"""
     assert pskToString(bytes([0x02, 0x01])) == 'secret'
@@ -104,20 +106,26 @@ def test_pskToString_simple():
     assert pskToString(bytes([0x03])) == 'simple2'
 
 
-@pytest.mark.unit
-def test_our_exit_zero_return_value():
+@pytest.mark.unitslow
+def test_our_exit_zero_return_value(capsys):
     """Test our_exit with a zero return value"""
     with pytest.raises(SystemExit) as pytest_wrapped_e:
         our_exit("Warning: Some message", 0)
+    out, err = capsys.readouterr()
+    assert re.search(r'Warning: Some message', out, re.MULTILINE)
+    assert err == ''
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 0
 
 
 @pytest.mark.unit
-def test_our_exit_non_zero_return_value():
+def test_our_exit_non_zero_return_value(capsys):
     """Test our_exit with a non-zero return value"""
     with pytest.raises(SystemExit) as pytest_wrapped_e:
         our_exit("Error: Some message", 1)
+    out, err = capsys.readouterr()
+    assert re.search(r'Error: Some message', out, re.MULTILINE)
+    assert err == ''
     assert pytest_wrapped_e.type == SystemExit
     assert pytest_wrapped_e.value.code == 1
 
@@ -152,7 +160,7 @@ def test_catchAndIgnore(caplog):
     assert re.search(r'Exception thrown in something', caplog.text, re.MULTILINE)
 
 
-@pytest.mark.unit
+@pytest.mark.unitslow
 def test_remove_keys_from_dict_empty_keys_empty_dict():
     """Test when keys and dict both are empty"""
     assert not remove_keys_from_dict((), {})
@@ -174,3 +182,53 @@ def test_remove_keys_from_dict_empty_keys():
 def test_remove_keys_from_dict():
     """Test remove_keys_from_dict()"""
     assert remove_keys_from_dict(('b'), {'a':1, 'b':2}) == {'a':1}
+
+
+@pytest.mark.unitslow
+def test_Timeout_not_found():
+    """Test Timeout()"""
+    to = Timeout(0.2)
+    attrs = ('foo')
+    to.waitForSet('bar', attrs)
+
+
+@pytest.mark.unitslow
+def test_Timeout_found():
+    """Test Timeout()"""
+    to = Timeout(0.2)
+    attrs = ()
+    to.waitForSet('bar', attrs)
+
+
+@pytest.mark.unitslow
+def test_hexstr():
+    """Test hexstr()"""
+    assert hexstr(b'123') == '31:32:33'
+    assert hexstr(b'') == ''
+
+
+@pytest.mark.unit
+def test_ipstr():
+    """Test ipstr()"""
+    assert ipstr(b'1234') == '49.50.51.52'
+    assert ipstr(b'') == ''
+
+
+@pytest.mark.unitslow
+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()
+
+
+@pytest.mark.unit
+def test_convert_mac_addr():
+    """Test convert_mac_addr()"""
+    assert convert_mac_addr('/c0gFyhb') == 'fd:cd:20:17:28:5b'
+    assert convert_mac_addr('') == ''

meshtastic/tunnel.py

@@ -17,61 +17,28 @@
 
 import logging
 import threading
+import platform
 from pubsub import pub
 
 from pytap2 import TapDevice
 
 from . import portnums_pb2
-
-# A new non standard log level that is lower level than DEBUG
-LOG_TRACE = 5
-
-# fixme - find a way to move onTunnelReceive inside of the class
-tunnelInstance = None
-
-"""A list of chatty UDP services we should never accidentally
-forward to our slow network"""
-udpBlacklist = {
-    1900,  # SSDP
-    5353,  # multicast DNS
-}
-
-"""A list of TCP services to block"""
-tcpBlacklist = {}
-
-"""A list of protocols we ignore"""
-protocolBlacklist = {
-    0x02,  # IGMP
-    0x80,  # Service-Specific Connection-Oriented Protocol in a Multilink and Connectionless Environment
-}
-
-
-def hexstr(barray):
-    """Print a string of hex digits"""
-    return ":".join('{:02x}'.format(x) for x in barray)
-
-
-def ipstr(barray):
-    """Print a string of ip digits"""
-    return ".".join('{}'.format(x) for x in barray)
-
-
-def readnet_u16(p, offset):
-    """Read big endian u16 (network byte order)"""
-    return p[offset] * 256 + p[offset + 1]
+from .util import ipstr, readnet_u16
+from .globals import Globals
 
 
 def onTunnelReceive(packet, interface):
-    """Callback for received tunneled messages from mesh
-
-    FIXME figure out how to do closures with methods in python"""
+    """Callback for received tunneled messages from mesh."""
+    logging.debug(f'in onTunnelReceive()')
+    our_globals = Globals.getInstance()
+    tunnelInstance = our_globals.get_tunnelInstance()
     tunnelInstance.onReceive(packet)
 
 
 class Tunnel:
     """A TUN based IP tunnel over meshtastic"""
 
-    def __init__(self, iface, subnet=None, netmask="255.255.0.0"):
+    def __init__(self, iface, subnet='10.115', netmask="255.255.0.0"):
         """
         Constructor
 
@@ -79,35 +46,69 @@ class Tunnel:
         subnet is used to construct our network number (normally 10.115.x.x)
         """
 
-        if subnet is None:
-            subnet = "10.115"
+        if not iface:
+            raise Exception("Tunnel() must have a interface")
 
         self.iface = iface
         self.subnetPrefix = subnet
 
-        global tunnelInstance
-        tunnelInstance = self
+        if platform.system() != 'Linux':
+            raise Exception("Tunnel() can only be run instantiated on a Linux system")
 
+        our_globals = Globals.getInstance()
+        our_globals.set_tunnelInstance(self)
+
+        """A list of chatty UDP services we should never accidentally
+        forward to our slow network"""
+        self.udpBlacklist = {
+            1900,  # SSDP
+            5353,  # multicast DNS
+        }
+
+        """A list of TCP services to block"""
+        self.tcpBlacklist = {
+            5900,  # VNC (Note: Only adding for testing purposes.)
+        }
+
+        """A list of protocols we ignore"""
+        self.protocolBlacklist = {
+            0x02,  # IGMP
+            0x80,  # Service-Specific Connection-Oriented Protocol in a Multilink and Connectionless Environment
+        }
+
+        # A new non standard log level that is lower level than DEBUG
+        self.LOG_TRACE = 5
+
+        # TODO: check if root?
         logging.info("Starting IP to mesh tunnel (you must be root for this *pre-alpha* "\
                      "feature to work).  Mesh members:")
 
         pub.subscribe(onTunnelReceive, "meshtastic.receive.data.IP_TUNNEL_APP")
         myAddr = self._nodeNumToIp(self.iface.myInfo.my_node_num)
 
-        for node in self.iface.nodes.values():
-            nodeId = node["user"]["id"]
-            ip = self._nodeNumToIp(node["num"])
-            logging.info(f"Node { nodeId } has IP address { ip }")
+        if self.iface.nodes:
+            for node in self.iface.nodes.values():
+                nodeId = node["user"]["id"]
+                ip = self._nodeNumToIp(node["num"])
+                logging.info(f"Node { nodeId } has IP address { ip }")
 
         logging.debug("creating TUN device with MTU=200")
         # FIXME - figure out real max MTU, it should be 240 - the overhead bytes for SubPacket and Data
-        self.tun = TapDevice(name="mesh")
-        self.tun.up()
-        self.tun.ifconfig(address=myAddr, netmask=netmask, mtu=200)
-        logging.debug(f"starting TUN reader, our IP address is {myAddr}")
-        self._rxThread = threading.Thread(
-            target=self.__tunReader, args=(), daemon=True)
-        self._rxThread.start()
+        self.tun = None
+        if self.iface.noProto:
+            logging.warning(f"Not creating a TapDevice() because it is disabled by noProto")
+        else:
+            self.tun = TapDevice(name="mesh")
+            self.tun.up()
+            self.tun.ifconfig(address=myAddr, netmask=netmask, mtu=200)
+
+        self._rxThread = None
+        if self.iface.noProto:
+            logging.warning(f"Not starting TUN reader because it is disabled by noProto")
+        else:
+            logging.debug(f"starting TUN reader, our IP address is {myAddr}")
+            self._rxThread = threading.Thread(target=self.__tunReader, args=(), daemon=True)
+            self._rxThread.start()
 
     def onReceive(self, packet):
         """onReceive"""
@@ -115,12 +116,12 @@ class Tunnel:
         if packet["from"] == self.iface.myInfo.my_node_num:
             logging.debug("Ignoring message we sent")
         else:
-            logging.debug(
-                f"Received mesh tunnel message type={type(p)} len={len(p)}")
+            logging.debug(f"Received mesh tunnel message type={type(p)} len={len(p)}")
             # we don't really need to check for filtering here (sender should have checked),
             # but this provides useful debug printing on types of packets received
-            if not self._shouldFilterPacket(p):
-                self.tun.write(p)
+            if not self.iface.noProto:
+                if not self._shouldFilterPacket(p):
+                    self.tun.write(p)
 
     def _shouldFilterPacket(self, p):
         """Given a packet, decode it and return true if it should be ignored"""
@@ -129,10 +130,9 @@ class Tunnel:
         destAddr = p[16:20]
         subheader = 20
         ignore = False  # Assume we will be forwarding the packet
-        if protocol in protocolBlacklist:
+        if protocol in self.protocolBlacklist:
             ignore = True
-            logging.log(
-                LOG_TRACE, f"Ignoring blacklisted protocol 0x{protocol:02x}")
+            logging.log(self.LOG_TRACE, f"Ignoring blacklisted protocol 0x{protocol:02x}")
         elif protocol == 0x01:  # ICMP
             icmpType = p[20]
             icmpCode = p[21]
@@ -145,19 +145,17 @@ class Tunnel:
         elif protocol == 0x11:  # UDP
             srcport = readnet_u16(p, subheader)
             destport = readnet_u16(p, subheader + 2)
-            if destport in udpBlacklist:
+            if destport in self.udpBlacklist:
                 ignore = True
-                logging.log(
-                    LOG_TRACE, f"ignoring blacklisted UDP port {destport}")
+                logging.log(self.LOG_TRACE, f"ignoring blacklisted UDP port {destport}")
             else:
-                logging.debug(
-                    f"forwarding udp srcport={srcport}, destport={destport}")
+                logging.debug(f"forwarding udp srcport={srcport}, destport={destport}")
         elif protocol == 0x06:  # TCP
             srcport = readnet_u16(p, subheader)
             destport = readnet_u16(p, subheader + 2)
-            if destport in tcpBlacklist:
+            if destport in self.tcpBlacklist:
                 ignore = True
-                logging.log(LOG_TRACE, f"ignoring blacklisted TCP port {destport}")
+                logging.log(self.LOG_TRACE, f"ignoring blacklisted TCP port {destport}")
             else:
                 logging.debug(f"forwarding tcp srcport={srcport}, destport={destport}")
         else:

meshtastic/util.py

@@ -4,6 +4,7 @@ import traceback
 from queue import Queue
 import os
 import sys
+import base64
 import time
 import platform
 import logging
@@ -169,8 +170,7 @@ class DeferredExecution():
                 o = self.queue.get()
                 o()
             except:
-                logging.error(
-                    f"Unexpected error in deferred execution {sys.exc_info()[0]}")
+                logging.error(f"Unexpected error in deferred execution {sys.exc_info()[0]}")
                 print(traceback.format_exc())
 
 
@@ -210,3 +210,27 @@ def remove_keys_from_dict(keys, adict):
         if key in adict:
             del newdict[key]
     return newdict
+
+
+def hexstr(barray):
+    """Print a string of hex digits"""
+    return ":".join('{:02x}'.format(x) for x in barray)
+
+
+def ipstr(barray):
+    """Print a string of ip digits"""
+    return ".".join('{}'.format(x) for x in barray)
+
+
+def readnet_u16(p, offset):
+    """Read big endian u16 (network byte order)"""
+    return p[offset] * 256 + p[offset + 1]
+
+
+def convert_mac_addr(val):
+    """Convert the base 64 encoded value to a mac address
+       val - base64 encoded value (ex: '/c0gFyhb'))
+       returns: a string formatted like a mac address (ex: 'fd:cd:20:17:28:5b')
+    """
+    val_as_bytes = base64.b64decode(val)
+    return hexstr(val_as_bytes)

vercel.json

@@ -0,0 +1,5 @@
+{
+    "github": {
+        "silent": true
+    }
+}