Refactor to remove pylint issues.

Since one of pylint's complains was that the globals module was
shadowing the built-in, and since the name `config` was already
is use in several modules, globals.py was renamed as mt_config.py.
All tests now pass, and the only remaining local pylint errors
relate to the protobuf code, I'm hoping this will make the PR
valid.
This commit is contained in:
Steve Holden
2024-04-10 17:42:44 +01:00
parent 0d57449030
commit a07e853f69
9 changed files with 291 additions and 290 deletions

View File

@@ -16,14 +16,14 @@ from pubsub import pub # type: ignore[import-untyped]
import meshtastic.test
import meshtastic.util
from meshtastic import globals
from meshtastic import mt_config
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
def onReceive(packet, interface):
"""Callback invoked when a packet arrives"""
args = globals.args
args = mt_config.args
try:
d = packet.get("decoded")
logging.debug(f"in onReceive() d:{d}")
@@ -67,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.camel_case}")
logging.debug(f"use camel:{mt_config.camel_case}")
# First validate the input
localConfig = node.localConfig
@@ -84,7 +84,7 @@ def getPref(node, comp_name):
break
if not found:
if globals.camel_case:
if mt_config.camel_case:
print(
f"{localConfig.__class__.__name__} and {moduleConfig.__class__.__name__} do not have an attribute {snake_name}."
)
@@ -103,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.camel_case:
if mt_config.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)}"
@@ -169,7 +169,7 @@ def setPref(config, comp_name, valStr) -> bool:
if e:
val = e.number
else:
if globals.camel_case:
if mt_config.camel_case:
print(
f"{name[0]}.{camel_name} does not have an enum called {val}, so you can not set it."
)
@@ -208,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.camel_case:
if mt_config.camel_case:
print(f"Set {prefix}{camel_name} to {valStr}")
else:
print(f"Set {prefix}{snake_name} to {valStr}")
@@ -223,7 +223,7 @@ def onConnected(interface):
False # Should we wait for an acknowledgment if we send to a remote node?
)
try:
args = globals.args
args = mt_config.args
# do not print this line if we are exporting the config
if not args.export_config:
@@ -474,7 +474,7 @@ def onConnected(interface):
print("Writing modified preferences to device")
node.writeConfig(field)
else:
if globals.camel_case:
if mt_config.camel_case:
print(
f"{node.localConfig.__class__.__name__} and {node.moduleConfig.__class__.__name__} do not have an attribute {pref[0]}."
)
@@ -587,7 +587,7 @@ def onConnected(interface):
# handle changing channels
if args.ch_add:
channelIndex = globals.channel_index
channelIndex = mt_config.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(
@@ -618,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")
globals.channel_index = ch.index
mt_config.channel_index = ch.index
if args.ch_del:
closeNow = True
channelIndex = globals.channel_index
channelIndex = mt_config.channel_index
if channelIndex is None:
meshtastic.util.our_exit(
"Warning: Need to specify '--ch-index' for '--ch-del'.", 1
@@ -639,7 +639,7 @@ def onConnected(interface):
def setSimpleConfig(modem_preset):
"""Set one of the simple modem_config"""
channelIndex = globals.channel_index
channelIndex = mt_config.channel_index
if channelIndex is not None and channelIndex > 0:
meshtastic.util.our_exit(
"Warning: Cannot set modem preset for non-primary channel", 1
@@ -674,7 +674,7 @@ def onConnected(interface):
if args.ch_set or args.ch_enable or args.ch_disable:
closeNow = True
channelIndex = globals.channel_index
channelIndex = mt_config.channel_index
if channelIndex is None:
meshtastic.util.our_exit("Warning: Need to specify '--ch-index'.", 1)
ch = interface.getNode(args.dest).channels[channelIndex]
@@ -829,7 +829,7 @@ def printConfig(config):
names = []
for field in config.message_type.fields:
tmp_name = f"{config_section.name}.{field.name}"
if globals.camel_case:
if mt_config.camel_case:
tmp_name = meshtastic.util.snake_to_camel(tmp_name)
names.append(tmp_name)
for temp_name in sorted(names):
@@ -874,7 +874,7 @@ def export_config(interface):
if owner_short:
configObj["owner_short"] = owner_short
if channel_url:
if globals.camel_case:
if mt_config.camel_case:
configObj["channelUrl"] = channel_url
else:
configObj["channel_url"] = channel_url
@@ -886,11 +886,11 @@ def export_config(interface):
# Convert inner keys to correct snake/camelCase
prefs = {}
for pref in config:
if globals.camel_case:
if mt_config.camel_case:
prefs[meshtastic.util.snake_to_camel(pref)] = config[pref]
else:
prefs[pref] = config[pref]
if globals.camel_case:
if mt_config.camel_case:
configObj["config"] = config
else:
configObj["config"] = config
@@ -902,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.camel_case:
if mt_config.camel_case:
configObj["module_config"] = prefs
else:
configObj["module_config"] = prefs
@@ -916,8 +916,8 @@ def export_config(interface):
def common():
"""Shared code for all of our command line wrappers"""
logfile = None
args = globals.args
parser = globals.parser
args = mt_config.args
parser = mt_config.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",
@@ -933,7 +933,7 @@ def common():
if args.ch_index is not None:
channelIndex = int(args.ch_index)
globals.channel_index = channelIndex
mt_config.channel_index = channelIndex
if not args.dest:
args.dest = BROADCAST_ADDR
@@ -968,7 +968,7 @@ def common():
# Note: using "line buffering"
# pylint: disable=R1732
logfile = open(args.seriallog, "w+", buffering=1, encoding="utf8")
globals.logfile = logfile
mt_config.logfile = logfile
subscribe()
if args.ble_scan:
@@ -1059,8 +1059,8 @@ def addConnectionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParse
def initParser():
"""Initialize the command line argument parsing."""
parser = globals.parser
args = globals.args
parser = mt_config.parser
args = mt_config.args
# The "Help" group includes the help option and other informational stuff about the CLI itself
outerHelpGroup = parser.add_argument_group('Help')
@@ -1426,8 +1426,8 @@ def initParser():
args = parser.parse_args()
globals.args = args
globals.parser = parser
mt_config.args = args
mt_config.parser = parser
def main():
@@ -1436,10 +1436,10 @@ def main():
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.")
globals.parser = parser
mt_config.parser = parser
initParser()
common()
logfile = globals.logfile
logfile = mt_config.logfile
if logfile:
logfile.close()
@@ -1447,11 +1447,11 @@ def main():
def tunnelMain():
"""Run a meshtastic IP tunnel"""
parser = argparse.ArgumentParser(add_help=False)
globals.parser = parser
mt_config.parser = parser
initParser()
args = globals.args
args = mt_config.args
args.tunnel = True
globals.args = args
mt_config.args = args
common()

View File

@@ -8,7 +8,7 @@ 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
plaster over the use of shared mt_config, but the coupling issues wil be dealt
with rather more easily once the code is simplified by this change.
"""
@@ -17,6 +17,7 @@ def reset():
"""
Restore the namespace to pristine condition.
"""
# pylint: disable=W0603
global args, parser, channel_index, logfile, tunnelInstance, camel_case
args = None
parser = None

View File

@@ -5,18 +5,18 @@ from unittest.mock import MagicMock
import pytest
from meshtastic import globals
from meshtastic import mt_config
from ..mesh_interface import MeshInterface
@pytest.fixture
def reset_globals():
"""Fixture to reset globals."""
def reset_mt_config():
"""Fixture to reset mt_config."""
parser = None
parser = argparse.ArgumentParser(add_help=False)
globals.reset()
globals.parser = parser
mt_config.reset()
mt_config.parser = parser
@pytest.fixture

View File

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

View File

File diff suppressed because it is too large Load Diff

View File

@@ -16,7 +16,7 @@ from ..util import Timeout
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_MeshInterface(capsys):
"""Test that we can instantiate a MeshInterface"""
iface = MeshInterface(noProto=True)
@@ -57,7 +57,7 @@ def test_MeshInterface(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getMyUser(iface_with_nodes):
"""Test getMyUser()"""
iface = iface_with_nodes
@@ -68,7 +68,7 @@ def test_getMyUser(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getLongName(iface_with_nodes):
"""Test getLongName()"""
iface = iface_with_nodes
@@ -78,7 +78,7 @@ def test_getLongName(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getShortName(iface_with_nodes):
"""Test getShortName()."""
iface = iface_with_nodes
@@ -88,7 +88,7 @@ def test_getShortName(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handlePacketFromRadio_no_from(capsys):
"""Test _handlePacketFromRadio with no 'from' in the mesh packet."""
iface = MeshInterface(noProto=True)
@@ -100,7 +100,7 @@ def test_handlePacketFromRadio_no_from(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handlePacketFromRadio_with_a_portnum(caplog):
"""Test _handlePacketFromRadio with a portnum
Since we have an attribute called 'from', we cannot simply 'set' it.
@@ -116,7 +116,7 @@ def test_handlePacketFromRadio_with_a_portnum(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handlePacketFromRadio_no_portnum(caplog):
"""Test _handlePacketFromRadio without a portnum"""
iface = MeshInterface(noProto=True)
@@ -128,7 +128,7 @@ def test_handlePacketFromRadio_no_portnum(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getNode_with_local():
"""Test getNode"""
iface = MeshInterface(noProto=True)
@@ -137,7 +137,7 @@ def test_getNode_with_local():
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getNode_not_local(caplog):
"""Test getNode not local"""
iface = MeshInterface(noProto=True)
@@ -150,7 +150,7 @@ def test_getNode_not_local(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getNode_not_local_timeout(capsys):
"""Test getNode not local, simulate timeout"""
iface = MeshInterface(noProto=True)
@@ -167,7 +167,7 @@ def test_getNode_not_local_timeout(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPosition(caplog):
"""Test sendPosition"""
iface = MeshInterface(noProto=True)
@@ -179,7 +179,7 @@ def test_sendPosition(caplog):
# TODO
# @pytest.mark.unit
# @pytest.mark.usefixtures("reset_globals")
# @pytest.mark.usefixtures("reset_mt_config")
# def test_close_with_heartbeatTimer(caplog):
# """Test close() with heartbeatTimer"""
# iface = MeshInterface(noProto=True)
@@ -197,7 +197,7 @@ def test_sendPosition(caplog):
# TODO
# @pytest.mark.unit
# @pytest.mark.usefixtures("reset_globals")
# @pytest.mark.usefixtures("reset_mt_config")
# def test_handleFromRadio_empty_payload(caplog):
# """Test _handleFromRadio"""
# iface = MeshInterface(noProto=True)
@@ -208,7 +208,7 @@ def test_sendPosition(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handleFromRadio_with_my_info(caplog):
"""Test _handleFromRadio with my_info"""
# Note: I captured the '--debug --info' for the bytes below.
@@ -233,7 +233,7 @@ def test_handleFromRadio_with_my_info(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handleFromRadio_with_node_info(caplog, capsys):
"""Test _handleFromRadio with node_info"""
# Note: I captured the '--debug --info' for the bytes below.
@@ -269,7 +269,7 @@ def test_handleFromRadio_with_node_info(caplog, capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handleFromRadio_with_node_info_tbeam1(caplog, capsys):
"""Test _handleFromRadio with node_info"""
# Note: Captured the '--debug --info' for the bytes below.
@@ -293,7 +293,7 @@ def test_handleFromRadio_with_node_info_tbeam1(caplog, capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_handleFromRadio_with_node_info_tbeam_with_bad_data(caplog):
"""Test _handleFromRadio with node_info with some bad data (issue#172) - ensure we do not throw exception"""
# Note: Captured the '--debug --info' for the bytes below.
@@ -305,7 +305,7 @@ def test_handleFromRadio_with_node_info_tbeam_with_bad_data(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_MeshInterface_sendToRadioImpl(caplog):
"""Test _sendToRadioImp()"""
iface = MeshInterface(noProto=True)
@@ -316,7 +316,7 @@ def test_MeshInterface_sendToRadioImpl(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_MeshInterface_sendToRadio_no_proto(caplog):
"""Test sendToRadio()"""
iface = MeshInterface()
@@ -327,7 +327,7 @@ def test_MeshInterface_sendToRadio_no_proto(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendData_too_long(caplog):
"""Test when data payload is too big"""
iface = MeshInterface(noProto=True)
@@ -352,7 +352,7 @@ def test_sendData_too_long(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendData_unknown_app(capsys):
"""Test sendData when unknown app"""
iface = MeshInterface(noProto=True)
@@ -366,7 +366,7 @@ def test_sendData_unknown_app(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPosition_with_a_position(caplog):
"""Test sendPosition when lat/long/alt"""
iface = MeshInterface(noProto=True)
@@ -378,7 +378,7 @@ def test_sendPosition_with_a_position(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_no_destination(capsys):
"""Test _sendPacket()"""
iface = MeshInterface(noProto=True)
@@ -392,7 +392,7 @@ def test_sendPacket_with_no_destination(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_as_int(caplog):
"""Test _sendPacket() with int as a destination"""
iface = MeshInterface(noProto=True)
@@ -403,7 +403,7 @@ def test_sendPacket_with_destination_as_int(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_starting_with_a_bang(caplog):
"""Test _sendPacket() with int as a destination"""
iface = MeshInterface(noProto=True)
@@ -414,7 +414,7 @@ def test_sendPacket_with_destination_starting_with_a_bang(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_as_BROADCAST_ADDR(caplog):
"""Test _sendPacket() with BROADCAST_ADDR as a destination"""
iface = MeshInterface(noProto=True)
@@ -425,7 +425,7 @@ def test_sendPacket_with_destination_as_BROADCAST_ADDR(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_as_LOCAL_ADDR_no_myInfo(capsys):
"""Test _sendPacket() with LOCAL_ADDR as a destination with no myInfo"""
iface = MeshInterface(noProto=True)
@@ -440,7 +440,7 @@ def test_sendPacket_with_destination_as_LOCAL_ADDR_no_myInfo(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_as_LOCAL_ADDR_with_myInfo(caplog):
"""Test _sendPacket() with LOCAL_ADDR as a destination with myInfo"""
iface = MeshInterface(noProto=True)
@@ -454,7 +454,7 @@ def test_sendPacket_with_destination_as_LOCAL_ADDR_with_myInfo(caplog):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_is_blank_with_nodes(capsys, iface_with_nodes):
"""Test _sendPacket() with '' as a destination with myInfo"""
iface = iface_with_nodes
@@ -469,7 +469,7 @@ def test_sendPacket_with_destination_is_blank_with_nodes(capsys, iface_with_node
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_sendPacket_with_destination_is_blank_without_nodes(caplog, iface_with_nodes):
"""Test _sendPacket() with '' as a destination with myInfo"""
iface = iface_with_nodes
@@ -481,7 +481,7 @@ def test_sendPacket_with_destination_is_blank_without_nodes(caplog, iface_with_n
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getMyNodeInfo():
"""Test getMyNodeInfo()"""
iface = MeshInterface(noProto=True)
@@ -496,7 +496,7 @@ def test_getMyNodeInfo():
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_generatePacketId(capsys):
"""Test _generatePacketId() when no currentPacketId (not connected)"""
iface = MeshInterface(noProto=True)
@@ -514,7 +514,7 @@ def test_generatePacketId(capsys):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_fixupPosition_empty_pos():
"""Test _fixupPosition()"""
iface = MeshInterface(noProto=True)
@@ -524,7 +524,7 @@ def test_fixupPosition_empty_pos():
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_fixupPosition_no_changes_needed():
"""Test _fixupPosition()"""
iface = MeshInterface(noProto=True)
@@ -534,7 +534,7 @@ def test_fixupPosition_no_changes_needed():
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_fixupPosition():
"""Test _fixupPosition()"""
iface = MeshInterface(noProto=True)
@@ -549,7 +549,7 @@ def test_fixupPosition():
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_nodeNumToId(iface_with_nodes):
"""Test _nodeNumToId()"""
iface = iface_with_nodes
@@ -559,7 +559,7 @@ def test_nodeNumToId(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_nodeNumToId_not_found(iface_with_nodes):
"""Test _nodeNumToId()"""
iface = iface_with_nodes
@@ -569,7 +569,7 @@ def test_nodeNumToId_not_found(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_nodeNumToId_to_all(iface_with_nodes):
"""Test _nodeNumToId()"""
iface = iface_with_nodes
@@ -579,7 +579,7 @@ def test_nodeNumToId_to_all(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getOrCreateByNum_minimal(iface_with_nodes):
"""Test _getOrCreateByNum()"""
iface = iface_with_nodes
@@ -589,7 +589,7 @@ def test_getOrCreateByNum_minimal(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getOrCreateByNum_not_found(iface_with_nodes):
"""Test _getOrCreateByNum()"""
iface = iface_with_nodes
@@ -600,7 +600,7 @@ def test_getOrCreateByNum_not_found(iface_with_nodes):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_getOrCreateByNum(iface_with_nodes):
"""Test _getOrCreateByNum()"""
iface = iface_with_nodes

View File

@@ -20,7 +20,7 @@ def test_StreamInterface():
# Note: This takes a bit, so moving from unit to slow
@pytest.mark.unitslow
@pytest.mark.usefixtures("reset_globals")
@pytest.mark.usefixtures("reset_mt_config")
def test_StreamInterface_with_noProto(caplog):
"""Test that we can instantiate a StreamInterface based on nonProto
and we can read/write bytes from a mocked stream
@@ -41,7 +41,7 @@ def test_StreamInterface_with_noProto(caplog):
### Tip: If you want to see the print output, run with '-s' flag:
### pytest -s meshtastic/tests/test_stream_interface.py::test_sendToRadioImpl
# @pytest.mark.unitslow
# @pytest.mark.usefixtures("reset_globals")
# @pytest.mark.usefixtures("reset_mt_config")
# def test_sendToRadioImpl(caplog):
# """Test _sendToRadioImpl()"""
#

View File

@@ -6,7 +6,7 @@ from unittest.mock import MagicMock, patch
import pytest
from meshtastic import globals
from meshtastic import mt_config
from ..tcp_interface import TCPInterface
from ..tunnel import Tunnel, onTunnelReceive
@@ -51,7 +51,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.tunnelInstance
assert tun == mt_config.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 +65,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.args = sys.argv
mt_config.args = sys.argv
packet = {"decoded": {"payload": "foo"}, "from": 2475227164}
a_mock = MagicMock()
a_mock.return_value = "Linux"
@@ -73,7 +73,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.tunnelInstance = tun
mt_config.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 +88,7 @@ def test_onTunnelReceive_from_someone_else(
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
sys.argv = [""]
globals.args = sys.argv
mt_config.args = sys.argv
packet = {"decoded": {"payload": "foo"}, "from": 123}
a_mock = MagicMock()
a_mock.return_value = "Linux"
@@ -96,7 +96,7 @@ def test_onTunnelReceive_from_someone_else(
with caplog.at_level(logging.DEBUG):
with patch("socket.socket"):
tun = Tunnel(iface)
globals.tunnelInstance = tun
mt_config.tunnelInstance = tun
onTunnelReceive(packet, iface)
assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)

View File

@@ -22,14 +22,14 @@ import threading
from pubsub import pub # type: ignore[import-untyped]
from pytap2 import TapDevice
from meshtastic import portnums_pb2, globals
from meshtastic import portnums_pb2, mt_config
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()")
tunnelInstance = globals.tunnelInstance
tunnelInstance = mt_config.tunnelInstance
tunnelInstance.onReceive(packet)
@@ -65,7 +65,7 @@ class Tunnel:
if platform.system() != "Linux":
raise Tunnel.TunnelError("Tunnel() can only be run instantiated on a Linux system")
globals.tunnelInstance = self
mt_config.tunnelInstance = self
"""A list of chatty UDP services we should never accidentally
forward to our slow network"""