mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2025-12-31 11:57:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

Refactor to avoid the use of a special global object.

Browse Source 
The global object formerly used is now replaced by direct use
of the namespace opf the globals module. This eliminates the
redundant getters and setters and simplifies the code for
future maintainers.

Note that the globals module name conflicts (harmlessly at
present) with a Python built-in function. A future commit
should rename it `config` to remove this clash and better
represent its intended purpose.
This commit is contained in:
Steve Holden
2024-04-10 09:56:38 +01:00
parent b280d0ba23
commit 067cddd354
8 changed files with 168 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
meshtastic/__main__.py
View File

@@ -16,16 +16,14 @@ from pubsub import pub # type: ignore[import-untyped]
import meshtastic.test
import meshtastic.util
from meshtastic import globals
from meshtastic import channel_pb2, config_pb2, portnums_pb2, remote_hardware, BROADCAST_ADDR
from meshtastic.version import get_active_version
from meshtastic.ble_interface import BLEInterface
from meshtastic.globals import Globals
def onReceive(packet, interface):
"""Callback invoked when a packet arrives"""
our_globals = Globals.getInstance()
args = our_globals.get_args()
args = globals.args
try:
d = packet.get("decoded")
logging.debug(f"in onReceive() d:{d}")
@@ -69,7 +67,7 @@ def getPref(node, comp_name):
# Note: protobufs has the keys in snake_case, so snake internally
snake_name = meshtastic.util.camel_to_snake(name[1])
logging.debug(f"snake_name:{snake_name} camel_name:{camel_name}")
logging.debug(f"use camel:{Globals.getInstance().get_camel_case()}")
logging.debug(f"use camel:{globals.camel_case}")
# First validate the input
localConfig = node.localConfig
@@ -86,7 +84,7 @@ def getPref(node, comp_name):
break
if not found:
if Globals.getInstance().get_camel_case():
if globals.camel_case:
print(
f"{localConfig.__class__.__name__} and {moduleConfig.__class__.__name__} do not have an attribute {snake_name}."
)
@@ -105,7 +103,7 @@ def getPref(node, comp_name):
config_values = getattr(config, config_type.name)
if not wholeField:
pref_value = getattr(config_values, pref.name)
if Globals.getInstance().get_camel_case():
if globals.camel_case:
print(f"{str(config_type.name)}.{camel_name}: {str(pref_value)}")
logging.debug(
f"{str(config_type.name)}.{camel_name}: {str(pref_value)}"
@@ -171,7 +169,7 @@ def setPref(config, comp_name, valStr) -> bool:
if e:
val = e.number
else:
if Globals.getInstance().get_camel_case():
if globals.camel_case:
print(
f"{name[0]}.{camel_name} does not have an enum called {val}, so you can not set it."
)
@@ -210,7 +208,7 @@ def setPref(config, comp_name, valStr) -> bool:
config_type.message_type.ignore_incoming.extend([val])
prefix = f"{name[0]}." if config_type.message_type is not None else ""
if Globals.getInstance().get_camel_case():
if globals.camel_case:
print(f"Set {prefix}{camel_name} to {valStr}")
else:
print(f"Set {prefix}{snake_name} to {valStr}")
@@ -225,8 +223,7 @@ def onConnected(interface):
False # Should we wait for an acknowledgment if we send to a remote node?
)
try:
our_globals = Globals.getInstance()
args = our_globals.get_args()
args = globals.args
# do not print this line if we are exporting the config
if not args.export_config:
@@ -477,7 +474,7 @@ def onConnected(interface):
print("Writing modified preferences to device")
node.writeConfig(field)
else:
if Globals.getInstance().get_camel_case():
if globals.camel_case:
print(
f"{node.localConfig.__class__.__name__} and {node.moduleConfig.__class__.__name__} do not have an attribute {pref[0]}."
)
@@ -590,7 +587,7 @@ def onConnected(interface):
# handle changing channels
if args.ch_add:
channelIndex = our_globals.get_channel_index()
channelIndex = globals.channel_index
if channelIndex is not None:
# Since we set the channel index after adding a channel, don't allow --ch-index
meshtastic.util.our_exit(
@@ -621,12 +618,12 @@ def onConnected(interface):
n.writeChannel(ch.index)
if channelIndex is None:
print(f"Setting newly-added channel's {ch.index} as '--ch-index' for further modifications")
our_globals.set_channel_index(ch.index)
globals.channel_index = ch.index
if args.ch_del:
closeNow = True
channelIndex = our_globals.get_channel_index()
channelIndex = globals.channel_index
if channelIndex is None:
meshtastic.util.our_exit(
"Warning: Need to specify '--ch-index' for '--ch-del'.", 1
@@ -642,7 +639,7 @@ def onConnected(interface):
def setSimpleConfig(modem_preset):
"""Set one of the simple modem_config"""
channelIndex = our_globals.get_channel_index()
channelIndex = globals.channel_index
if channelIndex is not None and channelIndex > 0:
meshtastic.util.our_exit(
"Warning: Cannot set modem preset for non-primary channel", 1
@@ -677,7 +674,7 @@ def onConnected(interface):
if args.ch_set or args.ch_enable or args.ch_disable:
closeNow = True
channelIndex = our_globals.get_channel_index()
channelIndex = globals.channel_index
if channelIndex is None:
meshtastic.util.our_exit("Warning: Need to specify '--ch-index'.", 1)
ch = interface.getNode(args.dest).channels[channelIndex]
@@ -832,7 +829,7 @@ def printConfig(config):
names = []
for field in config.message_type.fields:
tmp_name = f"{config_section.name}.{field.name}"
if Globals.getInstance().get_camel_case():
if globals.camel_case:
tmp_name = meshtastic.util.snake_to_camel(tmp_name)
names.append(tmp_name)
for temp_name in sorted(names):
@@ -877,7 +874,7 @@ def export_config(interface):
if owner_short:
configObj["owner_short"] = owner_short
if channel_url:
if Globals.getInstance().get_camel_case():
if globals.camel_case:
configObj["channelUrl"] = channel_url
else:
configObj["channel_url"] = channel_url
@@ -889,11 +886,11 @@ def export_config(interface):
# Convert inner keys to correct snake/camelCase
prefs = {}
for pref in config:
if Globals.getInstance().get_camel_case():
if globals.camel_case:
prefs[meshtastic.util.snake_to_camel(pref)] = config[pref]
else:
prefs[pref] = config[pref]
if Globals.getInstance().get_camel_case():
if globals.camel_case:
configObj["config"] = config
else:
configObj["config"] = config
@@ -905,7 +902,7 @@ def export_config(interface):
for pref in module_config:
if len(module_config[pref]) > 0:
prefs[pref] = module_config[pref]
if Globals.getInstance().get_camel_case():
if globals.camel_case:
configObj["module_config"] = prefs
else:
configObj["module_config"] = prefs
@@ -919,9 +916,8 @@ def export_config(interface):
def common():
"""Shared code for all of our command line wrappers"""
logfile = None
our_globals = Globals.getInstance()
args = our_globals.get_args()
parser = our_globals.get_parser()
args = globals.args
parser = globals.parser
logging.basicConfig(
level=logging.DEBUG if (args.debug or args.listen) else logging.INFO,
format="%(levelname)s file:%(filename)s %(funcName)s line:%(lineno)s %(message)s",
@@ -937,7 +933,7 @@ def common():
if args.ch_index is not None:
channelIndex = int(args.ch_index)
our_globals.set_channel_index(channelIndex)
globals.channel_index = channelIndex
if not args.dest:
args.dest = BROADCAST_ADDR
@@ -972,7 +968,7 @@ def common():
# Note: using "line buffering"
# pylint: disable=R1732
logfile = open(args.seriallog, "w+", buffering=1, encoding="utf8")
our_globals.set_logfile(logfile)
globals.logfile = logfile
subscribe()
if args.ble_scan:
@@ -1063,9 +1059,8 @@ def addConnectionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParse
def initParser():
"""Initialize the command line argument parsing."""
our_globals = Globals.getInstance()
parser = our_globals.get_parser()
args = our_globals.get_args()
parser = globals.parser
args = globals.args
# The "Help" group includes the help option and other informational stuff about the CLI itself
outerHelpGroup = parser.add_argument_group('Help')
@@ -1286,7 +1281,7 @@ def initParser():
)
group.add_argument(
"--request-telemetry",
"--request-telemetry",
help="Request telemetry from a node. "
"You need pass the destination ID as argument with '--dest'. "
"For repeaters, the nodeNum is required.",
@@ -1431,34 +1426,32 @@ def initParser():
args = parser.parse_args()
our_globals.set_args(args)
our_globals.set_parser(parser)
globals.args = args
globals.parser = parser
def main():
"""Perform command line meshtastic operations"""
our_globals = Globals.getInstance()
parser = argparse.ArgumentParser(
add_help=False,
epilog="If no connection arguments are specified, we search for a compatible serial device, "
"and if none is found, then attempt a TCP connection to localhost.")
our_globals.set_parser(parser)
globals.parser = parser
initParser()
common()
logfile = our_globals.get_logfile()
logfile = globals.logfile
if logfile:
logfile.close()
def tunnelMain():
"""Run a meshtastic IP tunnel"""
our_globals = Globals.getInstance()
parser = argparse.ArgumentParser(add_help=False)
our_globals.set_parser(parser)
globals.parser = parser
initParser()
args = our_globals.get_args()
args = globals.args
args.tunnel = True
our_globals.set_args(args)
globals.args = args
common()

114
meshtastic/globals.py
View File

@@ -1,96 +1,28 @@
"""Globals singleton class.
"""
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 instead.
The Global object is gone, as are all its setters and getters. Instead the
module itself is the singleton namespace, which can be imported into
whichever module is used. The associated tests have also been removed,
since we now rely on built in Python mechanisms.
This is intended to make the Python read more naturally, and to make the
intention of the code clearer and more compact. It is merely a sticking
plaster over the use of shared globals, but the coupling issues wil be dealt
with rather more easily once the code is simplified by this change.
"""
def reset():
"""
"""
global args, parser, channel_index, logfile, tunnelInstance, camel_case
args = None
parser = None
channel_index = None
logfile = None
tunnelInstance = None
# TODO: to migrate to camel_case for v1.3 change this value to True
camel_case = False
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") # pylint: disable=W0719
else:
Globals.__instance = self
self.args = None
self.parser = None
self.channel_index = None
self.logfile = None
self.tunnelInstance = None
# TODO: to migrate to camel_case for v1.3 change this value to True
self.camel_case = False
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.channel_index = None
self.logfile = None
self.tunnelInstance = None
# TODO: to migrate to camel_case for v1.3 change this value to True
self.camel_case = False
# setters
def set_args(self, args):
"""Set the args"""
self.args = args
def set_parser(self, parser):
"""Set the parser"""
self.parser = parser
def set_channel_index(self, channel_index):
"""Set the channel_index"""
self.channel_index = channel_index
def set_logfile(self, logfile):
"""Set the logfile"""
self.logfile = logfile
def set_tunnelInstance(self, tunnelInstance):
"""Set the tunnelInstance"""
self.tunnelInstance = tunnelInstance
def set_camel_case(self):
"""Force using camelCase for things like prefs/set/set"""
self.camel_case = True
# getters
def get_args(self):
"""Get args"""
return self.args
def get_parser(self):
"""Get parser"""
return self.parser
def get_channel_index(self):
"""Get channel_index"""
return self.channel_index
def get_logfile(self):
"""Get logfile"""
return self.logfile
def get_tunnelInstance(self):
"""Get tunnelInstance"""
return self.tunnelInstance
def get_camel_case(self):
"""Get whether or not to use camelCase"""
return self.camel_case
reset()

6
meshtastic/tests/conftest.py
View File

@@ -5,7 +5,7 @@ from unittest.mock import MagicMock
import pytest
from meshtastic.__main__ import Globals
from meshtastic import globals
from ..mesh_interface import MeshInterface
@@ -15,8 +15,8 @@ def reset_globals():
"""Fixture to reset globals."""
parser = None
parser = argparse.ArgumentParser(add_help=False)
Globals.getInstance().reset()
Globals.getInstance().set_parser(parser)
globals.reset()
globals.parser = parser
@pytest.fixture

25
meshtastic/tests/test_globals.py
View File

@@ -1,25 +0,0 @@
"""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

9
meshtastic/tests/test_init.py
View File

@@ -6,9 +6,8 @@ from unittest.mock import MagicMock
import pytest
from meshtastic import _onNodeInfoReceive, _onPositionReceive, _onTextReceive
from meshtastic import _onNodeInfoReceive, _onPositionReceive, _onTextReceive, globals
from ..globals import Globals
from ..serial_interface import SerialInterface
@@ -16,7 +15,7 @@ from ..serial_interface import SerialInterface
def test_init_onTextReceive_with_exception(caplog):
"""Test _onTextReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
globals.args = args
iface = MagicMock(autospec=SerialInterface)
packet = {}
with caplog.at_level(logging.DEBUG):
@@ -29,7 +28,7 @@ def test_init_onTextReceive_with_exception(caplog):
def test_init_onPositionReceive(caplog):
"""Test _onPositionReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
globals.args = args
iface = MagicMock(autospec=SerialInterface)
packet = {"from": "foo", "decoded": {"position": {}}}
with caplog.at_level(logging.DEBUG):
@@ -41,7 +40,7 @@ def test_init_onPositionReceive(caplog):
def test_init_onNodeInfoReceive(caplog, iface_with_nodes):
"""Test _onNodeInfoReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
globals.args = args
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
packet = {

192
meshtastic/tests/test_main.py
View File

@@ -11,7 +11,6 @@ from unittest.mock import mock_open, MagicMock, patch
import pytest
from meshtastic.__main__ import (
Globals,
export_config,
initParser,
main,
@@ -20,6 +19,7 @@ from meshtastic.__main__ import (
onReceive,
tunnelMain,
)
from meshtastic import globals
from ..channel_pb2 import Channel # pylint: disable=E0611
@@ -40,7 +40,7 @@ from ..tcp_interface import TCPInterface
def test_main_init_parser_no_args(capsys):
"""Test no arguments"""
sys.argv = [""]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
initParser()
out, err = capsys.readouterr()
assert out == ""
@@ -52,7 +52,7 @@ def test_main_init_parser_no_args(capsys):
def test_main_init_parser_version(capsys):
"""Test --version"""
sys.argv = ["", "--version"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
initParser()
@@ -68,7 +68,7 @@ def test_main_init_parser_version(capsys):
def test_main_main_version(capsys):
"""Test --version"""
sys.argv = ["", "--version"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -84,7 +84,7 @@ def test_main_main_version(capsys):
def test_main_main_no_args(capsys):
"""Test with no args"""
sys.argv = [""]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -99,7 +99,7 @@ def test_main_main_no_args(capsys):
def test_main_support(capsys):
"""Test --support"""
sys.argv = ["", "--support"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -119,11 +119,11 @@ def test_main_support(capsys):
def test_main_ch_index_no_devices(patched_find_ports, capsys):
"""Test --ch-index 1"""
sys.argv = ["", "--ch-index", "1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
assert Globals.getInstance().get_channel_index() == 1
assert globals.channel_index == 1
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
out, err = capsys.readouterr()
@@ -138,7 +138,7 @@ def test_main_ch_index_no_devices(patched_find_ports, capsys):
def test_main_test_no_ports(patched_find_ports, capsys):
"""Test --test with no hardware"""
sys.argv = ["", "--test"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -158,7 +158,7 @@ def test_main_test_no_ports(patched_find_ports, capsys):
def test_main_test_one_port(patched_find_ports, capsys):
"""Test --test with one fake port"""
sys.argv = ["", "--test"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -178,7 +178,7 @@ def test_main_test_one_port(patched_find_ports, capsys):
def test_main_test_two_ports_success(patched_test_all, capsys):
"""Test --test two fake ports and testAll() is a simulated success"""
sys.argv = ["", "--test"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -196,7 +196,7 @@ def test_main_test_two_ports_success(patched_test_all, capsys):
def test_main_test_two_ports_fails(patched_test_all, capsys):
"""Test --test two fake ports and testAll() is a simulated failure"""
sys.argv = ["", "--test"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
main()
@@ -213,7 +213,7 @@ def test_main_test_two_ports_fails(patched_test_all, capsys):
def test_main_info(capsys, caplog):
"""Test --info"""
sys.argv = ["", "--info"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -239,7 +239,7 @@ def test_main_info(capsys, caplog):
def test_main_info_with_permission_error(patched_getlogin, capsys, caplog):
"""Test --info"""
sys.argv = ["", "--info"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
patched_getlogin.return_value = "me"
@@ -264,7 +264,7 @@ def test_main_info_with_permission_error(patched_getlogin, capsys, caplog):
def test_main_info_with_tcp_interface(capsys):
"""Test --info"""
sys.argv = ["", "--info", "--host", "meshtastic.local"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=TCPInterface)
@@ -286,7 +286,7 @@ def test_main_info_with_tcp_interface(capsys):
def test_main_no_proto(capsys):
"""Test --noproto (using --info for output)"""
sys.argv = ["", "--info", "--noproto"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -317,7 +317,7 @@ def test_main_no_proto(capsys):
def test_main_info_with_seriallog_stdout(capsys):
"""Test --info"""
sys.argv = ["", "--info", "--seriallog", "stdout"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -339,7 +339,7 @@ def test_main_info_with_seriallog_stdout(capsys):
def test_main_info_with_seriallog_output_txt(capsys):
"""Test --info"""
sys.argv = ["", "--info", "--seriallog", "output.txt"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -363,7 +363,7 @@ def test_main_info_with_seriallog_output_txt(capsys):
def test_main_qr(capsys):
"""Test --qr"""
sys.argv = ["", "--qr"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
# TODO: could mock/check url
@@ -383,7 +383,7 @@ def test_main_qr(capsys):
def test_main_onConnected_exception(capsys):
"""Test the exception in onConnected"""
sys.argv = ["", "--qr"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
def throw_an_exception(junk):
raise Exception("Fake exception.") # pylint: disable=W0719
@@ -404,7 +404,7 @@ def test_main_onConnected_exception(capsys):
def test_main_nodes(capsys):
"""Test --nodes"""
sys.argv = ["", "--nodes"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -426,7 +426,7 @@ def test_main_nodes(capsys):
def test_main_set_owner_to_bob(capsys):
"""Test --set-owner bob"""
sys.argv = ["", "--set-owner", "bob"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface) as mo:
@@ -443,7 +443,7 @@ def test_main_set_owner_to_bob(capsys):
def test_main_set_owner_short_to_bob(capsys):
"""Test --set-owner-short bob"""
sys.argv = ["", "--set-owner-short", "bob"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface) as mo:
@@ -460,7 +460,7 @@ def test_main_set_owner_short_to_bob(capsys):
def test_main_set_canned_messages(capsys):
"""Test --set-canned-message"""
sys.argv = ["", "--set-canned-message", "foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface) as mo:
@@ -477,7 +477,7 @@ def test_main_set_canned_messages(capsys):
def test_main_get_canned_messages(capsys, caplog, iface_with_nodes):
"""Test --get-canned-message"""
sys.argv = ["", "--get-canned-message"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = iface_with_nodes
iface.localNode.cannedPluginMessage = "foo"
@@ -500,7 +500,7 @@ def test_main_get_canned_messages(capsys, caplog, iface_with_nodes):
def test_main_set_ham_to_KI123(capsys):
"""Test --set-ham KI123"""
sys.argv = ["", "--set-ham", "KI123"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -536,7 +536,7 @@ def test_main_set_ham_to_KI123(capsys):
def test_main_reboot(capsys):
"""Test --reboot"""
sys.argv = ["", "--reboot"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -562,7 +562,7 @@ def test_main_reboot(capsys):
def test_main_shutdown(capsys):
"""Test --shutdown"""
sys.argv = ["", "--shutdown"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -588,7 +588,7 @@ def test_main_shutdown(capsys):
def test_main_sendtext(capsys):
"""Test --sendtext"""
sys.argv = ["", "--sendtext", "hello"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -615,7 +615,7 @@ def test_main_sendtext(capsys):
def test_main_sendtext_with_channel(capsys):
"""Test --sendtext"""
sys.argv = ["", "--sendtext", "hello", "--ch-index", "1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
@@ -643,7 +643,7 @@ def test_main_sendtext_with_channel(capsys):
def test_main_sendtext_with_invalid_channel(caplog, capsys):
"""Test --sendtext"""
sys.argv = ["", "--sendtext", "hello", "--ch-index", "-1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
iface.localNode.getChannelByChannelIndex.return_value = None
@@ -667,7 +667,7 @@ def test_main_sendtext_with_invalid_channel(caplog, capsys):
def test_main_sendtext_with_invalid_channel_nine(caplog, capsys):
"""Test --sendtext"""
sys.argv = ["", "--sendtext", "hello", "--ch-index", "9"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
iface.localNode.getChannelByChannelIndex.return_value = None
@@ -696,7 +696,7 @@ def test_main_sendtext_with_invalid_channel_nine(caplog, capsys):
def test_main_sendtext_with_dest(mock_findPorts, mock_serial, mocked_open, mock_get, mock_set, capsys, caplog, iface_with_nodes):
"""Test --sendtext with --dest"""
sys.argv = ["", "--sendtext", "hello", "--dest", "foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
#iface = iface_with_nodes
#iface.myInfo.my_node_num = 2475227164
@@ -730,7 +730,7 @@ def test_main_sendtext_with_dest(mock_findPorts, mock_serial, mocked_open, mock_
def test_main_setlat(capsys):
"""Test --sendlat"""
sys.argv = ["", "--setlat", "37.5"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -765,7 +765,7 @@ def test_main_setlat(capsys):
def test_main_setlon(capsys):
"""Test --setlon"""
sys.argv = ["", "--setlon", "-122.1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -800,7 +800,7 @@ def test_main_setlon(capsys):
def test_main_setalt(capsys):
"""Test --setalt"""
sys.argv = ["", "--setalt", "51"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -835,7 +835,7 @@ def test_main_setalt(capsys):
def test_main_seturl(capsys):
"""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)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface) as mo:
@@ -856,7 +856,7 @@ def test_main_seturl(capsys):
def test_main_set_valid(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --set with valid field"""
sys.argv = ["", "--set", "network.wifi_ssid", "foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -881,7 +881,7 @@ def test_main_set_valid(mocked_findports, mocked_serial, mocked_open, mocked_get
def test_main_set_valid_wifi_psk(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --set with valid field"""
sys.argv = ["", "--set", "network.wifi_psk", "123456789"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -906,7 +906,7 @@ def test_main_set_valid_wifi_psk(mocked_findports, mocked_serial, mocked_open, m
def test_main_set_invalid_wifi_psk(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --set with an invalid value (psk must be 8 or more characters)"""
sys.argv = ["", "--set", "network.wifi_psk", "1234567"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -934,8 +934,8 @@ def test_main_set_invalid_wifi_psk(mocked_findports, mocked_serial, mocked_open,
def test_main_set_valid_camel_case(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --set with valid field"""
sys.argv = ["", "--set", "network.wifi_ssid", "foo"]
Globals.getInstance().set_args(sys.argv)
Globals.getInstance().set_camel_case()
globals.args = sys.argv
globals.camel_case = True
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -960,7 +960,7 @@ def test_main_set_valid_camel_case(mocked_findports, mocked_serial, mocked_open,
def test_main_set_with_invalid(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --set with invalid field"""
sys.argv = ["", "--set", "foo", "foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -986,7 +986,7 @@ def test_main_set_with_invalid(mocked_findports, mocked_serial, mocked_open, moc
def test_main_configure_with_snake_case(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --configure with valid file"""
sys.argv = ["", "--configure", "example_config.yaml"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -1019,7 +1019,7 @@ def test_main_configure_with_snake_case(mocked_findports, mocked_serial, mocked_
def test_main_configure_with_camel_case_keys(mocked_findports, mocked_serial, mocked_open, mocked_get, mocked_set, capsys):
"""Test --configure with valid file"""
sys.argv = ["", "--configure", "exampleConfig.yaml"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)
anode = Node(serialInterface, 1234567890, noProto=True)
@@ -1046,7 +1046,7 @@ def test_main_configure_with_camel_case_keys(mocked_findports, mocked_serial, mo
def test_main_ch_add_valid(capsys):
"""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)
globals.args = sys.argv
mocked_channel = MagicMock(autospec=Channel)
# TODO: figure out how to get it to print the channel name instead of MagicMock
@@ -1074,7 +1074,7 @@ def test_main_ch_add_valid(capsys):
def test_main_ch_add_invalid_name_too_long(capsys):
"""Test --ch-add with invalid channel name, name too long"""
sys.argv = ["", "--ch-add", "testingtestingtesting"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_channel = MagicMock(autospec=Channel)
# TODO: figure out how to get it to print the channel name instead of MagicMock
@@ -1105,7 +1105,7 @@ def test_main_ch_add_invalid_name_too_long(capsys):
def test_main_ch_add_but_name_already_exists(capsys):
"""Test --ch-add with a channel name that already exists"""
sys.argv = ["", "--ch-add", "testing"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
# set it up so we do not already have a channel named this
@@ -1131,7 +1131,7 @@ def test_main_ch_add_but_name_already_exists(capsys):
def test_main_ch_add_but_no_more_channels(capsys):
"""Test --ch-add with but there are no more channels"""
sys.argv = ["", "--ch-add", "testing"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
# set it up so we do not already have a channel named this
@@ -1159,7 +1159,7 @@ def test_main_ch_add_but_no_more_channels(capsys):
def test_main_ch_del(capsys):
"""Test --ch-del with valid secondary channel to be deleted"""
sys.argv = ["", "--ch-del", "--ch-index", "1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1180,7 +1180,7 @@ def test_main_ch_del(capsys):
def test_main_ch_del_no_ch_index_specified(capsys):
"""Test --ch-del without a valid ch-index"""
sys.argv = ["", "--ch-del"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1204,8 +1204,8 @@ def test_main_ch_del_no_ch_index_specified(capsys):
def test_main_ch_del_primary_channel(capsys):
"""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)
globals.args = sys.argv
globals.channel_index = 1
mocked_node = MagicMock(autospec=Node)
@@ -1229,7 +1229,7 @@ def test_main_ch_del_primary_channel(capsys):
def test_main_ch_enable_valid_secondary_channel(capsys):
"""Test --ch-enable with --ch-index"""
sys.argv = ["", "--ch-enable", "--ch-index", "1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1242,7 +1242,7 @@ def test_main_ch_enable_valid_secondary_channel(capsys):
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
assert globals.channel_index == 1
mo.assert_called()
@@ -1251,7 +1251,7 @@ def test_main_ch_enable_valid_secondary_channel(capsys):
def test_main_ch_disable_valid_secondary_channel(capsys):
"""Test --ch-disable with --ch-index"""
sys.argv = ["", "--ch-disable", "--ch-index", "1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1264,7 +1264,7 @@ def test_main_ch_disable_valid_secondary_channel(capsys):
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
assert globals.channel_index == 1
mo.assert_called()
@@ -1273,7 +1273,7 @@ def test_main_ch_disable_valid_secondary_channel(capsys):
def test_main_ch_enable_without_a_ch_index(capsys):
"""Test --ch-enable without --ch-index"""
sys.argv = ["", "--ch-enable"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1289,7 +1289,7 @@ def test_main_ch_enable_without_a_ch_index(capsys):
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
assert globals.channel_index is None
mo.assert_called()
@@ -1298,7 +1298,7 @@ def test_main_ch_enable_without_a_ch_index(capsys):
def test_main_ch_enable_primary_channel(capsys):
"""Test --ch-enable with --ch-index = 0"""
sys.argv = ["", "--ch-enable", "--ch-index", "0"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1314,7 +1314,7 @@ def test_main_ch_enable_primary_channel(capsys):
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
assert globals.channel_index == 0
mo.assert_called()
@@ -1327,7 +1327,7 @@ def test_main_ch_enable_primary_channel(capsys):
# '--ch-midfast', '--ch-shortslow', '--ch-shortfast']
# for range_option in range_options:
# sys.argv = ['', f"{range_option}" ]
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# mocked_node = MagicMock(autospec=Node)
#
@@ -1348,7 +1348,7 @@ def test_main_ch_enable_primary_channel(capsys):
def test_main_ch_longfast_on_non_primary_channel(capsys):
"""Test --ch-longfast --ch-index 1"""
sys.argv = ["", "--ch-longfast", "--ch-index", "1"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_node = MagicMock(autospec=Node)
@@ -1387,7 +1387,7 @@ def test_main_ch_longfast_on_non_primary_channel(capsys):
# def test_main_pos_fields_no_args(capsys):
# """Test --pos-fields no args (which shows settings)"""
# sys.argv = ['', '--pos-fields']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)
#
@@ -1419,7 +1419,7 @@ def test_main_ch_longfast_on_non_primary_channel(capsys):
# def test_main_pos_fields_arg_of_zero(capsys):
# """Test --pos-fields an arg of 0 (which shows list)"""
# sys.argv = ['', '--pos-fields', '0']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)
#
@@ -1454,7 +1454,7 @@ def test_main_ch_longfast_on_non_primary_channel(capsys):
# def test_main_pos_fields_valid_values(capsys):
# """Test --pos-fields with valid values"""
# sys.argv = ['', '--pos-fields', 'POS_GEO_SEP', 'POS_ALT_MSL']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)
#
@@ -1482,7 +1482,7 @@ def test_main_ch_longfast_on_non_primary_channel(capsys):
# def test_main_get_with_valid_values(capsys):
# """Test --get with valid values (with string, number, boolean)"""
# sys.argv = ['', '--get', 'ls_secs', '--get', 'wifi_ssid', '--get', 'fixed_position']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# with patch('meshtastic.serial_interface.SerialInterface') as mo:
#
@@ -1508,8 +1508,8 @@ def test_main_ch_longfast_on_non_primary_channel(capsys):
#def test_main_get_with_valid_values_camel(capsys, caplog):
# """Test --get with valid values (with string, number, boolean)"""
# sys.argv = ["", "--get", "lsSecs", "--get", "wifiSsid", "--get", "fixedPosition"]
# Globals.getInstance().set_args(sys.argv)
# Globals.getInstance().set_camel_case()
# globals.args = sys.argv
# globals.camel_case = True
#
# with caplog.at_level(logging.DEBUG):
# with patch("meshtastic.serial_interface.SerialInterface") as mo:
@@ -1534,7 +1534,7 @@ def test_main_ch_longfast_on_non_primary_channel(capsys):
def test_main_get_with_invalid(capsys):
"""Test --get with invalid field"""
sys.argv = ["", "--get", "foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
mocked_user_prefs = MagicMock()
mocked_user_prefs.DESCRIPTOR.fields_by_name.get.return_value = None
@@ -1561,7 +1561,7 @@ def test_main_get_with_invalid(capsys):
def test_main_onReceive_empty(caplog, capsys):
"""Test onReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
globals.args = args
iface = MagicMock(autospec=SerialInterface)
packet = {}
with caplog.at_level(logging.DEBUG):
@@ -1594,7 +1594,7 @@ def test_main_onReceive_with_sendtext(caplog, capsys):
is made in onReceive().
"""
sys.argv = ["", "--sendtext", "hello"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
# Note: 'TEXT_MESSAGE_APP' value is 1
packet = {
@@ -1625,7 +1625,7 @@ def test_main_onReceive_with_text(caplog, capsys):
"""Test onReceive with text"""
args = MagicMock()
args.sendtext.return_value = "foo"
Globals.getInstance().set_args(args)
globals.args = args
# Note: 'TEXT_MESSAGE_APP' value is 1
# Note: Some of this is faked below.
@@ -1659,7 +1659,7 @@ def test_main_onReceive_with_text(caplog, capsys):
def test_main_onConnection(capsys):
"""Test onConnection"""
sys.argv = [""]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
class TempTopic:
@@ -1727,7 +1727,7 @@ position_flags: 35"""
#@pytest.mark.usefixtures("reset_globals")
#def test_main_export_config_use_camel(capsys):
# """Test export_config() function directly"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
# iface = MagicMock(autospec=SerialInterface)
# with patch("meshtastic.serial_interface.SerialInterface", return_value=iface) as mo:
# mo.getLongName.return_value = "foo"
@@ -1774,7 +1774,7 @@ position_flags: 35"""
#def test_main_export_config_called_from_main(capsys):
# """Test --export-config"""
# sys.argv = ["", "--export-config"]
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# iface = MagicMock(autospec=SerialInterface)
# with patch("meshtastic.serial_interface.SerialInterface", return_value=iface) as mo:
@@ -1791,7 +1791,7 @@ position_flags: 35"""
def test_main_gpio_rd_no_gpio_channel(capsys):
"""Test --gpio_rd with no named gpio channel"""
sys.argv = ["", "--gpio-rd", "0x10", "--dest", "!foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=SerialInterface)
iface.localNode.getChannelByName.return_value = None
@@ -1810,7 +1810,7 @@ def test_main_gpio_rd_no_gpio_channel(capsys):
def test_main_gpio_rd_no_dest(capsys):
"""Test --gpio_rd with a named gpio channel but no dest was specified"""
sys.argv = ["", "--gpio-rd", "0x2000"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
channel = Channel(index=2, role=2)
channel.settings.psk = b"\x8a\x94y\x0e\xc6\xc9\x1e5\x91\x12@\xa60\xa8\xb43\x87\x00\xf2K\x0e\xe7\x7fAz\xcd\xf5\xb0\x900\xa84"
@@ -1844,7 +1844,7 @@ def test_main_gpio_rd_no_dest(capsys):
# # >>> print(hex(2**13))
# # 0x2000
# sys.argv = ['', '--gpio-rd', '0x1000', '--dest', '!1234']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# channel = Channel(index=1, role=1)
# channel.settings.modem_config = 3
@@ -1892,7 +1892,7 @@ def test_main_gpio_rd_no_dest(capsys):
# def test_main_gpio_rd_with_no_gpioMask(caplog, capsys):
# """Test --gpio_rd with a named gpio channel"""
# sys.argv = ['', '--gpio-rd', '0x1000', '--dest', '!1234']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# channel = Channel(index=1, role=1)
# channel.settings.modem_config = 3
@@ -1939,7 +1939,7 @@ def test_main_gpio_rd_no_dest(capsys):
# def test_main_gpio_watch(caplog, capsys):
# """Test --gpio_watch with a named gpio channel"""
# sys.argv = ['', '--gpio-watch', '0x1000', '--dest', '!1234']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# def my_sleep(amount):
# print(f'{amount}')
@@ -1993,7 +1993,7 @@ def test_main_gpio_rd_no_dest(capsys):
# def test_main_gpio_wrb(caplog, capsys):
# """Test --gpio_wrb with a named gpio channel"""
# sys.argv = ['', '--gpio-wrb', '4', '1', '--dest', '!1234']
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# channel = Channel(index=1, role=1)
# channel.settings.modem_config = 3
@@ -2057,7 +2057,7 @@ def test_main_gpio_rd_no_dest(capsys):
#@pytest.mark.usefixtures("reset_globals")
#def test_main_getPref_valid_field_camel(capsys):
# """Test getPref() with a valid field"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
# prefs = MagicMock()
# prefs.DESCRIPTOR.fields_by_name.get.return_value = "ls_secs"
# prefs.wifi_ssid = "foo"
@@ -2090,7 +2090,7 @@ def test_main_gpio_rd_no_dest(capsys):
#@pytest.mark.usefixtures("reset_globals")
#def test_main_getPref_valid_field_string_camel(capsys):
# """Test getPref() with a valid field and value as a string"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
# prefs = MagicMock()
# prefs.DESCRIPTOR.fields_by_name.get.return_value = "wifi_ssid"
# prefs.wifi_ssid = "foo"
@@ -2123,7 +2123,7 @@ def test_main_gpio_rd_no_dest(capsys):
#@pytest.mark.usefixtures("reset_globals")
#def test_main_getPref_valid_field_bool_camel(capsys):
# """Test getPref() with a valid field and value as a bool"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
# prefs = MagicMock()
# prefs.DESCRIPTOR.fields_by_name.get.return_value = "fixed_position"
# prefs.wifi_ssid = "foo"
@@ -2172,7 +2172,7 @@ def test_main_gpio_rd_no_dest(capsys):
#@pytest.mark.usefixtures("reset_globals")
#def test_main_getPref_invalid_field_camel(capsys):
# """Test getPref() with an invalid field"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
#
# class Field:
# """Simple class for testing."""
@@ -2267,7 +2267,7 @@ def test_main_gpio_rd_no_dest(capsys):
# @pytest.mark.usefixtures("reset_globals")
# def test_main_setPref_valid_field_invalid_enum_camel(capsys, caplog):
# """Test setPref() with a valid field but invalid enum value"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
#
# radioConfig = RadioConfig()
# prefs = radioConfig.preferences
@@ -2303,7 +2303,7 @@ def test_main_gpio_rd_no_dest(capsys):
# @pytest.mark.usefixtures("reset_globals")
# def test_main_setPref_valid_field_valid_enum_camel(capsys, caplog):
# """Test setPref() with a valid field and valid enum value"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
#
# # charge_current
# # some valid values: MA100 MA1000 MA1080
@@ -2354,7 +2354,7 @@ def test_main_gpio_rd_no_dest(capsys):
#@pytest.mark.usefixtures("reset_globals")
#def test_main_setPref_invalid_field_camel(capsys):
# """Test setPref() with a invalid field"""
# Globals.getInstance().set_camel_case()
# globals.camel_case = True
#
# class Field:
# """Simple class for testing."""
@@ -2435,7 +2435,7 @@ def test_main_gpio_rd_no_dest(capsys):
def test_main_ch_set_psk_no_ch_index(capsys):
"""Test --ch-set psk"""
sys.argv = ["", "--ch-set", "psk", "foo", "--host", "meshtastic.local"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=TCPInterface)
with patch("meshtastic.tcp_interface.TCPInterface", return_value=iface) as mo:
@@ -2464,7 +2464,7 @@ def test_main_ch_set_psk_with_ch_index(capsys):
"--ch-index",
"0",
]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
iface = MagicMock(autospec=TCPInterface)
with patch("meshtastic.tcp_interface.TCPInterface", return_value=iface) as mo:
@@ -2492,7 +2492,7 @@ def test_main_ch_set_psk_with_ch_index(capsys):
# "--ch-index",
# "0",
# ]
# Globals.getInstance().set_args(sys.argv)
# globals.args = sys.argv
#
# iface = MagicMock(autospec=TCPInterface)
# with patch("meshtastic.tcp_interface.TCPInterface", return_value=iface) as mo:
@@ -2520,7 +2520,7 @@ def test_onNode(capsys):
def test_tunnel_no_args(capsys):
"""Test tunnel no arguments"""
sys.argv = [""]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
with pytest.raises(SystemExit) as pytest_wrapped_e:
tunnelMain()
assert pytest_wrapped_e.type == SystemExit
@@ -2539,7 +2539,7 @@ def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, caplog, capsys)
a_mock.return_value = "Linux"
mock_platform_system.side_effect = a_mock
sys.argv = ["", "--tunnel"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
print(f"platform.system():{platform.system()}")
with caplog.at_level(logging.DEBUG):
with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -2562,7 +2562,7 @@ def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, caplog, capsys)
a_mock.return_value = "Linux"
mock_platform_system.side_effect = a_mock
sys.argv = ["", "--subnet", "foo"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
print(f"platform.system():{platform.system()}")
with caplog.at_level(logging.DEBUG):
with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -2597,7 +2597,7 @@ def test_tunnel_tunnel_arg(
a_mock.return_value = "Linux"
mock_platform_system.side_effect = a_mock
sys.argv = ["", "--tunnel"]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
serialInterface = SerialInterface(noProto=True)

13
meshtastic/tests/test_tunnel.py
View File

@@ -1,5 +1,5 @@
"""Meshtastic unit tests for tunnel.py"""
from meshtastic import globals
import logging
import re
import sys
@@ -7,7 +7,6 @@ from unittest.mock import MagicMock, patch
import pytest
from ..globals import Globals
from ..tcp_interface import TCPInterface
from ..tunnel import Tunnel, onTunnelReceive
@@ -51,7 +50,7 @@ def test_Tunnel_with_interface(mock_platform_system, caplog, iface_with_nodes):
with caplog.at_level(logging.WARNING):
with patch("socket.socket"):
tun = Tunnel(iface)
assert tun == Globals.getInstance().get_tunnelInstance()
assert tun == globals.tunnelInstance
iface.close()
assert re.search(r"Not creating a TapDevice()", caplog.text, re.MULTILINE)
assert re.search(r"Not starting TUN reader", caplog.text, re.MULTILINE)
@@ -65,7 +64,7 @@ def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
sys.argv = [""]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
packet = {"decoded": {"payload": "foo"}, "from": 2475227164}
a_mock = MagicMock()
a_mock.return_value = "Linux"
@@ -73,7 +72,7 @@ def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with
with caplog.at_level(logging.DEBUG):
with patch("socket.socket"):
tun = Tunnel(iface)
Globals.getInstance().set_tunnelInstance(tun)
globals.tunnelInstance = tun
onTunnelReceive(packet, iface)
assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)
assert re.search(r"Ignoring message we sent", caplog.text, re.MULTILINE)
@@ -88,7 +87,7 @@ def test_onTunnelReceive_from_someone_else(
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
sys.argv = [""]
Globals.getInstance().set_args(sys.argv)
globals.args = sys.argv
packet = {"decoded": {"payload": "foo"}, "from": 123}
a_mock = MagicMock()
a_mock.return_value = "Linux"
@@ -96,7 +95,7 @@ def test_onTunnelReceive_from_someone_else(
with caplog.at_level(logging.DEBUG):
with patch("socket.socket"):
tun = Tunnel(iface)
Globals.getInstance().set_tunnelInstance(tun)
globals.tunnelInstance = tun
onTunnelReceive(packet, iface)
assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)

9
meshtastic/tunnel.py
View File

@@ -22,16 +22,14 @@ import threading
from pubsub import pub # type: ignore[import-untyped]
from pytap2 import TapDevice
from meshtastic import portnums_pb2
from meshtastic.globals import Globals
from meshtastic import portnums_pb2, globals
from meshtastic.util import ipstr, readnet_u16
def onTunnelReceive(packet, interface): # pylint: disable=W0613
"""Callback for received tunneled messages from mesh."""
logging.debug(f"in onTunnelReceive()")
our_globals = Globals.getInstance()
tunnelInstance = our_globals.get_tunnelInstance()
tunnelInstance = globals.tunnelInstance
tunnelInstance.onReceive(packet)
@@ -67,8 +65,7 @@ class Tunnel:
if platform.system() != "Linux":
raise Tunnel.TunnelError("Tunnel() can only be run instantiated on a Linux system")
our_globals = Globals.getInstance()
our_globals.set_tunnelInstance(self)
globals.tunnelInstance = self
"""A list of chatty UDP services we should never accidentally
forward to our slow network"""