Compare commits

...

23 Commits

Author SHA1 Message Date
mkinney
52834e9966 Update setup.py 2022-01-12 17:25:00 -08:00
mkinney
63c60d4cea Merge pull request #229 from mkinney/check_for_multiple_mac_conversions
suggested fix from MitchConner912 for not converting mac address more…
2022-01-12 17:13:14 -08:00
Mike Kinney
6a2a9d2093 suggested fix from MitchConner912 for not converting mac address more than once 2022-01-12 17:10:51 -08:00
mkinney
1410448808 Merge pull request #228 from mkinney/more_unit_tests
add more unit tests
2022-01-12 17:06:24 -08:00
Mike Kinney
ad8f2222db cover a few more lines 2022-01-12 16:50:29 -08:00
Mike Kinney
48265e73b1 no need to import pygatt here 2022-01-12 15:54:07 -08:00
Mike Kinney
f3139a8aa0 add more unit tests 2022-01-12 15:50:16 -08:00
mkinney
8d68e36703 Merge pull request #227 from mkinney/pylint_fixes
fix the consider-using-f-string warnings
2022-01-12 14:57:50 -08:00
Mike Kinney
d2d93fbe80 no longer need to disable this pylint warning 2022-01-12 14:50:10 -08:00
Mike Kinney
ed8510468d no need to disable these pylint warnings anymore 2022-01-12 14:47:49 -08:00
Mike Kinney
e7680e07c2 fix pylint global-statement warnings 2022-01-12 14:46:21 -08:00
Mike Kinney
0f89baa36e fix the pylint unused-argument warnings 2022-01-12 14:41:49 -08:00
Mike Kinney
48ed7690af fix the consider-using-f-string warnings 2022-01-12 13:46:01 -08:00
mkinney
59b94ea650 Merge pull request #226 from mkinney/add_get_hw_example
add example how you can get the hwModel using api
2022-01-12 10:59:43 -08:00
Mike Kinney
5b992734fb add example how you can get the hwModel using api 2022-01-12 10:57:03 -08:00
mkinney
3b74b911f8 Merge pull request #224 from mkinney/add_set_owner_example
simple example showing how you can set the long and short name
2022-01-12 10:26:22 -08:00
Mike Kinney
b6570e3c27 simple example showing how you can set the long and short name 2022-01-12 10:23:51 -08:00
mkinney
e1e1664b96 Merge pull request #223 from mkinney/make_pygatt_linux_only
make pygatt linux only in requirements.txt
2022-01-12 10:23:06 -08:00
Mike Kinney
cb1913dfc3 make pygatt linux only in requirements.txt 2022-01-12 10:15:06 -08:00
Sacha Weatherstone
b813a6f8c5 Update requirements.txt 2022-01-12 14:39:20 +11:00
mkinney
0b662318e1 Merge pull request #222 from mkinney/testing_on_mac_air
fixes for working on mac air
2022-01-11 16:48:01 -08:00
Mike Kinney
a6ccc1a246 add conditional lib if linux 2022-01-11 16:42:25 -08:00
Mike Kinney
bc17e9b389 fixes for working on mac air 2022-01-11 16:36:39 -08:00
25 changed files with 484 additions and 195 deletions

View File

@@ -23,7 +23,7 @@ ignore-patterns=mqtt_pb2.py,channel_pb2.py,environmental_measurement_pb2.py,admi
# no Warning level messages displayed, use"--disable=all --enable=classes # no Warning level messages displayed, use"--disable=all --enable=classes
# --disable=W" # --disable=W"
# #
disable=invalid-name,fixme,logging-fstring-interpolation,too-many-statements,too-many-branches,too-many-locals,no-member,f-string-without-interpolation,protected-access,no-self-use,pointless-string-statement,too-few-public-methods,consider-using-f-string,broad-except,no-else-return,unused-argument,global-statement,global-variable-not-assigned,too-many-boolean-expressions,no-else-raise,bare-except,c-extension-no-member disable=invalid-name,fixme,logging-fstring-interpolation,too-many-statements,too-many-branches,too-many-locals,no-member,f-string-without-interpolation,protected-access,no-self-use,pointless-string-statement,too-few-public-methods,broad-except,no-else-return,no-else-raise,bare-except
[BASIC] [BASIC]

20
examples/get_hw.py Normal file
View File

@@ -0,0 +1,20 @@
"""Simple program to demo how to use meshtastic library.
To run: python examples/get_hw.py
"""
import sys
import meshtastic
import meshtastic.serial_interface
# simple arg check
if len(sys.argv) != 1:
print(f"usage: {sys.argv[0]}")
print("Print the hardware model for the local node.")
sys.exit(3)
iface = meshtastic.serial_interface.SerialInterface()
if iface.nodes:
for n in iface.nodes.values():
if n['num'] == iface.myInfo.my_node_num:
print(n['user']['hwModel'])
iface.close()

20
examples/set_owner.py Normal file
View File

@@ -0,0 +1,20 @@
"""Simple program to demo how to use meshtastic library.
To run: python examples/set_owner.py Bobby 333
"""
import sys
import meshtastic
import meshtastic.serial_interface
# simple arg check
if len(sys.argv) < 2:
print(f"usage: {sys.argv[0]} long_name [short_name]")
sys.exit(3)
iface = meshtastic.serial_interface.SerialInterface()
long_name = sys.argv[1]
short_name = None
if len(sys.argv) > 2:
short_name = sys.argv[2]
iface.localNode.setOwner(long_name, short_name)
iface.close()

View File

@@ -72,7 +72,6 @@ from typing import *
import serial import serial
import timeago import timeago
import google.protobuf.json_format import google.protobuf.json_format
import pygatt
from pubsub import pub from pubsub import pub
from dotmap import DotMap from dotmap import DotMap
from tabulate import tabulate from tabulate import tabulate
@@ -83,6 +82,7 @@ from meshtastic import (mesh_pb2, portnums_pb2, apponly_pb2, admin_pb2,
environmental_measurement_pb2, remote_hardware_pb2, environmental_measurement_pb2, remote_hardware_pb2,
channel_pb2, radioconfig_pb2, util) channel_pb2, radioconfig_pb2, util)
# Note: To follow PEP224, comments should be after the module variable. # Note: To follow PEP224, comments should be after the module variable.
LOCAL_ADDR = "^local" LOCAL_ADDR = "^local"
@@ -129,6 +129,7 @@ def _onTextReceive(iface, asDict):
# #
# Usually btw this problem is caused by apps sending binary data but setting the payload type to # Usually btw this problem is caused by apps sending binary data but setting the payload type to
# text. # text.
logging.debug(f'in _onTextReceive() asDict:{asDict}')
try: try:
asBytes = asDict["decoded"]["payload"] asBytes = asDict["decoded"]["payload"]
asDict["decoded"]["text"] = asBytes.decode("utf-8") asDict["decoded"]["text"] = asBytes.decode("utf-8")
@@ -139,22 +140,30 @@ def _onTextReceive(iface, asDict):
def _onPositionReceive(iface, asDict): def _onPositionReceive(iface, asDict):
"""Special auto parsing for received messages""" """Special auto parsing for received messages"""
p = asDict["decoded"]["position"] logging.debug(f'in _onPositionReceive() asDict:{asDict}')
iface._fixupPosition(p) if 'decoded' in asDict:
# update node DB as needed if 'position' in asDict['decoded'] and 'from' in asDict:
iface._getOrCreateByNum(asDict["from"])["position"] = p p = asDict["decoded"]["position"]
logging.debug(f'p:{p}')
p = iface._fixupPosition(p)
logging.debug(f'after fixup p:{p}')
# update node DB as needed
iface._getOrCreateByNum(asDict["from"])["position"] = p
def _onNodeInfoReceive(iface, asDict): def _onNodeInfoReceive(iface, asDict):
"""Special auto parsing for received messages""" """Special auto parsing for received messages"""
p = asDict["decoded"]["user"] logging.debug(f'in _onNodeInfoReceive() asDict:{asDict}')
# decode user protobufs and update nodedb, provide decoded version as "position" in the published msg if 'decoded' in asDict:
# update node DB as needed if 'user' in asDict['decoded'] and 'from' in asDict:
n = iface._getOrCreateByNum(asDict["from"]) p = asDict["decoded"]["user"]
n["user"] = p # decode user protobufs and update nodedb, provide decoded version as "position" in the published msg
# We now have a node ID, make sure it is uptodate in that table # update node DB as needed
iface.nodes[p["id"]] = n n = iface._getOrCreateByNum(asDict["from"])
_receiveInfoUpdate(iface, asDict) n["user"] = p
# We now have a node ID, make sure it is uptodate in that table
iface.nodes[p["id"]] = n
_receiveInfoUpdate(iface, asDict)
def _receiveInfoUpdate(iface, asDict): def _receiveInfoUpdate(iface, asDict):

View File

@@ -40,7 +40,7 @@ def onReceive(packet, interface):
rxSnr = packet['rxSnr'] rxSnr = packet['rxSnr']
hopLimit = packet['hopLimit'] hopLimit = packet['hopLimit']
print(f"message: {msg}") print(f"message: {msg}")
reply = "got msg \'{}\' with rxSnr: {} and hopLimit: {}".format(msg, rxSnr, hopLimit) reply = f"got msg \'{msg}\' with rxSnr: {rxSnr} and hopLimit: {hopLimit}"
print("Sending reply: ", reply) print("Sending reply: ", reply)
interface.sendText(reply) interface.sendText(reply)
@@ -48,7 +48,7 @@ def onReceive(packet, interface):
print(f'Warning: There is no field {ex} in the packet.') print(f'Warning: There is no field {ex} in the packet.')
def onConnection(interface, topic=pub.AUTO_TOPIC): def onConnection(interface, topic=pub.AUTO_TOPIC): # pylint: disable=W0613
"""Callback invoked when we connect/disconnect from a radio""" """Callback invoked when we connect/disconnect from a radio"""
print(f"Connection changed: {topic.getName()}") print(f"Connection changed: {topic.getName()}")
@@ -192,7 +192,7 @@ def onConnected(interface):
else: else:
print(f"Setting position fields to {allFields}") print(f"Setting position fields to {allFields}")
setPref(prefs, 'position_flags', ('%d' % allFields)) setPref(prefs, 'position_flags', f'{allFields:d}')
print("Writing modified preferences to device") print("Writing modified preferences to device")
interface.getNode(args.dest).writeConfig() interface.getNode(args.dest).writeConfig()

View File

