diff --git a/README.md b/README.md index 8b80dbb..ac8f760 100644 --- a/README.md +++ b/README.md @@ -244,4 +244,6 @@ pytest -m smokewifi meshtastic/test/test_smoke_wifi.py::test_smokewifi_info ``` pytest --cov=meshtastic +# or if want html coverage report +pytest --cov-report html --cov=meshtastic ``` diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 3460574..24217bb 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -7,7 +7,6 @@ import platform import logging import sys import time -import os import yaml from pubsub import pub import pyqrcode @@ -18,22 +17,19 @@ from .ble_interface import BLEInterface from . import test, remote_hardware from . import portnums_pb2, channel_pb2, mesh_pb2, radioconfig_pb2 from . import tunnel -from .util import support_info, our_exit +from .util import support_info, our_exit, genPSK256, fromPSK, fromStr +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' -"""The command line arguments""" -args = None - -"""The parser for arguments""" -parser = argparse.ArgumentParser() - channelIndex = 0 def onReceive(packet, interface): """Callback invoked when a packet arrives""" + our_globals = Globals.getInstance() + args = our_globals.get_args() try: d = packet.get('decoded') @@ -63,61 +59,6 @@ def onConnection(interface, topic=pub.AUTO_TOPIC): print(f"Connection changed: {topic.getName()}") -trueTerms = {"t", "true", "yes"} -falseTerms = {"f", "false", "no"} - - -def genPSK256(): - """Generate a random preshared key""" - return os.urandom(32) - - -def fromPSK(valstr): - """A special version of fromStr that assumes the user is trying to set a PSK. - In that case we also allow "none", "default" or "random" (to have python generate one), or simpleN - """ - if valstr == "random": - return genPSK256() - elif valstr == "none": - return bytes([0]) # Use the 'no encryption' PSK - elif valstr == "default": - return bytes([1]) # Use default channel psk - elif valstr.startswith("simple"): - # Use one of the single byte encodings - return bytes([int(valstr[6:]) + 1]) - else: - return fromStr(valstr) - - -def fromStr(valstr): - """try to parse as int, float or bool (and fallback to a string as last resort) - - Returns: an int, bool, float, str or byte array (for strings of hex digits) - - Args: - valstr (string): A user provided string - """ - if len(valstr) == 0: # Treat an emptystring as an empty bytes - val = bytes() - elif valstr.startswith('0x'): - # if needed convert to string with asBytes.decode('utf-8') - val = bytes.fromhex(valstr[2:]) - elif valstr in trueTerms: - val = True - elif valstr in falseTerms: - val = False - else: - try: - val = int(valstr) - except ValueError: - try: - val = float(valstr) - except ValueError: - val = valstr # Not a float or an int, assume string - - return val - - never = 0xffffffff oneday = 24 * 60 * 60 @@ -197,7 +138,8 @@ def onConnected(interface): """Callback invoked when we connect to a radio""" closeNow = False # Should we drop the connection after we finish? try: - global args + our_globals = Globals.getInstance() + args = our_globals.get_args() print("Connected to radio") @@ -564,7 +506,9 @@ def subscribe(): def common(): """Shared code for all of our command line wrappers""" - global args + our_globals = Globals.getInstance() + args = our_globals.get_args() + parser = our_globals.get_parser() logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) if len(sys.argv) == 1: @@ -641,7 +585,9 @@ def common(): def initParser(): """ Initialize the command line argument parsing.""" - global parser, args + our_globals = Globals.getInstance() + parser = our_globals.get_parser() + args = our_globals.get_args() parser.add_argument( "--configure", @@ -806,19 +752,28 @@ def initParser(): "--support", action='store_true', help="Show support info (useful when troubleshooting an issue)") args = parser.parse_args() + our_globals.set_args(args) + our_globals.set_parser(parser) def main(): """Perform command line meshtastic operations""" + our_globals = Globals.getInstance() + parser = argparse.ArgumentParser() + our_globals.set_parser(parser) initParser() common() def tunnelMain(): """Run a meshtastic IP tunnel""" - global args + our_globals = Globals.getInstance() + parser = argparse.ArgumentParser() + our_globals.set_parser(parser) initParser() + args = our_globals.get_args() args.tunnel = True + our_globals.set_args(args) common() diff --git a/meshtastic/globals.py b/meshtastic/globals.py new file mode 100644 index 0000000..c255e81 --- /dev/null +++ b/meshtastic/globals.py @@ -0,0 +1,44 @@ +"""Globals singleton class. + + Instead of using a global, stuff your variables in this "trash can". + This is not much better than using python's globals, but it allows + us to better test meshtastic. Plus, there are some weird python + global issues/gotcha that we can hopefully avoid by using this + class in stead. +""" + +class Globals: + """Globals class is a Singleton.""" + __instance = None + + @staticmethod + def getInstance(): + """Get an instance of the Globals class.""" + if Globals.__instance is None: + Globals() + return Globals.__instance + + def __init__(self): + """Constructor for the Globals CLass""" + if Globals.__instance is not None: + raise Exception("This class is a singleton") + else: + Globals.__instance = self + self.args = None + self.parser = None + + def set_args(self, args): + """Set the args""" + self.args = args + + def set_parser(self, parser): + """Set the parser""" + self.parser = parser + + def get_args(self): + """Get args""" + return self.args + + def get_parser(self): + """Get parser""" + return self.parser diff --git a/meshtastic/test/test_globals.py b/meshtastic/test/test_globals.py new file mode 100644 index 0000000..da9d08b --- /dev/null +++ b/meshtastic/test/test_globals.py @@ -0,0 +1,25 @@ +"""Meshtastic unit tests for globals.py +""" + +import pytest + +from ..globals import Globals + + +@pytest.mark.unit +def test_globals_get_instaance(): + """Test that we can instantiate a Globals instance""" + ourglobals = Globals.getInstance() + ourglobals2 = Globals.getInstance() + assert ourglobals == ourglobals2 + + +@pytest.mark.unit +def test_globals_there_can_be_only_one(): + """Test that we can cannot create two Globals instances""" + # if we have an instance, delete it + Globals.getInstance() + with pytest.raises(Exception) as pytest_wrapped_e: + # try to create another instance + Globals() + assert pytest_wrapped_e.type == Exception diff --git a/meshtastic/test/test_main.py b/meshtastic/test/test_main.py new file mode 100644 index 0000000..c745d5a --- /dev/null +++ b/meshtastic/test/test_main.py @@ -0,0 +1,43 @@ +"""Meshtastic unit tests for __main__.py""" + +import sys +import argparse +import re + +import pytest + +from meshtastic.__main__ import initParser, Globals + + +@pytest.mark.unit +def test_main_init_parser_no_args(capsys): + """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) + initParser() + out, err = capsys.readouterr() + assert out == '' + assert err == '' + + +@pytest.mark.unit +def test_main_init_parser_version(capsys): + """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) + with pytest.raises(SystemExit) as pytest_wrapped_e: + initParser() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 0 + out, err = capsys.readouterr() + assert re.match(r'[0-9]+\.[0-9]+\.[0-9]', out) + assert err == '' diff --git a/meshtastic/test/test_util.py b/meshtastic/test/test_util.py index cf5f638..20e095d 100644 --- a/meshtastic/test/test_util.py +++ b/meshtastic/test/test_util.py @@ -4,7 +4,45 @@ import re import pytest -from meshtastic.util import fixme, stripnl, pskToString, our_exit, support_info +from meshtastic.util import fixme, stripnl, pskToString, our_exit, support_info, genPSK256, fromStr, fromPSK + + +@pytest.mark.unit +def test_genPSK256(): + """Test genPSK256""" + assert genPSK256() != '' + + +@pytest.mark.unit +def test_fromStr(): + """Test fromStr""" + assert fromStr('') == b'' + assert fromStr('0x12') == b'\x12' + assert fromStr('t') + assert fromStr('T') + assert fromStr('true') + assert fromStr('True') + assert fromStr('yes') + assert fromStr('Yes') + assert fromStr('f') is False + assert fromStr('F') is False + assert fromStr('false') is False + assert fromStr('False') is False + assert fromStr('no') is False + assert fromStr('No') is False + assert fromStr('100.01') == 100.01 + assert fromStr('123') == 123 + assert fromStr('abc') == 'abc' + + +@pytest.mark.unit +def test_fromPSK(): + """Test fromPSK""" + assert fromPSK('random') != '' + assert fromPSK('none') == b'\x00' + assert fromPSK('default') == b'\x01' + assert fromPSK('simple22') == b'\x17' + assert fromPSK('trash') == 'trash' @pytest.mark.unit diff --git a/meshtastic/util.py b/meshtastic/util.py index 77a69b8..dc9d0b2 100644 --- a/meshtastic/util.py +++ b/meshtastic/util.py @@ -2,6 +2,7 @@ """ import traceback from queue import Queue +import os import sys import time import platform @@ -15,6 +16,55 @@ import pkg_resources blacklistVids = dict.fromkeys([0x1366]) +def genPSK256(): + """Generate a random preshared key""" + return os.urandom(32) + + +def fromPSK(valstr): + """A special version of fromStr that assumes the user is trying to set a PSK. + In that case we also allow "none", "default" or "random" (to have python generate one), or simpleN + """ + if valstr == "random": + return genPSK256() + elif valstr == "none": + return bytes([0]) # Use the 'no encryption' PSK + elif valstr == "default": + return bytes([1]) # Use default channel psk + elif valstr.startswith("simple"): + # Use one of the single byte encodings + return bytes([int(valstr[6:]) + 1]) + else: + return fromStr(valstr) + + +def fromStr(valstr): + """try to parse as int, float or bool (and fallback to a string as last resort) + Returns: an int, bool, float, str or byte array (for strings of hex digits) + + Args: + valstr (string): A user provided string + """ + if len(valstr) == 0: # Treat an emptystring as an empty bytes + val = bytes() + elif valstr.startswith('0x'): + # if needed convert to string with asBytes.decode('utf-8') + val = bytes.fromhex(valstr[2:]) + elif valstr.lower() in {"t", "true", "yes"}: + val = True + elif valstr.lower() in {"f", "false", "no"}: + val = False + else: + try: + val = int(valstr) + except ValueError: + try: + val = float(valstr) + except ValueError: + val = valstr # Not a float or an int, assume string + return val + + def pskToString(psk: bytes): """Given an array of PSK bytes, decode them into a human readable (but privacy protecting) string""" if len(psk) == 0: