diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index 116ada3..9e5e19c 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -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): diff --git a/meshtastic/tests/test_main.py b/meshtastic/tests/test_main.py index 44478ff..474435b 100644 --- a/meshtastic/tests/test_main.py +++ b/meshtastic/tests/test_main.py @@ -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 == '' diff --git a/meshtastic/tests/test_mesh_interface.py b/meshtastic/tests/test_mesh_interface.py index 6a100e5..dd9f4ff 100644 --- a/meshtastic/tests/test_mesh_interface.py +++ b/meshtastic/tests/test_mesh_interface.py @@ -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' diff --git a/meshtastic/tests/test_node.py b/meshtastic/tests/test_node.py index 02e0f19..e2ab56a 100644 --- a/meshtastic/tests/test_node.py +++ b/meshtastic/tests/test_node.py @@ -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 diff --git a/meshtastic/tests/test_remote_hardware.py b/meshtastic/tests/test_remote_hardware.py index 80e6e16..1bff9ef 100644 --- a/meshtastic/tests/test_remote_hardware.py +++ b/meshtastic/tests/test_remote_hardware.py @@ -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 == '' diff --git a/meshtastic/tests/test_serial_interface.py b/meshtastic/tests/test_serial_interface.py index 10b4dbc..e89b46e 100644 --- a/meshtastic/tests/test_serial_interface.py +++ b/meshtastic/tests/test_serial_interface.py @@ -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 diff --git a/meshtastic/tests/test_stream_interface.py b/meshtastic/tests/test_stream_interface.py index dc25525..d87580e 100644 --- a/meshtastic/tests/test_stream_interface.py +++ b/meshtastic/tests/test_stream_interface.py @@ -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) diff --git a/meshtastic/tests/test_util.py b/meshtastic/tests/test_util.py index 0cb0ffe..8ca96cc 100644 --- a/meshtastic/tests/test_util.py +++ b/meshtastic/tests/test_util.py @@ -107,19 +107,25 @@ def test_pskToString_simple(): @pytest.mark.unitslow -def test_our_exit_zero_return_value(): +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