mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2026-01-13 10:17:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #164 from mkinney/unit_tests

Browse Source 
Unit tests
This commit is contained in:
mkinney
2021-12-16 09:23:50 -08:00
committed by GitHub
parent 182a9b89f3 2fcfdeb04f
commit 876a0a13dd
15 changed files with 327 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
meshtastic/__init__.py
View File

@@ -91,9 +91,10 @@ BROADCAST_NUM = 0xffffffff
BROADCAST_ADDR = "^all"
"""The numeric buildnumber (shared with android apps) specifying the level of device code we are guaranteed to understand
"""The numeric buildnumber (shared with android apps) specifying the
level of device code we are guaranteed to understand
format is Mmmss (where M is 1+the numeric major number. i.e. 20120 means 1.1.20
format is Mmmss (where M is 1+the numeric major number. i.e. 20120 means 1.1.20
"""
OUR_APP_VERSION = 20200

14
meshtastic/__main__.py
View File

@@ -13,14 +13,12 @@ import pyqrcode
import pkg_resources
import meshtastic.util
import meshtastic.test
from .tcp_interface import TCPInterface
from .ble_interface import BLEInterface
from . import remote_hardware
from . import portnums_pb2, channel_pb2, radioconfig_pb2
from .globals import Globals
"""We only import the tunnel code if we are on a platform that can run it"""
"""We only import the tunnel code if we are on a platform that can run it. """
have_tunnel = platform.system() == 'Linux'
def onReceive(packet, interface):
@@ -377,7 +375,9 @@ def onConnected(interface):
print(f"Deleting channel {channelIndex}")
ch = getNode().deleteChannel(channelIndex)
ch_changes = [args.ch_longslow, args.ch_longfast, args.ch_mediumslow, args.ch_mediumfast, args.ch_shortslow, args.ch_shortfast]
ch_changes = [args.ch_longslow, args.ch_longfast,
args.ch_mediumslow, args.ch_mediumfast,
args.ch_shortslow, args.ch_shortfast]
any_primary_channel_changes = any(x for x in ch_changes)
if args.ch_set or any_primary_channel_changes or args.ch_enable or args.ch_disable:
closeNow = True
@@ -575,9 +575,9 @@ def common():
subscribe()
if args.ble:
client = BLEInterface(args.ble, debugOut=logfile)
client = meshtastic.ble_interface.BLEInterface(args.ble, debugOut=logfile, noProto=args.noproto)
elif args.host:
client = TCPInterface(
client = meshtastic.tcp_interface.TCPInterface(
args.host, debugOut=logfile, noProto=args.noproto)
else:
client = meshtastic.serial_interface.SerialInterface(
@@ -595,7 +595,7 @@ def common():
def initParser():
""" Initialize the command line argument parsing."""
"""Initialize the command line argument parsing."""
our_globals = Globals.getInstance()
parser = our_globals.get_parser()
args = our_globals.get_args()

38
meshtastic/ble_interface.py
View File

@@ -1,4 +1,4 @@
""" Bluetooth interface
"""Bluetooth interface
"""
import logging
import pygatt
@@ -15,22 +15,27 @@ FROMNUM_UUID = "ed9da18c-a800-4f66-a670-aa7547e34453"
class BLEInterface(MeshInterface):
"""A not quite ready - FIXME - BLE interface to devices"""
def __init__(self, address, debugOut=None):
def __init__(self, address, noProto=False, debugOut=None):
self.address = address
self.adapter = pygatt.GATTToolBackend() # BGAPIBackend()
self.adapter.start()
logging.debug(f"Connecting to {self.address}")
self.device = self.adapter.connect(address)
if not noProto:
self.adapter = pygatt.GATTToolBackend() # BGAPIBackend()
self.adapter.start()
logging.debug(f"Connecting to {self.address}")
self.device = self.adapter.connect(address)
else:
self.adapter = None
self.device = None
logging.debug("Connected to device")
# fromradio = self.device.char_read(FROMRADIO_UUID)
MeshInterface.__init__(self, debugOut=debugOut)
MeshInterface.__init__(self, debugOut=debugOut, noProto=noProto)
self._readFromRadio() # read the initial responses
def handle_data(handle, data):
self._handleFromRadio(data)
self.device.subscribe(FROMNUM_UUID, callback=handle_data)
if self.device:
self.device.subscribe(FROMNUM_UUID, callback=handle_data)
def _sendToRadioImpl(self, toRadio):
"""Send a ToRadio protobuf to the device"""
@@ -40,12 +45,15 @@ class BLEInterface(MeshInterface):
def close(self):
MeshInterface.close(self)
self.adapter.stop()
if self.adapter:
self.adapter.stop()
def _readFromRadio(self):
wasEmpty = False
while not wasEmpty:
b = self.device.char_read(FROMRADIO_UUID)
wasEmpty = len(b) == 0
if not wasEmpty:
self._handleFromRadio(b)
if not self.noProto:
wasEmpty = False
while not wasEmpty:
if self.device:
b = self.device.char_read(FROMRADIO_UUID)
wasEmpty = len(b) == 0
if not wasEmpty:
self._handleFromRadio(b)

145
meshtastic/mesh_interface.py
View File

@@ -39,7 +39,8 @@ class MeshInterface:
"""Constructor
Keyword Arguments:
noProto -- If True, don't try to run our protocol on the link - just be a dumb serial client.
noProto -- If True, don't try to run our protocol on the
link - just be a dumb serial client.
"""
self.debugOut = debugOut
self.nodes = None # FIXME
@@ -92,54 +93,55 @@ class MeshInterface:
def showNodes(self, includeSelf=True, file=sys.stdout):
"""Show table summary of nodes in mesh"""
def formatFloat(value, precision=2, unit=''):
"""Format a float value with precsion."""
return f'{value:.{precision}f}{unit}' if value else None
def getLH(ts):
"""Format last heard"""
return datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') if ts else None
def getTimeAgo(ts):
"""Format how long ago have we heard from this node (aka timeago)."""
return timeago.format(datetime.fromtimestamp(ts), datetime.now()) if ts else None
rows = []
for node in self.nodes.values():
if not includeSelf and node['num'] == self.localNode.nodeNum:
continue
if self.nodes:
for node in self.nodes.values():
if not includeSelf and node['num'] == self.localNode.nodeNum:
continue
row = {"N": 0}
row = {"N": 0}
user = node.get('user')
if user:
row.update({
"User": user['longName'],
"AKA": user['shortName'],
"ID": user['id'],
})
pos = node.get('position')
if pos:
row.update({
"Latitude": formatFloat(pos.get("latitude"), 4, "°"),
"Longitude": formatFloat(pos.get("longitude"), 4, "°"),
"Altitude": formatFloat(pos.get("altitude"), 0, " m"),
"Battery": formatFloat(pos.get("batteryLevel"), 2, "%"),
})
user = node.get('user')
if user:
row.update({
"User": user['longName'],
"AKA": user['shortName'],
"ID": user['id'],
"SNR": formatFloat(node.get("snr"), 2, " dB"),
"LastHeard": getLH(node.get("lastHeard")),
"Since": getTimeAgo(node.get("lastHeard")),
})
pos = node.get('position')
if pos:
row.update({
"Latitude": formatFloat(pos.get("latitude"), 4, "°"),
"Longitude": formatFloat(pos.get("longitude"), 4, "°"),
"Altitude": formatFloat(pos.get("altitude"), 0, " m"),
"Battery": formatFloat(pos.get("batteryLevel"), 2, "%"),
})
rows.append(row)
row.update({
"SNR": formatFloat(node.get("snr"), 2, " dB"),
"LastHeard": getLH(node.get("lastHeard")),
"Since": getTimeAgo(node.get("lastHeard")),
})
rows.append(row)
# Why doesn't this way work?
#rows.sort(key=lambda r: r.get('LastHeard', '0000'), reverse=True)
rows.sort(key=lambda r: r.get('LastHeard') or '0000', reverse=True)
for i, row in enumerate(rows):
row['N'] = i+1
table = tabulate(rows, headers='keys', missingval='N/A',
tablefmt='fancy_grid')
table = tabulate(rows, headers='keys', missingval='N/A', tablefmt='fancy_grid')
print(table)
return table
@@ -162,18 +164,24 @@ class MeshInterface:
hopLimit=defaultHopLimit,
onResponse=None,
channelIndex=0):
"""Send a utf8 string to some other node, if the node has a display it will also be shown on the device.
"""Send a utf8 string to some other node, if the node has a display it
will also be shown on the device.
Arguments:
text {string} -- The text to send
Keyword Arguments:
destinationId {nodeId or nodeNum} -- where to send this message (default: {BROADCAST_ADDR})
portNum -- the application portnum (similar to IP port numbers) of the destination, see portnums.proto for a list
wantAck -- True if you want the message sent in a reliable manner (with retries and ack/nak provided for delivery)
wantResponse -- True if you want the service on the other side to send an application layer response
destinationId {nodeId or nodeNum} -- where to send this
message (default: {BROADCAST_ADDR})
portNum -- the application portnum (similar to IP port numbers)
of the destination, see portnums.proto for a list
wantAck -- True if you want the message sent in a reliable manner
(with retries and ack/nak provided for delivery)
wantResponse -- True if you want the service on the other side to
send an application layer response
Returns the sent packet. The id field will be populated in this packet and can be used to track future message acks/naks.
Returns the sent packet. The id field will be populated in this packet
and can be used to track future message acks/naks.
"""
return self.sendData(text.encode("utf-8"), destinationId,
portNum=portnums_pb2.PortNum.TEXT_MESSAGE_APP,
@@ -192,15 +200,23 @@ class MeshInterface:
"""Send a data packet to some other node
Keyword Arguments:
data -- the data to send, either as an array of bytes or as a protobuf (which will be automatically serialized to bytes)
destinationId {nodeId or nodeNum} -- where to send this message (default: {BROADCAST_ADDR})
portNum -- the application portnum (similar to IP port numbers) of the destination, see portnums.proto for a list
wantAck -- True if you want the message sent in a reliable manner (with retries and ack/nak provided for delivery)
wantResponse -- True if you want the service on the other side to send an application layer response
onResponse -- A closure of the form funct(packet), that will be called when a response packet arrives
(or the transaction is NAKed due to non receipt)
data -- the data to send, either as an array of bytes or
as a protobuf (which will be automatically
serialized to bytes)
destinationId {nodeId or nodeNum} -- where to send this
message (default: {BROADCAST_ADDR})
portNum -- the application portnum (similar to IP port numbers)
of the destination, see portnums.proto for a list
wantAck -- True if you want the message sent in a reliable
manner (with retries and ack/nak provided for delivery)
wantResponse -- True if you want the service on the other
side to send an application layer response
onResponse -- A closure of the form funct(packet), that will be
called when a response packet arrives (or the transaction
is NAKed due to non receipt)
Returns the sent packet. The id field will be populated in this packet and can be used to track future message acks/naks.
Returns the sent packet. The id field will be populated in this packet
and can be used to track future message acks/naks.
"""
if getattr(data, "SerializeToString", None):
logging.debug(f"Serializing protobuf as data: {stripnl(data)}")
@@ -224,16 +240,18 @@ class MeshInterface:
self._addResponseHandler(p.id, onResponse)
return p
def sendPosition(self, latitude=0.0, longitude=0.0, altitude=0, timeSec=0, destinationId=BROADCAST_ADDR, wantAck=False, wantResponse=False):
def sendPosition(self, latitude=0.0, longitude=0.0, altitude=0, timeSec=0,
destinationId=BROADCAST_ADDR, wantAck=False, wantResponse=False):
"""
Send a position packet to some other node (normally a broadcast)
Also, the device software will notice this packet and use it to automatically set its notion of
the local position.
Also, the device software will notice this packet and use it to automatically
set its notion of the local position.
If timeSec is not specified (recommended), we will use the local machine time.
Returns the sent packet. The id field will be populated in this packet and can be used to track future message acks/naks.
Returns the sent packet. The id field will be populated in this packet and
can be used to track future message acks/naks.
"""
p = mesh_pb2.Position()
if latitude != 0.0:
@@ -294,8 +312,8 @@ class MeshInterface:
meshPacket.want_ack = wantAck
meshPacket.hop_limit = hopLimit
# if the user hasn't set an ID for this packet (likely and recommended), we should pick a new unique ID
# so the message can be tracked.
# if the user hasn't set an ID for this packet (likely and recommended),
# we should pick a new unique ID so the message can be tracked.
if meshPacket.id == 0:
meshPacket.id = self._generatePacketId()
@@ -306,8 +324,7 @@ class MeshInterface:
def waitForConfig(self):
"""Block until radio config is received. Returns True if config has been received."""
success = self._timeout.waitForSet(self, attrs=('myInfo', 'nodes')
) and self.localNode.waitForConfig()
success = self._timeout.waitForSet(self, attrs=('myInfo', 'nodes')) and self.localNode.waitForConfig()
if not success:
raise Exception("Timed out waiting for interface config")
@@ -409,8 +426,7 @@ class MeshInterface:
def _sendToRadio(self, toRadio):
"""Send a ToRadio protobuf to the device"""
if self.noProto:
logging.warning(
f"Not sending packet because protocol use is disabled by noProto")
logging.warning(f"Not sending packet because protocol use is disabled by noProto")
else:
#logging.debug(f"Sending toRadio: {stripnl(toRadio)}")
self._sendToRadioImpl(toRadio)
@@ -421,7 +437,8 @@ class MeshInterface:
def _handleConfigComplete(self):
"""
Done with initial config messages, now send regular MeshPackets to ask for settings and channels
Done with initial config messages, now send regular MeshPackets
to ask for settings and channels
"""
self.localNode.requestConfig()
@@ -442,7 +459,7 @@ class MeshInterface:
failmsg = None
# Check for app too old
if self.myInfo.min_app_version > OUR_APP_VERSION:
failmsg = "This device needs a newer python client, please \"pip install --upgrade meshtastic\". "\
failmsg = "This device needs a newer python client, run 'pip install --upgrade meshtastic'."\
"For more information see https://tinyurl.com/5bjsxu32"
# check for firmware too old
@@ -470,13 +487,15 @@ class MeshInterface:
publishingThread.queueWork(lambda: pub.sendMessage("meshtastic.node.updated",
node=node, interface=self))
elif fromRadio.config_complete_id == self.configId:
# we ignore the config_complete_id, it is unneeded for our stream API fromRadio.config_complete_id
# we ignore the config_complete_id, it is unneeded for our
# stream API fromRadio.config_complete_id
logging.debug(f"Config complete ID {self.configId}")
self._handleConfigComplete()
elif fromRadio.HasField("packet"):
self._handlePacketFromRadio(fromRadio.packet)
elif fromRadio.rebooted:
# Tell clients the device went away. Careful not to call the overridden subclass version that closes the serial port
# Tell clients the device went away. Careful not to call the overridden
# subclass version that closes the serial port
MeshInterface._disconnected(self)
self._startConfig() # redownload the node db etc...
@@ -570,8 +589,9 @@ class MeshInterface:
# byte array.
decoded["payload"] = meshPacket.decoded.payload
# UNKNOWN_APP is the default protobuf portnum value, and therefore if not set it will not be populated at all
# to make API usage easier, set it to prevent confusion
# UNKNOWN_APP is the default protobuf portnum value, and therefore if not
# set it will not be populated at all to make API usage easier, set
# it to prevent confusion
if not "portnum" in decoded:
decoded["portnum"] = portnums_pb2.PortNum.Name(
portnums_pb2.PortNum.UNKNOWN_APP)
@@ -580,8 +600,9 @@ class MeshInterface:
topic = f"meshtastic.receive.data.{portnum}"
# decode position protobufs and update nodedb, provide decoded version as "position" in the published msg
# move the following into a 'decoders' API that clients could register?
# decode position protobufs and update nodedb, provide decoded version
# as "position" in the published msg move the following into a 'decoders'
# API that clients could register?
portNumInt = meshPacket.decoded.portnum # we want portnum as an int
handler = protocols.get(portNumInt)
# The decoded protobuf as a dictionary (if we understand this message)

62
meshtastic/node.py
View File

@@ -1,4 +1,4 @@
""" Node class
"""Node class
"""
import logging
@@ -24,7 +24,7 @@ class Node:
self.partialChannels = None
def showChannels(self):
"""Show human readable description of our channels"""
"""Show human readable description of our channels."""
print("Channels:")
if self.channels:
for c in self.channels:
@@ -47,9 +47,7 @@ class Node:
self.showChannels()
def requestConfig(self):
"""
Send regular MeshPackets to ask for settings and channels
"""
"""Send regular MeshPackets to ask for settings and channels."""
self.radioConfig = None
self.channels = None
self.partialChannels = [] # We keep our channels in a temp array until finished
@@ -104,9 +102,11 @@ class Node:
self.writeChannel(index, adminIndex=adminIndex)
index += 1
# if we are updating the local node, we might end up *moving* the admin channel index as we are writing
# if we are updating the local node, we might end up
# *moving* the admin channel index as we are writing
if (self.iface.localNode == self) and index >= adminIndex:
# We've now passed the old location for admin index (and writen it), so we can start finding it by name again
# We've now passed the old location for admin index
# (and writen it), so we can start finding it by name again
adminIndex = 0
def getChannelByName(self, name):
@@ -165,8 +165,7 @@ class Node:
return self._sendAdmin(p)
def getURL(self, includeAll: bool = True):
"""The sharable URL that describes the current channel
"""
"""The sharable URL that describes the current channel"""
# Only keep the primary/secondary channels, assume primary is first
channelSet = apponly_pb2.ChannelSet()
if self.channels:
@@ -212,9 +211,8 @@ class Node:
i = i + 1
def _requestSettings(self):
"""
Done with initial config messages, now send regular MeshPackets to ask for settings
"""
"""Done with initial config messages, now send regular
MeshPackets to ask for settings."""
p = admin_pb2.AdminMessage()
p.get_radio_request = True
@@ -233,26 +231,20 @@ class Node:
# Show progress message for super slow operations
if self != self.iface.localNode:
logging.info(
"Requesting preferences from remote node (this could take a while)")
print("Requesting preferences from remote node (this could take a while)")
return self._sendAdmin(p,
wantResponse=True,
onResponse=onResponse)
return self._sendAdmin(p, wantResponse=True, onResponse=onResponse)
def exitSimulator(self):
"""
Tell a simulator node to exit (this message is ignored for other nodes)
"""
"""Tell a simulator node to exit (this message
is ignored for other nodes)"""
p = admin_pb2.AdminMessage()
p.exit_simulator = True
return self._sendAdmin(p)
def reboot(self, secs: int = 10):
"""
Tell the node to reboot
"""
"""Tell the node to reboot."""
p = admin_pb2.AdminMessage()
p.reboot_seconds = secs
logging.info(f"Telling node to reboot in {secs} seconds")
@@ -263,6 +255,7 @@ class Node:
"""Fixup indexes and add disabled channels as needed"""
# Add extra disabled channels as needed
# TODO: These 2 lines seem to not do anything.
for index, ch in enumerate(self.channels):
ch.index = index # fixup indexes
@@ -281,21 +274,19 @@ class Node:
index += 1
def _requestChannel(self, channelNum: int):
"""
Done with initial config messages, now send regular MeshPackets to ask for settings
"""
"""Done with initial config messages, now send regular
MeshPackets to ask for settings"""
p = admin_pb2.AdminMessage()
p.get_channel_request = channelNum + 1
# Show progress message for super slow operations
if self != self.iface.localNode:
logging.info(
f"Requesting channel {channelNum} info from remote node (this could take a while)")
logging.info(f"Requesting channel {channelNum} info from remote node (this could take a while)")
else:
logging.debug(f"Requesting channel {channelNum}")
def onResponse(p):
"""A closure to handle the response packet"""
"""A closure to handle the response packet for requesting a channel"""
c = p["decoded"]["admin"]["raw"].get_channel_response
self.partialChannels.append(c)
self._timeout.reset() # We made foreward progress
@@ -305,9 +296,9 @@ class Node:
# for stress testing, we can always download all channels
fastChannelDownload = True
# Once we see a response that has NO settings, assume we are at the end of channels and stop fetching
quitEarly = (
c.role == channel_pb2.Channel.Role.DISABLED) and fastChannelDownload
# Once we see a response that has NO settings, assume
# we are at the end of channels and stop fetching
quitEarly = (c.role == channel_pb2.Channel.Role.DISABLED) and fastChannelDownload
if quitEarly or index >= self.iface.myInfo.max_channels - 1:
logging.debug("Finished downloading channels")
@@ -320,13 +311,10 @@ class Node:
else:
self._requestChannel(index + 1)
return self._sendAdmin(p,
wantResponse=True,
onResponse=onResponse)
return self._sendAdmin(p, wantResponse=True, onResponse=onResponse)
def _sendAdmin(self, p: admin_pb2.AdminMessage, wantResponse=False,
onResponse=None,
adminIndex=0):
onResponse=None, adminIndex=0):
"""Send an admin message to the specified node (or the local node if destNodeNum is zero)"""
if adminIndex == 0: # unless a special channel index was used, we want to use the admin index

