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
# --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]

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 timeago
import google.protobuf.json_format
import pygatt
from pubsub import pub
from dotmap import DotMap
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,
channel_pb2, radioconfig_pb2, util)
# Note: To follow PEP224, comments should be after the module variable.
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
# text.
logging.debug(f'in _onTextReceive() asDict:{asDict}')
try:
asBytes = asDict["decoded"]["payload"]
asDict["decoded"]["text"] = asBytes.decode("utf-8")
@@ -139,22 +140,30 @@ def _onTextReceive(iface, asDict):
def _onPositionReceive(iface, asDict):
"""Special auto parsing for received messages"""
p = asDict["decoded"]["position"]
iface._fixupPosition(p)
# update node DB as needed
iface._getOrCreateByNum(asDict["from"])["position"] = p
logging.debug(f'in _onPositionReceive() asDict:{asDict}')
if 'decoded' in asDict:
if 'position' in asDict['decoded'] and 'from' in asDict:
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):
"""Special auto parsing for received messages"""
p = asDict["decoded"]["user"]
# decode user protobufs and update nodedb, provide decoded version as "position" in the published msg
# update node DB as needed
n = iface._getOrCreateByNum(asDict["from"])
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)
logging.debug(f'in _onNodeInfoReceive() asDict:{asDict}')
if 'decoded' in asDict:
if 'user' in asDict['decoded'] and 'from' in asDict:
p = asDict["decoded"]["user"]
# decode user protobufs and update nodedb, provide decoded version as "position" in the published msg
# update node DB as needed
n = iface._getOrCreateByNum(asDict["from"])
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):

View File

