From e7991cfc003d7350184d7fe5a6463f5c43d2f12a Mon Sep 17 00:00:00 2001 From: Mike Kinney Date: Thu, 9 Dec 2021 11:26:24 -0800 Subject: [PATCH 1/3] refactor globals to singleton --- README.md | 2 ++ meshtastic/__main__.py | 60 ++++++++++++++++++++++++++++++------ meshtastic/test/test_main.py | 55 +++++++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+), 10 deletions(-) create mode 100644 meshtastic/test/test_main.py 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..9b037d4 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -23,17 +23,43 @@ from .util import support_info, our_exit """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 +class Settings: + """Settings class is a Singleton.""" + __instance = None + @staticmethod + def getInstance(): + """Get an instance of the Settings class.""" + if Settings.__instance is None: + Settings() + return Settings.__instance + def __init__(self): + """Constructor for the Settings CLass""" + if Settings.__instance is not None: + raise Exception("This class is a singleton") + else: + Settings.__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 + def onReceive(packet, interface): """Callback invoked when a packet arrives""" + settings = Settings.getInstance() + args = settings.get_args() try: d = packet.get('decoded') @@ -197,7 +223,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 + settings = Settings.getInstance() + args = settings.get_args() print("Connected to radio") @@ -564,7 +591,9 @@ def subscribe(): def common(): """Shared code for all of our command line wrappers""" - global args + settings = Settings.getInstance() + args = settings.get_args() + parser = settings.get_parser() logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) if len(sys.argv) == 1: @@ -641,7 +670,9 @@ def common(): def initParser(): """ Initialize the command line argument parsing.""" - global parser, args + settings = Settings.getInstance() + parser = settings.get_parser() + args = settings.get_args() parser.add_argument( "--configure", @@ -806,19 +837,28 @@ def initParser(): "--support", action='store_true', help="Show support info (useful when troubleshooting an issue)") args = parser.parse_args() + settings.set_args(args) + settings.set_parser(parser) def main(): """Perform command line meshtastic operations""" + settings = Settings.getInstance() + parser = argparse.ArgumentParser() + settings.set_parser(parser) initParser() common() def tunnelMain(): """Run a meshtastic IP tunnel""" - global args + settings = Settings.getInstance() + parser = argparse.ArgumentParser() + settings.set_parser(parser) initParser() + args = settings.get_args() args.tunnel = True + settings.set_args(args) common() diff --git a/meshtastic/test/test_main.py b/meshtastic/test/test_main.py new file mode 100644 index 0000000..186e8de --- /dev/null +++ b/meshtastic/test/test_main.py @@ -0,0 +1,55 @@ +"""Meshtastic unit tests for __main__.py""" + +import sys +import argparse +import re + +import pytest + +from meshtastic.__main__ import initParser, Settings + + +"""The command line arguments""" +#args = None + +"""The parser for arguments""" +#parser = None + +#@pytest.fixture +#def patched_env(monkeypatch): +# monkeypatch.args = None +# #monkeypatch.sys.argv = [''] +# monkeypatch.parser = None + +@pytest.mark.unit +def test_main_no_args(capsys): + """Test no arguments""" + sys.argv = [''] + args = sys.argv + settings = Settings.getInstance() + parser = argparse.ArgumentParser() + settings.set_parser(parser) + settings.set_args(args) + initParser() + out, err = capsys.readouterr() + assert out == '' + assert err == '' + + +@pytest.mark.unit +def test_main_version(capsys): + """Test --version""" + sys.argv = ['', '--version'] + args = sys.argv + parser = None + parser = argparse.ArgumentParser() + settings = Settings.getInstance() + settings.set_parser(parser) + settings.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 == '' From 54fc1a92e74eb3abf4f21ede593045f3432d6ca5 Mon Sep 17 00:00:00 2001 From: Mike Kinney Date: Thu, 9 Dec 2021 11:46:33 -0800 Subject: [PATCH 2/3] refactor Settings() to Globals() --- meshtastic/__main__.py | 67 ++++++++++----------------------- meshtastic/globals.py | 44 ++++++++++++++++++++++ meshtastic/test/test_globals.py | 25 ++++++++++++ meshtastic/test/test_main.py | 26 ++++--------- 4 files changed, 95 insertions(+), 67 deletions(-) create mode 100644 meshtastic/globals.py create mode 100644 meshtastic/test/test_globals.py diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 9b037d4..5d2db44 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -19,47 +19,18 @@ 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 .globals import Globals """We only import the tunnel code if we are on a platform that can run it""" have_tunnel = platform.system() == 'Linux' channelIndex = 0 -class Settings: - """Settings class is a Singleton.""" - __instance = None - @staticmethod - def getInstance(): - """Get an instance of the Settings class.""" - if Settings.__instance is None: - Settings() - return Settings.__instance - def __init__(self): - """Constructor for the Settings CLass""" - if Settings.__instance is not None: - raise Exception("This class is a singleton") - else: - Settings.__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 - def onReceive(packet, interface): """Callback invoked when a packet arrives""" - settings = Settings.getInstance() - args = settings.get_args() + our_globals = Globals.getInstance() + args = our_globals.get_args() try: d = packet.get('decoded') @@ -223,8 +194,8 @@ def onConnected(interface): """Callback invoked when we connect to a radio""" closeNow = False # Should we drop the connection after we finish? try: - settings = Settings.getInstance() - args = settings.get_args() + our_globals = Globals.getInstance() + args = our_globals.get_args() print("Connected to radio") @@ -591,9 +562,9 @@ def subscribe(): def common(): """Shared code for all of our command line wrappers""" - settings = Settings.getInstance() - args = settings.get_args() - parser = settings.get_parser() + 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: @@ -670,9 +641,9 @@ def common(): def initParser(): """ Initialize the command line argument parsing.""" - settings = Settings.getInstance() - parser = settings.get_parser() - args = settings.get_args() + our_globals = Globals.getInstance() + parser = our_globals.get_parser() + args = our_globals.get_args() parser.add_argument( "--configure", @@ -837,28 +808,28 @@ def initParser(): "--support", action='store_true', help="Show support info (useful when troubleshooting an issue)") args = parser.parse_args() - settings.set_args(args) - settings.set_parser(parser) + our_globals.set_args(args) + our_globals.set_parser(parser) def main(): """Perform command line meshtastic operations""" - settings = Settings.getInstance() + our_globals = Globals.getInstance() parser = argparse.ArgumentParser() - settings.set_parser(parser) + our_globals.set_parser(parser) initParser() common() def tunnelMain(): """Run a meshtastic IP tunnel""" - settings = Settings.getInstance() + our_globals = Globals.getInstance() parser = argparse.ArgumentParser() - settings.set_parser(parser) + our_globals.set_parser(parser) initParser() - args = settings.get_args() + args = our_globals.get_args() args.tunnel = True - settings.set_args(args) + 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..aba911c --- /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() + 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""" + # ensure we have at least one instance created + 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 index 186e8de..bc9a3a5 100644 --- a/meshtastic/test/test_main.py +++ b/meshtastic/test/test_main.py @@ -6,30 +6,18 @@ import re import pytest -from meshtastic.__main__ import initParser, Settings +from meshtastic.__main__ import initParser, Globals -"""The command line arguments""" -#args = None - -"""The parser for arguments""" -#parser = None - -#@pytest.fixture -#def patched_env(monkeypatch): -# monkeypatch.args = None -# #monkeypatch.sys.argv = [''] -# monkeypatch.parser = None - @pytest.mark.unit def test_main_no_args(capsys): """Test no arguments""" sys.argv = [''] args = sys.argv - settings = Settings.getInstance() + our_globals = Globals.getInstance() parser = argparse.ArgumentParser() - settings.set_parser(parser) - settings.set_args(args) + our_globals.set_parser(parser) + our_globals.set_args(args) initParser() out, err = capsys.readouterr() assert out == '' @@ -43,9 +31,9 @@ def test_main_version(capsys): args = sys.argv parser = None parser = argparse.ArgumentParser() - settings = Settings.getInstance() - settings.set_parser(parser) - settings.set_args(args) + 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 From f587698d2a691031a0d0d62d4086b4f72cbe9d67 Mon Sep 17 00:00:00 2001 From: Mike Kinney Date: Thu, 9 Dec 2021 12:26:17 -0800 Subject: [PATCH 3/3] move functions from main to util and add unit tests --- meshtastic/__main__.py | 58 +-------------------------------- meshtastic/test/test_globals.py | 4 +-- meshtastic/test/test_main.py | 4 +-- meshtastic/test/test_util.py | 40 ++++++++++++++++++++++- meshtastic/util.py | 50 ++++++++++++++++++++++++++++ 5 files changed, 94 insertions(+), 62 deletions(-) diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 5d2db44..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,7 +17,7 @@ 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""" @@ -60,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 diff --git a/meshtastic/test/test_globals.py b/meshtastic/test/test_globals.py index aba911c..da9d08b 100644 --- a/meshtastic/test/test_globals.py +++ b/meshtastic/test/test_globals.py @@ -9,7 +9,7 @@ from ..globals import Globals @pytest.mark.unit def test_globals_get_instaance(): """Test that we can instantiate a Globals instance""" - ourglobals = Globals() + ourglobals = Globals.getInstance() ourglobals2 = Globals.getInstance() assert ourglobals == ourglobals2 @@ -17,7 +17,7 @@ def test_globals_get_instaance(): @pytest.mark.unit def test_globals_there_can_be_only_one(): """Test that we can cannot create two Globals instances""" - # ensure we have at least one instance created + # if we have an instance, delete it Globals.getInstance() with pytest.raises(Exception) as pytest_wrapped_e: # try to create another instance diff --git a/meshtastic/test/test_main.py b/meshtastic/test/test_main.py index bc9a3a5..c745d5a 100644 --- a/meshtastic/test/test_main.py +++ b/meshtastic/test/test_main.py @@ -10,7 +10,7 @@ from meshtastic.__main__ import initParser, Globals @pytest.mark.unit -def test_main_no_args(capsys): +def test_main_init_parser_no_args(capsys): """Test no arguments""" sys.argv = [''] args = sys.argv @@ -25,7 +25,7 @@ def test_main_no_args(capsys): @pytest.mark.unit -def test_main_version(capsys): +def test_main_init_parser_version(capsys): """Test --version""" sys.argv = ['', '--version'] args = sys.argv 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: