add unit test for --set-owner and --set-ham; refactor setting encryption after

meshtastic/__main__.py

@@ -56,10 +56,6 @@ 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"""
 
@@ -236,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

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

@@ -10,6 +10,7 @@ import pytest
 from meshtastic.__main__ import initParser, main, Globals
 
 from ..serial_interface import SerialInterface
+from ..node import Node
 
 
 @pytest.mark.unit
@@ -278,3 +279,61 @@ def test_main_nodes(capsys):
         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()