@@ -40,7 +40,7 @@ def onReceive(packet, interface):
rxSnr = packet['rxSnr']
hopLimit = packet['hopLimit']
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)
interface.sendText(reply)
@@ -48,7 +48,7 @@ def onReceive(packet, interface):
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"""
print(f"Connection changed: {topic.getName()}")
@@ -192,7 +192,7 @@ def onConnected(interface):
else:
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")
interface.getNode(args.dest).writeConfig()

View File

@@ -1,10 +1,16 @@
"""Bluetooth interface
"""
import logging
import pygatt
import platform
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
TORADIO_UUID = "f75c76d2-129e-4dad-a1dd-7866124401e7"
@@ -16,6 +22,8 @@ class BLEInterface(MeshInterface):
"""A not quite ready - FIXME - BLE interface to devices"""
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
if not noProto:
self.adapter = pygatt.GATTToolBackend() # BGAPIBackend()
@@ -31,7 +39,7 @@ class BLEInterface(MeshInterface):
self._readFromRadio() # read the initial responses
def handle_data(handle, data):
def handle_data(handle, data): # pylint: disable=W0613
self._handleFromRadio(data)
if self.device:

View File

@@ -73,7 +73,7 @@ class MeshInterface:
logging.error(f'Traceback: {traceback}')
self.close()
def showInfo(self, file=sys.stdout):
def showInfo(self, file=sys.stdout): # pylint: disable=W0613
"""Show human readable summary about this object"""
owner = f"Owner: {self.getLongName()} ({self.getShortName()})"
myinfo = ''
@@ -100,7 +100,7 @@ class MeshInterface:
print(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"""
def formatFloat(value, precision=2, unit=''):
"""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)
p = bytearray([START2] * 32)
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()
@@ -90,8 +89,7 @@ class StreamInterface(MeshInterface):
self.stream.write(b)
self.stream.flush()
# 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):
"""Read an array of bytes from our stream"""

View File

@@ -33,6 +33,12 @@ class TCPInterface(StreamInterface):
StreamInterface.__init__(self, debugOut=debugOut, noProto=noProto,
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):
"""Connect to socket"""
server_address = (self.hostname, self.portNumber)
@@ -48,7 +54,7 @@ class TCPInterface(StreamInterface):
self._wantExit = True
if not self.socket is None:
try:
self.socket.shutdown(socket.SHUT_RDWR)
self._socket_shutdown()
except:
pass # Ignore errors in shutdown, because we might have a race with the server
self.socket.close()

View File

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

View File

@@ -1,12 +1,15 @@
"""Meshtastic unit tests for ble_interface.py"""
from unittest.mock import patch
import pytest
from ..ble_interface import BLEInterface
@pytest.mark.unit
def test_BLEInterface():
@patch('platform.system', return_value='Linux')
def test_BLEInterface(mock_platform):
"""Test that we can instantiate a BLEInterface"""
iface = BLEInterface('foo', debugOut=True, noProto=True)
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
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"""
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
def test_main_init_parser_version(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_init_parser_version(capsys):
"""Test --version"""
sys.argv = ['', '--version']
Globals.getInstance().set_args(sys.argv)
@@ -48,7 +50,8 @@ def test_main_init_parser_version(capsys, reset_globals):
@pytest.mark.unit
def test_main_main_version(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_main_version(capsys):
"""Test --version"""
sys.argv = ['', '--version']
Globals.getInstance().set_args(sys.argv)
@@ -63,7 +66,8 @@ def test_main_main_version(capsys, reset_globals):
@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"""
sys.argv = ['']
Globals.getInstance().set_args(sys.argv)
@@ -77,7 +81,8 @@ def test_main_main_no_args(reset_globals, capsys):
@pytest.mark.unit
def test_main_support(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_support(capsys):
"""Test --support"""
sys.argv = ['', '--support']
Globals.getInstance().set_args(sys.argv)
@@ -95,8 +100,9 @@ def test_main_support(capsys, reset_globals):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@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"""
sys.argv = ['', '--ch-index', '1']
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.usefixtures("reset_globals")
@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"""
sys.argv = ['', '--test']
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.usefixtures("reset_globals")
@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"""
sys.argv = ['', '--test']
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.usefixtures("reset_globals")
@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"""
sys.argv = ['', '--test']
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.usefixtures("reset_globals")
@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"""
sys.argv = ['', '--test']
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
def test_main_info(capsys, caplog, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_info(capsys, caplog):
"""Test --info"""
sys.argv = ['', '--info']
Globals.getInstance().set_args(sys.argv)
@@ -201,8 +212,9 @@ def test_main_info(capsys, caplog, reset_globals):
@pytest.mark.unit
@pytest.mark.usefixtures("reset_globals")
@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"""
sys.argv = ['', '--info']
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
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"""
sys.argv = ['', '--info', '--host', 'meshtastic.local']
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)
#@pytest.mark.unit
#def test_main_info_with_ble_interface(capsys, reset_globals):
#def test_main_info_with_ble_interface(capsys):
# """Test --info"""
# sys.argv = ['', '--info', '--ble', 'foo']
# Globals.getInstance().set_args(sys.argv)
@@ -263,7 +276,8 @@ def test_main_info_with_tcp_interface(capsys, reset_globals):
@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)"""
sys.argv = ['', '--info', '--noproto']
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
def my_sleep(amount):
print(f'amount:{amount}')
sys.exit(0)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface):
@@ -290,7 +305,8 @@ def test_main_no_proto(capsys, reset_globals):
@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"""
sys.argv = ['', '--info', '--seriallog', 'stdout']
Globals.getInstance().set_args(sys.argv)
@@ -309,7 +325,8 @@ def test_main_info_with_seriallog_stdout(capsys, reset_globals):
@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"""
sys.argv = ['', '--info', '--seriallog', 'output.txt']
Globals.getInstance().set_args(sys.argv)
@@ -330,7 +347,8 @@ def test_main_info_with_seriallog_output_txt(capsys, reset_globals):
@pytest.mark.unit
def test_main_qr(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_qr(capsys):
"""Test --qr"""
sys.argv = ['', '--qr']
Globals.getInstance().set_args(sys.argv)
@@ -349,7 +367,8 @@ def test_main_qr(capsys, reset_globals):
@pytest.mark.unit
def test_main_nodes(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_nodes(capsys):
"""Test --nodes"""
sys.argv = ['', '--nodes']
Globals.getInstance().set_args(sys.argv)
@@ -368,7 +387,8 @@ def test_main_nodes(capsys, reset_globals):
@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"""
sys.argv = ['', '--set-owner', 'bob']
Globals.getInstance().set_args(sys.argv)
@@ -384,7 +404,8 @@ def test_main_set_owner_to_bob(capsys, reset_globals):
@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"""
sys.argv = ['', '--set-ham', 'KI123']
Globals.getInstance().set_args(sys.argv)
@@ -393,7 +414,7 @@ def test_main_set_ham_to_KI123(capsys, reset_globals):
def mock_turnOffEncryptionOnPrimaryChannel():
print('inside mocked turnOffEncryptionOnPrimaryChannel')
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.setOwner.side_effect = mock_setOwner
@@ -412,7 +433,8 @@ def test_main_set_ham_to_KI123(capsys, reset_globals):
@pytest.mark.unit
def test_main_reboot(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_reboot(capsys):
"""Test --reboot"""
sys.argv = ['', '--reboot']
Globals.getInstance().set_args(sys.argv)
@@ -435,7 +457,8 @@ def test_main_reboot(capsys, reset_globals):
@pytest.mark.unit
def test_main_sendtext(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_sendtext(capsys):
"""Test --sendtext"""
sys.argv = ['', '--sendtext', 'hello']
Globals.getInstance().set_args(sys.argv)
@@ -443,6 +466,7 @@ def test_main_sendtext(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface)
def mock_sendText(text, dest, wantAck, channelIndex):
print('inside mocked sendText')
print(f'{text} {dest} {wantAck} {channelIndex}')
iface.sendText.side_effect = mock_sendText
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
def test_main_sendtext_with_channel(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_sendtext_with_channel(capsys):
"""Test --sendtext"""
sys.argv = ['', '--sendtext', 'hello', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv)
@@ -464,6 +489,7 @@ def test_main_sendtext_with_channel(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface)
def mock_sendText(text, dest, wantAck, channelIndex):
print('inside mocked sendText')
print(f'{text} {dest} {wantAck} {channelIndex}')
iface.sendText.side_effect = mock_sendText
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
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"""
sys.argv = ['', '--sendtext', 'hello', '--ch-index', '-1']
Globals.getInstance().set_args(sys.argv)
@@ -499,7 +526,8 @@ def test_main_sendtext_with_invalid_channel(caplog, capsys, reset_globals):
@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"""
sys.argv = ['', '--sendtext', 'hello', '--ch-index', '9']
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
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"""
sys.argv = ['', '--sendtext', 'hello', '--dest', 'foo']
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
def test_main_sendping(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_sendping(capsys):
"""Test --sendping"""
sys.argv = ['', '--sendping']
Globals.getInstance().set_args(sys.argv)
@@ -554,6 +584,7 @@ def test_main_sendping(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface)
def mock_sendData(payload, dest, portNum, wantAck, wantResponse):
print('inside mocked sendData')
print(f'{payload} {dest} {portNum} {wantAck} {wantResponse}')
iface.sendData.side_effect = mock_sendData
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
def test_main_setlat(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_setlat(capsys):
"""Test --sendlat"""
sys.argv = ['', '--setlat', '37.5']
Globals.getInstance().set_args(sys.argv)
@@ -580,6 +612,7 @@ def test_main_setlat(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface)
def mock_sendPosition(lat, lon, alt):
print('inside mocked sendPosition')
print(f'{lat} {lon} {alt}')
iface.sendPosition.side_effect = mock_sendPosition
iface.localNode.return_value = mocked_node
@@ -596,7 +629,8 @@ def test_main_setlat(capsys, reset_globals):
@pytest.mark.unit
def test_main_setlon(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_setlon(capsys):
"""Test --setlon"""
sys.argv = ['', '--setlon', '-122.1']
Globals.getInstance().set_args(sys.argv)
@@ -609,6 +643,7 @@ def test_main_setlon(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface)
def mock_sendPosition(lat, lon, alt):
print('inside mocked sendPosition')
print(f'{lat} {lon} {alt}')
iface.sendPosition.side_effect = mock_sendPosition
iface.localNode.return_value = mocked_node
@@ -625,7 +660,8 @@ def test_main_setlon(capsys, reset_globals):
@pytest.mark.unit
def test_main_setalt(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_setalt(capsys):
"""Test --setalt"""
sys.argv = ['', '--setalt', '51']
Globals.getInstance().set_args(sys.argv)
@@ -638,6 +674,7 @@ def test_main_setalt(capsys, reset_globals):
iface = MagicMock(autospec=SerialInterface)
def mock_sendPosition(lat, lon, alt):
print('inside mocked sendPosition')
print(f'{lat} {lon} {alt}')
iface.sendPosition.side_effect = mock_sendPosition
iface.localNode.return_value = mocked_node
@@ -654,7 +691,8 @@ def test_main_setalt(capsys, reset_globals):
@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"""
sys.argv = ['', '--set-team', 'CYAN']
Globals.getInstance().set_args(sys.argv)
@@ -662,6 +700,7 @@ def test_main_set_team_valid(capsys, reset_globals):
mocked_node = MagicMock(autospec=Node)
def mock_setOwner(team):
print('inside mocked setOwner')
print(f'{team}')
mocked_node.setOwner.side_effect = mock_setOwner
iface = MagicMock(autospec=SerialInterface)
@@ -682,7 +721,8 @@ def test_main_set_team_valid(capsys, reset_globals):
@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"""
sys.argv = ['', '--set-team', 'NOTCYAN']
Globals.getInstance().set_args(sys.argv)
@@ -705,7 +745,8 @@ def test_main_set_team_invalid(capsys, reset_globals):
@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)"""
sys.argv = ['', '--seturl', 'https://www.meshtastic.org/d/#CgUYAyIBAQ']
Globals.getInstance().set_args(sys.argv)
@@ -720,7 +761,8 @@ def test_main_seturl(capsys, reset_globals):
@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"""
sys.argv = ['', '--set', 'wifi_ssid', 'foo']
Globals.getInstance().set_args(sys.argv)
@@ -740,7 +782,8 @@ def test_main_set_valid(capsys, reset_globals):
@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"""
sys.argv = ['', '--set', 'foo', 'foo']
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
@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"""
sys.argv = ['', '--configure', 'example_config.yaml']
Globals.getInstance().set_args(sys.argv)
@@ -790,7 +834,8 @@ def test_main_configure(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-add', 'testing']
Globals.getInstance().set_args(sys.argv)
@@ -817,7 +862,8 @@ def test_main_ch_add_valid(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-add', 'testingtestingtesting']
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
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"""
sys.argv = ['', '--ch-add', 'testing']
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
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"""
sys.argv = ['', '--ch-add', 'testing']
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
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"""
sys.argv = ['', '--ch-del', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv)
@@ -919,7 +968,8 @@ def test_main_ch_del(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-del']
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
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"""
sys.argv = ['', '--ch-del', '--ch-index', '0']
Globals.getInstance().set_args(sys.argv)
@@ -966,7 +1017,8 @@ def test_main_ch_del_primary_channel(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-enable', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv)
@@ -987,7 +1039,8 @@ def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-disable', '--ch-index', '1']
Globals.getInstance().set_args(sys.argv)
@@ -1008,7 +1061,8 @@ def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-enable']
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
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"""
sys.argv = ['', '--ch-enable', '--ch-index', '0']
Globals.getInstance().set_args(sys.argv)
@@ -1056,7 +1111,8 @@ def test_main_ch_enable_primary_channel(capsys, reset_globals):
@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."""
range_options = ['--ch-longslow', '--ch-longfast', '--ch-mediumslow',
'--ch-mediumfast', '--ch-shortslow', '--ch-shortfast']
@@ -1079,7 +1135,8 @@ def test_main_ch_range_options(capsys, reset_globals):
@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"""
sys.argv = ['', '--ch-longfast', '--ch-index', '1']
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
@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)"""
sys.argv = ['', '--pos-fields']
Globals.getInstance().set_args(sys.argv)
@@ -1146,7 +1204,8 @@ def test_main_pos_fields_no_args(capsys, reset_globals):
@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)"""
sys.argv = ['', '--pos-fields', '0']
Globals.getInstance().set_args(sys.argv)
@@ -1179,7 +1238,8 @@ def test_main_pos_fields_arg_of_zero(capsys, reset_globals):
@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"""
sys.argv = ['', '--pos-fields', 'POS_GEO_SEP', 'POS_ALT_MSL']
Globals.getInstance().set_args(sys.argv)
@@ -1205,7 +1265,8 @@ def test_main_pos_fields_valid_values(capsys, reset_globals):
@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)"""
sys.argv = ['', '--get', 'ls_secs', '--get', 'wifi_ssid', '--get', 'fixed_position']
Globals.getInstance().set_args(sys.argv)
@@ -1229,7 +1290,8 @@ def test_main_get_with_valid_values(capsys, reset_globals):
@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"""
sys.argv = ['', '--get', 'foo']
Globals.getInstance().set_args(sys.argv)
@@ -1254,7 +1316,8 @@ def test_main_get_with_invalid(capsys, reset_globals):
@pytest.mark.unit
def test_main_setchan(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_main_setchan(capsys):
"""Test --setchan (deprecated)"""
sys.argv = ['', '--setchan', 'a', 'b']
Globals.getInstance().set_args(sys.argv)
@@ -1271,7 +1334,8 @@ def test_main_setchan(capsys, reset_globals):
@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"""
args = MagicMock()
Globals.getInstance().set_args(args)
@@ -1297,7 +1361,8 @@ def test_main_onReceive_empty(caplog, reset_globals, capsys):
# }
@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
The entire point of this test is to make sure the interface.close() call
is made in onReceive().
@@ -1332,7 +1397,8 @@ def test_main_onReceive_with_sendtext(caplog, capsys, reset_globals):
@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
"""
args = MagicMock()
@@ -1371,7 +1437,8 @@ def test_main_onReceive_with_text(caplog, capsys, reset_globals):
@pytest.mark.unit
def test_main_onConnection(reset_globals, capsys):
@pytest.mark.usefixtures("reset_globals")
def test_main_onConnection(capsys):
"""Test onConnection"""
sys.argv = ['']
Globals.getInstance().set_args(sys.argv)
@@ -1389,7 +1456,8 @@ def test_main_onConnection(reset_globals, capsys):
@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"""
iface = MagicMock(autospec=SerialInterface)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -1425,7 +1493,8 @@ position_flags: 35"""
@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"""
sys.argv = ['', '--export-config']
Globals.getInstance().set_args(sys.argv)
@@ -1441,7 +1510,8 @@ def test_main_export_config_called_from_main(capsys, reset_globals):
@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"""
sys.argv = ['', '--gpio-rd', '0x10', '--dest', '!foo']
Globals.getInstance().set_args(sys.argv)
@@ -1459,7 +1529,8 @@ def test_main_gpio_rd_no_gpio_channel(capsys, reset_globals):
@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"""
sys.argv = ['', '--gpio-rd', '0x2000']
Globals.getInstance().set_args(sys.argv)
@@ -1481,7 +1552,8 @@ def test_main_gpio_rd_no_dest(capsys, reset_globals):
@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"""
# 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)
@@ -1538,7 +1610,8 @@ def test_main_gpio_rd(caplog, capsys, reset_globals):
@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"""
prefs = MagicMock()
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
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"""
prefs = MagicMock()
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
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"""
prefs = MagicMock()
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
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"""
class Field:
@@ -1614,7 +1690,8 @@ def test_main_getPref_invalid_field(capsys, reset_globals):
@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"""
class Field:
@@ -1636,7 +1713,8 @@ def test_main_setPref_valid_field_int(capsys, reset_globals):
@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"""
@@ -1667,7 +1745,8 @@ def test_main_setPref_invalid_field(capsys, reset_globals):
@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 """
sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local']
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
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 """
sys.argv = ['', '--ch-set', 'psk', 'foo', '--host', 'meshtastic.local', '--ch-index', '0']
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
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"""
sys.argv = ['', '--ch-set', 'name', 'foo', '--host', 'meshtastic.local', '--ch-index', '0']
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
def test_onNode(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_onNode(capsys):
"""Test onNode"""
onNode('foo')
out, err = capsys.readouterr()
@@ -1728,7 +1810,8 @@ def test_onNode(capsys, reset_globals):
@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"""
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.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=[])
@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)"""
a_mock = MagicMock()
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.usefixtures("reset_globals")
@patch('meshtastic.util.findPorts', return_value=[])
@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)"""
a_mock = MagicMock()
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.usefixtures("reset_globals")
@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)"""
# Override the time.sleep so there is no loop
def my_sleep(amount):
print(f'{amount}')
sys.exit(3)
a_mock = MagicMock()

View File

@@ -15,7 +15,8 @@ from ..util import Timeout
@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"""
iface = MeshInterface(noProto=True)
anode = Node('foo', 'bar')
@@ -56,7 +57,8 @@ def test_MeshInterface(capsys, reset_globals):
@pytest.mark.unit
def test_getMyUser(reset_globals, iface_with_nodes):
@pytest.mark.usefixtures("reset_globals")
def test_getMyUser(iface_with_nodes):
"""Test getMyUser()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -66,7 +68,8 @@ def test_getMyUser(reset_globals, iface_with_nodes):
@pytest.mark.unit
def test_getLongName(reset_globals, iface_with_nodes):
@pytest.mark.usefixtures("reset_globals")
def test_getLongName(iface_with_nodes):
"""Test getLongName()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -75,7 +78,8 @@ def test_getLongName(reset_globals, iface_with_nodes):
@pytest.mark.unit
def test_getShortName(reset_globals, iface_with_nodes):
@pytest.mark.usefixtures("reset_globals")
def test_getShortName(iface_with_nodes):
"""Test getShortName()."""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -84,7 +88,8 @@ def test_getShortName(reset_globals, iface_with_nodes):
@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."""
iface = MeshInterface(noProto=True)
meshPacket = mesh_pb2.MeshPacket()
@@ -95,7 +100,8 @@ def test_handlePacketFromRadio_no_from(capsys, reset_globals):
@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
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.
@@ -110,7 +116,8 @@ def test_handlePacketFromRadio_with_a_portnum(caplog, reset_globals):
@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"""
iface = MeshInterface(noProto=True)
meshPacket = mesh_pb2.MeshPacket()
@@ -121,7 +128,8 @@ def test_handlePacketFromRadio_no_portnum(caplog, reset_globals):
@pytest.mark.unit
def test_getNode_with_local(reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_getNode_with_local():
"""Test getNode"""
iface = MeshInterface(noProto=True)
anode = iface.getNode(LOCAL_ADDR)
@@ -129,7 +137,8 @@ def test_getNode_with_local(reset_globals):
@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"""
iface = MeshInterface(noProto=True)
anode = MagicMock(autospec=Node)
@@ -141,7 +150,8 @@ def test_getNode_not_local(reset_globals, caplog):
@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"""
iface = MeshInterface(noProto=True)
anode = MagicMock(autospec=Node)
@@ -157,7 +167,8 @@ def test_getNode_not_local_timeout(reset_globals, capsys):
@pytest.mark.unit
def test_sendPosition(reset_globals, caplog):
@pytest.mark.usefixtures("reset_globals")
def test_sendPosition(caplog):
"""Test sendPosition"""
iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG):
@@ -167,7 +178,8 @@ def test_sendPosition(reset_globals, caplog):
@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"""
iface = MeshInterface(noProto=True)
anode = Node('foo', 'bar')
@@ -183,7 +195,8 @@ def test_close_with_heartbeatTimer(reset_globals, caplog):
@pytest.mark.unit
def test_handleFromRadio_empty_payload(reset_globals, caplog):
@pytest.mark.usefixtures("reset_globals")
def test_handleFromRadio_empty_payload(caplog):
"""Test _handleFromRadio"""
iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG):
@@ -193,7 +206,8 @@ def test_handleFromRadio_empty_payload(reset_globals, caplog):
@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"""
# Note: I captured the '--debug --info' for the bytes below.
# It "translates" to this:
@@ -218,7 +232,8 @@ def test_handleFromRadio_with_my_info(reset_globals, caplog):
@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"""
# Note: I captured the '--debug --info' for the bytes below.
# It "translates" to this:
@@ -254,7 +269,8 @@ def test_handleFromRadio_with_node_info(reset_globals, caplog, capsys):
@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"""
# Note: Captured the '--debug --info' for the bytes below.
# pylint: disable=C0301
@@ -277,7 +293,8 @@ def test_handleFromRadio_with_node_info_tbeam1(reset_globals, caplog, capsys):
@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"""
# 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@'
@@ -288,7 +305,8 @@ def test_handleFromRadio_with_node_info_tbeam_with_bad_data(reset_globals, caplo
@pytest.mark.unit
def test_MeshInterface_sendToRadioImpl(caplog, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_MeshInterface_sendToRadioImpl(caplog):
"""Test _sendToRadioImp()"""
iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG):
@@ -298,7 +316,8 @@ def test_MeshInterface_sendToRadioImpl(caplog, reset_globals):
@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()"""
iface = MeshInterface()
with caplog.at_level(logging.DEBUG):
@@ -308,7 +327,8 @@ def test_MeshInterface_sendToRadio_no_proto(caplog, reset_globals):
@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"""
iface = MeshInterface(noProto=True)
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
def test_sendData_unknown_app(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_sendData_unknown_app(capsys):
"""Test sendData when unknown app"""
iface = MeshInterface(noProto=True)
with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -345,7 +366,8 @@ def test_sendData_unknown_app(capsys, reset_globals):
@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"""
iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG):
@@ -356,7 +378,8 @@ def test_sendPosition_with_a_position(caplog, reset_globals):
@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()"""
iface = MeshInterface(noProto=True)
with pytest.raises(SystemExit) as pytest_wrapped_e:
@@ -369,7 +392,8 @@ def test_sendPacket_with_no_destination(capsys, reset_globals):
@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"""
iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG):
@@ -379,7 +403,8 @@ def test_sendPacket_with_destination_as_int(caplog, reset_globals):
@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"""
iface = MeshInterface(noProto=True)
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
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"""
iface = MeshInterface(noProto=True)
with caplog.at_level(logging.DEBUG):
@@ -399,7 +425,8 @@ def test_sendPacket_with_destination_as_BROADCAST_ADDR(caplog, reset_globals):
@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"""
iface = MeshInterface(noProto=True)
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
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"""
iface = MeshInterface(noProto=True)
myInfo = MagicMock()
iface.myInfo = myInfo
iface.myInfo.my_node_num = 1
with caplog.at_level(logging.DEBUG):
meshPacket = mesh_pb2.MeshPacket()
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
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"""
iface = iface_with_nodes
meshPacket = mesh_pb2.MeshPacket()
@@ -439,7 +469,8 @@ def test_sendPacket_with_destination_is_blank_with_nodes(capsys, reset_globals,
@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"""
iface = iface_with_nodes
iface.nodes = None
@@ -450,7 +481,8 @@ def test_sendPacket_with_destination_is_blank_without_nodes(caplog, reset_global
@pytest.mark.unit
def test_getMyNodeInfo(reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_getMyNodeInfo():
"""Test getMyNodeInfo()"""
iface = MeshInterface(noProto=True)
anode = iface.getNode(LOCAL_ADDR)
@@ -464,7 +496,8 @@ def test_getMyNodeInfo(reset_globals):
@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)"""
iface = MeshInterface(noProto=True)
# 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
def test_fixupPosition_empty_pos(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_fixupPosition_empty_pos():
"""Test _fixupPosition()"""
iface = MeshInterface(noProto=True)
pos = {}
@@ -488,7 +522,8 @@ def test_fixupPosition_empty_pos(capsys, reset_globals):
@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()"""
iface = MeshInterface(noProto=True)
pos = {"latitude": 101, "longitude": 102}
@@ -497,7 +532,8 @@ def test_fixupPosition_no_changes_needed(capsys, reset_globals):
@pytest.mark.unit
def test_fixupPosition(capsys, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_fixupPosition():
"""Test _fixupPosition()"""
iface = MeshInterface(noProto=True)
pos = {"latitudeI": 1010000000, "longitudeI": 1020000000}
@@ -509,7 +545,8 @@ def test_fixupPosition(capsys, reset_globals):
@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()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -518,7 +555,8 @@ def test_nodeNumToId(capsys, reset_globals, iface_with_nodes):
@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()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -527,7 +565,8 @@ def test_nodeNumToId_not_found(capsys, reset_globals, iface_with_nodes):
@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()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -536,7 +575,8 @@ def test_nodeNumToId_to_all(capsys, reset_globals, iface_with_nodes):
@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()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -545,7 +585,8 @@ def test_getOrCreateByNum_minimal(capsys, reset_globals, iface_with_nodes):
@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()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -555,7 +596,8 @@ def test_getOrCreateByNum_not_found(capsys, reset_globals, iface_with_nodes):
@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()"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -581,7 +623,7 @@ def test_exit_with_exception(caplog):
@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"""
with caplog.at_level(logging.DEBUG):
iface = iface_with_nodes
@@ -592,7 +634,7 @@ def test_showNodes_exclude_self(capsys, caplog, reset_globals, iface_with_nodes)
@pytest.mark.unitslow
def test_waitForConfig(caplog, capsys):
def test_waitForConfig(capsys):
"""Test waitForConfig()"""
iface = MeshInterface(noProto=True)
# override how long to wait
@@ -606,7 +648,7 @@ def test_waitForConfig(caplog, capsys):
@pytest.mark.unit
def test_waitConnected_raises_an_exception(caplog, capsys):
def test_waitConnected_raises_an_exception(capsys):
"""Test waitConnected()"""
iface = MeshInterface(noProto=True)
with pytest.raises(Exception) as pytest_wrapped_e:
@@ -619,7 +661,7 @@ def test_waitConnected_raises_an_exception(caplog, capsys):
@pytest.mark.unit
def test_waitConnected_isConnected_timeout(caplog, capsys):
def test_waitConnected_isConnected_timeout(capsys):
"""Test waitConnected()"""
with pytest.raises(Exception) as pytest_wrapped_e:
iface = MeshInterface()

View File

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

View File

@@ -23,6 +23,10 @@ def test_SerialInterface_single_port(mocked_findPorts, mocked_serial, mocked_ope
iface.close()
mocked_findPorts.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()
assert re.search(r'Nodes in mesh', 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
@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
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:
## pytest -s meshtastic/tests/test_stream_interface.py::test_sendToRadioImpl
@pytest.mark.unitslow
def test_sendToRadioImpl(caplog, reset_globals):
@pytest.mark.usefixtures("reset_globals")
def test_sendToRadioImpl(caplog):
"""Test _sendToRadioImpl()"""
# def add_header(b):

View File

@@ -28,7 +28,24 @@ def test_TCPInterface(capsys):
@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"""
with patch('socket.socket'):
iface = TCPInterface(hostname='localhost', noProto=True, connectNow=False)

View File

@@ -14,7 +14,7 @@ from ..globals import Globals
@pytest.mark.unit
@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"""
a_mock = MagicMock()
a_mock.return_value = 'notLinux'
@@ -29,7 +29,7 @@ def test_Tunnel_on_non_linux_system(mock_platform_system, reset_globals):
@pytest.mark.unit
@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"""
a_mock = MagicMock()
a_mock.return_value = 'Linux'
@@ -41,7 +41,7 @@ def test_Tunnel_without_interface(mock_platform_system, reset_globals):
@pytest.mark.unitslow
@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"""
iface = iface_with_nodes
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
@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"""
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -81,7 +81,7 @@ def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, reset_glob
@pytest.mark.unit
@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"""
iface = iface_with_nodes
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
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -119,7 +119,7 @@ def test_shouldFilterPacket_random(mock_platform_system, caplog, reset_globals,
@pytest.mark.unitslow
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -137,7 +137,7 @@ def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, reset_glo
@pytest.mark.unitslow
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -156,7 +156,7 @@ def test_shouldFilterPacket_icmp(mock_platform_system, caplog, reset_globals, if
@pytest.mark.unit
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -175,7 +175,7 @@ def test_shouldFilterPacket_udp(mock_platform_system, caplog, reset_globals, ifa
@pytest.mark.unitslow
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -196,7 +196,7 @@ def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, reset_
@pytest.mark.unit
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -215,7 +215,7 @@ def test_shouldFilterPacket_tcp(mock_platform_system, caplog, reset_globals, ifa
@pytest.mark.unitslow
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -236,7 +236,7 @@ def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, reset_
@pytest.mark.unitslow
@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()"""
iface = iface_with_nodes
iface.noProto = True
@@ -252,7 +252,7 @@ def test_ipToNodeId_none(mock_platform_system, caplog, reset_globals, iface_with
@pytest.mark.unitslow
@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()"""
iface = iface_with_nodes
iface.noProto = True

View File

@@ -242,10 +242,12 @@ def test_readnet_u16():
def test_findPorts_when_none_found(patch_comports):
"""Test findPorts()"""
assert not findPorts()
patch_comports.assert_called()
@pytest.mark.unitslow
def test_convert_mac_addr():
"""Test convert_mac_addr()"""
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('') == ''

View File

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

View File

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

View File

@@ -4,7 +4,6 @@ protobuf
dotmap
pexpect
pyqrcode
pygatt
tabulate
timeago
webencodings
@@ -18,3 +17,4 @@ pyyaml
pytap2
pdoc3
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
setup(
name="meshtastic",
version="1.2.53",
version="1.2.54",
description="Python API & client shell for talking to Meshtastic devices",
long_description=long_description,
long_description_content_type="text/markdown",
@@ -33,7 +33,8 @@ setup(
include_package_data=True,
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",
"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={
'tunnel': ["pytap2>=2.0.0"]
},