Merge pull request #161 from mkinney/work_on_channel

sort possible options/tests, figured out how to mock so --info can be…

bin/prerelease-tests.sh

@@ -12,9 +12,6 @@ bin/run.sh --set is_router false
 # TODO: This does not seem to work.
 echo setting channel
 bin/run.sh --seturl "https://www.meshtastic.org/c/#GAMiENTxuzogKQdZ8Lz_q89Oab8qB0RlZmF1bHQ="
-echo setting time
-# '--settime' seems to be deprecated
-#bin/run.sh --settime
 echo setting owner
 bin/run.sh --set-owner "Test Build"
 echo setting position

meshtastic/__main__.py

@@ -13,8 +13,6 @@ import pyqrcode
 import pkg_resources
 import meshtastic.util
 import meshtastic.test
-from meshtastic.serial_interface import SerialInterface
-from .serial_interface import SerialInterface
 from .tcp_interface import TCPInterface
 from .ble_interface import BLEInterface
 from . import remote_hardware
@@ -58,20 +56,19 @@ def onConnection(interface, topic=pub.AUTO_TOPIC):
     print(f"Connection changed: {topic.getName()}")
 
 
-never = 0xffffffff
-oneday = 24 * 60 * 60
-
-
 def getPref(attributes, name):
     """Get a channel or preferences value"""
 
     objDesc = attributes.DESCRIPTOR
     field = objDesc.fields_by_name.get(name)
     if not field:
-        print(f"{attributes.__class__.__name__} doesn't have an attribute called {name}, so you can not get it.")
-        print(f"Choices are:")
+        print(f"{attributes.__class__.__name__} does not have an attribute called {name}, so you can not get it.")
+        print(f"Choices in sorted order are:")
+        names = []
         for f in objDesc.fields:
-            print(f"  {f.name}")
+            names.append(f'{f.name}')
+        for temp_name in sorted(names):
+            print(f"    {temp_name}")
         return
 
     # okay - try to read the value
@@ -94,10 +91,13 @@ def setPref(attributes, name, valStr):
     objDesc = attributes.DESCRIPTOR
     field = objDesc.fields_by_name.get(name)
     if not field:
-        print(f"{attributes.__class__.__name__} doesn't have an attribute called {name}, so you can not set it.")
-        print(f"Choices are:")
+        print(f"{attributes.__class__.__name__} does not have an attribute called {name}, so you can not set it.")
+        print(f"Choices in sorted order are:")
+        names = []
         for f in objDesc.fields:
-            print(f"  {f.name}")
+            names.append(f'{f.name}')
+        for temp_name in sorted(names):
+            print(f"    {temp_name}")
         return
 
     val = meshtastic.util.fromStr(valStr)
@@ -110,10 +110,13 @@ def setPref(attributes, name, valStr):
         if e:
             val = e.number
         else:
-            print(f"{name} doesn't have an enum called {val}, so you can not set it.")
-            print(f"Choices are:")
+            print(f"{name} does not have an enum called {val}, so you can not set it.")
+            print(f"Choices in sorted order are:")
+            names = []
             for f in enumType.values:
-                print(f"  {f.name}")
+                names.append(f'{f.name}')
+            for temp_name in sorted(names):
+                print(f"    {temp_name}")
             return
 
     # okay - try to read the value
@@ -229,14 +232,10 @@ def onConnected(interface):
 
         if args.set_ham:
             closeNow = True
-            print(
-                f"Setting HAM ID to {args.set_ham} and turning off encryption")
+            print(f"Setting HAM ID to {args.set_ham} and turning off encryption")
             getNode().setOwner(args.set_ham, is_licensed=True)
-            # Must turn off crypt on primary channel
-            ch = getNode().channels[0]
-            ch.settings.psk = meshtastic.util.fromPSK("none")
-            print(f"Writing modified channels to device")
-            getNode().writeChannel(0)
+            # Must turn off encryption on primary channel
+            getNode().turnOffEncryptionOnPrimaryChannel()
 
         if args.reboot:
             closeNow = True
@@ -478,8 +477,7 @@ def onConnected(interface):
 
             # Handle the int/float/bool arguments
             for pref in args.get:
-                getPref(
-                    prefs, pref[0])
+                getPref(prefs, pref[0])
 
             print("Completed getting preferences")
 
@@ -591,7 +589,7 @@ def common():
                 client = TCPInterface(
                     args.host, debugOut=logfile, noProto=args.noproto)
             else:
-                client = SerialInterface(
+                client = meshtastic.serial_interface.SerialInterface(
                     args.port, debugOut=logfile, noProto=args.noproto)
 
             # We assume client is fully connected now

meshtastic/node.py

@@ -5,7 +5,7 @@ import logging
 import base64
 from google.protobuf.json_format import MessageToJson
 from . import portnums_pb2, apponly_pb2, admin_pb2, channel_pb2
-from .util import pskToString, stripnl, Timeout, our_exit
+from .util import pskToString, stripnl, Timeout, our_exit, fromPSK
 
 
 class Node:
@@ -56,6 +56,12 @@ class Node:
 
         self._requestSettings()
 
+    def turnOffEncryptionOnPrimaryChannel(self):
+        """Turn off encryption on primary channel."""
+        self.channels[0].settings.psk = fromPSK("none")
+        print("Writing modified channels to device")
+        self.writeChannel(0)
+
     def waitForConfig(self):
         """Block until radio config is received. Returns True if config has been received."""
         return self._timeout.waitForSet(self, attrs=('radioConfig', 'channels'))

meshtastic/tests/test_main.py

@@ -4,11 +4,13 @@ import sys
 import argparse
 import re
 
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 import pytest
 
 from meshtastic.__main__ import initParser, main, Globals
-#from meshtastic.serial_interface import SerialInterface
+
+from ..serial_interface import SerialInterface
+from ..node import Node
 
 
 @pytest.mark.unit
@@ -202,31 +204,225 @@ def test_main_test_two_ports_fails(patched_find_ports, patched_test_all):
     assert pytest_wrapped_e.value.code == 1
     # TODO: why does this fail? patched_find_ports.assert_called()
     patched_test_all.assert_called()
-#
-#
-#@pytest.mark.unit
-#@patch('meshtastic.stream_interface.StreamInterface.__init__')
-#@patch('serial.Serial')
-#@patch('meshtastic.serial_interface.SerialInterface')
-#@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1'])
-#def test_main_info_one_port(patched_find_ports, patched_serial_interface,
-#                            patched_serial_serial, patched_stream_interface_constructor):
-#    """Test --info one fake port"""
-#    iface = MagicMock()
-#    patched_serial_interface.return_value = iface
-#    astream = MagicMock()
-#    patched_serial_serial = astream
-#    siface = MagicMock()
-#    patched_stream_interface_constructor = siface
-#    sys.argv = ['', '--info']
-#    args = sys.argv
-#    parser = None
-#    parser = argparse.ArgumentParser()
-#    our_globals = Globals.getInstance()
-#    our_globals.set_parser(parser)
-#    our_globals.set_args(args)
-#    main()
-#    patched_find_ports.assert_called()
-#    patched_serial_interface.assert_called()
-#    patched_serial_serial.assert_called()
-#    patched_stream_interface_constructor
+
+
+@pytest.mark.unit
+def test_main_info(capsys):
+    """Test --info"""
+    sys.argv = ['', '--info']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    iface = MagicMock(autospec=SerialInterface)
+    def mock_showInfo():
+        print('inside mocked showInfo')
+    iface.showInfo.side_effect = mock_showInfo
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        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
+def test_main_qr(capsys):
+    """Test --qr"""
+    sys.argv = ['', '--qr']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    iface = MagicMock(autospec=SerialInterface)
+    # TODO: could mock/check url
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'Primary channel URL', out, re.MULTILINE)
+        # if a qr code is generated it will have lots of these
+        assert re.search(r'\[7m', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_nodes(capsys):
+    """Test --nodes"""
+    sys.argv = ['', '--nodes']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    iface = MagicMock(autospec=SerialInterface)
+    def mock_showNodes():
+        print('inside mocked showNodes')
+    iface.showNodes.side_effect = mock_showNodes
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'inside mocked showNodes', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_set_owner_to_bob(capsys):
+    """Test --set-owner bob"""
+    sys.argv = ['', '--set-owner', 'bob']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    iface = MagicMock(autospec=SerialInterface)
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'Setting device owner to bob', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_set_ham_to_KI123(capsys):
+    """Test --set-ham KI123"""
+    sys.argv = ['', '--set-ham', 'KI123']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    our_globals.set_target_node(None)
+
+    mocked_node = MagicMock(autospec=Node)
+    def mock_turnOffEncryptionOnPrimaryChannel():
+        print('inside mocked turnOffEncryptionOnPrimaryChannel')
+    def mock_setOwner(name, is_licensed):
+        print('inside mocked setOwner')
+    mocked_node.turnOffEncryptionOnPrimaryChannel.side_effect = mock_turnOffEncryptionOnPrimaryChannel
+    mocked_node.setOwner.side_effect = mock_setOwner
+
+    iface = MagicMock(autospec=SerialInterface)
+    iface.getNode.return_value = mocked_node
+
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'Setting HAM ID to KI123', out, re.MULTILINE)
+        assert re.search(r'inside mocked setOwner', out, re.MULTILINE)
+        assert re.search(r'inside mocked turnOffEncryptionOnPrimaryChannel', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_reboot(capsys):
+    """Test --reboot"""
+    sys.argv = ['', '--reboot']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    our_globals.set_target_node(None)
+
+    mocked_node = MagicMock(autospec=Node)
+    def mock_reboot():
+        print('inside mocked reboot')
+    mocked_node.reboot.side_effect = mock_reboot
+
+    iface = MagicMock(autospec=SerialInterface)
+    iface.getNode.return_value = mocked_node
+
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'inside mocked reboot', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_sendtext(capsys):
+    """Test --sendtext"""
+    sys.argv = ['', '--sendtext', 'hello']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    our_globals.set_target_node(None)
+
+    iface = MagicMock(autospec=SerialInterface)
+    def mock_sendText(text, dest, wantAck):
+        print('inside mocked sendText')
+    iface.sendText.side_effect = mock_sendText
+
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'Sending text message', out, re.MULTILINE)
+        assert re.search(r'inside mocked sendText', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()
+
+
+@pytest.mark.unit
+def test_main_sendping(capsys):
+    """Test --sendping"""
+    sys.argv = ['', '--sendping']
+    args = sys.argv
+    parser = None
+    parser = argparse.ArgumentParser()
+    our_globals = Globals.getInstance()
+    our_globals.set_parser(parser)
+    our_globals.set_args(args)
+    our_globals.set_target_node(None)
+
+    iface = MagicMock(autospec=SerialInterface)
+    def mock_sendData(payload, dest, portNum, wantAck, wantResponse):
+        print('inside mocked sendData')
+    iface.sendData.side_effect = mock_sendData
+
+    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
+        main()
+        out, err = capsys.readouterr()
+        print('out:', out)
+        print('err:', err)
+        assert re.search(r'Connected to radio', out, re.MULTILINE)
+        assert re.search(r'Sending ping message', out, re.MULTILINE)
+        assert re.search(r'inside mocked sendData', out, re.MULTILINE)
+        assert err == ''
+        mo.assert_called()

meshtastic/tests/test_smoke1.py

@@ -48,6 +48,30 @@ def test_smoke1_sendping():
     assert return_value == 0
 
 
+@pytest.mark.smoke1
+def test_get_with_invalid_setting():
+    """Test '--get a_bad_setting'."""
+    return_value, out = subprocess.getstatusoutput('meshtastic --get a_bad_setting')
+    assert re.search(r'Choices in sorted order', out)
+    assert return_value == 0
+
+
+@pytest.mark.smoke1
+def test_set_with_invalid_setting():
+    """Test '--set a_bad_setting'."""
+    return_value, out = subprocess.getstatusoutput('meshtastic --set a_bad_setting foo')
+    assert re.search(r'Choices in sorted order', out)
+    assert return_value == 0
+
+
+@pytest.mark.smoke1
+def test_ch_set_with_invalid_settingpatch_find_ports():
+    """Test '--ch-set with a_bad_setting'."""
+    return_value, out = subprocess.getstatusoutput('meshtastic --ch-set invalid_setting foo --ch-index 0')
+    assert re.search(r'Choices in sorted order', out)
+    assert return_value == 0
+
+
 @pytest.mark.smoke1
 def test_smoke1_pos_fields():
     """Test --pos-fields (with some values POS_ALTITUDE POS_ALT_MSL POS_BATTERY)"""