mirror of
https://github.com/meshtastic/python.git
synced 2025-12-26 01:17:51 -05:00
Compare commits
23 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
52834e9966 | ||
|
|
63c60d4cea | ||
|
|
6a2a9d2093 | ||
|
|
1410448808 | ||
|
|
ad8f2222db | ||
|
|
48265e73b1 | ||
|
|
f3139a8aa0 | ||
|
|
8d68e36703 | ||
|
|
d2d93fbe80 | ||
|
|
ed8510468d | ||
|
|
e7680e07c2 | ||
|
|
0f89baa36e | ||
|
|
48ed7690af | ||
|
|
59b94ea650 | ||
|
|
5b992734fb | ||
|
|
3b74b911f8 | ||
|
|
b6570e3c27 | ||
|
|
e1e1664b96 | ||
|
|
cb1913dfc3 | ||
|
|
b813a6f8c5 | ||
|
|
0b662318e1 | ||
|
|
a6ccc1a246 | ||
|
|
bc17e9b389 |
@@ -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
20
examples/get_hw.py
Normal 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
20
examples/set_owner.py
Normal 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()
|
||||
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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"""
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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()
|
||||
|
||||
61
meshtastic/tests/test_init.py
Normal file
61
meshtastic/tests/test_init.py
Normal 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)
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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')
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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('') == ''
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -4,7 +4,6 @@ protobuf
|
||||
dotmap
|
||||
pexpect
|
||||
pyqrcode
|
||||
pygatt
|
||||
tabulate
|
||||
timeago
|
||||
webencodings
|
||||
@@ -18,3 +17,4 @@ pyyaml
|
||||
pytap2
|
||||
pdoc3
|
||||
pypubsub
|
||||
pygatt; platform_system == "Linux"
|
||||
|
||||
5
setup.py
5
setup.py
@@ -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"]
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user