30
meshtastic/stream_interface.py
View File

@@ -1,4 +1,4 @@
""" Stream Interface base class
"""Stream Interface base class
"""
import logging
import threading
@@ -25,7 +25,8 @@ class StreamInterface(MeshInterface):
Keyword Arguments:
devPath {string} -- A filepath to a device, i.e. /dev/ttyUSB0 (default: {None})
debugOut {stream} -- If a stream is provided, any debug serial output from the device will be emitted to that stream. (default: {None})
debugOut {stream} -- If a stream is provided, any debug serial output from the
device will be emitted to that stream. (default: {None})
Raises:
Exception: [description]
@@ -53,12 +54,14 @@ class StreamInterface(MeshInterface):
def connect(self):
"""Connect to our radio
Normally this is called automatically by the constructor, but if you passed in connectNow=False you can manually
start the reading thread later.
Normally this is called automatically by the constructor, but if you
passed in connectNow=False you can manually start the reading thread later.
"""
# Send some bogus UART characters to force a sleeping device to wake, and if the reading statemachine was parsing a bad packet make sure
# we write enought start bytes to force it to resync (we don't use START1 because we want to ensure it is looking for START1)
# Send some bogus UART characters to force a sleeping device to wake, and
# if the reading statemachine was parsing a bad packet make sure
# we write enought start bytes to force it to resync (we don't use START1
# because we want to ensure it is looking for START1)
p = bytearray([START2] * 32)
self._writeBytes(p)
time.sleep(0.1) # wait 100ms to give device time to start running
@@ -105,7 +108,8 @@ class StreamInterface(MeshInterface):
"""Close a connection to the device"""
logging.debug("Closing stream")
MeshInterface.close(self)
# pyserial cancel_read doesn't seem to work, therefore we ask the reader thread to close things for us
# pyserial cancel_read doesn't seem to work, therefore we ask the
# reader thread to close things for us
self._wantExit = True
if self._rxThread != threading.current_thread():
self._rxThread.join() # wait for it to exit
@@ -151,8 +155,7 @@ class StreamInterface(MeshInterface):
try:
self._handleFromRadio(self._rxBuf[HEADER_LEN:])
except Exception as ex:
logging.error(
f"Error while handling message from radio {ex}")
logging.error(f"Error while handling message from radio {ex}")
traceback.print_exc()
self._rxBuf = empty
else:
@@ -160,15 +163,12 @@ class StreamInterface(MeshInterface):
pass
except serial.SerialException as ex:
if not self._wantExit: # We might intentionally get an exception during shutdown
logging.warning(
f"Meshtastic serial port disconnected, disconnecting... {ex}")
logging.warning(f"Meshtastic serial port disconnected, disconnecting... {ex}")
except OSError as ex:
if not self._wantExit: # We might intentionally get an exception during shutdown
logging.error(
f"Unexpected OSError, terminating meshtastic reader... {ex}")
logging.error(f"Unexpected OSError, terminating meshtastic reader... {ex}")
except Exception as ex:
logging.error(
f"Unexpected exception, terminating meshtastic reader... {ex}")
logging.error(f"Unexpected exception, terminating meshtastic reader... {ex}")
finally:
logging.debug("reader is exiting")
self._disconnected()

9
meshtastic/tcp_interface.py
View File

@@ -1,4 +1,4 @@
""" TCPInterface class for interfacing with http endpoint
"""TCPInterface class for interfacing with http endpoint
"""
import logging
import socket
@@ -9,7 +9,8 @@ from .stream_interface import StreamInterface
class TCPInterface(StreamInterface):
"""Interface class for meshtastic devices over a TCP link"""
def __init__(self, hostname: AnyStr, debugOut=None, noProto=False, connectNow=True, portNumber=4403):
def __init__(self, hostname: AnyStr, debugOut=None, noProto=False,
connectNow=True, portNumber=4403):
"""Constructor, opens a connection to a specified IP address/hostname
Keyword Arguments:
@@ -33,8 +34,8 @@ class TCPInterface(StreamInterface):
"""Close a connection to the device"""
logging.debug("Closing TCP stream")
StreamInterface.close(self)
# Sometimes the socket read might be blocked in the reader thread. Therefore we force the shutdown by closing
# the socket here
# Sometimes the socket read might be blocked in the reader thread.
# Therefore we force the shutdown by closing the socket here
self._wantExit = True
if not self.socket is None:
try:

4
meshtastic/test.py
View File

@@ -1,5 +1,5 @@
""" With two radios connected serially, send and receive test
messages and report back if successful.
"""With two radios connected serially, send and receive test
messages and report back if successful.
"""
import logging
import time

12
meshtastic/tests/test_ble_interface.py Normal file
View File

@@ -0,0 +1,12 @@
"""Meshtastic unit tests for ble_interface.py"""
import pytest
from ..ble_interface import BLEInterface
@pytest.mark.unit
def test_BLEInterface():
"""Test that we can instantiate a BLEInterface"""
iface = BLEInterface('foo', debugOut=True, noProto=True)
iface.close()

92
meshtastic/tests/test_main.py
View File

@@ -7,9 +7,11 @@ import re
from unittest.mock import patch, MagicMock
import pytest
from meshtastic.__main__ import initParser, main, Globals
from meshtastic.__main__ import initParser, main, Globals, onReceive, onConnection
import meshtastic.radioconfig_pb2
from ..serial_interface import SerialInterface
from ..tcp_interface import TCPInterface
from ..ble_interface import BLEInterface
from ..node import Node
from ..channel_pb2 import Channel
@@ -184,6 +186,44 @@ def test_main_info(capsys, reset_globals):
mo.assert_called()
@pytest.mark.unit
def test_main_info_with_tcp_interface(capsys, reset_globals):
"""Test --info"""
sys.argv = ['', '--info', '--host', 'meshtastic.local']
Globals.getInstance().set_args(sys.argv)
iface = MagicMock(autospec=TCPInterface)
def mock_showInfo():
print('inside mocked showInfo')
iface.showInfo.side_effect = mock_showInfo
with patch('meshtastic.tcp_interface.TCPInterface', return_value=iface) as mo:
main()
out, err = capsys.readouterr()
assert re.search(r'Connected to radio', out, re.MULTILINE)
assert re.search(r'inside mocked showInfo', out, re.MULTILINE)
assert err == ''
mo.assert_called()
@pytest.mark.unit
def test_main_info_with_ble_interface(capsys, reset_globals):
"""Test --info"""
sys.argv = ['', '--info', '--ble', 'foo']
Globals.getInstance().set_args(sys.argv)
iface = MagicMock(autospec=BLEInterface)
def mock_showInfo():
print('inside mocked showInfo')
iface.showInfo.side_effect = mock_showInfo
with patch('meshtastic.ble_interface.BLEInterface', return_value=iface) as mo:
main()
out, err = capsys.readouterr()
assert re.search(r'Connected to radio', out, re.MULTILINE)
assert re.search(r'inside mocked showInfo', out, re.MULTILINE)
assert err == ''
mo.assert_called()
@pytest.mark.unit
def test_main_no_proto(capsys, reset_globals):
"""Test --noproto (using --info for output)"""
@@ -377,6 +417,27 @@ def test_main_sendtext(capsys, reset_globals):
mo.assert_called()
@pytest.mark.unit
def test_main_sendtext_with_dest(capsys, reset_globals):
"""Test --sendtext with --dest"""
sys.argv = ['', '--sendtext', 'hello', '--dest', 'foo']
Globals.getInstance().set_args(sys.argv)
iface = MagicMock(autospec=SerialInterface)
def mock_sendText(text, dest, wantAck):
print('inside mocked sendText')
iface.sendText.side_effect = mock_sendText
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
main()
out, err = capsys.readouterr()
assert re.search(r'Connected to radio', out, re.MULTILINE)
assert re.search(r'Sending text message', out, re.MULTILINE)
assert re.search(r'inside mocked sendText', out, re.MULTILINE)
assert err == ''
mo.assert_called()
@pytest.mark.unit
def test_main_sendping(capsys, reset_globals):
"""Test --sendping"""
@@ -1106,3 +1167,32 @@ def test_main_setchan(capsys, reset_globals):
main()
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
@pytest.mark.unit
def test_main_onReceive_empty(reset_globals):
"""Test onReceive"""
sys.argv = ['']
Globals.getInstance().set_args(sys.argv)
iface = MagicMock(autospec=SerialInterface)
packet = {'decoded': 'foo'}
onReceive(packet, iface)
# TODO: how do we know we actually called it?
@pytest.mark.unit
def test_main_onConnection(reset_globals, capsys):
"""Test onConnection"""
sys.argv = ['']
Globals.getInstance().set_args(sys.argv)
iface = MagicMock(autospec=SerialInterface)
class TempTopic:
""" temp class for topic """
def getName(self):
""" return the fake name of a topic"""
return 'foo'
mytopic = TempTopic()
onConnection(iface, mytopic)
out, err = capsys.readouterr()
assert re.search(r'Connection changed: foo', out, re.MULTILINE)
assert err == ''

3
meshtastic/tests/test_mesh_interface.py
View File

@@ -13,6 +13,8 @@ def test_MeshInterface(capsys):
iface = MeshInterface(noProto=True)
iface.showInfo()
iface.localNode.showInfo()
iface.showNodes()
iface.close()
out, err = capsys.readouterr()
assert re.search(r'Owner: None \(None\)', out, re.MULTILINE)
assert re.search(r'Nodes', out, re.MULTILINE)
@@ -20,4 +22,3 @@ def test_MeshInterface(capsys):
assert re.search(r'Channels', out, re.MULTILINE)
assert re.search(r'Primary channel URL', out, re.MULTILINE)
assert err == ''
iface.close()

34
meshtastic/tests/test_node.py Normal file
View File

@@ -0,0 +1,34 @@
"""Meshtastic unit tests for node.py"""
import re
from unittest.mock import patch, MagicMock
import pytest
from ..node import Node
from ..serial_interface import SerialInterface
from ..admin_pb2 import AdminMessage
@pytest.mark.unit
def test_node(capsys):
"""Test that we can instantiate a Node"""
anode = Node('foo', 'bar')
anode.showChannels()
anode.showInfo()
out, err = capsys.readouterr()
assert re.search(r'Preferences', out)
assert re.search(r'Channels', out)
assert re.search(r'Primary channel URL', out)
assert err == ''
@pytest.mark.unit
def test_node_reqquestConfig():
"""Test run requestConfig"""
iface = MagicMock(autospec=SerialInterface)
amesg = MagicMock(autospec=AdminMessage)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
with patch('meshtastic.admin_pb2.AdminMessage', return_value=amesg):
anode = Node(mo, 'bar')
anode.requestConfig()