@@ -1,10 +1,16 @@
"""Bluetooth interface """Bluetooth interface
""" """
import logging import logging
import pygatt import platform
from meshtastic.mesh_interface import MeshInterface from meshtastic.mesh_interface import MeshInterface
from meshtastic.util import our_exit
if platform.system() == 'Linux':
# pylint: disable=E0401
import pygatt
# Our standard BLE characteristics # Our standard BLE characteristics
TORADIO_UUID = "f75c76d2-129e-4dad-a1dd-7866124401e7" TORADIO_UUID = "f75c76d2-129e-4dad-a1dd-7866124401e7"
@@ -16,6 +22,8 @@ class BLEInterface(MeshInterface):
"""A not quite ready - FIXME - BLE interface to devices""" """A not quite ready - FIXME - BLE interface to devices"""
def __init__(self, address, noProto=False, debugOut=None): def __init__(self, address, noProto=False, debugOut=None):
if platform.system() != 'Linux':
our_exit("Linux is the only platform with experimental BLE support.", 1)
self.address = address self.address = address
if not noProto: if not noProto:
self.adapter = pygatt.GATTToolBackend() # BGAPIBackend() self.adapter = pygatt.GATTToolBackend() # BGAPIBackend()
@@ -31,7 +39,7 @@ class BLEInterface(MeshInterface):
self._readFromRadio() # read the initial responses self._readFromRadio() # read the initial responses
def handle_data(handle, data): def handle_data(handle, data): # pylint: disable=W0613
self._handleFromRadio(data) self._handleFromRadio(data)
if self.device: if self.device:

View File

@@ -73,7 +73,7 @@ class MeshInterface:
logging.error(f'Traceback: {traceback}') logging.error(f'Traceback: {traceback}')
self.close() self.close()
def showInfo(self, file=sys.stdout): def showInfo(self, file=sys.stdout): # pylint: disable=W0613
"""Show human readable summary about this object""" """Show human readable summary about this object"""
owner = f"Owner: {self.getLongName()} ({self.getShortName()})" owner = f"Owner: {self.getLongName()} ({self.getShortName()})"
myinfo = '' myinfo = ''
@@ -100,7 +100,7 @@ class MeshInterface:
print(infos) print(infos)
return infos return infos
def showNodes(self, includeSelf=True, file=sys.stdout): def showNodes(self, includeSelf=True, file=sys.stdout): # pylint: disable=W0613
"""Show table summary of nodes in mesh""" """Show table summary of nodes in mesh"""
def formatFloat(value, precision=2, unit=''): def formatFloat(value, precision=2, unit=''):
"""Format a float value with precsion.""" """Format a float value with precsion."""

View File

@@ -62,8 +62,7 @@ class StreamInterface(MeshInterface):
# because we want to ensure it is looking for START1) # because we want to ensure it is looking for START1)
p = bytearray([START2] * 32) p = bytearray([START2] * 32)
self._writeBytes(p) self._writeBytes(p)
if not self.noProto: time.sleep(0.1) # wait 100ms to give device time to start running
time.sleep(0.1) # wait 100ms to give device time to start running
self._rxThread.start() self._rxThread.start()
@@ -90,8 +89,7 @@ class StreamInterface(MeshInterface):
self.stream.write(b) self.stream.write(b)
self.stream.flush() self.stream.flush()
# we sleep here to give the TBeam a chance to work # we sleep here to give the TBeam a chance to work
if not self.noProto: time.sleep(0.1)
time.sleep(0.1)
def _readBytes(self, length): def _readBytes(self, length):
"""Read an array of bytes from our stream""" """Read an array of bytes from our stream"""

View File

@@ -33,6 +33,12 @@ class TCPInterface(StreamInterface):
StreamInterface.__init__(self, debugOut=debugOut, noProto=noProto, StreamInterface.__init__(self, debugOut=debugOut, noProto=noProto,
connectNow=connectNow) connectNow=connectNow)
def _socket_shutdown(self):
"""Shutdown the socket.
Note: Broke out this line so the exception could be unit tested.
"""
self.socket.shutdown(socket.SHUT_RDWR)
def myConnect(self): def myConnect(self):
"""Connect to socket""" """Connect to socket"""
server_address = (self.hostname, self.portNumber) server_address = (self.hostname, self.portNumber)
@@ -48,7 +54,7 @@ class TCPInterface(StreamInterface):
self._wantExit = True self._wantExit = True
if not self.socket is None: if not self.socket is None:
try: try:
self.socket.shutdown(socket.SHUT_RDWR) self._socket_shutdown()
except: except:
pass # Ignore errors in shutdown, because we might have a race with the server pass # Ignore errors in shutdown, because we might have a race with the server
self.socket.close() self.socket.close()

View File

