diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 9556670..116ada3 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -498,7 +498,10 @@ def onConnected(interface): from . import tunnel # Even if others said we could close, stay open if the user asked for a tunnel closeNow = False - tunnel.Tunnel(interface, subnet=args.tunnel_net) + if interface.noProto: + logging.warning(f"Not starting Tunnel - disabled by noProto") + else: + tunnel.Tunnel(interface, subnet=args.tunnel_net) # if the user didn't ask for serial debugging output, we might want to exit after we've done our operation if (not args.seriallog) and closeNow: diff --git a/meshtastic/stream_interface.py b/meshtastic/stream_interface.py index 948054c..8b81f9c 100644 --- a/meshtastic/stream_interface.py +++ b/meshtastic/stream_interface.py @@ -39,10 +39,7 @@ class StreamInterface(MeshInterface): self._wantExit = False # FIXME, figure out why daemon=True causes reader thread to exit too early - if noProto: - self._rxThread = None - else: - self._rxThread = threading.Thread(target=self.__reader, args=(), daemon=True) + self._rxThread = threading.Thread(target=self.__reader, args=(), daemon=True) MeshInterface.__init__(self, debugOut=debugOut, noProto=noProto) @@ -68,8 +65,7 @@ class StreamInterface(MeshInterface): if not self.noProto: time.sleep(0.1) # wait 100ms to give device time to start running - if not self.noProto: - self._rxThread.start() + self._rxThread.start() self._startConfig() @@ -121,9 +117,8 @@ class StreamInterface(MeshInterface): # pyserial cancel_read doesn't seem to work, therefore we ask the # reader thread to close things for us self._wantExit = True - if not self.noProto: - if self._rxThread != threading.current_thread(): - self._rxThread.join() # wait for it to exit + if self._rxThread != threading.current_thread(): + self._rxThread.join() # wait for it to exit def __reader(self): """The reader thread that reads bytes from our stream""" diff --git a/meshtastic/tests/test_main.py b/meshtastic/tests/test_main.py index 831474c..44478ff 100644 --- a/meshtastic/tests/test_main.py +++ b/meshtastic/tests/test_main.py @@ -1696,8 +1696,9 @@ def test_tunnel_no_args(capsys, reset_globals): @pytest.mark.unit +@patch('meshtastic.util.findPorts', return_value=[]) @patch('platform.system') -def test_tunnel_tunnel_arg(mock_platform_system, capsys, reset_globals): +def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, patched_find_ports, caplog, capsys, reset_globals): """Test tunnel with tunnel arg (act like we are on a linux system)""" a_mock = MagicMock() a_mock.return_value = 'Linux' @@ -1705,18 +1706,21 @@ def test_tunnel_tunnel_arg(mock_platform_system, capsys, reset_globals): sys.argv = ['', '--tunnel'] Globals.getInstance().set_args(sys.argv) print(f'platform.system():{platform.system()}') - with pytest.raises(SystemExit) as pytest_wrapped_e: - tunnelMain() - mock_platform_system.assert_called() - assert pytest_wrapped_e.type == SystemExit - assert pytest_wrapped_e.value.code == 1 - _, err = capsys.readouterr() - assert not re.search(r'usage: ', err, re.MULTILINE) + with caplog.at_level(logging.DEBUG): + with pytest.raises(SystemExit) as pytest_wrapped_e: + tunnelMain() + mock_platform_system.assert_called() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + assert re.search(r'Warning: No Meshtastic devices detected', out, re.MULTILINE) + assert err == '' @pytest.mark.unit +@patch('meshtastic.util.findPorts', return_value=[]) @patch('platform.system') -def test_tunnel_subnet_arg(mock_platform_system, capsys, reset_globals): +def test_tunnel_subnet_arg_with_no_devices(mock_platform_system, patched_find_ports, caplog, capsys, reset_globals): """Test tunnel with subnet arg (act like we are on a linux system)""" a_mock = MagicMock() a_mock.return_value = 'Linux' @@ -1724,11 +1728,41 @@ def test_tunnel_subnet_arg(mock_platform_system, capsys, reset_globals): sys.argv = ['', '--subnet', 'foo'] Globals.getInstance().set_args(sys.argv) print(f'platform.system():{platform.system()}') - with pytest.raises(SystemExit) as pytest_wrapped_e: - tunnelMain() - mock_platform_system.assert_called() - assert pytest_wrapped_e.type == SystemExit - assert pytest_wrapped_e.value.code == 1 - out, err = capsys.readouterr() - assert not re.search(r'usage: ', err, re.MULTILINE) - assert re.search(r'Warning: No Meshtastic devices detected', out, re.MULTILINE) + with caplog.at_level(logging.DEBUG): + with pytest.raises(SystemExit) as pytest_wrapped_e: + tunnelMain() + mock_platform_system.assert_called() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + out, err = capsys.readouterr() + assert re.search(r'Warning: No Meshtastic devices detected', out, re.MULTILINE) + assert err == '' + + +@pytest.mark.unit +@patch('platform.system') +def test_tunnel_tunnel_arg(mock_platform_system, caplog, reset_globals, iface_with_nodes): + """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): + sys.exit(3) + + a_mock = MagicMock() + a_mock.return_value = 'Linux' + 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 + + with caplog.at_level(logging.DEBUG): + with patch('meshtastic.serial_interface.SerialInterface', return_value=iface): + with patch('time.sleep', side_effect=my_sleep): + with pytest.raises(SystemExit) as pytest_wrapped_e: + tunnelMain() + mock_platform_system.assert_called() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 3 + assert re.search(r'Not starting Tunnel', caplog.text, re.MULTILINE) diff --git a/meshtastic/tests/test_stream_interface.py b/meshtastic/tests/test_stream_interface.py index dbbf044..dc25525 100644 --- a/meshtastic/tests/test_stream_interface.py +++ b/meshtastic/tests/test_stream_interface.py @@ -1,7 +1,7 @@ """Meshtastic unit tests for stream_interface.py""" import logging -#import re +import re from unittest.mock import MagicMock import pytest @@ -37,45 +37,45 @@ def test_StreamInterface_with_noProto(caplog, reset_globals): ## Note: This takes a bit, so moving from unit to slow ## 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): -# """Test _sendToRadioImpl()""" -# -## def add_header(b): -## """Add header stuffs for radio""" -## bufLen = len(b) -## header = bytes([START1, START2, (bufLen >> 8) & 0xff, bufLen & 0xff]) -## return header + b -# -# # captured raw bytes of a Heltec2.1 radio with 2 channels (primary and a secondary channel named "gpio") -# raw_1_my_info = b'\x1a,\x08\xdc\x8c\xd5\xc5\x02\x18\r2\x0e1.2.49.5354c49P\x15]\xe1%\x17Eh\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01' -# raw_2_node_info = b'"9\x08\xdc\x8c\xd5\xc5\x02\x12(\n\t!28b5465c\x12\x0cUnknown 465c\x1a\x03?5C"\x06$o(\xb5F\\0\n\x1a\x02 1%M<\xc6a' -# # pylint: disable=C0301 -# raw_3_node_info = b'"C\x08\xa4\x8c\xd5\xc5\x02\x12(\n\t!28b54624\x12\x0cUnknown 4624\x1a\x03?24"\x06$o(\xb5F$0\n\x1a\x07 5MH<\xc6a%G<\xc6a=\x00\x00\xc0@' -# raw_4_complete = b'@\xcf\xe5\xd1\x8c\x0e' -# # pylint: disable=C0301 -# raw_5_prefs = b'Z6\r\\F\xb5(\x15\\F\xb5("\x1c\x08\x06\x12\x13*\x11\n\x0f0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#\xb8\t\x015]$\xddk5\xd5\x7f!b=M<\xc6aP\x03`F' -# # pylint: disable=C0301 -# raw_6_channel0 = b'Z.\r\\F\xb5(\x15\\F\xb5("\x14\x08\x06\x12\x0b:\t\x12\x05\x18\x01"\x01\x01\x18\x015^$\xddk5\xd6\x7f!b=M<\xc6aP\x03`F' -# # pylint: disable=C0301 -# raw_7_channel1 = b'ZS\r\\F\xb5(\x15\\F\xb5("9\x08\x06\x120:.\x08\x01\x12(" \xb4&\xb3\xc7\x06\xd8\xe39%\xba\xa5\xee\x8eH\x06\xf6\xf4H\xe8\xd5\xc1[ao\xb5Y\\\xb4"\xafmi*\x04gpio\x18\x025_$\xddk5\xd7\x7f!b=M<\xc6aP\x03`F' -# raw_8_channel2 = b'Z)\r\\F\xb5(\x15\\F\xb5("\x0f\x08\x06\x12\x06:\x04\x08\x02\x12\x005`$\xddk5\xd8\x7f!b=M<\xc6aP\x03`F' -# raw_blank = b'' -# -# test_data = b'hello' -# stream = MagicMock() -# #stream.read.return_value = add_header(test_data) -# stream.read.side_effect = [ raw_1_my_info, raw_2_node_info, raw_3_node_info, raw_4_complete, -# raw_5_prefs, raw_6_channel0, raw_7_channel1, raw_8_channel2, -# raw_blank, raw_blank] -# toRadio = MagicMock() -# toRadio.SerializeToString.return_value = test_data -# with caplog.at_level(logging.DEBUG): -# iface = StreamInterface(noProto=True, connectNow=False) -# iface.stream = stream -# iface.connect() -# iface._sendToRadioImpl(toRadio) -# 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) +@pytest.mark.unitslow +def test_sendToRadioImpl(caplog, reset_globals): + """Test _sendToRadioImpl()""" + +# def add_header(b): +# """Add header stuffs for radio""" +# bufLen = len(b) +# header = bytes([START1, START2, (bufLen >> 8) & 0xff, bufLen & 0xff]) +# return header + b + + # captured raw bytes of a Heltec2.1 radio with 2 channels (primary and a secondary channel named "gpio") + raw_1_my_info = b'\x1a,\x08\xdc\x8c\xd5\xc5\x02\x18\r2\x0e1.2.49.5354c49P\x15]\xe1%\x17Eh\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01' + raw_2_node_info = b'"9\x08\xdc\x8c\xd5\xc5\x02\x12(\n\t!28b5465c\x12\x0cUnknown 465c\x1a\x03?5C"\x06$o(\xb5F\\0\n\x1a\x02 1%M<\xc6a' + # pylint: disable=C0301 + raw_3_node_info = b'"C\x08\xa4\x8c\xd5\xc5\x02\x12(\n\t!28b54624\x12\x0cUnknown 4624\x1a\x03?24"\x06$o(\xb5F$0\n\x1a\x07 5MH<\xc6a%G<\xc6a=\x00\x00\xc0@' + raw_4_complete = b'@\xcf\xe5\xd1\x8c\x0e' + # pylint: disable=C0301 + raw_5_prefs = b'Z6\r\\F\xb5(\x15\\F\xb5("\x1c\x08\x06\x12\x13*\x11\n\x0f0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#\xb8\t\x015]$\xddk5\xd5\x7f!b=M<\xc6aP\x03`F' + # pylint: disable=C0301 + raw_6_channel0 = b'Z.\r\\F\xb5(\x15\\F\xb5("\x14\x08\x06\x12\x0b:\t\x12\x05\x18\x01"\x01\x01\x18\x015^$\xddk5\xd6\x7f!b=M<\xc6aP\x03`F' + # pylint: disable=C0301 + raw_7_channel1 = b'ZS\r\\F\xb5(\x15\\F\xb5("9\x08\x06\x120:.\x08\x01\x12(" \xb4&\xb3\xc7\x06\xd8\xe39%\xba\xa5\xee\x8eH\x06\xf6\xf4H\xe8\xd5\xc1[ao\xb5Y\\\xb4"\xafmi*\x04gpio\x18\x025_$\xddk5\xd7\x7f!b=M<\xc6aP\x03`F' + raw_8_channel2 = b'Z)\r\\F\xb5(\x15\\F\xb5("\x0f\x08\x06\x12\x06:\x04\x08\x02\x12\x005`$\xddk5\xd8\x7f!b=M<\xc6aP\x03`F' + raw_blank = b'' + + test_data = b'hello' + stream = MagicMock() + #stream.read.return_value = add_header(test_data) + stream.read.side_effect = [ raw_1_my_info, raw_2_node_info, raw_3_node_info, raw_4_complete, + raw_5_prefs, raw_6_channel0, raw_7_channel1, raw_8_channel2, + raw_blank, raw_blank] + toRadio = MagicMock() + toRadio.SerializeToString.return_value = test_data + with caplog.at_level(logging.DEBUG): + iface = StreamInterface(noProto=True, connectNow=False) + iface.stream = stream + iface.connect() + iface._sendToRadioImpl(toRadio) + 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) diff --git a/meshtastic/tests/test_util.py b/meshtastic/tests/test_util.py index 5679d1a..23bf371 100644 --- a/meshtastic/tests/test_util.py +++ b/meshtastic/tests/test_util.py @@ -3,13 +3,14 @@ import re import logging +from unittest.mock import patch import pytest from meshtastic.util import (fixme, stripnl, pskToString, our_exit, support_info, genPSK256, fromStr, fromPSK, quoteBooleans, catchAndIgnore, remove_keys_from_dict, Timeout, hexstr, - ipstr, readnet_u16) + ipstr, readnet_u16, findPorts) @pytest.mark.unit @@ -211,3 +212,10 @@ def test_ipstr(): def test_readnet_u16(): """Test readnet_u16()""" assert readnet_u16(b'123456', 2) == 13108 + + +@pytest.mark.unit +@patch('serial.tools.list_ports.comports', return_value=[]) +def test_findPorts_when_none_found(patch_comports): + """Test findPorts()""" + assert not findPorts()