mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2026-01-20 05:37:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #200 from mkinney/tuning_tests

Browse Source 
Tuning tests
This commit is contained in:
mkinney
2021-12-31 12:30:00 -08:00
committed by GitHub
parent bd697864e4 475ddcc8dd
commit 9d0bc09e0f
11 changed files with 254 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@@ -16,7 +16,7 @@ lint:
# show the slowest unit tests
slow:
pytest --durations=0
pytest --durations=5
# run the coverage report and open results in a browser
cov:

2
meshtastic/__main__.py
View File

@@ -42,7 +42,7 @@ def onReceive(packet, interface):
interface.sendText(reply)
except Exception as ex:
print(ex)
print(f'Warning: There is no field {ex} in the packet.')
def onConnection(interface, topic=pub.AUTO_TOPIC):

52
meshtastic/tests/test_main.py
View File

@@ -63,7 +63,7 @@ def test_main_main_version(capsys, reset_globals):
@pytest.mark.unit
def test_main_main_no_args(reset_globals):
def test_main_main_no_args(reset_globals, capsys):
"""Test with no args"""
sys.argv = ['']
Globals.getInstance().set_args(sys.argv)
@@ -72,6 +72,8 @@ def test_main_main_no_args(reset_globals):
main()
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
_, err = capsys.readouterr()
assert re.search(r'usage:', err, re.MULTILINE)
@pytest.mark.unit
@@ -112,7 +114,7 @@ def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals):
@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=[])
def test_main_test_no_ports(patched_find_ports, reset_globals):
def test_main_test_no_ports(patched_find_ports, reset_globals, capsys):
"""Test --test with no hardware"""
sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv)
@@ -123,11 +125,14 @@ def test_main_test_no_ports(patched_find_ports, reset_globals):
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
patched_find_ports.assert_called()
out, err = capsys.readouterr()
assert re.search(r'Warning: Must have at least two devices connected to USB', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1'])
def test_main_test_one_port(patched_find_ports, reset_globals):
def test_main_test_one_port(patched_find_ports, reset_globals, capsys):
"""Test --test with one fake port"""
sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv)
@@ -138,12 +143,14 @@ def test_main_test_one_port(patched_find_ports, reset_globals):
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
patched_find_ports.assert_called()
out, err = capsys.readouterr()
assert re.search(r'Warning: Must have at least two devices connected to USB', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@patch('meshtastic.test.testAll', return_value=True)
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1', '/dev/ttyFake2'])
def test_main_test_two_ports_success(patched_find_ports, patched_test_all, reset_globals):
def test_main_test_two_ports_success(patched_test_all, reset_globals, capsys):
"""Test --test two fake ports and testAll() is a simulated success"""
sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv)
@@ -152,14 +159,15 @@ def test_main_test_two_ports_success(patched_find_ports, patched_test_all, reset
main()
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 0
# TODO: why does this fail? patched_find_ports.assert_called()
patched_test_all.assert_called()
out, err = capsys.readouterr()
assert re.search(r'Test was a success.', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@patch('meshtastic.test.testAll', return_value=False)
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1', '/dev/ttyFake2'])
def test_main_test_two_ports_fails(patched_find_ports, patched_test_all, reset_globals):
def test_main_test_two_ports_fails(patched_test_all, reset_globals, capsys):
"""Test --test two fake ports and testAll() is a simulated failure"""
sys.argv = ['', '--test']
Globals.getInstance().set_args(sys.argv)
@@ -168,8 +176,10 @@ def test_main_test_two_ports_fails(patched_find_ports, patched_test_all, reset_g
main()
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
# TODO: why does this fail? patched_find_ports.assert_called()
patched_test_all.assert_called()
out, err = capsys.readouterr()
assert re.search(r'Test was not successful.', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@@ -1233,18 +1243,23 @@ def test_main_setchan(capsys, reset_globals):
main()
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
_, err = capsys.readouterr()
assert re.search(r'usage:', err, re.MULTILINE)
@pytest.mark.unit
def test_main_onReceive_empty(caplog, reset_globals):
def test_main_onReceive_empty(caplog, reset_globals, capsys):
"""Test onReceive"""
sys.argv = ['']
Globals.getInstance().set_args(sys.argv)
args = MagicMock()
Globals.getInstance().set_args(args)
iface = MagicMock(autospec=SerialInterface)
packet = {'decoded': 'foo'}
packet = {}
with caplog.at_level(logging.DEBUG):
onReceive(packet, iface)
assert re.search(r'in onReceive', caplog.text, re.MULTILINE)
out, err = capsys.readouterr()
assert re.search(r"Warning: There is no field 'to' in the packet.", out, re.MULTILINE)
assert err == ''
# TODO: use this captured position app message (might want/need in the future)
@@ -1259,7 +1274,7 @@ def test_main_onReceive_empty(caplog, reset_globals):
# }
@pytest.mark.unit
def test_main_onReceive_with_sendtext(caplog, reset_globals):
def test_main_onReceive_with_sendtext(caplog, capsys, reset_globals):
"""Test onReceive with sendtext
The entire point of this test is to make sure the interface.close() call
is made in onReceive().
@@ -1288,6 +1303,9 @@ def test_main_onReceive_with_sendtext(caplog, reset_globals):
onReceive(packet, iface)
assert re.search(r'in onReceive', caplog.text, re.MULTILINE)
mo.assert_called()
out, err = capsys.readouterr()
assert re.search(r'Sending text message hello to', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@@ -1741,7 +1759,7 @@ def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, patched_find_po
@pytest.mark.unit
@patch('platform.system')
def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, iface_with_nodes):
def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, 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):
@@ -1752,7 +1770,6 @@ def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, iface_wi
mock_platform_system.side_effect = a_mock
sys.argv = ['', '--tunnel']
Globals.getInstance().set_args(sys.argv)
print(f'platform.system():{platform.system()}')
iface = iface_with_nodes
iface.myInfo.my_node_num = 2475227164
@@ -1766,3 +1783,6 @@ def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, iface_wi
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 3
assert re.search(r'Not starting Tunnel', caplog.text, re.MULTILINE)
out, err = capsys.readouterr()
assert re.search(r'Connected to radio', out, re.MULTILINE)
assert err == ''

1
meshtastic/tests/test_mesh_interface.py
View File

@@ -60,7 +60,6 @@ def test_getMyUser(reset_globals, iface_with_nodes):
iface.myInfo.my_node_num = 2475227164
myuser = iface.getMyUser()
print(f'myuser:{myuser}')
assert myuser is not None
assert myuser["id"] == '!9388f81c'

20
meshtastic/tests/test_node.py
View File

@@ -29,7 +29,7 @@ def test_node(capsys):
@pytest.mark.unit
def test_node_reqquestConfig():
def test_node_requestConfig(capsys):
"""Test run requestConfig"""
iface = MagicMock(autospec=SerialInterface)
amesg = MagicMock(autospec=AdminMessage)
@@ -37,6 +37,9 @@ def test_node_reqquestConfig():
with patch('meshtastic.admin_pb2.AdminMessage', return_value=amesg):
anode = Node(mo, 'bar')
anode.requestConfig()
out, err = capsys.readouterr()
assert re.search(r'Requesting preferences from remote node', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@@ -106,13 +109,16 @@ def test_reboot(caplog):
@pytest.mark.unit
def test_setURL_empty_url():
def test_setURL_empty_url(capsys):
"""Test reboot"""
anode = Node('foo', 'bar', noProto=True)
with pytest.raises(SystemExit) as pytest_wrapped_e:
anode.setURL('')
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
out, err = capsys.readouterr()
assert re.search(r'Warning: No RadioConfig has been read', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@@ -133,7 +139,7 @@ def test_setURL_valid_URL(caplog):
@pytest.mark.unit
def test_setURL_valid_URL_but_no_settings(caplog):
def test_setURL_valid_URL_but_no_settings(caplog, capsys):
"""Test setURL"""
iface = MagicMock(autospec=SerialInterface)
url = "https://www.meshtastic.org/d/#"
@@ -143,6 +149,9 @@ def test_setURL_valid_URL_but_no_settings(caplog):
anode.setURL(url)
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
out, err = capsys.readouterr()
assert re.search(r'Warning: There were no settings', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit
@@ -623,7 +632,7 @@ def test_writeConfig(caplog):
@pytest.mark.unit
def test_requestChannel_not_localNode(caplog):
def test_requestChannel_not_localNode(caplog, capsys):
"""Test _requestChannel()"""
iface = MagicMock(autospec=SerialInterface)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -633,6 +642,9 @@ def test_requestChannel_not_localNode(caplog):
with caplog.at_level(logging.DEBUG):
anode._requestChannel(0)
assert re.search(r'Requesting channel 0 info from remote node', caplog.text, re.MULTILINE)
out, err = capsys.readouterr()
assert re.search(r'Requesting channel 0 info', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit

5
meshtastic/tests/test_remote_hardware.py
View File

@@ -79,7 +79,7 @@ def test_watchGPIOs(caplog):
@pytest.mark.unit
def test_sendHardware_no_nodeid():
def test_sendHardware_no_nodeid(capsys):
"""Test sending no nodeid to _sendHardware()"""
iface = MagicMock(autospec=SerialInterface)
with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
@@ -87,3 +87,6 @@ def test_sendHardware_no_nodeid():
rhw = RemoteHardwareClient(mo)
rhw._sendHardware(None, None)
assert pytest_wrapped_e.type == SystemExit
out, err = capsys.readouterr()
assert re.search(r'Warning: Must use a destination node ID', out)
assert err == ''

8
meshtastic/tests/test_serial_interface.py
View File

@@ -11,7 +11,7 @@ from ..serial_interface import SerialInterface
@pytest.mark.unit
@patch('serial.Serial')
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyUSBfake'])
def test_SerialInterface_single_port(mocked_findPorts, mocked_serial):
def test_SerialInterface_single_port(mocked_findPorts, mocked_serial, capsys):
"""Test that we can instantiate a SerialInterface with a single port"""
iface = SerialInterface(noProto=True)
iface.showInfo()
@@ -19,6 +19,12 @@ def test_SerialInterface_single_port(mocked_findPorts, mocked_serial):
iface.close()
mocked_findPorts.assert_called()
mocked_serial.assert_called()
out, err = capsys.readouterr()
assert re.search(r'Nodes in mesh', out, re.MULTILINE)
assert re.search(r'Preferences', out, re.MULTILINE)
assert re.search(r'Channels', out, re.MULTILINE)
assert re.search(r'Primary channel', out, re.MULTILINE)
assert err == ''
@pytest.mark.unit

1
meshtastic/tests/test_stream_interface.py
View File

@@ -78,4 +78,3 @@ def test_sendToRadioImpl(caplog, reset_globals):
assert re.search(r'Sending: ', caplog.text, re.MULTILINE)
assert re.search(r'reading character', caplog.text, re.MULTILINE)
assert re.search(r'In reader loop', caplog.text, re.MULTILINE)
print(caplog.text)

171
meshtastic/tests/test_tunnel.py
View File

@@ -39,7 +39,7 @@ def test_Tunnel_without_interface(mock_platform_system, reset_globals):
assert pytest_wrapped_e.type == Exception
@pytest.mark.unit
@pytest.mark.unitslow
@patch('platform.system')
def test_Tunnel_with_interface(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test that we can not instantiate a Tunnel without a valid interface"""
@@ -58,7 +58,7 @@ def test_Tunnel_with_interface(mock_platform_system, caplog, reset_globals, ifac
assert re.search(r'Not sending packet', caplog.text, re.MULTILINE)
@pytest.mark.unit
@pytest.mark.unitslow
@patch('platform.system')
def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test onTunnelReceive"""
@@ -97,3 +97,170 @@ def test_onTunnelReceive_from_someone_else(mock_platform_system, caplog, reset_g
Globals.getInstance().set_tunnelInstance(tun)
onTunnelReceive(packet, iface)
assert re.search(r'in onTunnelReceive', caplog.text, re.MULTILINE)
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_random(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# random packet
packet = b'1234567890123456789012345678901234567890'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert not ignore
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# faked IGMP
packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert ignore
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_icmp(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# faked ICMP
packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert re.search(r'forwarding ICMP message', caplog.text, re.MULTILINE)
assert not ignore
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_udp(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# faked UDP
packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert re.search(r'forwarding udp', caplog.text, re.MULTILINE)
assert not ignore
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# faked UDP
packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07\x6c\x07\x6c\x00\x00\x00'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
# Note: custom logging level
LOG_TRACE = 5
with caplog.at_level(LOG_TRACE):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert re.search(r'ignoring blacklisted UDP', caplog.text, re.MULTILINE)
assert ignore
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_tcp(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# faked TCP
packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert re.search(r'forwarding tcp', caplog.text, re.MULTILINE)
assert not ignore
@pytest.mark.unit
@patch('platform.system')
def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _shouldFilterPacket()"""
iface = iface_with_nodes
iface.noProto = True
# faked TCP
packet = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x17\x0c\x17\x0c\x00\x00\x00'
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
# Note: custom logging level
LOG_TRACE = 5
with caplog.at_level(LOG_TRACE):
with patch('socket.socket'):
tun = Tunnel(iface)
ignore = tun._shouldFilterPacket(packet)
assert re.search(r'ignoring blacklisted TCP', caplog.text, re.MULTILINE)
assert ignore
@pytest.mark.unit
@patch('platform.system')
def test_ipToNodeId_none(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _ipToNodeId()"""
iface = iface_with_nodes
iface.noProto = True
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
nodeid = tun._ipToNodeId('something not useful')
assert nodeid is None
@pytest.mark.unit
@patch('platform.system')
def test_ipToNodeId_all(mock_platform_system, caplog, reset_globals, iface_with_nodes):
"""Test _ipToNodeId()"""
iface = iface_with_nodes
iface.noProto = True
a_mock = MagicMock()
a_mock.return_value = 'Linux'
mock_platform_system.side_effect = a_mock
with caplog.at_level(logging.DEBUG):
with patch('socket.socket'):
tun = Tunnel(iface)
nodeid = tun._ipToNodeId(b'\x00\x00\xff\xff')
assert nodeid == '^all'

26
meshtastic/tests/test_util.py
View File

@@ -41,7 +41,7 @@ def test_fromStr():
assert fromStr('abc') == 'abc'
@pytest.mark.unit
@pytest.mark.unitslow
def test_quoteBooleans():
"""Test quoteBooleans"""
assert quoteBooleans('') == ''
@@ -94,7 +94,7 @@ def test_pskToString_one_byte_non_zero_value():
assert pskToString(bytes([0x01])) == 'default'
@pytest.mark.unit
@pytest.mark.unitslow
def test_pskToString_many_bytes():
"""Test pskToString many bytes"""
assert pskToString(bytes([0x02, 0x01])) == 'secret'
@@ -106,20 +106,26 @@ def test_pskToString_simple():
assert pskToString(bytes([0x03])) == 'simple2'
@pytest.mark.unit
def test_our_exit_zero_return_value():
@pytest.mark.unitslow
def test_our_exit_zero_return_value(capsys):
"""Test our_exit with a zero return value"""
with pytest.raises(SystemExit) as pytest_wrapped_e:
our_exit("Warning: Some message", 0)
out, err = capsys.readouterr()
assert re.search(r'Warning: Some message', out, re.MULTILINE)
assert err == ''
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 0
@pytest.mark.unit
def test_our_exit_non_zero_return_value():
def test_our_exit_non_zero_return_value(capsys):
"""Test our_exit with a non-zero return value"""
with pytest.raises(SystemExit) as pytest_wrapped_e:
our_exit("Error: Some message", 1)
out, err = capsys.readouterr()
assert re.search(r'Error: Some message', out, re.MULTILINE)
assert err == ''
assert pytest_wrapped_e.type == SystemExit
assert pytest_wrapped_e.value.code == 1
@@ -154,7 +160,7 @@ def test_catchAndIgnore(caplog):
assert re.search(r'Exception thrown in something', caplog.text, re.MULTILINE)
@pytest.mark.unit
@pytest.mark.unitslow
def test_remove_keys_from_dict_empty_keys_empty_dict():
"""Test when keys and dict both are empty"""
assert not remove_keys_from_dict((), {})
@@ -178,7 +184,7 @@ def test_remove_keys_from_dict():
assert remove_keys_from_dict(('b'), {'a':1, 'b':2}) == {'a':1}
@pytest.mark.unit
@pytest.mark.unitslow
def test_Timeout_not_found():
"""Test Timeout()"""
to = Timeout(0.2)
@@ -186,7 +192,7 @@ def test_Timeout_not_found():
to.waitForSet('bar', attrs)
@pytest.mark.unit
@pytest.mark.unitslow
def test_Timeout_found():
"""Test Timeout()"""
to = Timeout(0.2)
@@ -194,7 +200,7 @@ def test_Timeout_found():
to.waitForSet('bar', attrs)
@pytest.mark.unit
@pytest.mark.unitslow
def test_hexstr():
"""Test hexstr()"""
assert hexstr(b'123') == '31:32:33'
@@ -208,7 +214,7 @@ def test_ipstr():
assert ipstr(b'') == ''
@pytest.mark.unit
@pytest.mark.unitslow
def test_readnet_u16():
"""Test readnet_u16()"""
assert readnet_u16(b'123456', 2) == 13108

6
meshtastic/tunnel.py
View File

@@ -66,7 +66,9 @@ class Tunnel:
}
"""A list of TCP services to block"""
self.tcpBlacklist = {}
self.tcpBlacklist = {
5900, # VNC (Note: Only adding for testing purposes.)
}
"""A list of protocols we ignore"""
self.protocolBlacklist = {
@@ -117,7 +119,7 @@ class Tunnel:
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
if not self.iface.noProto: # could move this one line down later
if not self.iface.noProto:
if not self._shouldFilterPacket(p):
self.tun.write(p)