6
meshtastic/tests/test_util.py
View File

@@ -84,6 +84,12 @@ def test_pskToString_many_bytes():
assert pskToString(bytes([0x02, 0x01])) == 'secret'
@pytest.mark.unit
def test_pskToString_simple():
"""Test pskToString simple"""
assert pskToString(bytes([0x03])) == 'simple2'
@pytest.mark.unit
def test_our_exit_zero_return_value():
"""Test our_exit with a zero return value"""

30
meshtastic/tunnel.py
View File

@@ -1,4 +1,4 @@
""" Code for IP tunnel over a mesh
"""Code for IP tunnel over a mesh
# Note python-pytuntap was too buggy
# using pip3 install pytap2
@@ -88,8 +88,8 @@ class Tunnel:
global tunnelInstance
tunnelInstance = self
logging.info(
"Starting IP to mesh tunnel (you must be root for this *pre-alpha* feature to work). Mesh members:")
logging.info("Starting IP to mesh tunnel (you must be root for this *pre-alpha* "\
"feature to work). Mesh members:")
pub.subscribe(onTunnelReceive, "meshtastic.receive.data.IP_TUNNEL_APP")
myAddr = self._nodeNumToIp(self.iface.myInfo.my_node_num)
@@ -117,8 +117,8 @@ class Tunnel:
else:
logging.debug(
f"Received mesh tunnel message type={type(p)} len={len(p)}")
# we don't really need to check for filtering here (sender should have checked), but this provides
# useful debug printing on types of packets received
# we don't really need to check for filtering here (sender should have checked),
# but this provides useful debug printing on types of packets received
if not self._shouldFilterPacket(p):
self.tun.write(p)
@@ -137,8 +137,8 @@ class Tunnel:
icmpType = p[20]
icmpCode = p[21]
checksum = p[22:24]
logging.debug(
f"forwarding ICMP message src={ipstr(srcaddr)}, dest={ipstr(destAddr)}, type={icmpType}, code={icmpCode}, checksum={checksum}")
# pylint: disable=line-too-long
logging.debug(f"forwarding ICMP message src={ipstr(srcaddr)}, dest={ipstr(destAddr)}, type={icmpType}, code={icmpCode}, checksum={checksum}")
# reply to pings (swap src and dest but keep rest of packet unchanged)
#pingback = p[:12]+p[16:20]+p[12:16]+p[20:]
# tap.write(pingback)
@@ -157,14 +157,12 @@ class Tunnel:
destport = readnet_u16(p, subheader + 2)
if destport in tcpBlacklist:
ignore = True
logging.log(
LOG_TRACE, f"ignoring blacklisted TCP port {destport}")
logging.log(LOG_TRACE, f"ignoring blacklisted TCP port {destport}")
else:
logging.debug(
f"forwarding tcp srcport={srcport}, destport={destport}")
logging.debug(f"forwarding tcp srcport={srcport}, destport={destport}")
else:
logging.warning(
f"forwarding unexpected protocol 0x{protocol:02x}, src={ipstr(srcaddr)}, dest={ipstr(destAddr)}")
logging.warning(f"forwarding unexpected protocol 0x{protocol:02x}, "\
"src={ipstr(srcaddr)}, dest={ipstr(destAddr)}")
return ignore
@@ -200,13 +198,11 @@ class Tunnel:
"""Forward the provided IP packet into the mesh"""
nodeId = self._ipToNodeId(destAddr)
if nodeId is not None:
logging.debug(
f"Forwarding packet bytelen={len(p)} dest={ipstr(destAddr)}, destNode={nodeId}")
logging.debug(f"Forwarding packet bytelen={len(p)} dest={ipstr(destAddr)}, destNode={nodeId}")
self.iface.sendData(
p, nodeId, portnums_pb2.IP_TUNNEL_APP, wantAck=False)
else:
logging.warning(
f"Dropping packet because no node found for destIP={ipstr(destAddr)}")
logging.warning(f"Dropping packet because no node found for destIP={ipstr(destAddr)}")
def close(self):
"""Close"""

11
meshtastic/util.py
View File

@@ -1,4 +1,4 @@
""" Utility functions.
"""Utility functions.
"""
import traceback
from queue import Queue
@@ -39,7 +39,8 @@ def fromPSK(valstr):
def fromStr(valstr):
"""try to parse as int, float or bool (and fallback to a string as last resort)
"""Try to parse as int, float or bool (and fallback to a string as last resort)
Returns: an int, bool, float, str or byte array (for strings of hex digits)
Args:
@@ -82,13 +83,13 @@ def pskToString(psk: bytes):
def stripnl(s):
"""remove newlines from a string (and remove extra whitespace)"""
"""Remove newlines from a string (and remove extra whitespace)"""
s = str(s).replace("\n", " ")
return ' '.join(s.split())
def fixme(message):
"""raise an exception for things that needs to be fixed"""
"""Raise an exception for things that needs to be fixed"""
raise Exception(f"FIXME: {message}")
@@ -174,7 +175,7 @@ def our_exit(message, return_value = 1):
def support_info():
"""Print out info that is helping in support of the cli."""
"""Print out info that helps troubleshooting of the cli."""
print('')
print('If having issues with meshtastic cli or python library')
print('or wish to make feature requests, visit:')