diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 7d7f9c0..a241d45 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -16,13 +16,13 @@ import meshtastic.test from .tcp_interface import TCPInterface from .ble_interface import BLEInterface from . import remote_hardware -from . import portnums_pb2, channel_pb2, mesh_pb2, radioconfig_pb2 +from . import portnums_pb2, channel_pb2, radioconfig_pb2 from .globals import Globals + """We only import the tunnel code if we are on a platform that can run it""" have_tunnel = platform.system() == 'Linux' - def onReceive(packet, interface): """Callback invoked when a packet arrives""" our_globals = Globals.getInstance() @@ -156,7 +156,6 @@ def onConnected(interface): alt = 0 lat = 0.0 lon = 0.0 - timeval = 0 # always set time, but based on the local clock prefs = interface.localNode.radioConfig.preferences if args.setalt: alt = int(args.setalt) @@ -173,7 +172,7 @@ def onConnected(interface): print("Setting device position") # can include lat/long/alt etc: latitude = 37.5, longitude = -122.1 - interface.sendPosition(lat, lon, alt, timeval) + interface.sendPosition(lat, lon, alt) interface.localNode.writeConfig() elif not args.no_time: # We normally provide a current time to the mesh when we connect @@ -220,14 +219,14 @@ def onConnected(interface): if args.set_team: closeNow = True try: - v_team = mesh_pb2.Team.Value(args.set_team.upper()) + v_team = meshtastic.mesh_pb2.Team.Value(args.set_team.upper()) except ValueError: v_team = 0 print(f"ERROR: Team \'{args.set_team}\' not found.") - print("Try a team name from the list below, or CLEAR for unaffiliated:") - print(mesh_pb2.Team.keys()) + print("Try a team name from the sorted list below, or use 'CLEAR' for unaffiliated:") + print(sorted(meshtastic.mesh_pb2.Team.keys())) else: - print(f"Setting team to {mesh_pb2.Team.Name(v_team)}") + print(f"Setting team to {meshtastic.mesh_pb2.Team.Name(v_team)}") getNode().setOwner(team=v_team) if args.set_ham: @@ -291,8 +290,7 @@ def onConnected(interface): # Handle the int/float/bool arguments for pref in args.set: - setPref( - prefs, pref[0], pref[1]) + setPref(prefs, pref[0], pref[1]) print("Writing modified preferences to device") getNode().writeConfig() @@ -314,7 +312,6 @@ def onConnected(interface): alt = 0 lat = 0.0 lon = 0.0 - timeval = 0 # always set time, but based on the local clock prefs = interface.localNode.radioConfig.preferences if 'alt' in configuration['location']: @@ -330,7 +327,7 @@ def onConnected(interface): prefs.fixed_position = True print(f"Fixing longitude at {lon} degrees") print("Setting device position") - interface.sendPosition(lat, lon, alt, timeval) + interface.sendPosition(lat, lon, alt) interface.localNode.writeConfig() if 'user_prefs' in configuration: @@ -348,14 +345,14 @@ def onConnected(interface): if args.ch_add: closeNow = True - n = getNode() if len(args.ch_add) > 10: meshtastic.util.our_exit("Warning: Channel name must be shorter. Channel not added.") + n = getNode() ch = n.getChannelByName(args.ch_add) if ch: - logging.error( - f"This node already has a '{args.ch_add}' channel - no changes.") + meshtastic.util.our_exit(f"Warning: This node already has a '{args.ch_add}' channel. No changes were made.") else: + # get the first channel that is disabled (i.e., available) ch = n.getDisabledChannel() if not ch: meshtastic.util.our_exit("Warning: No free channels were found") @@ -421,28 +418,22 @@ def onConnected(interface): # handle the simple channel set commands if args.ch_longslow: - setSimpleChannel( - channel_pb2.ChannelSettings.ModemConfig.Bw125Cr48Sf4096) + setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw125Cr48Sf4096) if args.ch_longfast: - setSimpleChannel( - channel_pb2.ChannelSettings.ModemConfig.Bw31_25Cr48Sf512) + setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw31_25Cr48Sf512) if args.ch_mediumslow: - setSimpleChannel( - channel_pb2.ChannelSettings.ModemConfig.Bw250Cr46Sf2048) + setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw250Cr46Sf2048) if args.ch_mediumfast: - setSimpleChannel( - channel_pb2.ChannelSettings.ModemConfig.Bw250Cr47Sf1024) + setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw250Cr47Sf1024) if args.ch_shortslow: - setSimpleChannel( - channel_pb2.ChannelSettings.ModemConfig.Bw125Cr45Sf128) + setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw125Cr45Sf128) if args.ch_shortfast: - setSimpleChannel( - channel_pb2.ChannelSettings.ModemConfig.Bw500Cr45Sf128) + setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw500Cr45Sf128) # Handle the channel settings for pref in (args.ch_set or []): @@ -734,7 +725,7 @@ def initParser(): "--setlon", help="Set device longitude (allows use without GPS)") parser.add_argument( - "--pos-fields", help="Specify position message fields. Use '0' for list of valid values. "\ + "--pos-fields", help="Specify fields to send when sending a position. Use no argument for a list of valid values. "\ "Can pass multiple values as a space separated list like "\ "this: '--pos-fields POS_ALTITUDE POS_ALT_MSL'", nargs="*", action="store") diff --git a/meshtastic/globals.py b/meshtastic/globals.py index b00683c..05049c1 100644 --- a/meshtastic/globals.py +++ b/meshtastic/globals.py @@ -30,6 +30,13 @@ class Globals: self.target_node = None self.channel_index = None + def reset(self): + """Reset all of our globals. If you add a member, add it to this method, too.""" + self.args = None + self.parser = None + self.target_node = None + self.channel_index = None + def set_args(self, args): """Set the args""" self.args = args diff --git a/meshtastic/tests/conftest.py b/meshtastic/tests/conftest.py new file mode 100644 index 0000000..3bcaecb --- /dev/null +++ b/meshtastic/tests/conftest.py @@ -0,0 +1,15 @@ +"""Common pytest code (place for fixtures).""" + +import argparse + +import pytest + +from meshtastic.__main__ import Globals + +@pytest.fixture +def reset_globals(): + """Fixture to reset globals.""" + parser = None + parser = argparse.ArgumentParser() + Globals.getInstance().reset() + Globals.getInstance().set_parser(parser) diff --git a/meshtastic/tests/test_main.py b/meshtastic/tests/test_main.py index 8e6c457..bd05a97 100644 --- a/meshtastic/tests/test_main.py +++ b/meshtastic/tests/test_main.py @@ -1,7 +1,6 @@ """Meshtastic unit tests for __main__.py""" import sys -import argparse import re from unittest.mock import patch, MagicMock @@ -11,17 +10,15 @@ from meshtastic.__main__ import initParser, main, Globals from ..serial_interface import SerialInterface from ..node import Node +from ..radioconfig_pb2 import RadioConfig +from ..channel_pb2 import Channel @pytest.mark.unit -def test_main_init_parser_no_args(capsys): +def test_main_init_parser_no_args(capsys, reset_globals): """Test no arguments""" sys.argv = [''] - args = sys.argv - our_globals = Globals.getInstance() - parser = argparse.ArgumentParser() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) initParser() out, err = capsys.readouterr() assert out == '' @@ -29,15 +26,11 @@ def test_main_init_parser_no_args(capsys): @pytest.mark.unit -def test_main_init_parser_version(capsys): +def test_main_init_parser_version(capsys, reset_globals): """Test --version""" sys.argv = ['', '--version'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: initParser() assert pytest_wrapped_e.type == SystemExit @@ -48,15 +41,11 @@ def test_main_init_parser_version(capsys): @pytest.mark.unit -def test_main_main_version(capsys): +def test_main_main_version(capsys, reset_globals): """Test --version""" sys.argv = ['', '--version'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -67,15 +56,11 @@ def test_main_main_version(capsys): @pytest.mark.unit -def test_main_main_no_args(): +def test_main_main_no_args(reset_globals): """Test with no args""" sys.argv = [''] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -83,15 +68,11 @@ def test_main_main_no_args(): @pytest.mark.unit -def test_main_support(capsys): +def test_main_support(capsys, reset_globals): """Test --support""" sys.argv = ['', '--support'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -106,20 +87,14 @@ def test_main_support(capsys): @pytest.mark.unit @patch('meshtastic.util.findPorts', return_value=[]) -def test_main_ch_index_no_devices(patched_find_ports, capsys): +def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals): """Test --ch-index 1""" sys.argv = ['', '--ch-index', '1'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) - assert our_globals.get_target_node() is None - assert our_globals.get_channel_index() is None + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: main() - assert our_globals.get_channel_index() == 1 + assert Globals.getInstance().get_channel_index() == 1 assert pytest_wrapped_e.type == SystemExit assert pytest_wrapped_e.value.code == 1 out, err = capsys.readouterr() @@ -130,16 +105,12 @@ def test_main_ch_index_no_devices(patched_find_ports, capsys): @pytest.mark.unit @patch('meshtastic.util.findPorts', return_value=[]) -def test_main_test_no_ports(patched_find_ports): +def test_main_test_no_ports(patched_find_ports, reset_globals): """Test --test with no hardware""" sys.argv = ['', '--test'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) - assert our_globals.get_target_node() is None + Globals.getInstance().set_args(sys.argv) + + assert Globals.getInstance().get_target_node() is None with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -149,16 +120,12 @@ def test_main_test_no_ports(patched_find_ports): @pytest.mark.unit @patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1']) -def test_main_test_one_port(patched_find_ports): +def test_main_test_one_port(patched_find_ports, reset_globals): """Test --test with one fake port""" sys.argv = ['', '--test'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) - assert our_globals.get_target_node() is None + Globals.getInstance().set_args(sys.argv) + + assert Globals.getInstance().get_target_node() is None with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -169,15 +136,11 @@ def test_main_test_one_port(patched_find_ports): @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): +def test_main_test_two_ports_success(patched_find_ports, patched_test_all, reset_globals): """Test --test two fake ports and testAll() is a simulated success""" sys.argv = ['', '--test'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -189,15 +152,11 @@ def test_main_test_two_ports_success(patched_find_ports, patched_test_all): @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): +def test_main_test_two_ports_fails(patched_find_ports, patched_test_all, reset_globals): """Test --test two fake ports and testAll() is a simulated failure""" sys.argv = ['', '--test'] - args = sys.argv - parser = None - parser = argparse.ArgumentParser() - our_globals = Globals.getInstance() - our_globals.set_parser(parser) - our_globals.set_args(args) + Globals.getInstance().set_args(sys.argv) + with pytest.raises(SystemExit) as pytest_wrapped_e: main() assert pytest_wrapped_e.type == SystemExit @@ -207,15 +166,11 @@ def test_main_test_two_ports_fails(patched_find_ports, patched_test_all): @pytest.mark.unit -def test_main_info(capsys): +def test_main_info(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) + iface = MagicMock(autospec=SerialInterface) def mock_showInfo(): print('inside mocked showInfo') @@ -232,15 +187,11 @@ def test_main_info(capsys): @pytest.mark.unit -def test_main_qr(capsys): +def test_main_qr(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) + iface = MagicMock(autospec=SerialInterface) # TODO: could mock/check url with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: @@ -257,15 +208,11 @@ def test_main_qr(capsys): @pytest.mark.unit -def test_main_nodes(capsys): +def test_main_nodes(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) + iface = MagicMock(autospec=SerialInterface) def mock_showNodes(): print('inside mocked showNodes') @@ -282,15 +229,11 @@ def test_main_nodes(capsys): @pytest.mark.unit -def test_main_set_owner_to_bob(capsys): +def test_main_set_owner_to_bob(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) + iface = MagicMock(autospec=SerialInterface) with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: main() @@ -304,16 +247,10 @@ def test_main_set_owner_to_bob(capsys): @pytest.mark.unit -def test_main_set_ham_to_KI123(capsys): +def test_main_set_ham_to_KI123(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) mocked_node = MagicMock(autospec=Node) def mock_turnOffEncryptionOnPrimaryChannel(): @@ -340,16 +277,10 @@ def test_main_set_ham_to_KI123(capsys): @pytest.mark.unit -def test_main_reboot(capsys): +def test_main_reboot(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) mocked_node = MagicMock(autospec=Node) def mock_reboot(): @@ -371,16 +302,10 @@ def test_main_reboot(capsys): @pytest.mark.unit -def test_main_sendtext(capsys): +def test_main_sendtext(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) iface = MagicMock(autospec=SerialInterface) def mock_sendText(text, dest, wantAck): @@ -400,16 +325,10 @@ def test_main_sendtext(capsys): @pytest.mark.unit -def test_main_sendping(capsys): +def test_main_sendping(capsys, reset_globals): """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) + Globals.getInstance().set_args(sys.argv) iface = MagicMock(autospec=SerialInterface) def mock_sendData(payload, dest, portNum, wantAck, wantResponse): @@ -426,3 +345,591 @@ def test_main_sendping(capsys): assert re.search(r'inside mocked sendData', out, re.MULTILINE) assert err == '' mo.assert_called() + + +@pytest.mark.unit +def test_main_setlat(capsys, reset_globals): + """Test --sendlat""" + sys.argv = ['', '--setlat', '37.5'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + def mock_writeConfig(): + print('inside mocked writeConfig') + mocked_node.writeConfig.side_effect = mock_writeConfig + + iface = MagicMock(autospec=SerialInterface) + def mock_sendPosition(lat, lon, alt): + print('inside mocked sendPosition') + iface.sendPosition.side_effect = mock_sendPosition + iface.localNode.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'Fixing latitude', out, re.MULTILINE) + assert re.search(r'Setting device position', out, re.MULTILINE) + assert re.search(r'inside mocked sendPosition', out, re.MULTILINE) + # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_setlon(capsys, reset_globals): + """Test --setlon""" + sys.argv = ['', '--setlon', '-122.1'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + def mock_writeConfig(): + print('inside mocked writeConfig') + mocked_node.writeConfig.side_effect = mock_writeConfig + + iface = MagicMock(autospec=SerialInterface) + def mock_sendPosition(lat, lon, alt): + print('inside mocked sendPosition') + iface.sendPosition.side_effect = mock_sendPosition + iface.localNode.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'Fixing longitude', out, re.MULTILINE) + assert re.search(r'Setting device position', out, re.MULTILINE) + assert re.search(r'inside mocked sendPosition', out, re.MULTILINE) + # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_setalt(capsys, reset_globals): + """Test --setalt""" + sys.argv = ['', '--setalt', '51'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + def mock_writeConfig(): + print('inside mocked writeConfig') + mocked_node.writeConfig.side_effect = mock_writeConfig + + iface = MagicMock(autospec=SerialInterface) + def mock_sendPosition(lat, lon, alt): + print('inside mocked sendPosition') + iface.sendPosition.side_effect = mock_sendPosition + iface.localNode.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'Fixing altitude', out, re.MULTILINE) + assert re.search(r'Setting device position', out, re.MULTILINE) + assert re.search(r'inside mocked sendPosition', out, re.MULTILINE) + # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_set_team_valid(capsys, reset_globals): + """Test --set-team""" + sys.argv = ['', '--set-team', 'CYAN'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + def mock_setOwner(team): + print('inside mocked setOwner') + mocked_node.setOwner.side_effect = mock_setOwner + + iface = MagicMock(autospec=SerialInterface) + iface.localNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with patch('meshtastic.mesh_pb2.Team') as mm: + mm.Name.return_value = 'FAKENAME' + mm.Value.return_value = 'FAKEVAL' + 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 team to', out, re.MULTILINE) + assert err == '' + mo.assert_called() + mm.Name.assert_called() + mm.Value.assert_called() + + +@pytest.mark.unit +def test_main_set_team_invalid(capsys, reset_globals): + """Test --set-team using an invalid team name""" + sys.argv = ['', '--set-team', 'NOTCYAN'] + Globals.getInstance().set_args(sys.argv) + + iface = MagicMock(autospec=SerialInterface) + + def throw_an_exception(exc): + raise ValueError("Fake exception.") + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with patch('meshtastic.mesh_pb2.Team') as mm: + mm.Value.side_effect = throw_an_exception + 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'ERROR: Team', out, re.MULTILINE) + assert err == '' + mo.assert_called() + mm.Value.assert_called() + + +@pytest.mark.unit +def test_main_seturl(capsys, reset_globals): + """Test --seturl (url used below is what is generated after a factory_reset)""" + sys.argv = ['', '--seturl', 'https://www.meshtastic.org/d/#CgUYAyIBAQ'] + Globals.getInstance().set_args(sys.argv) + + 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 err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_set_valid(capsys, reset_globals): + """Test --set with valid field""" + sys.argv = ['', '--set', 'wifi_ssid', 'foo'] + Globals.getInstance().set_args(sys.argv) + + mocked_user_prefs = MagicMock(autospec=RadioConfig.UserPreferences) + mocked_user_prefs.phone_timeout_secs.return_value = 900 + mocked_user_prefs.ls_secs.return_value = 300 + + mocked_node = MagicMock(autospec=Node) + mocked_node.radioConfig.preferences = ( mocked_user_prefs ) + + 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'Set wifi_ssid to foo', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_set_with_invalid(capsys, reset_globals): + """Test --set with invalid field""" + sys.argv = ['', '--set', 'foo', 'foo'] + Globals.getInstance().set_args(sys.argv) + + mocked_user_prefs = MagicMock() + mocked_user_prefs.DESCRIPTOR.fields_by_name.get.return_value = None + + mocked_node = MagicMock(autospec=Node) + mocked_node.radioConfig.preferences = ( mocked_user_prefs ) + + 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'does not have an attribute called foo', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_configure(capsys, reset_globals): + """Test --configure with valid file""" + sys.argv = ['', '--configure', 'example_config.yaml'] + Globals.getInstance().set_args(sys.argv) + + mocked_user_prefs = MagicMock(autospec=RadioConfig.UserPreferences) + mocked_user_prefs.phone_timeout_secs.return_value = 900 + mocked_user_prefs.ls_secs.return_value = 300 + + mocked_node = MagicMock(autospec=Node) + mocked_node.radioConfig.preferences = ( mocked_user_prefs ) + + 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 device owner', out, re.MULTILINE) + assert re.search(r'Setting channel url', out, re.MULTILINE) + assert re.search(r'Fixing altitude', out, re.MULTILINE) + assert re.search(r'Fixing latitude', out, re.MULTILINE) + assert re.search(r'Fixing longitude', out, re.MULTILINE) + assert re.search(r'Writing modified preferences', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_add_valid(capsys, reset_globals): + """Test --ch-add with valid channel name, and that channel name does not already exist""" + sys.argv = ['', '--ch-add', 'testing'] + Globals.getInstance().set_args(sys.argv) + + mocked_channel = MagicMock(autospec=Channel) + # TODO: figure out how to get it to print the channel name instead of MagicMock + + mocked_node = MagicMock(autospec=Node) + # set it up so we do not already have a channel named this + mocked_node.getChannelByName.return_value = False + # set it up so we have free channels + mocked_node.getDisabledChannel.return_value = mocked_channel + + 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'Writing modified channels to device', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_add_invalid_name_too_long(capsys, reset_globals): + """Test --ch-add with invalid channel name, name too long""" + sys.argv = ['', '--ch-add', 'testingtestingtesting'] + Globals.getInstance().set_args(sys.argv) + + mocked_channel = MagicMock(autospec=Channel) + # TODO: figure out how to get it to print the channel name instead of MagicMock + + mocked_node = MagicMock(autospec=Node) + # set it up so we do not already have a channel named this + mocked_node.getChannelByName.return_value = False + # set it up so we have free channels + mocked_node.getDisabledChannel.return_value = mocked_channel + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: Channel name must be shorter', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_add_but_name_already_exists(capsys, reset_globals): + """Test --ch-add with a channel name that already exists""" + sys.argv = ['', '--ch-add', 'testing'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + # set it up so we do not already have a channel named this + mocked_node.getChannelByName.return_value = True + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: This node already has', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_add_but_no_more_channels(capsys, reset_globals): + """Test --ch-add with but there are no more channels""" + sys.argv = ['', '--ch-add', 'testing'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + # set it up so we do not already have a channel named this + mocked_node.getChannelByName.return_value = False + # set it up so we have free channels + mocked_node.getDisabledChannel.return_value = None + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: No free channels were found', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_del(capsys, reset_globals): + """Test --ch-del with valid secondary channel to be deleted""" + sys.argv = ['', '--ch-del', '--ch-index', '1'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + 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'Deleting channel', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_del_no_ch_index_specified(capsys, reset_globals): + """Test --ch-del without a valid ch-index""" + sys.argv = ['', '--ch-del'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: Need to specify', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_del_primary_channel(capsys, reset_globals): + """Test --ch-del on ch-index=0""" + sys.argv = ['', '--ch-del', '--ch-index', '0'] + Globals.getInstance().set_args(sys.argv) + Globals.getInstance().set_channel_index(1) + + mocked_node = MagicMock(autospec=Node) + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: Cannot delete primary channel', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals): + """Test --ch-enable with --ch-index""" + sys.argv = ['', '--ch-enable', '--ch-index', '1'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + 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'Writing modified channels', out, re.MULTILINE) + assert err == '' + assert Globals.getInstance().get_channel_index() == 1 + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals): + """Test --ch-disable with --ch-index""" + sys.argv = ['', '--ch-disable', '--ch-index', '1'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + 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'Writing modified channels', out, re.MULTILINE) + assert err == '' + assert Globals.getInstance().get_channel_index() == 1 + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_enable_without_a_ch_index(capsys, reset_globals): + """Test --ch-enable without --ch-index""" + sys.argv = ['', '--ch-enable'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: Need to specify', out, re.MULTILINE) + assert err == '' + assert Globals.getInstance().get_channel_index() is None + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_enable_primary_channel(capsys, reset_globals): + """Test --ch-enable with --ch-index = 0""" + sys.argv = ['', '--ch-enable', '--ch-index', '0'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: Cannot enable/disable PRIMARY', out, re.MULTILINE) + assert err == '' + assert Globals.getInstance().get_channel_index() == 0 + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_range_options(capsys, reset_globals): + """Test changing the various range options.""" + range_options = ['--ch-longslow', '--ch-longfast', '--ch-mediumslow', + '--ch-mediumfast', '--ch-shortslow', '--ch-shortfast'] + for range_option in range_options: + sys.argv = ['', f"{range_option}" ] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + 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'Writing modified channels', out, re.MULTILINE) + assert err == '' + mo.assert_called() + + +@pytest.mark.unit +def test_main_ch_longsfast_on_non_primary_channel(capsys, reset_globals): + """Test --ch-longfast --ch-index 1""" + sys.argv = ['', '--ch-longfast', '--ch-index', '1'] + Globals.getInstance().set_args(sys.argv) + + mocked_node = MagicMock(autospec=Node) + + iface = MagicMock(autospec=SerialInterface) + iface.getNode.return_value = mocked_node + + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: + with pytest.raises(SystemExit) as pytest_wrapped_e: + main() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + print('out:', out) + print('err:', err) + assert re.search(r'Connected to radio', out, re.MULTILINE) + assert re.search(r'Warning: Standard channel settings', out, re.MULTILINE) + assert err == '' + mo.assert_called()