@@ -63,6 +63,7 @@ def testSend(fromInterface, toInterface, isBroadcast=False, asBinary=False, want
Returns: Returns:
boolean -- True for success boolean -- True for success
""" """
# pylint: disable=W0603
global receivedPackets global receivedPackets
receivedPackets = [] receivedPackets = []
fromNode = fromInterface.myInfo.my_node_num fromNode = fromInterface.myInfo.my_node_num
@@ -74,6 +75,7 @@ def testSend(fromInterface, toInterface, isBroadcast=False, asBinary=False, want
logging.debug( logging.debug(
f"Sending test wantAck={wantAck} packet from {fromNode} to {toNode}") f"Sending test wantAck={wantAck} packet from {fromNode} to {toNode}")
# pylint: disable=W0603
global sendingInterface global sendingInterface
sendingInterface = fromInterface sendingInterface = fromInterface
if not asBinary: if not asBinary:
@@ -94,6 +96,7 @@ def runTests(numTests=50, wantAck=False, maxFailures=0):
numFail = 0 numFail = 0
numSuccess = 0 numSuccess = 0
for _ in range(numTests): for _ in range(numTests):
# pylint: disable=W0603
global testNumber global testNumber
testNumber = testNumber + 1 testNumber = testNumber + 1
isBroadcast = True isBroadcast = True
@@ -152,6 +155,7 @@ def testAll(numTests=5):
pub.subscribe(onConnection, "meshtastic.connection") pub.subscribe(onConnection, "meshtastic.connection")
pub.subscribe(onReceive, "meshtastic.receive") pub.subscribe(onReceive, "meshtastic.receive")
# pylint: disable=W0603
global interfaces global interfaces
interfaces = list(map(lambda port: SerialInterface( interfaces = list(map(lambda port: SerialInterface(
port, debugOut=openDebugLog(port), connectNow=True), ports)) port, debugOut=openDebugLog(port), connectNow=True), ports))

View File

@@ -1,12 +1,15 @@
"""Meshtastic unit tests for ble_interface.py""" """Meshtastic unit tests for ble_interface.py"""
from unittest.mock import patch
import pytest import pytest
from ..ble_interface import BLEInterface from ..ble_interface import BLEInterface
@pytest.mark.unit @pytest.mark.unit
def test_BLEInterface(): @patch('platform.system', return_value='Linux')
def test_BLEInterface(mock_platform):
"""Test that we can instantiate a BLEInterface""" """Test that we can instantiate a BLEInterface"""
iface = BLEInterface('foo', debugOut=True, noProto=True) iface = BLEInterface('foo', debugOut=True, noProto=True)
iface.close() iface.close()
mock_platform.assert_called()

View File

@@ -0,0 +1,61 @@
"""Meshtastic unit tests for __init__.py"""
import re
import logging
from unittest.mock import MagicMock
import pytest
from meshtastic.__init__ import _onTextReceive, _onPositionReceive, _onNodeInfoReceive
from ..serial_interface import SerialInterface
from ..globals import Globals
@pytest.mark.unit
def test_init_onTextReceive_with_exception(caplog):
"""Test _onTextReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
iface = MagicMock(autospec=SerialInterface)
packet = {}
with caplog.at_level(logging.DEBUG):
_onTextReceive(iface, packet)
assert re.search(r'in _onTextReceive', caplog.text, re.MULTILINE)
assert re.search(r'Malformatted', caplog.text, re.MULTILINE)
@pytest.mark.unit
def test_init_onPositionReceive(caplog):
"""Test _onPositionReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
iface = MagicMock(autospec=SerialInterface)
packet = {
'from': 'foo',
'decoded': {
'position': {}
}
}
with caplog.at_level(logging.DEBUG):
_onPositionReceive(iface, packet)
assert re.search(r'in _onPositionReceive', caplog.text, re.MULTILINE)
@pytest.mark.unit
def test_init_onNodeInfoReceive(caplog, iface_with_nodes):
"""Test _onNodeInfoReceive"""
args = MagicMock()
Globals.getInstance().set_args(args)
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
packet = {
'from': 'foo',
'decoded': {
'user': {
'id': 'bar',
},
}
}
with caplog.at_level(logging.DEBUG):
_onNodeInfoReceive(iface, packet)
assert re.search(r'in _onNodeInfoReceive', caplog.text, re.MULTILINE)

View File

@@ -22,7 +22,8 @@ from ..remote_hardware import onGPIOreceive
@pytest.mark.unit @pytest.mark.unit
def test_main_init_parser_no_args(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_init_parser_no_args(capsys):
"""Test no arguments""" """Test no arguments"""
sys.argv = [''] sys.argv = ['']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -33,7 +34,8 @@ def test_main_init_parser_no_args(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_init_parser_version(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_init_parser_version(capsys):
"""Test --version""" """Test --version"""
sys.argv = ['', '--version'] sys.argv = ['', '--version']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -48,7 +50,8 @@ def test_main_init_parser_version(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_main_version(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_main_version(capsys):
"""Test --version""" """Test --version"""
sys.argv = ['', '--version'] sys.argv = ['', '--version']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -63,7 +66,8 @@ def test_main_main_version(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_main_no_args(reset_globals, capsys): @pytest.mark.usefixtures("reset_globals")
def test_main_main_no_args(capsys):
"""Test with no args""" """Test with no args"""
sys.argv = [''] sys.argv = ['']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -77,7 +81,8 @@ def test_main_main_no_args(reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_main_support(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_support(capsys):
"""Test --support""" """Test --support"""
sys.argv = ['', '--support'] sys.argv = ['', '--support']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -95,8 +100,9 @@ def test_main_support(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=[]) @patch('meshtastic.util.findPorts', return_value=[])
def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals): def test_main_ch_index_no_devices(patched_find_ports, capsys):
"""Test --ch-index 1""" """Test --ch-index 1"""
sys.argv = ['', '--ch-index', '1'] sys.argv = ['', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -113,8 +119,9 @@ def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=[]) @patch('meshtastic.util.findPorts', return_value=[])
def test_main_test_no_ports(patched_find_ports, reset_globals, capsys): def test_main_test_no_ports(patched_find_ports, capsys):
"""Test --test with no hardware""" """Test --test with no hardware"""
sys.argv = ['', '--test'] sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -130,8 +137,9 @@ def test_main_test_no_ports(patched_find_ports, reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1']) @patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1'])
def test_main_test_one_port(patched_find_ports, reset_globals, capsys): def test_main_test_one_port(patched_find_ports, capsys):
"""Test --test with one fake port""" """Test --test with one fake port"""
sys.argv = ['', '--test'] sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -147,8 +155,9 @@ def test_main_test_one_port(patched_find_ports, reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.test.testAll', return_value=True) @patch('meshtastic.test.testAll', return_value=True)
def test_main_test_two_ports_success(patched_test_all, reset_globals, capsys): def test_main_test_two_ports_success(patched_test_all, capsys):
"""Test --test two fake ports and testAll() is a simulated success""" """Test --test two fake ports and testAll() is a simulated success"""
sys.argv = ['', '--test'] sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -164,8 +173,9 @@ def test_main_test_two_ports_success(patched_test_all, reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.test.testAll', return_value=False) @patch('meshtastic.test.testAll', return_value=False)
def test_main_test_two_ports_fails(patched_test_all, reset_globals, capsys): def test_main_test_two_ports_fails(patched_test_all, capsys):
"""Test --test two fake ports and testAll() is a simulated failure""" """Test --test two fake ports and testAll() is a simulated failure"""
sys.argv = ['', '--test'] sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -181,7 +191,8 @@ def test_main_test_two_ports_fails(patched_test_all, reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_main_info(capsys, caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_info(capsys, caplog):
"""Test --info""" """Test --info"""
sys.argv = ['', '--info'] sys.argv = ['', '--info']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -201,8 +212,9 @@ def test_main_info(capsys, caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('os.getlogin') @patch('os.getlogin')
def test_main_info_with_permission_error(patched_getlogin, capsys, caplog, reset_globals): def test_main_info_with_permission_error(patched_getlogin, capsys, caplog):
"""Test --info""" """Test --info"""
sys.argv = ['', '--info'] sys.argv = ['', '--info']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -224,7 +236,8 @@ def test_main_info_with_permission_error(patched_getlogin, capsys, caplog, reset
@pytest.mark.unit @pytest.mark.unit
def test_main_info_with_tcp_interface(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_info_with_tcp_interface(capsys):
"""Test --info""" """Test --info"""
sys.argv = ['', '--info', '--host', 'meshtastic.local'] sys.argv = ['', '--info', '--host', 'meshtastic.local']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -244,7 +257,7 @@ def test_main_info_with_tcp_interface(capsys, reset_globals):
# TODO: comment out ble (for now) # TODO: comment out ble (for now)
#@pytest.mark.unit #@pytest.mark.unit
#def test_main_info_with_ble_interface(capsys, reset_globals): #def test_main_info_with_ble_interface(capsys):
# """Test --info""" # """Test --info"""
# sys.argv = ['', '--info', '--ble', 'foo'] # sys.argv = ['', '--info', '--ble', 'foo']
# Globals.getInstance().set_args(sys.argv) # Globals.getInstance().set_args(sys.argv)
@@ -263,7 +276,8 @@ def test_main_info_with_tcp_interface(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_no_proto(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_no_proto(capsys):
"""Test --noproto (using --info for output)""" """Test --noproto (using --info for output)"""
sys.argv = ['', '--info', '--noproto'] sys.argv = ['', '--info', '--noproto']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -275,6 +289,7 @@ def test_main_no_proto(capsys, reset_globals):
# Override the time.sleep so there is no loop # Override the time.sleep so there is no loop
def my_sleep(amount): def my_sleep(amount):
print(f'amount:{amount}')
sys.exit(0) sys.exit(0)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface): with patch('meshtastic.serial_interface.SerialInterface', return_value=iface):
@@ -290,7 +305,8 @@ def test_main_no_proto(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_info_with_seriallog_stdout(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_info_with_seriallog_stdout(capsys):
"""Test --info""" """Test --info"""
sys.argv = ['', '--info', '--seriallog', 'stdout'] sys.argv = ['', '--info', '--seriallog', 'stdout']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -309,7 +325,8 @@ def test_main_info_with_seriallog_stdout(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_info_with_seriallog_output_txt(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_info_with_seriallog_output_txt(capsys):
"""Test --info""" """Test --info"""
sys.argv = ['', '--info', '--seriallog', 'output.txt'] sys.argv = ['', '--info', '--seriallog', 'output.txt']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -330,7 +347,8 @@ def test_main_info_with_seriallog_output_txt(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_qr(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_qr(capsys):
"""Test --qr""" """Test --qr"""
sys.argv = ['', '--qr'] sys.argv = ['', '--qr']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -349,7 +367,8 @@ def test_main_qr(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_nodes(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_nodes(capsys):
"""Test --nodes""" """Test --nodes"""
sys.argv = ['', '--nodes'] sys.argv = ['', '--nodes']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -368,7 +387,8 @@ def test_main_nodes(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_set_owner_to_bob(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_set_owner_to_bob(capsys):
"""Test --set-owner bob""" """Test --set-owner bob"""
sys.argv = ['', '--set-owner', 'bob'] sys.argv = ['', '--set-owner', 'bob']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -384,7 +404,8 @@ def test_main_set_owner_to_bob(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_set_ham_to_KI123(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_set_ham_to_KI123(capsys):
"""Test --set-ham KI123""" """Test --set-ham KI123"""
sys.argv = ['', '--set-ham', 'KI123'] sys.argv = ['', '--set-ham', 'KI123']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -393,7 +414,7 @@ def test_main_set_ham_to_KI123(capsys, reset_globals):
def mock_turnOffEncryptionOnPrimaryChannel(): def mock_turnOffEncryptionOnPrimaryChannel():
print('inside mocked turnOffEncryptionOnPrimaryChannel') print('inside mocked turnOffEncryptionOnPrimaryChannel')
def mock_setOwner(name, is_licensed): def mock_setOwner(name, is_licensed):
print('inside mocked setOwner') print(f'inside mocked setOwner name:{name} is_licensed:{is_licensed}')
mocked_node.turnOffEncryptionOnPrimaryChannel.side_effect = mock_turnOffEncryptionOnPrimaryChannel mocked_node.turnOffEncryptionOnPrimaryChannel.side_effect = mock_turnOffEncryptionOnPrimaryChannel
mocked_node.setOwner.side_effect = mock_setOwner mocked_node.setOwner.side_effect = mock_setOwner
@@ -412,7 +433,8 @@ def test_main_set_ham_to_KI123(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_reboot(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_reboot(capsys):
"""Test --reboot""" """Test --reboot"""
sys.argv = ['', '--reboot'] sys.argv = ['', '--reboot']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -435,7 +457,8 @@ def test_main_reboot(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_sendtext(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_sendtext(capsys):
"""Test --sendtext""" """Test --sendtext"""
sys.argv = ['', '--sendtext', 'hello'] sys.argv = ['', '--sendtext', 'hello']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -443,6 +466,7 @@ def test_main_sendtext(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
def mock_sendText(text, dest, wantAck, channelIndex): def mock_sendText(text, dest, wantAck, channelIndex):
print('inside mocked sendText') print('inside mocked sendText')
print(f'{text} {dest} {wantAck} {channelIndex}')
iface.sendText.side_effect = mock_sendText iface.sendText.side_effect = mock_sendText
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -456,7 +480,8 @@ def test_main_sendtext(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_sendtext_with_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_sendtext_with_channel(capsys):
"""Test --sendtext""" """Test --sendtext"""
sys.argv = ['', '--sendtext', 'hello', '--ch-index', '1'] sys.argv = ['', '--sendtext', 'hello', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -464,6 +489,7 @@ def test_main_sendtext_with_channel(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
def mock_sendText(text, dest, wantAck, channelIndex): def mock_sendText(text, dest, wantAck, channelIndex):
print('inside mocked sendText') print('inside mocked sendText')
print(f'{text} {dest} {wantAck} {channelIndex}')
iface.sendText.side_effect = mock_sendText iface.sendText.side_effect = mock_sendText
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -478,7 +504,8 @@ def test_main_sendtext_with_channel(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_sendtext_with_invalid_channel(caplog, capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_sendtext_with_invalid_channel(caplog, capsys):
"""Test --sendtext""" """Test --sendtext"""
sys.argv = ['', '--sendtext', 'hello', '--ch-index', '-1'] sys.argv = ['', '--sendtext', 'hello', '--ch-index', '-1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -499,7 +526,8 @@ def test_main_sendtext_with_invalid_channel(caplog, capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_sendtext_with_invalid_channel_nine(caplog, capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_sendtext_with_invalid_channel_nine(caplog, capsys):
"""Test --sendtext""" """Test --sendtext"""
sys.argv = ['', '--sendtext', 'hello', '--ch-index', '9'] sys.argv = ['', '--sendtext', 'hello', '--ch-index', '9']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -520,7 +548,8 @@ def test_main_sendtext_with_invalid_channel_nine(caplog, capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_sendtext_with_dest(capsys, caplog, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_main_sendtext_with_dest(capsys, caplog, iface_with_nodes):
"""Test --sendtext with --dest""" """Test --sendtext with --dest"""
sys.argv = ['', '--sendtext', 'hello', '--dest', 'foo'] sys.argv = ['', '--sendtext', 'hello', '--dest', 'foo']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -546,7 +575,8 @@ def test_main_sendtext_with_dest(capsys, caplog, reset_globals, iface_with_nodes
@pytest.mark.unit @pytest.mark.unit
def test_main_sendping(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_sendping(capsys):
"""Test --sendping""" """Test --sendping"""
sys.argv = ['', '--sendping'] sys.argv = ['', '--sendping']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -554,6 +584,7 @@ def test_main_sendping(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
def mock_sendData(payload, dest, portNum, wantAck, wantResponse): def mock_sendData(payload, dest, portNum, wantAck, wantResponse):
print('inside mocked sendData') print('inside mocked sendData')
print(f'{payload} {dest} {portNum} {wantAck} {wantResponse}')
iface.sendData.side_effect = mock_sendData iface.sendData.side_effect = mock_sendData
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -567,7 +598,8 @@ def test_main_sendping(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_setlat(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_setlat(capsys):
"""Test --sendlat""" """Test --sendlat"""
sys.argv = ['', '--setlat', '37.5'] sys.argv = ['', '--setlat', '37.5']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -580,6 +612,7 @@ def test_main_setlat(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
def mock_sendPosition(lat, lon, alt): def mock_sendPosition(lat, lon, alt):
print('inside mocked sendPosition') print('inside mocked sendPosition')
print(f'{lat} {lon} {alt}')
iface.sendPosition.side_effect = mock_sendPosition iface.sendPosition.side_effect = mock_sendPosition
iface.localNode.return_value = mocked_node iface.localNode.return_value = mocked_node
@@ -596,7 +629,8 @@ def test_main_setlat(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_setlon(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_setlon(capsys):
"""Test --setlon""" """Test --setlon"""
sys.argv = ['', '--setlon', '-122.1'] sys.argv = ['', '--setlon', '-122.1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -609,6 +643,7 @@ def test_main_setlon(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
def mock_sendPosition(lat, lon, alt): def mock_sendPosition(lat, lon, alt):
print('inside mocked sendPosition') print('inside mocked sendPosition')
print(f'{lat} {lon} {alt}')
iface.sendPosition.side_effect = mock_sendPosition iface.sendPosition.side_effect = mock_sendPosition
iface.localNode.return_value = mocked_node iface.localNode.return_value = mocked_node
@@ -625,7 +660,8 @@ def test_main_setlon(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_setalt(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_setalt(capsys):
"""Test --setalt""" """Test --setalt"""
sys.argv = ['', '--setalt', '51'] sys.argv = ['', '--setalt', '51']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -638,6 +674,7 @@ def test_main_setalt(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
def mock_sendPosition(lat, lon, alt): def mock_sendPosition(lat, lon, alt):
print('inside mocked sendPosition') print('inside mocked sendPosition')
print(f'{lat} {lon} {alt}')
iface.sendPosition.side_effect = mock_sendPosition iface.sendPosition.side_effect = mock_sendPosition
iface.localNode.return_value = mocked_node iface.localNode.return_value = mocked_node
@@ -654,7 +691,8 @@ def test_main_setalt(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_set_team_valid(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_set_team_valid(capsys):
"""Test --set-team""" """Test --set-team"""
sys.argv = ['', '--set-team', 'CYAN'] sys.argv = ['', '--set-team', 'CYAN']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -662,6 +700,7 @@ def test_main_set_team_valid(capsys, reset_globals):
mocked_node = MagicMock(autospec=Node) mocked_node = MagicMock(autospec=Node)
def mock_setOwner(team): def mock_setOwner(team):
print('inside mocked setOwner') print('inside mocked setOwner')
print(f'{team}')
mocked_node.setOwner.side_effect = mock_setOwner mocked_node.setOwner.side_effect = mock_setOwner
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
@@ -682,7 +721,8 @@ def test_main_set_team_valid(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_set_team_invalid(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_set_team_invalid(capsys):
"""Test --set-team using an invalid team name""" """Test --set-team using an invalid team name"""
sys.argv = ['', '--set-team', 'NOTCYAN'] sys.argv = ['', '--set-team', 'NOTCYAN']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -705,7 +745,8 @@ def test_main_set_team_invalid(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_seturl(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_seturl(capsys):
"""Test --seturl (url used below is what is generated after a factory_reset)""" """Test --seturl (url used below is what is generated after a factory_reset)"""
sys.argv = ['', '--seturl', 'https://www.meshtastic.org/d/#CgUYAyIBAQ'] sys.argv = ['', '--seturl', 'https://www.meshtastic.org/d/#CgUYAyIBAQ']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -720,7 +761,8 @@ def test_main_seturl(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_set_valid(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_set_valid(capsys):
"""Test --set with valid field""" """Test --set with valid field"""
sys.argv = ['', '--set', 'wifi_ssid', 'foo'] sys.argv = ['', '--set', 'wifi_ssid', 'foo']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -740,7 +782,8 @@ def test_main_set_valid(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_set_with_invalid(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_set_with_invalid(capsys):
"""Test --set with invalid field""" """Test --set with invalid field"""
sys.argv = ['', '--set', 'foo', 'foo'] sys.argv = ['', '--set', 'foo', 'foo']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -765,7 +808,8 @@ def test_main_set_with_invalid(capsys, reset_globals):
# TODO: write some negative --configure tests # TODO: write some negative --configure tests
@pytest.mark.unit @pytest.mark.unit
def test_main_configure(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_configure(capsys):
"""Test --configure with valid file""" """Test --configure with valid file"""
sys.argv = ['', '--configure', 'example_config.yaml'] sys.argv = ['', '--configure', 'example_config.yaml']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -790,7 +834,8 @@ def test_main_configure(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_add_valid(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_add_valid(capsys):
"""Test --ch-add with valid channel name, and that channel name does not already exist""" """Test --ch-add with valid channel name, and that channel name does not already exist"""
sys.argv = ['', '--ch-add', 'testing'] sys.argv = ['', '--ch-add', 'testing']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -817,7 +862,8 @@ def test_main_ch_add_valid(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_add_invalid_name_too_long(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_add_invalid_name_too_long(capsys):
"""Test --ch-add with invalid channel name, name too long""" """Test --ch-add with invalid channel name, name too long"""
sys.argv = ['', '--ch-add', 'testingtestingtesting'] sys.argv = ['', '--ch-add', 'testingtestingtesting']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -847,7 +893,8 @@ def test_main_ch_add_invalid_name_too_long(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_add_but_name_already_exists(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_add_but_name_already_exists(capsys):
"""Test --ch-add with a channel name that already exists""" """Test --ch-add with a channel name that already exists"""
sys.argv = ['', '--ch-add', 'testing'] sys.argv = ['', '--ch-add', 'testing']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -872,7 +919,8 @@ def test_main_ch_add_but_name_already_exists(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_add_but_no_more_channels(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_add_but_no_more_channels(capsys):
"""Test --ch-add with but there are no more channels""" """Test --ch-add with but there are no more channels"""
sys.argv = ['', '--ch-add', 'testing'] sys.argv = ['', '--ch-add', 'testing']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -899,7 +947,8 @@ def test_main_ch_add_but_no_more_channels(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_del(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_del(capsys):
"""Test --ch-del with valid secondary channel to be deleted""" """Test --ch-del with valid secondary channel to be deleted"""
sys.argv = ['', '--ch-del', '--ch-index', '1'] sys.argv = ['', '--ch-del', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -919,7 +968,8 @@ def test_main_ch_del(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_del_no_ch_index_specified(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_del_no_ch_index_specified(capsys):
"""Test --ch-del without a valid ch-index""" """Test --ch-del without a valid ch-index"""
sys.argv = ['', '--ch-del'] sys.argv = ['', '--ch-del']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -942,7 +992,8 @@ def test_main_ch_del_no_ch_index_specified(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_del_primary_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_del_primary_channel(capsys):
"""Test --ch-del on ch-index=0""" """Test --ch-del on ch-index=0"""
sys.argv = ['', '--ch-del', '--ch-index', '0'] sys.argv = ['', '--ch-del', '--ch-index', '0']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -966,7 +1017,8 @@ def test_main_ch_del_primary_channel(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_enable_valid_secondary_channel(capsys):
"""Test --ch-enable with --ch-index""" """Test --ch-enable with --ch-index"""
sys.argv = ['', '--ch-enable', '--ch-index', '1'] sys.argv = ['', '--ch-enable', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -987,7 +1039,8 @@ def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_disable_valid_secondary_channel(capsys):
"""Test --ch-disable with --ch-index""" """Test --ch-disable with --ch-index"""
sys.argv = ['', '--ch-disable', '--ch-index', '1'] sys.argv = ['', '--ch-disable', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1008,7 +1061,8 @@ def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_enable_without_a_ch_index(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_enable_without_a_ch_index(capsys):
"""Test --ch-enable without --ch-index""" """Test --ch-enable without --ch-index"""
sys.argv = ['', '--ch-enable'] sys.argv = ['', '--ch-enable']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1032,7 +1086,8 @@ def test_main_ch_enable_without_a_ch_index(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_enable_primary_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_enable_primary_channel(capsys):
"""Test --ch-enable with --ch-index = 0""" """Test --ch-enable with --ch-index = 0"""
sys.argv = ['', '--ch-enable', '--ch-index', '0'] sys.argv = ['', '--ch-enable', '--ch-index', '0']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1056,7 +1111,8 @@ def test_main_ch_enable_primary_channel(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_range_options(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_range_options(capsys):
"""Test changing the various range options.""" """Test changing the various range options."""
range_options = ['--ch-longslow', '--ch-longfast', '--ch-mediumslow', range_options = ['--ch-longslow', '--ch-longfast', '--ch-mediumslow',
'--ch-mediumfast', '--ch-shortslow', '--ch-shortfast'] '--ch-mediumfast', '--ch-shortslow', '--ch-shortfast']
@@ -1079,7 +1135,8 @@ def test_main_ch_range_options(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_longsfast_on_non_primary_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_longsfast_on_non_primary_channel(capsys):
"""Test --ch-longfast --ch-index 1""" """Test --ch-longfast --ch-index 1"""
sys.argv = ['', '--ch-longfast', '--ch-index', '1'] sys.argv = ['', '--ch-longfast', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1116,7 +1173,8 @@ def test_main_ch_longsfast_on_non_primary_channel(capsys, reset_globals):
# POS_TIMESTAMP 256 # POS_TIMESTAMP 256
@pytest.mark.unit @pytest.mark.unit
def test_main_pos_fields_no_args(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_pos_fields_no_args(capsys):
"""Test --pos-fields no args (which shows settings)""" """Test --pos-fields no args (which shows settings)"""
sys.argv = ['', '--pos-fields'] sys.argv = ['', '--pos-fields']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1146,7 +1204,8 @@ def test_main_pos_fields_no_args(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_pos_fields_arg_of_zero(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_pos_fields_arg_of_zero(capsys):
"""Test --pos-fields an arg of 0 (which shows list)""" """Test --pos-fields an arg of 0 (which shows list)"""
sys.argv = ['', '--pos-fields', '0'] sys.argv = ['', '--pos-fields', '0']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1179,7 +1238,8 @@ def test_main_pos_fields_arg_of_zero(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_pos_fields_valid_values(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_pos_fields_valid_values(capsys):
"""Test --pos-fields with valid values""" """Test --pos-fields with valid values"""
sys.argv = ['', '--pos-fields', 'POS_GEO_SEP', 'POS_ALT_MSL'] sys.argv = ['', '--pos-fields', 'POS_GEO_SEP', 'POS_ALT_MSL']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1205,7 +1265,8 @@ def test_main_pos_fields_valid_values(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_get_with_valid_values(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_get_with_valid_values(capsys):
"""Test --get with valid values (with string, number, boolean)""" """Test --get with valid values (with string, number, boolean)"""
sys.argv = ['', '--get', 'ls_secs', '--get', 'wifi_ssid', '--get', 'fixed_position'] sys.argv = ['', '--get', 'ls_secs', '--get', 'wifi_ssid', '--get', 'fixed_position']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1229,7 +1290,8 @@ def test_main_get_with_valid_values(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_get_with_invalid(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_get_with_invalid(capsys):
"""Test --get with invalid field""" """Test --get with invalid field"""
sys.argv = ['', '--get', 'foo'] sys.argv = ['', '--get', 'foo']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1254,7 +1316,8 @@ def test_main_get_with_invalid(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_setchan(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_setchan(capsys):
"""Test --setchan (deprecated)""" """Test --setchan (deprecated)"""
sys.argv = ['', '--setchan', 'a', 'b'] sys.argv = ['', '--setchan', 'a', 'b']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1271,7 +1334,8 @@ def test_main_setchan(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_onReceive_empty(caplog, reset_globals, capsys): @pytest.mark.usefixtures("reset_globals")
def test_main_onReceive_empty(caplog, capsys):
"""Test onReceive""" """Test onReceive"""
args = MagicMock() args = MagicMock()
Globals.getInstance().set_args(args) Globals.getInstance().set_args(args)
@@ -1297,7 +1361,8 @@ def test_main_onReceive_empty(caplog, reset_globals, capsys):
# } # }
@pytest.mark.unit @pytest.mark.unit
def test_main_onReceive_with_sendtext(caplog, capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_onReceive_with_sendtext(caplog, capsys):
"""Test onReceive with sendtext """Test onReceive with sendtext
The entire point of this test is to make sure the interface.close() call The entire point of this test is to make sure the interface.close() call
is made in onReceive(). is made in onReceive().
@@ -1332,7 +1397,8 @@ def test_main_onReceive_with_sendtext(caplog, capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_onReceive_with_text(caplog, capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_onReceive_with_text(caplog, capsys):
"""Test onReceive with text """Test onReceive with text
""" """
args = MagicMock() args = MagicMock()
@@ -1371,7 +1437,8 @@ def test_main_onReceive_with_text(caplog, capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_onConnection(reset_globals, capsys): @pytest.mark.usefixtures("reset_globals")
def test_main_onConnection(capsys):
"""Test onConnection""" """Test onConnection"""
sys.argv = [''] sys.argv = ['']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1389,7 +1456,8 @@ def test_main_onConnection(reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_main_export_config(reset_globals, capsys): @pytest.mark.usefixtures("reset_globals")
def test_main_export_config(capsys):
"""Test export_config() function directly""" """Test export_config() function directly"""
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo: with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -1425,7 +1493,8 @@ position_flags: 35"""
@pytest.mark.unit @pytest.mark.unit
def test_main_export_config_called_from_main(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_export_config_called_from_main(capsys):
"""Test --export-config""" """Test --export-config"""
sys.argv = ['', '--export-config'] sys.argv = ['', '--export-config']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1441,7 +1510,8 @@ def test_main_export_config_called_from_main(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_gpio_rd_no_gpio_channel(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_gpio_rd_no_gpio_channel(capsys):
"""Test --gpio_rd with no named gpio channel""" """Test --gpio_rd with no named gpio channel"""
sys.argv = ['', '--gpio-rd', '0x10', '--dest', '!foo'] sys.argv = ['', '--gpio-rd', '0x10', '--dest', '!foo']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1459,7 +1529,8 @@ def test_main_gpio_rd_no_gpio_channel(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_gpio_rd_no_dest(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_gpio_rd_no_dest(capsys):
"""Test --gpio_rd with a named gpio channel but no dest was specified""" """Test --gpio_rd with a named gpio channel but no dest was specified"""
sys.argv = ['', '--gpio-rd', '0x2000'] sys.argv = ['', '--gpio-rd', '0x2000']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1481,7 +1552,8 @@ def test_main_gpio_rd_no_dest(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_gpio_rd(caplog, capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_gpio_rd(caplog, capsys):
"""Test --gpio_rd with a named gpio channel""" """Test --gpio_rd with a named gpio channel"""
# Note: On the Heltec v2.1, there is a GPIO pin GPIO 13 that does not have a # Note: On the Heltec v2.1, there is a GPIO pin GPIO 13 that does not have a
# red arrow (meaning ok to use for our purposes) # red arrow (meaning ok to use for our purposes)
@@ -1538,7 +1610,8 @@ def test_main_gpio_rd(caplog, capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_getPref_valid_field(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_getPref_valid_field(capsys):
"""Test getPref() with a valid field""" """Test getPref() with a valid field"""
prefs = MagicMock() prefs = MagicMock()
prefs.DESCRIPTOR.fields_by_name.get.return_value = 'ls_secs' prefs.DESCRIPTOR.fields_by_name.get.return_value = 'ls_secs'
@@ -1553,7 +1626,8 @@ def test_main_getPref_valid_field(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_getPref_valid_field_string(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_getPref_valid_field_string(capsys):
"""Test getPref() with a valid field and value as a string""" """Test getPref() with a valid field and value as a string"""
prefs = MagicMock() prefs = MagicMock()
prefs.DESCRIPTOR.fields_by_name.get.return_value = 'wifi_ssid' prefs.DESCRIPTOR.fields_by_name.get.return_value = 'wifi_ssid'
@@ -1568,7 +1642,8 @@ def test_main_getPref_valid_field_string(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_getPref_valid_field_bool(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_getPref_valid_field_bool(capsys):
"""Test getPref() with a valid field and value as a bool""" """Test getPref() with a valid field and value as a bool"""
prefs = MagicMock() prefs = MagicMock()
prefs.DESCRIPTOR.fields_by_name.get.return_value = 'fixed_position' prefs.DESCRIPTOR.fields_by_name.get.return_value = 'fixed_position'
@@ -1583,7 +1658,8 @@ def test_main_getPref_valid_field_bool(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_getPref_invalid_field(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_getPref_invalid_field(capsys):
"""Test getPref() with an invalid field""" """Test getPref() with an invalid field"""
class Field: class Field:
@@ -1614,7 +1690,8 @@ def test_main_getPref_invalid_field(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_setPref_valid_field_int(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_setPref_valid_field_int(capsys):
"""Test setPref() with a valid field""" """Test setPref() with a valid field"""
class Field: class Field:
@@ -1636,7 +1713,8 @@ def test_main_setPref_valid_field_int(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_setPref_invalid_field(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_setPref_invalid_field(capsys):
"""Test setPref() with a invalid field""" """Test setPref() with a invalid field"""
@@ -1667,7 +1745,8 @@ def test_main_setPref_invalid_field(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_set_psk_no_ch_index(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_set_psk_no_ch_index(capsys):
"""Test --ch-set psk """ """Test --ch-set psk """
sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local'] sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1686,7 +1765,8 @@ def test_main_ch_set_psk_no_ch_index(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_set_psk_with_ch_index(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_set_psk_with_ch_index(capsys):
"""Test --ch-set psk """ """Test --ch-set psk """
sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local', '--ch-index', '0'] sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local', '--ch-index', '0']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1702,7 +1782,8 @@ def test_main_ch_set_psk_with_ch_index(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_main_ch_set_name_with_ch_index(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_main_ch_set_name_with_ch_index(capsys):
"""Test --ch-set setting other than psk""" """Test --ch-set setting other than psk"""
sys.argv = ['', '--ch-set', 'name', 'foo', '--host', 'meshtastic.local', '--ch-index', '0'] sys.argv = ['', '--ch-set', 'name', 'foo', '--host', 'meshtastic.local', '--ch-index', '0']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1719,7 +1800,8 @@ def test_main_ch_set_name_with_ch_index(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_onNode(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_onNode(capsys):
"""Test onNode""" """Test onNode"""
onNode('foo') onNode('foo')
out, err = capsys.readouterr() out, err = capsys.readouterr()
@@ -1728,7 +1810,8 @@ def test_onNode(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_tunnel_no_args(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_tunnel_no_args(capsys):
"""Test tunnel no arguments""" """Test tunnel no arguments"""
sys.argv = [''] sys.argv = ['']
Globals.getInstance().set_args(sys.argv) Globals.getInstance().set_args(sys.argv)
@@ -1741,9 +1824,10 @@ def test_tunnel_no_args(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=[]) @patch('meshtastic.util.findPorts', return_value=[])
@patch('platform.system') @patch('platform.system')
def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, patched_find_ports, caplog, capsys, reset_globals): def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, caplog, capsys):
"""Test tunnel with tunnel arg (act like we are on a linux system)""" """Test tunnel with tunnel arg (act like we are on a linux system)"""
a_mock = MagicMock() a_mock = MagicMock()
a_mock.return_value = 'Linux' a_mock.return_value = 'Linux'
@@ -1763,9 +1847,10 @@ def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, patched_find_po
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=[]) @patch('meshtastic.util.findPorts', return_value=[])
@patch('platform.system') @patch('platform.system')
def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, patched_find_ports, caplog, capsys, reset_globals): def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, caplog, capsys):
"""Test tunnel with subnet arg (act like we are on a linux system)""" """Test tunnel with subnet arg (act like we are on a linux system)"""
a_mock = MagicMock() a_mock = MagicMock()
a_mock.return_value = 'Linux' a_mock.return_value = 'Linux'
@@ -1785,11 +1870,13 @@ def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, patched_find_po
@pytest.mark.unit @pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@patch('platform.system') @patch('platform.system')
def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, iface_with_nodes, capsys): def test_tunnel_tunnel_arg(mock_platform_system, caplog, iface_with_nodes, capsys):
"""Test tunnel with tunnel arg (act like we are on a linux system)""" """Test tunnel with tunnel arg (act like we are on a linux system)"""
# Override the time.sleep so there is no loop # Override the time.sleep so there is no loop
def my_sleep(amount): def my_sleep(amount):
print(f'{amount}')
sys.exit(3) sys.exit(3)
a_mock = MagicMock() a_mock = MagicMock()

View File

@@ -15,7 +15,8 @@ from ..util import Timeout
@pytest.mark.unit @pytest.mark.unit
def test_MeshInterface(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_MeshInterface(capsys):
"""Test that we can instantiate a MeshInterface""" """Test that we can instantiate a MeshInterface"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -56,7 +57,8 @@ def test_MeshInterface(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_getMyUser(reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_getMyUser(iface_with_nodes):
"""Test getMyUser()""" """Test getMyUser()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -66,7 +68,8 @@ def test_getMyUser(reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_getLongName(reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_getLongName(iface_with_nodes):
"""Test getLongName()""" """Test getLongName()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -75,7 +78,8 @@ def test_getLongName(reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_getShortName(reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_getShortName(iface_with_nodes):
"""Test getShortName().""" """Test getShortName()."""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -84,7 +88,8 @@ def test_getShortName(reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_handlePacketFromRadio_no_from(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_handlePacketFromRadio_no_from(capsys):
"""Test _handlePacketFromRadio with no 'from' in the mesh packet.""" """Test _handlePacketFromRadio with no 'from' in the mesh packet."""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
meshPacket = mesh_pb2.MeshPacket() meshPacket = mesh_pb2.MeshPacket()
@@ -95,7 +100,8 @@ def test_handlePacketFromRadio_no_from(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_handlePacketFromRadio_with_a_portnum(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_handlePacketFromRadio_with_a_portnum(caplog):
"""Test _handlePacketFromRadio with a portnum """Test _handlePacketFromRadio with a portnum
Since we have an attribute called 'from', we cannot simply 'set' it. Since we have an attribute called 'from', we cannot simply 'set' it.
Had to implement a hack just to be able to test some code. Had to implement a hack just to be able to test some code.
@@ -110,7 +116,8 @@ def test_handlePacketFromRadio_with_a_portnum(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_handlePacketFromRadio_no_portnum(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_handlePacketFromRadio_no_portnum(caplog):
"""Test _handlePacketFromRadio without a portnum""" """Test _handlePacketFromRadio without a portnum"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
meshPacket = mesh_pb2.MeshPacket() meshPacket = mesh_pb2.MeshPacket()
@@ -121,7 +128,8 @@ def test_handlePacketFromRadio_no_portnum(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_getNode_with_local(reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_getNode_with_local():
"""Test getNode""" """Test getNode"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
anode = iface.getNode(LOCAL_ADDR) anode = iface.getNode(LOCAL_ADDR)
@@ -129,7 +137,8 @@ def test_getNode_with_local(reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_getNode_not_local(reset_globals, caplog): @pytest.mark.usefixtures("reset_globals")
def test_getNode_not_local(caplog):
"""Test getNode not local""" """Test getNode not local"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
anode = MagicMock(autospec=Node) anode = MagicMock(autospec=Node)
@@ -141,7 +150,8 @@ def test_getNode_not_local(reset_globals, caplog):
@pytest.mark.unit @pytest.mark.unit
def test_getNode_not_local_timeout(reset_globals, capsys): @pytest.mark.usefixtures("reset_globals")
def test_getNode_not_local_timeout(capsys):
"""Test getNode not local, simulate timeout""" """Test getNode not local, simulate timeout"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
anode = MagicMock(autospec=Node) anode = MagicMock(autospec=Node)
@@ -157,7 +167,8 @@ def test_getNode_not_local_timeout(reset_globals, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_sendPosition(reset_globals, caplog): @pytest.mark.usefixtures("reset_globals")
def test_sendPosition(caplog):
"""Test sendPosition""" """Test sendPosition"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -167,7 +178,8 @@ def test_sendPosition(reset_globals, caplog):
@pytest.mark.unit @pytest.mark.unit
def test_close_with_heartbeatTimer(reset_globals, caplog): @pytest.mark.usefixtures("reset_globals")
def test_close_with_heartbeatTimer(caplog):
"""Test close() with heartbeatTimer""" """Test close() with heartbeatTimer"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -183,7 +195,8 @@ def test_close_with_heartbeatTimer(reset_globals, caplog):
@pytest.mark.unit @pytest.mark.unit
def test_handleFromRadio_empty_payload(reset_globals, caplog): @pytest.mark.usefixtures("reset_globals")
def test_handleFromRadio_empty_payload(caplog):
"""Test _handleFromRadio""" """Test _handleFromRadio"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -193,7 +206,8 @@ def test_handleFromRadio_empty_payload(reset_globals, caplog):
@pytest.mark.unit @pytest.mark.unit
def test_handleFromRadio_with_my_info(reset_globals, caplog): @pytest.mark.usefixtures("reset_globals")
def test_handleFromRadio_with_my_info(caplog):
"""Test _handleFromRadio with my_info""" """Test _handleFromRadio with my_info"""
# Note: I captured the '--debug --info' for the bytes below. # Note: I captured the '--debug --info' for the bytes below.
# It "translates" to this: # It "translates" to this:
@@ -218,7 +232,8 @@ def test_handleFromRadio_with_my_info(reset_globals, caplog):
@pytest.mark.unit @pytest.mark.unit
def test_handleFromRadio_with_node_info(reset_globals, caplog, capsys): @pytest.mark.usefixtures("reset_globals")
def test_handleFromRadio_with_node_info(caplog, capsys):
"""Test _handleFromRadio with node_info""" """Test _handleFromRadio with node_info"""
# Note: I captured the '--debug --info' for the bytes below. # Note: I captured the '--debug --info' for the bytes below.
# It "translates" to this: # It "translates" to this:
@@ -254,7 +269,8 @@ def test_handleFromRadio_with_node_info(reset_globals, caplog, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_handleFromRadio_with_node_info_tbeam1(reset_globals, caplog, capsys): @pytest.mark.usefixtures("reset_globals")
def test_handleFromRadio_with_node_info_tbeam1(caplog, capsys):
"""Test _handleFromRadio with node_info""" """Test _handleFromRadio with node_info"""
# Note: Captured the '--debug --info' for the bytes below. # Note: Captured the '--debug --info' for the bytes below.
# pylint: disable=C0301 # pylint: disable=C0301
@@ -277,7 +293,8 @@ def test_handleFromRadio_with_node_info_tbeam1(reset_globals, caplog, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_handleFromRadio_with_node_info_tbeam_with_bad_data(reset_globals, caplog, capsys): @pytest.mark.usefixtures("reset_globals")
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""" """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. # Note: Captured the '--debug --info' for the bytes below.
from_radio_bytes = b'"\x17\x08\xdc\x8a\x8a\xae\x02\x12\x08"\x06\x00\x00\x00\x00\x00\x00\x1a\x00=\x00\x00\xb8@' from_radio_bytes = b'"\x17\x08\xdc\x8a\x8a\xae\x02\x12\x08"\x06\x00\x00\x00\x00\x00\x00\x1a\x00=\x00\x00\xb8@'
@@ -288,7 +305,8 @@ def test_handleFromRadio_with_node_info_tbeam_with_bad_data(reset_globals, caplo
@pytest.mark.unit @pytest.mark.unit
def test_MeshInterface_sendToRadioImpl(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_MeshInterface_sendToRadioImpl(caplog):
"""Test _sendToRadioImp()""" """Test _sendToRadioImp()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -298,7 +316,8 @@ def test_MeshInterface_sendToRadioImpl(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_MeshInterface_sendToRadio_no_proto(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_MeshInterface_sendToRadio_no_proto(caplog):
"""Test sendToRadio()""" """Test sendToRadio()"""
iface = MeshInterface() iface = MeshInterface()
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -308,7 +327,8 @@ def test_MeshInterface_sendToRadio_no_proto(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendData_too_long(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendData_too_long(caplog):
"""Test when data payload is too big""" """Test when data payload is too big"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
some_large_text = b'This is a long text that will be too long for send text.' some_large_text = b'This is a long text that will be too long for send text.'
@@ -332,7 +352,8 @@ def test_sendData_too_long(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendData_unknown_app(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendData_unknown_app(capsys):
"""Test sendData when unknown app""" """Test sendData when unknown app"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with pytest.raises(SystemExit) as pytest_wrapped_e: with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -345,7 +366,8 @@ def test_sendData_unknown_app(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendPosition_with_a_position(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPosition_with_a_position(caplog):
"""Test sendPosition when lat/long/alt""" """Test sendPosition when lat/long/alt"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -356,7 +378,8 @@ def test_sendPosition_with_a_position(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_no_destination(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_no_destination(capsys):
"""Test _sendPacket()""" """Test _sendPacket()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with pytest.raises(SystemExit) as pytest_wrapped_e: with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -369,7 +392,8 @@ def test_sendPacket_with_no_destination(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_as_int(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_as_int(caplog):
"""Test _sendPacket() with int as a destination""" """Test _sendPacket() with int as a destination"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -379,7 +403,8 @@ def test_sendPacket_with_destination_as_int(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_starting_with_a_bang(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_starting_with_a_bang(caplog):
"""Test _sendPacket() with int as a destination""" """Test _sendPacket() with int as a destination"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -389,7 +414,8 @@ def test_sendPacket_with_destination_starting_with_a_bang(caplog, reset_globals)
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_as_BROADCAST_ADDR(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_as_BROADCAST_ADDR(caplog):
"""Test _sendPacket() with BROADCAST_ADDR as a destination""" """Test _sendPacket() with BROADCAST_ADDR as a destination"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
@@ -399,7 +425,8 @@ def test_sendPacket_with_destination_as_BROADCAST_ADDR(caplog, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_as_LOCAL_ADDR_no_myInfo(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_as_LOCAL_ADDR_no_myInfo(capsys):
"""Test _sendPacket() with LOCAL_ADDR as a destination with no myInfo""" """Test _sendPacket() with LOCAL_ADDR as a destination with no myInfo"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with pytest.raises(SystemExit) as pytest_wrapped_e: with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -413,11 +440,13 @@ def test_sendPacket_with_destination_as_LOCAL_ADDR_no_myInfo(capsys, reset_globa
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_as_LOCAL_ADDR_with_myInfo(caplog, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_as_LOCAL_ADDR_with_myInfo(caplog):
"""Test _sendPacket() with LOCAL_ADDR as a destination with myInfo""" """Test _sendPacket() with LOCAL_ADDR as a destination with myInfo"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
myInfo = MagicMock() myInfo = MagicMock()
iface.myInfo = myInfo iface.myInfo = myInfo
iface.myInfo.my_node_num = 1
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
meshPacket = mesh_pb2.MeshPacket() meshPacket = mesh_pb2.MeshPacket()
iface._sendPacket(meshPacket, destinationId=LOCAL_ADDR) iface._sendPacket(meshPacket, destinationId=LOCAL_ADDR)
@@ -425,7 +454,8 @@ def test_sendPacket_with_destination_as_LOCAL_ADDR_with_myInfo(caplog, reset_glo
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_is_blank_with_nodes(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_is_blank_with_nodes(capsys, iface_with_nodes):
"""Test _sendPacket() with '' as a destination with myInfo""" """Test _sendPacket() with '' as a destination with myInfo"""
iface = iface_with_nodes iface = iface_with_nodes
meshPacket = mesh_pb2.MeshPacket() meshPacket = mesh_pb2.MeshPacket()
@@ -439,7 +469,8 @@ def test_sendPacket_with_destination_is_blank_with_nodes(capsys, reset_globals,
@pytest.mark.unit @pytest.mark.unit
def test_sendPacket_with_destination_is_blank_without_nodes(caplog, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_sendPacket_with_destination_is_blank_without_nodes(caplog, iface_with_nodes):
"""Test _sendPacket() with '' as a destination with myInfo""" """Test _sendPacket() with '' as a destination with myInfo"""
iface = iface_with_nodes iface = iface_with_nodes
iface.nodes = None iface.nodes = None
@@ -450,7 +481,8 @@ def test_sendPacket_with_destination_is_blank_without_nodes(caplog, reset_global
@pytest.mark.unit @pytest.mark.unit
def test_getMyNodeInfo(reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_getMyNodeInfo():
"""Test getMyNodeInfo()""" """Test getMyNodeInfo()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
anode = iface.getNode(LOCAL_ADDR) anode = iface.getNode(LOCAL_ADDR)
@@ -464,7 +496,8 @@ def test_getMyNodeInfo(reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_generatePacketId(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_generatePacketId(capsys):
"""Test _generatePacketId() when no currentPacketId (not connected)""" """Test _generatePacketId() when no currentPacketId (not connected)"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
# not sure when this condition would ever happen... but we can simulate it # not sure when this condition would ever happen... but we can simulate it
@@ -479,7 +512,8 @@ def test_generatePacketId(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_fixupPosition_empty_pos(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_fixupPosition_empty_pos():
"""Test _fixupPosition()""" """Test _fixupPosition()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
pos = {} pos = {}
@@ -488,7 +522,8 @@ def test_fixupPosition_empty_pos(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_fixupPosition_no_changes_needed(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_fixupPosition_no_changes_needed():
"""Test _fixupPosition()""" """Test _fixupPosition()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
pos = {"latitude": 101, "longitude": 102} pos = {"latitude": 101, "longitude": 102}
@@ -497,7 +532,8 @@ def test_fixupPosition_no_changes_needed(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_fixupPosition(capsys, reset_globals): @pytest.mark.usefixtures("reset_globals")
def test_fixupPosition():
"""Test _fixupPosition()""" """Test _fixupPosition()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
pos = {"latitudeI": 1010000000, "longitudeI": 1020000000} pos = {"latitudeI": 1010000000, "longitudeI": 1020000000}
@@ -509,7 +545,8 @@ def test_fixupPosition(capsys, reset_globals):
@pytest.mark.unit @pytest.mark.unit
def test_nodeNumToId(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_nodeNumToId(iface_with_nodes):
"""Test _nodeNumToId()""" """Test _nodeNumToId()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -518,7 +555,8 @@ def test_nodeNumToId(capsys, reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_nodeNumToId_not_found(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_nodeNumToId_not_found(iface_with_nodes):
"""Test _nodeNumToId()""" """Test _nodeNumToId()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -527,7 +565,8 @@ def test_nodeNumToId_not_found(capsys, reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_nodeNumToId_to_all(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_nodeNumToId_to_all(iface_with_nodes):
"""Test _nodeNumToId()""" """Test _nodeNumToId()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -536,7 +575,8 @@ def test_nodeNumToId_to_all(capsys, reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_getOrCreateByNum_minimal(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_getOrCreateByNum_minimal(iface_with_nodes):
"""Test _getOrCreateByNum()""" """Test _getOrCreateByNum()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -545,7 +585,8 @@ def test_getOrCreateByNum_minimal(capsys, reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_getOrCreateByNum_not_found(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_getOrCreateByNum_not_found(iface_with_nodes):
"""Test _getOrCreateByNum()""" """Test _getOrCreateByNum()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -555,7 +596,8 @@ def test_getOrCreateByNum_not_found(capsys, reset_globals, iface_with_nodes):
@pytest.mark.unit @pytest.mark.unit
def test_getOrCreateByNum(capsys, reset_globals, iface_with_nodes): @pytest.mark.usefixtures("reset_globals")
def test_getOrCreateByNum(iface_with_nodes):
"""Test _getOrCreateByNum()""" """Test _getOrCreateByNum()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -581,7 +623,7 @@ def test_exit_with_exception(caplog):
@pytest.mark.unit @pytest.mark.unit
def test_showNodes_exclude_self(capsys, caplog, reset_globals, iface_with_nodes): def test_showNodes_exclude_self(capsys, caplog, iface_with_nodes):
"""Test that we hit that continue statement""" """Test that we hit that continue statement"""
with caplog.at_level(logging.DEBUG): with caplog.at_level(logging.DEBUG):
iface = iface_with_nodes iface = iface_with_nodes
@@ -592,7 +634,7 @@ def test_showNodes_exclude_self(capsys, caplog, reset_globals, iface_with_nodes)
@pytest.mark.unitslow @pytest.mark.unitslow
def test_waitForConfig(caplog, capsys): def test_waitForConfig(capsys):
"""Test waitForConfig()""" """Test waitForConfig()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
# override how long to wait # override how long to wait
@@ -606,7 +648,7 @@ def test_waitForConfig(caplog, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_waitConnected_raises_an_exception(caplog, capsys): def test_waitConnected_raises_an_exception(capsys):
"""Test waitConnected()""" """Test waitConnected()"""
iface = MeshInterface(noProto=True) iface = MeshInterface(noProto=True)
with pytest.raises(Exception) as pytest_wrapped_e: with pytest.raises(Exception) as pytest_wrapped_e:
@@ -619,7 +661,7 @@ def test_waitConnected_raises_an_exception(caplog, capsys):
@pytest.mark.unit @pytest.mark.unit
def test_waitConnected_isConnected_timeout(caplog, capsys): def test_waitConnected_isConnected_timeout(capsys):
"""Test waitConnected()""" """Test waitConnected()"""
with pytest.raises(Exception) as pytest_wrapped_e: with pytest.raises(Exception) as pytest_wrapped_e:
iface = MeshInterface() iface = MeshInterface()

View File

@@ -150,7 +150,7 @@ def test_setURL_valid_URL(caplog):
@pytest.mark.unit @pytest.mark.unit
def test_setURL_valid_URL_but_no_settings(caplog, capsys): def test_setURL_valid_URL_but_no_settings(capsys):
"""Test setURL""" """Test setURL"""
iface = MagicMock(autospec=SerialInterface) iface = MagicMock(autospec=SerialInterface)
url = "https://www.meshtastic.org/d/#" url = "https://www.meshtastic.org/d/#"
@@ -430,7 +430,7 @@ def test_deleteChannel_secondary_with_admin_channel_before_testing():
@pytest.mark.unit @pytest.mark.unit
def test_getChannelByName(capsys): def test_getChannelByName():
"""Get a channel by the name.""" """Get a channel by the name."""
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -457,7 +457,7 @@ def test_getChannelByName(capsys):
@pytest.mark.unit @pytest.mark.unit
def test_getChannelByName_invalid_name(capsys): def test_getChannelByName_invalid_name():
"""Get a channel by the name but one that is not present.""" """Get a channel by the name but one that is not present."""
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -484,7 +484,7 @@ def test_getChannelByName_invalid_name(capsys):
@pytest.mark.unit @pytest.mark.unit
def test_getDisabledChannel(capsys): def test_getDisabledChannel():
"""Get the first disabled channel.""" """Get the first disabled channel."""
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -514,7 +514,7 @@ def test_getDisabledChannel(capsys):
@pytest.mark.unit @pytest.mark.unit
def test_getDisabledChannel_where_all_channels_are_used(capsys): def test_getDisabledChannel_where_all_channels_are_used():
"""Get the first disabled channel.""" """Get the first disabled channel."""
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -538,7 +538,7 @@ def test_getDisabledChannel_where_all_channels_are_used(capsys):
@pytest.mark.unit @pytest.mark.unit
def test_getAdminChannelIndex(capsys): def test_getAdminChannelIndex():
"""Get the 'admin' channel index.""" """Get the 'admin' channel index."""
anode = Node('foo', 'bar') anode = Node('foo', 'bar')
@@ -565,7 +565,7 @@ def test_getAdminChannelIndex(capsys):
@pytest.mark.unit @pytest.mark.unit
def test_getAdminChannelIndex_when_no_admin_named_channel(capsys): def test_getAdminChannelIndex_when_no_admin_named_channel():
"""Get the 'admin' channel when there is not one.""" """Get the 'admin' channel when there is not one."""
anode = Node('foo', 'bar') anode = Node('foo', 'bar')

View File

@@ -23,6 +23,10 @@ def test_SerialInterface_single_port(mocked_findPorts, mocked_serial, mocked_ope
iface.close() iface.close()
mocked_findPorts.assert_called() mocked_findPorts.assert_called()
mocked_serial.assert_called() mocked_serial.assert_called()
mocked_open.assert_called()
mock_get.assert_called()
mock_set.assert_called()
mock_sleep.assert_called()
out, err = capsys.readouterr() out, err = capsys.readouterr()
assert re.search(r'Nodes in mesh', out, re.MULTILINE) assert re.search(r'Nodes in mesh', out, re.MULTILINE)
assert re.search(r'Preferences', out, re.MULTILINE) assert re.search(r'Preferences', out, re.MULTILINE)

View File

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

View File

@@ -28,7 +28,24 @@ def test_TCPInterface(capsys):
@pytest.mark.unit @pytest.mark.unit
def test_TCPInterface_without_connecting(capsys): def test_TCPInterface_exception():
"""Test that we can instantiate a TCPInterface"""
def throw_an_exception():
raise ValueError("Fake exception.")
with patch('meshtastic.tcp_interface.TCPInterface._socket_shutdown') as mock_shutdown:
mock_shutdown.side_effect = throw_an_exception
with patch('socket.socket') as mock_socket:
iface = TCPInterface(hostname='localhost', noProto=True)
iface.myConnect()
iface.close()
assert mock_socket.called
assert mock_shutdown.called
@pytest.mark.unit
def test_TCPInterface_without_connecting():
"""Test that we can instantiate a TCPInterface with connectNow as false""" """Test that we can instantiate a TCPInterface with connectNow as false"""
with patch('socket.socket'): with patch('socket.socket'):
iface = TCPInterface(hostname='localhost', noProto=True, connectNow=False) iface = TCPInterface(hostname='localhost', noProto=True, connectNow=False)

View File

@@ -14,7 +14,7 @@ from ..globals import Globals
@pytest.mark.unit @pytest.mark.unit
@patch('platform.system') @patch('platform.system')
def test_Tunnel_on_non_linux_system(mock_platform_system, reset_globals): def test_Tunnel_on_non_linux_system(mock_platform_system):
"""Test that we cannot instantiate a Tunnel on a non Linux system""" """Test that we cannot instantiate a Tunnel on a non Linux system"""
a_mock = MagicMock() a_mock = MagicMock()
a_mock.return_value = 'notLinux' a_mock.return_value = 'notLinux'
@@ -29,7 +29,7 @@ def test_Tunnel_on_non_linux_system(mock_platform_system, reset_globals):
@pytest.mark.unit @pytest.mark.unit
@patch('platform.system') @patch('platform.system')
def test_Tunnel_without_interface(mock_platform_system, reset_globals): def test_Tunnel_without_interface(mock_platform_system):
"""Test that we can not instantiate a Tunnel without a valid interface""" """Test that we can not instantiate a Tunnel without a valid interface"""
a_mock = MagicMock() a_mock = MagicMock()
a_mock.return_value = 'Linux' a_mock.return_value = 'Linux'
@@ -41,7 +41,7 @@ def test_Tunnel_without_interface(mock_platform_system, reset_globals):
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_Tunnel_with_interface(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_Tunnel_with_interface(mock_platform_system, caplog, iface_with_nodes):
"""Test that we can not instantiate a Tunnel without a valid interface""" """Test that we can not instantiate a Tunnel without a valid interface"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -60,7 +60,7 @@ def test_Tunnel_with_interface(mock_platform_system, caplog, reset_globals, ifac
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with_nodes):
"""Test onTunnelReceive""" """Test onTunnelReceive"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -81,7 +81,7 @@ def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, reset_glob
@pytest.mark.unit @pytest.mark.unit
@patch('platform.system') @patch('platform.system')
def test_onTunnelReceive_from_someone_else(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_onTunnelReceive_from_someone_else(mock_platform_system, caplog, iface_with_nodes):
"""Test onTunnelReceive""" """Test onTunnelReceive"""
iface = iface_with_nodes iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164 iface.myInfo.my_node_num = 2475227164
@@ -101,7 +101,7 @@ def test_onTunnelReceive_from_someone_else(mock_platform_system, caplog, reset_g
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_random(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_random(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -119,7 +119,7 @@ def test_shouldFilterPacket_random(mock_platform_system, caplog, reset_globals,
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -137,7 +137,7 @@ def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, reset_glo
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_icmp(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_icmp(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -156,7 +156,7 @@ def test_shouldFilterPacket_icmp(mock_platform_system, caplog, reset_globals, if
@pytest.mark.unit @pytest.mark.unit
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_udp(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_udp(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -175,7 +175,7 @@ def test_shouldFilterPacket_udp(mock_platform_system, caplog, reset_globals, ifa
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -196,7 +196,7 @@ def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, reset_
@pytest.mark.unit @pytest.mark.unit
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_tcp(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_tcp(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -215,7 +215,7 @@ def test_shouldFilterPacket_tcp(mock_platform_system, caplog, reset_globals, ifa
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, iface_with_nodes):
"""Test _shouldFilterPacket()""" """Test _shouldFilterPacket()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -236,7 +236,7 @@ def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, reset_
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_ipToNodeId_none(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_ipToNodeId_none(mock_platform_system, caplog, iface_with_nodes):
"""Test _ipToNodeId()""" """Test _ipToNodeId()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True
@@ -252,7 +252,7 @@ def test_ipToNodeId_none(mock_platform_system, caplog, reset_globals, iface_with
@pytest.mark.unitslow @pytest.mark.unitslow
@patch('platform.system') @patch('platform.system')
def test_ipToNodeId_all(mock_platform_system, caplog, reset_globals, iface_with_nodes): def test_ipToNodeId_all(mock_platform_system, caplog, iface_with_nodes):
"""Test _ipToNodeId()""" """Test _ipToNodeId()"""
iface = iface_with_nodes iface = iface_with_nodes
iface.noProto = True iface.noProto = True

View File

@@ -242,10 +242,12 @@ def test_readnet_u16():
def test_findPorts_when_none_found(patch_comports): def test_findPorts_when_none_found(patch_comports):
"""Test findPorts()""" """Test findPorts()"""
assert not findPorts() assert not findPorts()
patch_comports.assert_called()
@pytest.mark.unitslow @pytest.mark.unitslow
def test_convert_mac_addr(): def test_convert_mac_addr():
"""Test convert_mac_addr()""" """Test convert_mac_addr()"""
assert convert_mac_addr('/c0gFyhb') == 'fd:cd:20:17:28:5b' assert convert_mac_addr('/c0gFyhb') == 'fd:cd:20:17:28:5b'
assert convert_mac_addr('fd:cd:20:17:28:5b') == 'fd:cd:20:17:28:5b'
assert convert_mac_addr('') == '' assert convert_mac_addr('') == ''

View File

@@ -27,7 +27,7 @@ from meshtastic.util import ipstr, readnet_u16
from meshtastic.globals import Globals from meshtastic.globals import Globals
def onTunnelReceive(packet, interface): def onTunnelReceive(packet, interface): # pylint: disable=W0613
"""Callback for received tunneled messages from mesh.""" """Callback for received tunneled messages from mesh."""
logging.debug(f'in onTunnelReceive()') logging.debug(f'in onTunnelReceive()')
our_globals = Globals.getInstance() our_globals = Globals.getInstance()

View File

@@ -3,6 +3,7 @@
import traceback import traceback
from queue import Queue from queue import Queue
import os import os
import re
import sys import sys
import base64 import base64
import time import time
@@ -189,17 +190,16 @@ def support_info():
print('or wish to make feature requests, visit:') print('or wish to make feature requests, visit:')
print('https://github.com/meshtastic/Meshtastic-python/issues') print('https://github.com/meshtastic/Meshtastic-python/issues')
print('When adding an issue, be sure to include the following info:') print('When adding an issue, be sure to include the following info:')
print(' System: {0}'.format(platform.system())) print(f' System: {platform.system()}')
print(' Platform: {0}'.format(platform.platform())) print(f' Platform: {platform.platform()}')
print(' Release: {0}'.format(platform.uname().release)) print(f' Release: {platform.uname().release}')
print(' Machine: {0}'.format(platform.uname().machine)) print(f' Machine: {platform.uname().machine}')
print(' Encoding (stdin): {0}'.format(sys.stdin.encoding)) print(f' Encoding (stdin): {sys.stdin.encoding}')
print(' Encoding (stdout): {0}'.format(sys.stdout.encoding)) print(f' Encoding (stdout): {sys.stdout.encoding}')
the_version = pkg_resources.get_distribution("meshtastic").version the_version = pkg_resources.get_distribution("meshtastic").version
print(' meshtastic: v{0}'.format(the_version)) print(f' meshtastic: v{the_version}')
print(' Executable: {0}'.format(sys.argv[0])) print(f' Executable: {sys.argv[0]}')
print(' Python: {0} {1} {2}'.format(platform.python_version(), print(f' Python: {platform.python_version()} {platform.python_implementation()} {platform.python_compiler()}')
platform.python_implementation(), platform.python_compiler()))
print('') print('')
print('Please add the output from the command: meshtastic --info') print('Please add the output from the command: meshtastic --info')
@@ -221,12 +221,12 @@ def remove_keys_from_dict(keys, adict):
def hexstr(barray): def hexstr(barray):
"""Print a string of hex digits""" """Print a string of hex digits"""
return ":".join('{:02x}'.format(x) for x in barray) return ":".join(f'{x:02x}' for x in barray)
def ipstr(barray): def ipstr(barray):
"""Print a string of ip digits""" """Print a string of ip digits"""
return ".".join('{}'.format(x) for x in barray) return ".".join(f'{x}' for x in barray)
def readnet_u16(p, offset): def readnet_u16(p, offset):
@@ -239,5 +239,7 @@ def convert_mac_addr(val):
val - base64 encoded value (ex: '/c0gFyhb')) val - base64 encoded value (ex: '/c0gFyhb'))
returns: a string formatted like a mac address (ex: 'fd:cd:20:17:28:5b') returns: a string formatted like a mac address (ex: 'fd:cd:20:17:28:5b')
""" """
val_as_bytes = base64.b64decode(val) if not re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", val):
return hexstr(val_as_bytes) val_as_bytes = base64.b64decode(val)
return hexstr(val_as_bytes)
return val

View File

@@ -2,6 +2,9 @@
addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples and not smokevirt" addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples and not smokevirt"
filterwarnings =
ignore::DeprecationWarning
markers = markers =
unit: marks tests as unit tests unit: marks tests as unit tests
unitslow: marks slow unit tests unitslow: marks slow unit tests

View File

@@ -4,7 +4,6 @@ protobuf
dotmap dotmap
pexpect pexpect
pyqrcode pyqrcode
pygatt
tabulate tabulate
timeago timeago
webencodings webencodings
@@ -18,3 +17,4 @@ pyyaml
pytap2 pytap2
pdoc3 pdoc3
pypubsub pypubsub
pygatt; platform_system == "Linux"

View File

@@ -12,7 +12,7 @@ with open("README.md", "r") as fh:
# This call to setup() does all the work # This call to setup() does all the work
setup( setup(
name="meshtastic", name="meshtastic",
version="1.2.53", version="1.2.54",
description="Python API & client shell for talking to Meshtastic devices", description="Python API & client shell for talking to Meshtastic devices",
long_description=long_description, long_description=long_description,
long_description_content_type="text/markdown", long_description_content_type="text/markdown",
@@ -33,7 +33,8 @@ setup(
include_package_data=True, include_package_data=True,
install_requires=["pyserial>=3.4", "protobuf>=3.13.0", install_requires=["pyserial>=3.4", "protobuf>=3.13.0",
"pypubsub>=4.0.3", "dotmap>=1.3.14", "pexpect>=4.6.0", "pyqrcode>=1.2.1", "pypubsub>=4.0.3", "dotmap>=1.3.14", "pexpect>=4.6.0", "pyqrcode>=1.2.1",
"pygatt>=4.0.5", "tabulate>=0.8.9", "timeago>=1.0.15", "pyyaml"], "tabulate>=0.8.9", "timeago>=1.0.15", "pyyaml",
"pygatt>=4.0.5 ; platform_system=='Linux'"],
extras_require={ extras_require={
'tunnel': ["pytap2>=2.0.0"] 'tunnel': ["pytap2>=2.0.0"]
}, },