Module meshtastic.tests.test_main

Meshtastic unit tests for main.py

"""Meshtastic unit tests for __main__.py"""

import sys
import os
import re

from unittest.mock import patch, MagicMock
import pytest

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


@pytest.mark.unit
def test_main_init_parser_no_args(capsys, reset_globals):
    """Test no arguments"""
    sys.argv = ['']
    Globals.getInstance().set_args(sys.argv)
    initParser()
    out, err = capsys.readouterr()
    assert out == ''
    assert err == ''


@pytest.mark.unit
def test_main_init_parser_version(capsys, reset_globals):
    """Test --version"""
    sys.argv = ['', '--version']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        initParser()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 0
    out, err = capsys.readouterr()
    assert re.match(r'[0-9]+\.[0-9]+\.[0-9]', out)
    assert err == ''


@pytest.mark.unit
def test_main_main_version(capsys, reset_globals):
    """Test --version"""
    sys.argv = ['', '--version']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 0
    out, err = capsys.readouterr()
    assert re.match(r'[0-9]+\.[0-9]+\.[0-9]', out)
    assert err == ''


@pytest.mark.unit
def test_main_main_no_args(reset_globals):
    """Test with no args"""
    sys.argv = ['']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 1


@pytest.mark.unit
def test_main_support(capsys, reset_globals):
    """Test --support"""
    sys.argv = ['', '--support']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 0
    out, err = capsys.readouterr()
    assert re.search(r'System', out, re.MULTILINE)
    assert re.search(r'Platform', out, re.MULTILINE)
    assert re.search(r'Machine', out, re.MULTILINE)
    assert re.search(r'Executable', out, re.MULTILINE)
    assert err == ''


@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=[])
def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals):
    """Test --ch-index 1"""
    sys.argv = ['', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert Globals.getInstance().get_channel_index() == 1
    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 == ''
    patched_find_ports.assert_called()


@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=[])
def test_main_test_no_ports(patched_find_ports, reset_globals):
    """Test --test with no hardware"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    assert Globals.getInstance().get_target_node() is None
    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 1
    patched_find_ports.assert_called()


@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1'])
def test_main_test_one_port(patched_find_ports, reset_globals):
    """Test --test with one fake port"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    assert Globals.getInstance().get_target_node() is None
    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 1
    patched_find_ports.assert_called()


@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):
    """Test --test two fake ports and testAll() is a simulated success"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        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()


@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):
    """Test --test two fake ports and testAll() is a simulated failure"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        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()


@pytest.mark.unit
def test_main_info(capsys, reset_globals):
    """Test --info"""
    sys.argv = ['', '--info']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo
    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'inside mocked showInfo', out, re.MULTILINE)
        assert err == ''
        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)"""
    sys.argv = ['', '--info', '--noproto']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo

    # Override the time.sleep so there is no loop
    def my_sleep(amount):
        sys.exit(0)

    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:
                main()
            assert pytest_wrapped_e.type == SystemExit
            assert pytest_wrapped_e.value.code == 0
            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 == ''


@pytest.mark.unit
def test_main_info_with_seriallog_stdout(capsys, reset_globals):
    """Test --info"""
    sys.argv = ['', '--info', '--seriallog', 'stdout']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo
    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'inside mocked showInfo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_info_with_seriallog_output_txt(capsys, reset_globals):
    """Test --info"""
    sys.argv = ['', '--info', '--seriallog', 'output.txt']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo
    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'inside mocked showInfo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()
    # do some cleanup
    os.remove('output.txt')


@pytest.mark.unit
def test_main_qr(capsys, reset_globals):
    """Test --qr"""
    sys.argv = ['', '--qr']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    # TODO: could mock/check url
    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'Primary channel URL', out, re.MULTILINE)
        # if a qr code is generated it will have lots of these
        assert re.search(r'\[7m', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_nodes(capsys, reset_globals):
    """Test --nodes"""
    sys.argv = ['', '--nodes']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showNodes():
        print('inside mocked showNodes')
    iface.showNodes.side_effect = mock_showNodes
    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'inside mocked showNodes', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_set_owner_to_bob(capsys, reset_globals):
    """Test --set-owner bob"""
    sys.argv = ['', '--set-owner', 'bob']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    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'Setting device owner to bob', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_set_ham_to_KI123(capsys, reset_globals):
    """Test --set-ham KI123"""
    sys.argv = ['', '--set-ham', 'KI123']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_turnOffEncryptionOnPrimaryChannel():
        print('inside mocked turnOffEncryptionOnPrimaryChannel')
    def mock_setOwner(name, is_licensed):
        print('inside mocked setOwner')
    mocked_node.turnOffEncryptionOnPrimaryChannel.side_effect = mock_turnOffEncryptionOnPrimaryChannel
    mocked_node.setOwner.side_effect = mock_setOwner

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Setting HAM ID to KI123', out, re.MULTILINE)
        assert re.search(r'inside mocked setOwner', out, re.MULTILINE)
        assert re.search(r'inside mocked turnOffEncryptionOnPrimaryChannel', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_reboot(capsys, reset_globals):
    """Test --reboot"""
    sys.argv = ['', '--reboot']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_reboot():
        print('inside mocked reboot')
    mocked_node.reboot.side_effect = mock_reboot

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'inside mocked reboot', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_sendtext(capsys, reset_globals):
    """Test --sendtext"""
    sys.argv = ['', '--sendtext', 'hello']
    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_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"""
    sys.argv = ['', '--sendping']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendData(payload, dest, portNum, wantAck, wantResponse):
        print('inside mocked sendData')
    iface.sendData.side_effect = mock_sendData

    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 ping message', out, re.MULTILINE)
        assert re.search(r'inside mocked sendData', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_setlat(capsys, reset_globals):
    """Test --sendlat"""
    sys.argv = ['', '--setlat', '37.5']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_writeConfig():
        print('inside mocked writeConfig')
    mocked_node.writeConfig.side_effect = mock_writeConfig

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendPosition(lat, lon, alt):
        print('inside mocked sendPosition')
    iface.sendPosition.side_effect = mock_sendPosition
    iface.localNode.return_value = mocked_node

    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'Fixing latitude', out, re.MULTILINE)
        assert re.search(r'Setting device position', out, re.MULTILINE)
        assert re.search(r'inside mocked sendPosition', out, re.MULTILINE)
        # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_setlon(capsys, reset_globals):
    """Test --setlon"""
    sys.argv = ['', '--setlon', '-122.1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_writeConfig():
        print('inside mocked writeConfig')
    mocked_node.writeConfig.side_effect = mock_writeConfig

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendPosition(lat, lon, alt):
        print('inside mocked sendPosition')
    iface.sendPosition.side_effect = mock_sendPosition
    iface.localNode.return_value = mocked_node

    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'Fixing longitude', out, re.MULTILINE)
        assert re.search(r'Setting device position', out, re.MULTILINE)
        assert re.search(r'inside mocked sendPosition', out, re.MULTILINE)
        # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_setalt(capsys, reset_globals):
    """Test --setalt"""
    sys.argv = ['', '--setalt', '51']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_writeConfig():
        print('inside mocked writeConfig')
    mocked_node.writeConfig.side_effect = mock_writeConfig

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendPosition(lat, lon, alt):
        print('inside mocked sendPosition')
    iface.sendPosition.side_effect = mock_sendPosition
    iface.localNode.return_value = mocked_node

    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'Fixing altitude', out, re.MULTILINE)
        assert re.search(r'Setting device position', out, re.MULTILINE)
        assert re.search(r'inside mocked sendPosition', out, re.MULTILINE)
        # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_set_team_valid(capsys, reset_globals):
    """Test --set-team"""
    sys.argv = ['', '--set-team', 'CYAN']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_setOwner(team):
        print('inside mocked setOwner')
    mocked_node.setOwner.side_effect = mock_setOwner

    iface = MagicMock(autospec=SerialInterface)
    iface.localNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with patch('meshtastic.mesh_pb2.Team') as mm:
            mm.Name.return_value = 'FAKENAME'
            mm.Value.return_value = 'FAKEVAL'
            main()
            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'Setting team to', out, re.MULTILINE)
            assert err == ''
            mo.assert_called()
            mm.Name.assert_called()
            mm.Value.assert_called()


@pytest.mark.unit
def test_main_set_team_invalid(capsys, reset_globals):
    """Test --set-team using an invalid team name"""
    sys.argv = ['', '--set-team', 'NOTCYAN']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)

    def throw_an_exception(exc):
        raise ValueError("Fake exception.")

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with patch('meshtastic.mesh_pb2.Team') as mm:
            mm.Value.side_effect = throw_an_exception
            main()
            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'ERROR: Team', out, re.MULTILINE)
            assert err == ''
            mo.assert_called()
            mm.Value.assert_called()


@pytest.mark.unit
def test_main_seturl(capsys, reset_globals):
    """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)

    iface = MagicMock(autospec=SerialInterface)
    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 err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_set_valid(capsys, reset_globals):
    """Test --set with valid field"""
    sys.argv = ['', '--set', 'wifi_ssid', 'foo']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Set wifi_ssid to foo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_set_with_invalid(capsys, reset_globals):
    """Test --set with invalid field"""
    sys.argv = ['', '--set', 'foo', 'foo']
    Globals.getInstance().set_args(sys.argv)

    mocked_user_prefs = MagicMock()
    mocked_user_prefs.DESCRIPTOR.fields_by_name.get.return_value = None

    mocked_node = MagicMock(autospec=Node)
    mocked_node.radioConfig.preferences = ( mocked_user_prefs )

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'does not have an attribute called foo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


# TODO: write some negative --configure tests
@pytest.mark.unit
def test_main_configure(capsys, reset_globals):
    """Test --configure with valid file"""
    sys.argv = ['', '--configure', 'example_config.yaml']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Setting device owner', out, re.MULTILINE)
        assert re.search(r'Setting channel url', out, re.MULTILINE)
        assert re.search(r'Fixing altitude', out, re.MULTILINE)
        assert re.search(r'Fixing latitude', out, re.MULTILINE)
        assert re.search(r'Fixing longitude', out, re.MULTILINE)
        assert re.search(r'Writing modified preferences', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_add_valid(capsys, reset_globals):
    """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)

    mocked_channel = MagicMock(autospec=Channel)
    # TODO: figure out how to get it to print the channel name instead of MagicMock

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = False
    # set it up so we have free channels
    mocked_node.getDisabledChannel.return_value = mocked_channel

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Writing modified channels to device', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_add_invalid_name_too_long(capsys, reset_globals):
    """Test --ch-add with invalid channel name, name too long"""
    sys.argv = ['', '--ch-add', 'testingtestingtesting']
    Globals.getInstance().set_args(sys.argv)

    mocked_channel = MagicMock(autospec=Channel)
    # TODO: figure out how to get it to print the channel name instead of MagicMock

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = False
    # set it up so we have free channels
    mocked_node.getDisabledChannel.return_value = mocked_channel

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Channel name must be shorter', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_add_but_name_already_exists(capsys, reset_globals):
    """Test --ch-add with a channel name that already exists"""
    sys.argv = ['', '--ch-add', 'testing']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = True

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: This node already has', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_add_but_no_more_channels(capsys, reset_globals):
    """Test --ch-add with but there are no more channels"""
    sys.argv = ['', '--ch-add', 'testing']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = False
    # set it up so we have free channels
    mocked_node.getDisabledChannel.return_value = None

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: No free channels were found', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_del(capsys, reset_globals):
    """Test --ch-del with valid secondary channel to be deleted"""
    sys.argv = ['', '--ch-del', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Deleting channel', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_del_no_ch_index_specified(capsys, reset_globals):
    """Test --ch-del without a valid ch-index"""
    sys.argv = ['', '--ch-del']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Need to specify', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_del_primary_channel(capsys, reset_globals):
    """Test --ch-del on ch-index=0"""
    sys.argv = ['', '--ch-del', '--ch-index', '0']
    Globals.getInstance().set_args(sys.argv)
    Globals.getInstance().set_channel_index(1)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Cannot delete primary channel', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals):
    """Test --ch-enable with --ch-index"""
    sys.argv = ['', '--ch-enable', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Writing modified channels', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() == 1
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals):
    """Test --ch-disable with --ch-index"""
    sys.argv = ['', '--ch-disable', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Writing modified channels', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() == 1
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_enable_without_a_ch_index(capsys, reset_globals):
    """Test --ch-enable without --ch-index"""
    sys.argv = ['', '--ch-enable']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Need to specify', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() is None
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_enable_primary_channel(capsys, reset_globals):
    """Test --ch-enable with --ch-index = 0"""
    sys.argv = ['', '--ch-enable', '--ch-index', '0']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Cannot enable/disable PRIMARY', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() == 0
        mo.assert_called()


@pytest.mark.unit
def test_main_ch_range_options(capsys, reset_globals):
    """Test changing the various range options."""
    range_options = ['--ch-longslow', '--ch-longfast', '--ch-mediumslow',
                     '--ch-mediumfast', '--ch-shortslow', '--ch-shortfast']
    for range_option in range_options:
        sys.argv = ['', f"{range_option}" ]
        Globals.getInstance().set_args(sys.argv)

        mocked_node = MagicMock(autospec=Node)

        iface = MagicMock(autospec=SerialInterface)
        iface.getNode.return_value = mocked_node

        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'Writing modified channels', out, re.MULTILINE)
            assert err == ''
            mo.assert_called()


@pytest.mark.unit
def test_main_ch_longsfast_on_non_primary_channel(capsys, reset_globals):
    """Test --ch-longfast --ch-index 1"""
    sys.argv = ['', '--ch-longfast', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Standard channel settings', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


# PositionFlags:
# Misc info that might be helpful (this info will grow stale, just
# a snapshot of the values.) The radioconfig_pb2.PositionFlags.Name and bit values are:
# POS_UNDEFINED 0
# POS_ALTITUDE 1
# POS_ALT_MSL 2
# POS_GEO_SEP 4
# POS_DOP 8
# POS_HVDOP 16
# POS_BATTERY 32
# POS_SATINVIEW 64
# POS_SEQ_NOS 128
# POS_TIMESTAMP 256

@pytest.mark.unit
def test_main_pos_fields_no_args(capsys, reset_globals):
    """Test --pos-fields no args (which shows settings)"""
    sys.argv = ['', '--pos-fields']
    Globals.getInstance().set_args(sys.argv)

    pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:
        with patch('meshtastic.radioconfig_pb2.PositionFlags', return_value=pos_flags) as mrc:
            # kind of cheating here, we are setting up the node
            mocked_node = MagicMock(autospec=Node)
            anode = mocked_node()
            anode.radioConfig.preferences.position_flags = 35
            Globals.getInstance().set_target_node(anode)

            mrc.values.return_value = [0, 1, 2, 4, 8, 16, 32, 64, 128, 256]
            # Note: When you use side_effect and a list, each call will use a value from the front of the list then
            # remove that value from the list. If there are three values in the list, we expect it to be called
            # three times.
            mrc.Name.side_effect = [ 'POS_ALTITUDE', 'POS_ALT_MSL', 'POS_BATTERY' ]

            main()

            mrc.Name.assert_called()
            mrc.values.assert_called()
            mo.assert_called()

            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'POS_ALTITUDE POS_ALT_MSL POS_BATTERY', out, re.MULTILINE)
            assert err == ''


@pytest.mark.unit
def test_main_pos_fields_arg_of_zero(capsys, reset_globals):
    """Test --pos-fields an arg of 0 (which shows list)"""
    sys.argv = ['', '--pos-fields', '0']
    Globals.getInstance().set_args(sys.argv)

    pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:
        with patch('meshtastic.radioconfig_pb2.PositionFlags', return_value=pos_flags) as mrc:

            def throw_value_error_exception(exc):
                raise ValueError()
            mrc.Value.side_effect = throw_value_error_exception
            mrc.keys.return_value = [ 'POS_UNDEFINED', 'POS_ALTITUDE', 'POS_ALT_MSL',
                                      'POS_GEO_SEP', 'POS_DOP', 'POS_HVDOP', 'POS_BATTERY',
                                      'POS_SATINVIEW', 'POS_SEQ_NOS', 'POS_TIMESTAMP']

            main()

            mrc.Value.assert_called()
            mrc.keys.assert_called()
            mo.assert_called()

            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'ERROR: supported position fields are:', out, re.MULTILINE)
            assert re.search(r"['POS_UNDEFINED', 'POS_ALTITUDE', 'POS_ALT_MSL', 'POS_GEO_SEP',"\
                              "'POS_DOP', 'POS_HVDOP', 'POS_BATTERY', 'POS_SATINVIEW', 'POS_SEQ_NOS',"\
                              "'POS_TIMESTAMP']", out, re.MULTILINE)
            assert err == ''


@pytest.mark.unit
def test_main_pos_fields_valid_values(capsys, reset_globals):
    """Test --pos-fields with valid values"""
    sys.argv = ['', '--pos-fields', 'POS_GEO_SEP', 'POS_ALT_MSL']
    Globals.getInstance().set_args(sys.argv)

    pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:
        with patch('meshtastic.radioconfig_pb2.PositionFlags', return_value=pos_flags) as mrc:

            mrc.Value.side_effect = [ 4, 2 ]

            main()

            mrc.Value.assert_called()
            mo.assert_called()

            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'Setting position fields to 6', out, re.MULTILINE)
            assert re.search(r'Set position_flags to 6', out, re.MULTILINE)
            assert re.search(r'Writing modified preferences to device', out, re.MULTILINE)
            assert err == ''


@pytest.mark.unit
def test_main_get_with_valid_values(capsys, reset_globals):
    """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)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:

        # kind of cheating here, we are setting up the node
        mocked_node = MagicMock(autospec=Node)
        anode = mocked_node()
        anode.radioConfig.preferences.wifi_ssid = 'foo'
        anode.radioConfig.preferences.ls_secs = 300
        anode.radioConfig.preferences.fixed_position = False
        Globals.getInstance().set_target_node(anode)

        main()

        mo.assert_called()

        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'ls_secs: 300', out, re.MULTILINE)
        assert re.search(r'wifi_ssid: foo', out, re.MULTILINE)
        assert re.search(r'fixed_position: False', out, re.MULTILINE)
        assert err == ''


@pytest.mark.unit
def test_main_get_with_invalid(capsys, reset_globals):
    """Test --get with invalid field"""
    sys.argv = ['', '--get', 'foo']
    Globals.getInstance().set_args(sys.argv)

    mocked_user_prefs = MagicMock()
    mocked_user_prefs.DESCRIPTOR.fields_by_name.get.return_value = None

    mocked_node = MagicMock(autospec=Node)
    mocked_node.radioConfig.preferences = ( mocked_user_prefs )

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'does not have an attribute called foo', out, re.MULTILINE)
        assert re.search(r'Choices in sorted order are', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()


@pytest.mark.unit
def test_main_setchan(capsys, reset_globals):
    """Test --setchan (deprecated)"""
    sys.argv = ['', '--setchan', 'a', 'b']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface):
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            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 == ''

Functions

def test_main_ch_add_but_name_already_exists(capsys, reset_globals)

Test –ch-add with a channel name that already exists

Expand source code

@pytest.mark.unit
def test_main_ch_add_but_name_already_exists(capsys, reset_globals):
    """Test --ch-add with a channel name that already exists"""
    sys.argv = ['', '--ch-add', 'testing']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = True

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: This node already has', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_add_but_no_more_channels(capsys, reset_globals)

Test –ch-add with but there are no more channels

Expand source code

@pytest.mark.unit
def test_main_ch_add_but_no_more_channels(capsys, reset_globals):
    """Test --ch-add with but there are no more channels"""
    sys.argv = ['', '--ch-add', 'testing']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = False
    # set it up so we have free channels
    mocked_node.getDisabledChannel.return_value = None

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: No free channels were found', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_add_invalid_name_too_long(capsys, reset_globals)

Test –ch-add with invalid channel name, name too long

Expand source code

@pytest.mark.unit
def test_main_ch_add_invalid_name_too_long(capsys, reset_globals):
    """Test --ch-add with invalid channel name, name too long"""
    sys.argv = ['', '--ch-add', 'testingtestingtesting']
    Globals.getInstance().set_args(sys.argv)

    mocked_channel = MagicMock(autospec=Channel)
    # TODO: figure out how to get it to print the channel name instead of MagicMock

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = False
    # set it up so we have free channels
    mocked_node.getDisabledChannel.return_value = mocked_channel

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Channel name must be shorter', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_add_valid(capsys, reset_globals)

Test –ch-add with valid channel name, and that channel name does not already exist

Expand source code

@pytest.mark.unit
def test_main_ch_add_valid(capsys, reset_globals):
    """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)

    mocked_channel = MagicMock(autospec=Channel)
    # TODO: figure out how to get it to print the channel name instead of MagicMock

    mocked_node = MagicMock(autospec=Node)
    # set it up so we do not already have a channel named this
    mocked_node.getChannelByName.return_value = False
    # set it up so we have free channels
    mocked_node.getDisabledChannel.return_value = mocked_channel

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Writing modified channels to device', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_del(capsys, reset_globals)

Test –ch-del with valid secondary channel to be deleted

Expand source code

@pytest.mark.unit
def test_main_ch_del(capsys, reset_globals):
    """Test --ch-del with valid secondary channel to be deleted"""
    sys.argv = ['', '--ch-del', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Deleting channel', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_del_no_ch_index_specified(capsys, reset_globals)

Test –ch-del without a valid ch-index

Expand source code

@pytest.mark.unit
def test_main_ch_del_no_ch_index_specified(capsys, reset_globals):
    """Test --ch-del without a valid ch-index"""
    sys.argv = ['', '--ch-del']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Need to specify', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_del_primary_channel(capsys, reset_globals)

Test –ch-del on ch-index=0

Expand source code

@pytest.mark.unit
def test_main_ch_del_primary_channel(capsys, reset_globals):
    """Test --ch-del on ch-index=0"""
    sys.argv = ['', '--ch-del', '--ch-index', '0']
    Globals.getInstance().set_args(sys.argv)
    Globals.getInstance().set_channel_index(1)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Cannot delete primary channel', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals)

Test –ch-disable with –ch-index

Expand source code

@pytest.mark.unit
def test_main_ch_disable_valid_secondary_channel(capsys, reset_globals):
    """Test --ch-disable with --ch-index"""
    sys.argv = ['', '--ch-disable', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Writing modified channels', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() == 1
        mo.assert_called()

def test_main_ch_enable_primary_channel(capsys, reset_globals)

Test –ch-enable with –ch-index = 0

Expand source code

@pytest.mark.unit
def test_main_ch_enable_primary_channel(capsys, reset_globals):
    """Test --ch-enable with --ch-index = 0"""
    sys.argv = ['', '--ch-enable', '--ch-index', '0']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Cannot enable/disable PRIMARY', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() == 0
        mo.assert_called()

def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals)

Test –ch-enable with –ch-index

Expand source code

@pytest.mark.unit
def test_main_ch_enable_valid_secondary_channel(capsys, reset_globals):
    """Test --ch-enable with --ch-index"""
    sys.argv = ['', '--ch-enable', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Writing modified channels', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() == 1
        mo.assert_called()

def test_main_ch_enable_without_a_ch_index(capsys, reset_globals)

Test –ch-enable without –ch-index

Expand source code

@pytest.mark.unit
def test_main_ch_enable_without_a_ch_index(capsys, reset_globals):
    """Test --ch-enable without --ch-index"""
    sys.argv = ['', '--ch-enable']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Need to specify', out, re.MULTILINE)
        assert err == ''
        assert Globals.getInstance().get_channel_index() is None
        mo.assert_called()

def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals)

Test –ch-index 1

Expand source code

@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=[])
def test_main_ch_index_no_devices(patched_find_ports, capsys, reset_globals):
    """Test --ch-index 1"""
    sys.argv = ['', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert Globals.getInstance().get_channel_index() == 1
    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 == ''
    patched_find_ports.assert_called()

def test_main_ch_longsfast_on_non_primary_channel(capsys, reset_globals)

Test –ch-longfast –ch-index 1

Expand source code

@pytest.mark.unit
def test_main_ch_longsfast_on_non_primary_channel(capsys, reset_globals):
    """Test --ch-longfast --ch-index 1"""
    sys.argv = ['', '--ch-longfast', '--ch-index', '1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1
        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'Warning: Standard channel settings', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_ch_range_options(capsys, reset_globals)

Test changing the various range options.

Expand source code

@pytest.mark.unit
def test_main_ch_range_options(capsys, reset_globals):
    """Test changing the various range options."""
    range_options = ['--ch-longslow', '--ch-longfast', '--ch-mediumslow',
                     '--ch-mediumfast', '--ch-shortslow', '--ch-shortfast']
    for range_option in range_options:
        sys.argv = ['', f"{range_option}" ]
        Globals.getInstance().set_args(sys.argv)

        mocked_node = MagicMock(autospec=Node)

        iface = MagicMock(autospec=SerialInterface)
        iface.getNode.return_value = mocked_node

        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'Writing modified channels', out, re.MULTILINE)
            assert err == ''
            mo.assert_called()

def test_main_configure(capsys, reset_globals)

Test –configure with valid file

Expand source code

@pytest.mark.unit
def test_main_configure(capsys, reset_globals):
    """Test --configure with valid file"""
    sys.argv = ['', '--configure', 'example_config.yaml']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Setting device owner', out, re.MULTILINE)
        assert re.search(r'Setting channel url', out, re.MULTILINE)
        assert re.search(r'Fixing altitude', out, re.MULTILINE)
        assert re.search(r'Fixing latitude', out, re.MULTILINE)
        assert re.search(r'Fixing longitude', out, re.MULTILINE)
        assert re.search(r'Writing modified preferences', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_get_with_invalid(capsys, reset_globals)

Test –get with invalid field

Expand source code

@pytest.mark.unit
def test_main_get_with_invalid(capsys, reset_globals):
    """Test --get with invalid field"""
    sys.argv = ['', '--get', 'foo']
    Globals.getInstance().set_args(sys.argv)

    mocked_user_prefs = MagicMock()
    mocked_user_prefs.DESCRIPTOR.fields_by_name.get.return_value = None

    mocked_node = MagicMock(autospec=Node)
    mocked_node.radioConfig.preferences = ( mocked_user_prefs )

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'does not have an attribute called foo', out, re.MULTILINE)
        assert re.search(r'Choices in sorted order are', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_get_with_valid_values(capsys, reset_globals)

Test –get with valid values (with string, number, boolean)

Expand source code

@pytest.mark.unit
def test_main_get_with_valid_values(capsys, reset_globals):
    """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)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:

        # kind of cheating here, we are setting up the node
        mocked_node = MagicMock(autospec=Node)
        anode = mocked_node()
        anode.radioConfig.preferences.wifi_ssid = 'foo'
        anode.radioConfig.preferences.ls_secs = 300
        anode.radioConfig.preferences.fixed_position = False
        Globals.getInstance().set_target_node(anode)

        main()

        mo.assert_called()

        out, err = capsys.readouterr()
        assert re.search(r'Connected to radio', out, re.MULTILINE)
        assert re.search(r'ls_secs: 300', out, re.MULTILINE)
        assert re.search(r'wifi_ssid: foo', out, re.MULTILINE)
        assert re.search(r'fixed_position: False', out, re.MULTILINE)
        assert err == ''

def test_main_info(capsys, reset_globals)

Test –info

Expand source code

@pytest.mark.unit
def test_main_info(capsys, reset_globals):
    """Test --info"""
    sys.argv = ['', '--info']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo
    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'inside mocked showInfo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_info_with_ble_interface(capsys, reset_globals)

Test –info

Expand source code

@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()

def test_main_info_with_seriallog_output_txt(capsys, reset_globals)

Test –info

Expand source code

@pytest.mark.unit
def test_main_info_with_seriallog_output_txt(capsys, reset_globals):
    """Test --info"""
    sys.argv = ['', '--info', '--seriallog', 'output.txt']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo
    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'inside mocked showInfo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()
    # do some cleanup
    os.remove('output.txt')

def test_main_info_with_seriallog_stdout(capsys, reset_globals)

Test –info

Expand source code

@pytest.mark.unit
def test_main_info_with_seriallog_stdout(capsys, reset_globals):
    """Test --info"""
    sys.argv = ['', '--info', '--seriallog', 'stdout']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo
    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'inside mocked showInfo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_info_with_tcp_interface(capsys, reset_globals)

Test –info

Expand source code

@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()

def test_main_init_parser_no_args(capsys, reset_globals)

Test no arguments

Expand source code

@pytest.mark.unit
def test_main_init_parser_no_args(capsys, reset_globals):
    """Test no arguments"""
    sys.argv = ['']
    Globals.getInstance().set_args(sys.argv)
    initParser()
    out, err = capsys.readouterr()
    assert out == ''
    assert err == ''

def test_main_init_parser_version(capsys, reset_globals)

Test –version

Expand source code

@pytest.mark.unit
def test_main_init_parser_version(capsys, reset_globals):
    """Test --version"""
    sys.argv = ['', '--version']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        initParser()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 0
    out, err = capsys.readouterr()
    assert re.match(r'[0-9]+\.[0-9]+\.[0-9]', out)
    assert err == ''

def test_main_main_no_args(reset_globals)

Test with no args

Expand source code

@pytest.mark.unit
def test_main_main_no_args(reset_globals):
    """Test with no args"""
    sys.argv = ['']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 1

def test_main_main_version(capsys, reset_globals)

Test –version

Expand source code

@pytest.mark.unit
def test_main_main_version(capsys, reset_globals):
    """Test --version"""
    sys.argv = ['', '--version']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 0
    out, err = capsys.readouterr()
    assert re.match(r'[0-9]+\.[0-9]+\.[0-9]', out)
    assert err == ''

def test_main_no_proto(capsys, reset_globals)

Test –noproto (using –info for output)

Expand source code

@pytest.mark.unit
def test_main_no_proto(capsys, reset_globals):
    """Test --noproto (using --info for output)"""
    sys.argv = ['', '--info', '--noproto']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showInfo():
        print('inside mocked showInfo')
    iface.showInfo.side_effect = mock_showInfo

    # Override the time.sleep so there is no loop
    def my_sleep(amount):
        sys.exit(0)

    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:
                main()
            assert pytest_wrapped_e.type == SystemExit
            assert pytest_wrapped_e.value.code == 0
            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 == ''

def test_main_nodes(capsys, reset_globals)

Test –nodes

Expand source code

@pytest.mark.unit
def test_main_nodes(capsys, reset_globals):
    """Test --nodes"""
    sys.argv = ['', '--nodes']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_showNodes():
        print('inside mocked showNodes')
    iface.showNodes.side_effect = mock_showNodes
    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'inside mocked showNodes', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_onConnection(reset_globals, capsys)

Test onConnection

Expand source code

@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 == ''

def test_main_onReceive_empty(reset_globals)

Test onReceive

Expand source code

@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?

def test_main_pos_fields_arg_of_zero(capsys, reset_globals)

Test –pos-fields an arg of 0 (which shows list)

Expand source code

@pytest.mark.unit
def test_main_pos_fields_arg_of_zero(capsys, reset_globals):
    """Test --pos-fields an arg of 0 (which shows list)"""
    sys.argv = ['', '--pos-fields', '0']
    Globals.getInstance().set_args(sys.argv)

    pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:
        with patch('meshtastic.radioconfig_pb2.PositionFlags', return_value=pos_flags) as mrc:

            def throw_value_error_exception(exc):
                raise ValueError()
            mrc.Value.side_effect = throw_value_error_exception
            mrc.keys.return_value = [ 'POS_UNDEFINED', 'POS_ALTITUDE', 'POS_ALT_MSL',
                                      'POS_GEO_SEP', 'POS_DOP', 'POS_HVDOP', 'POS_BATTERY',
                                      'POS_SATINVIEW', 'POS_SEQ_NOS', 'POS_TIMESTAMP']

            main()

            mrc.Value.assert_called()
            mrc.keys.assert_called()
            mo.assert_called()

            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'ERROR: supported position fields are:', out, re.MULTILINE)
            assert re.search(r"['POS_UNDEFINED', 'POS_ALTITUDE', 'POS_ALT_MSL', 'POS_GEO_SEP',"\
                              "'POS_DOP', 'POS_HVDOP', 'POS_BATTERY', 'POS_SATINVIEW', 'POS_SEQ_NOS',"\
                              "'POS_TIMESTAMP']", out, re.MULTILINE)
            assert err == ''

def test_main_pos_fields_no_args(capsys, reset_globals)

Test –pos-fields no args (which shows settings)

Expand source code

@pytest.mark.unit
def test_main_pos_fields_no_args(capsys, reset_globals):
    """Test --pos-fields no args (which shows settings)"""
    sys.argv = ['', '--pos-fields']
    Globals.getInstance().set_args(sys.argv)

    pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:
        with patch('meshtastic.radioconfig_pb2.PositionFlags', return_value=pos_flags) as mrc:
            # kind of cheating here, we are setting up the node
            mocked_node = MagicMock(autospec=Node)
            anode = mocked_node()
            anode.radioConfig.preferences.position_flags = 35
            Globals.getInstance().set_target_node(anode)

            mrc.values.return_value = [0, 1, 2, 4, 8, 16, 32, 64, 128, 256]
            # Note: When you use side_effect and a list, each call will use a value from the front of the list then
            # remove that value from the list. If there are three values in the list, we expect it to be called
            # three times.
            mrc.Name.side_effect = [ 'POS_ALTITUDE', 'POS_ALT_MSL', 'POS_BATTERY' ]

            main()

            mrc.Name.assert_called()
            mrc.values.assert_called()
            mo.assert_called()

            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'POS_ALTITUDE POS_ALT_MSL POS_BATTERY', out, re.MULTILINE)
            assert err == ''

def test_main_pos_fields_valid_values(capsys, reset_globals)

Test –pos-fields with valid values

Expand source code

@pytest.mark.unit
def test_main_pos_fields_valid_values(capsys, reset_globals):
    """Test --pos-fields with valid values"""
    sys.argv = ['', '--pos-fields', 'POS_GEO_SEP', 'POS_ALT_MSL']
    Globals.getInstance().set_args(sys.argv)

    pos_flags = MagicMock(autospec=meshtastic.radioconfig_pb2.PositionFlags)

    with patch('meshtastic.serial_interface.SerialInterface') as mo:
        with patch('meshtastic.radioconfig_pb2.PositionFlags', return_value=pos_flags) as mrc:

            mrc.Value.side_effect = [ 4, 2 ]

            main()

            mrc.Value.assert_called()
            mo.assert_called()

            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'Setting position fields to 6', out, re.MULTILINE)
            assert re.search(r'Set position_flags to 6', out, re.MULTILINE)
            assert re.search(r'Writing modified preferences to device', out, re.MULTILINE)
            assert err == ''

def test_main_qr(capsys, reset_globals)

Test –qr

Expand source code

@pytest.mark.unit
def test_main_qr(capsys, reset_globals):
    """Test --qr"""
    sys.argv = ['', '--qr']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    # TODO: could mock/check url
    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'Primary channel URL', out, re.MULTILINE)
        # if a qr code is generated it will have lots of these
        assert re.search(r'\[7m', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_reboot(capsys, reset_globals)

Test –reboot

Expand source code

@pytest.mark.unit
def test_main_reboot(capsys, reset_globals):
    """Test --reboot"""
    sys.argv = ['', '--reboot']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_reboot():
        print('inside mocked reboot')
    mocked_node.reboot.side_effect = mock_reboot

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'inside mocked reboot', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_sendping(capsys, reset_globals)

Test –sendping

Expand source code

@pytest.mark.unit
def test_main_sendping(capsys, reset_globals):
    """Test --sendping"""
    sys.argv = ['', '--sendping']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendData(payload, dest, portNum, wantAck, wantResponse):
        print('inside mocked sendData')
    iface.sendData.side_effect = mock_sendData

    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 ping message', out, re.MULTILINE)
        assert re.search(r'inside mocked sendData', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_sendtext(capsys, reset_globals)

Test –sendtext

Expand source code

@pytest.mark.unit
def test_main_sendtext(capsys, reset_globals):
    """Test --sendtext"""
    sys.argv = ['', '--sendtext', 'hello']
    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()

def test_main_sendtext_with_dest(capsys, reset_globals)

Test –sendtext with –dest

Expand source code

@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()

def test_main_set_ham_to_KI123(capsys, reset_globals)

Test –set-ham KI123

Expand source code

@pytest.mark.unit
def test_main_set_ham_to_KI123(capsys, reset_globals):
    """Test --set-ham KI123"""
    sys.argv = ['', '--set-ham', 'KI123']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_turnOffEncryptionOnPrimaryChannel():
        print('inside mocked turnOffEncryptionOnPrimaryChannel')
    def mock_setOwner(name, is_licensed):
        print('inside mocked setOwner')
    mocked_node.turnOffEncryptionOnPrimaryChannel.side_effect = mock_turnOffEncryptionOnPrimaryChannel
    mocked_node.setOwner.side_effect = mock_setOwner

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Setting HAM ID to KI123', out, re.MULTILINE)
        assert re.search(r'inside mocked setOwner', out, re.MULTILINE)
        assert re.search(r'inside mocked turnOffEncryptionOnPrimaryChannel', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_set_owner_to_bob(capsys, reset_globals)

Test –set-owner bob

Expand source code

@pytest.mark.unit
def test_main_set_owner_to_bob(capsys, reset_globals):
    """Test --set-owner bob"""
    sys.argv = ['', '--set-owner', 'bob']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)
    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'Setting device owner to bob', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_set_team_invalid(capsys, reset_globals)

Test –set-team using an invalid team name

Expand source code

@pytest.mark.unit
def test_main_set_team_invalid(capsys, reset_globals):
    """Test --set-team using an invalid team name"""
    sys.argv = ['', '--set-team', 'NOTCYAN']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)

    def throw_an_exception(exc):
        raise ValueError("Fake exception.")

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with patch('meshtastic.mesh_pb2.Team') as mm:
            mm.Value.side_effect = throw_an_exception
            main()
            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'ERROR: Team', out, re.MULTILINE)
            assert err == ''
            mo.assert_called()
            mm.Value.assert_called()

def test_main_set_team_valid(capsys, reset_globals)

Test –set-team

Expand source code

@pytest.mark.unit
def test_main_set_team_valid(capsys, reset_globals):
    """Test --set-team"""
    sys.argv = ['', '--set-team', 'CYAN']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_setOwner(team):
        print('inside mocked setOwner')
    mocked_node.setOwner.side_effect = mock_setOwner

    iface = MagicMock(autospec=SerialInterface)
    iface.localNode.return_value = mocked_node

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface) as mo:
        with patch('meshtastic.mesh_pb2.Team') as mm:
            mm.Name.return_value = 'FAKENAME'
            mm.Value.return_value = 'FAKEVAL'
            main()
            out, err = capsys.readouterr()
            assert re.search(r'Connected to radio', out, re.MULTILINE)
            assert re.search(r'Setting team to', out, re.MULTILINE)
            assert err == ''
            mo.assert_called()
            mm.Name.assert_called()
            mm.Value.assert_called()

def test_main_set_valid(capsys, reset_globals)

Test –set with valid field

Expand source code

@pytest.mark.unit
def test_main_set_valid(capsys, reset_globals):
    """Test --set with valid field"""
    sys.argv = ['', '--set', 'wifi_ssid', 'foo']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'Set wifi_ssid to foo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_set_with_invalid(capsys, reset_globals)

Test –set with invalid field

Expand source code

@pytest.mark.unit
def test_main_set_with_invalid(capsys, reset_globals):
    """Test --set with invalid field"""
    sys.argv = ['', '--set', 'foo', 'foo']
    Globals.getInstance().set_args(sys.argv)

    mocked_user_prefs = MagicMock()
    mocked_user_prefs.DESCRIPTOR.fields_by_name.get.return_value = None

    mocked_node = MagicMock(autospec=Node)
    mocked_node.radioConfig.preferences = ( mocked_user_prefs )

    iface = MagicMock(autospec=SerialInterface)
    iface.getNode.return_value = mocked_node

    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'does not have an attribute called foo', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_setalt(capsys, reset_globals)

Test –setalt

Expand source code

@pytest.mark.unit
def test_main_setalt(capsys, reset_globals):
    """Test --setalt"""
    sys.argv = ['', '--setalt', '51']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_writeConfig():
        print('inside mocked writeConfig')
    mocked_node.writeConfig.side_effect = mock_writeConfig

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendPosition(lat, lon, alt):
        print('inside mocked sendPosition')
    iface.sendPosition.side_effect = mock_sendPosition
    iface.localNode.return_value = mocked_node

    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'Fixing altitude', out, re.MULTILINE)
        assert re.search(r'Setting device position', out, re.MULTILINE)
        assert re.search(r'inside mocked sendPosition', out, re.MULTILINE)
        # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_setchan(capsys, reset_globals)

Test –setchan (deprecated)

Expand source code

@pytest.mark.unit
def test_main_setchan(capsys, reset_globals):
    """Test --setchan (deprecated)"""
    sys.argv = ['', '--setchan', 'a', 'b']
    Globals.getInstance().set_args(sys.argv)

    iface = MagicMock(autospec=SerialInterface)

    with patch('meshtastic.serial_interface.SerialInterface', return_value=iface):
        with pytest.raises(SystemExit) as pytest_wrapped_e:
            main()
        assert pytest_wrapped_e.type == SystemExit
        assert pytest_wrapped_e.value.code == 1

def test_main_setlat(capsys, reset_globals)

Test –sendlat

Expand source code

@pytest.mark.unit
def test_main_setlat(capsys, reset_globals):
    """Test --sendlat"""
    sys.argv = ['', '--setlat', '37.5']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_writeConfig():
        print('inside mocked writeConfig')
    mocked_node.writeConfig.side_effect = mock_writeConfig

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendPosition(lat, lon, alt):
        print('inside mocked sendPosition')
    iface.sendPosition.side_effect = mock_sendPosition
    iface.localNode.return_value = mocked_node

    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'Fixing latitude', out, re.MULTILINE)
        assert re.search(r'Setting device position', out, re.MULTILINE)
        assert re.search(r'inside mocked sendPosition', out, re.MULTILINE)
        # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_setlon(capsys, reset_globals)

Test –setlon

Expand source code

@pytest.mark.unit
def test_main_setlon(capsys, reset_globals):
    """Test --setlon"""
    sys.argv = ['', '--setlon', '-122.1']
    Globals.getInstance().set_args(sys.argv)

    mocked_node = MagicMock(autospec=Node)
    def mock_writeConfig():
        print('inside mocked writeConfig')
    mocked_node.writeConfig.side_effect = mock_writeConfig

    iface = MagicMock(autospec=SerialInterface)
    def mock_sendPosition(lat, lon, alt):
        print('inside mocked sendPosition')
    iface.sendPosition.side_effect = mock_sendPosition
    iface.localNode.return_value = mocked_node

    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'Fixing longitude', out, re.MULTILINE)
        assert re.search(r'Setting device position', out, re.MULTILINE)
        assert re.search(r'inside mocked sendPosition', out, re.MULTILINE)
        # TODO: Why does this not work? assert re.search(r'inside mocked writeConfig', out, re.MULTILINE)
        assert err == ''
        mo.assert_called()

def test_main_seturl(capsys, reset_globals)

Test –seturl (url used below is what is generated after a factory_reset)

Expand source code

@pytest.mark.unit
def test_main_seturl(capsys, reset_globals):
    """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)

    iface = MagicMock(autospec=SerialInterface)
    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 err == ''
        mo.assert_called()

def test_main_support(capsys, reset_globals)

Test –support

Expand source code

@pytest.mark.unit
def test_main_support(capsys, reset_globals):
    """Test --support"""
    sys.argv = ['', '--support']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 0
    out, err = capsys.readouterr()
    assert re.search(r'System', out, re.MULTILINE)
    assert re.search(r'Platform', out, re.MULTILINE)
    assert re.search(r'Machine', out, re.MULTILINE)
    assert re.search(r'Executable', out, re.MULTILINE)
    assert err == ''

def test_main_test_no_ports(patched_find_ports, reset_globals)

Test –test with no hardware

Expand source code

@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=[])
def test_main_test_no_ports(patched_find_ports, reset_globals):
    """Test --test with no hardware"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    assert Globals.getInstance().get_target_node() is None
    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 1
    patched_find_ports.assert_called()

def test_main_test_one_port(patched_find_ports, reset_globals)

Test –test with one fake port

Expand source code

@pytest.mark.unit
@patch('meshtastic.util.findPorts', return_value=['/dev/ttyFake1'])
def test_main_test_one_port(patched_find_ports, reset_globals):
    """Test --test with one fake port"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    assert Globals.getInstance().get_target_node() is None
    with pytest.raises(SystemExit) as pytest_wrapped_e:
        main()
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 1
    patched_find_ports.assert_called()

def test_main_test_two_ports_fails(patched_find_ports, patched_test_all, reset_globals)

Test –test two fake ports and testAll() is a simulated failure

Expand source code

@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):
    """Test --test two fake ports and testAll() is a simulated failure"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        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()

def test_main_test_two_ports_success(patched_find_ports, patched_test_all, reset_globals)

Test –test two fake ports and testAll() is a simulated success

Expand source code

@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):
    """Test --test two fake ports and testAll() is a simulated success"""
    sys.argv = ['', '--test']
    Globals.getInstance().set_args(sys.argv)

    with pytest.raises(SystemExit) as pytest_wrapped_e:
        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()

Classes

class Channel (**kwargs)

Abstract base class for protocol messages.

Protocol message classes are almost always generated by the protocol compiler. These generated types subclass Message and implement the methods shown below.

Ancestors

• google.protobuf.message.Message

Class variables

var DESCRIPTOR

var DISABLED

var INDEX_FIELD_NUMBER

var PRIMARY

var ROLE_FIELD_NUMBER

var Role

var SECONDARY

var SETTINGS_FIELD_NUMBER

Static methods

def FromString(s)

Expand source code

def FromString(s):
  message = cls()
  message.MergeFromString(s)
  return message

def RegisterExtension(extension_handle)

Expand source code

def RegisterExtension(extension_handle):
  extension_handle.containing_type = cls.DESCRIPTOR
  # TODO(amauryfa): Use cls.MESSAGE_FACTORY.pool when available.
  # pylint: disable=protected-access
  cls.DESCRIPTOR.file.pool._AddExtensionDescriptor(extension_handle)
  _AttachFieldHelpers(cls, extension_handle)

Instance variables

var index

Getter for index.

Expand source code

def getter(self):
  # TODO(protobuf-team): This may be broken since there may not be
  # default_value.  Combine with has_default_value somehow.
  return self._fields.get(field, default_value)

var role

Getter for role.

Expand source code

def getter(self):
  # TODO(protobuf-team): This may be broken since there may not be
  # default_value.  Combine with has_default_value somehow.
  return self._fields.get(field, default_value)

var settings

Getter for settings.

Expand source code

def getter(self):
  field_value = self._fields.get(field)
  if field_value is None:
    # Construct a new object to represent this field.
    field_value = field._default_constructor(self)

    # Atomically check if another thread has preempted us and, if not, swap
    # in the new object we just created.  If someone has preempted us, we
    # take that object and discard ours.
    # WARNING:  We are relying on setdefault() being atomic.  This is true
    #   in CPython but we haven't investigated others.  This warning appears
    #   in several other locations in this file.
    field_value = self._fields.setdefault(field, field_value)
  return field_value

Methods

def ByteSize(self)

Expand source code

def ByteSize(self):
  if not self._cached_byte_size_dirty:
    return self._cached_byte_size

  size = 0
  descriptor = self.DESCRIPTOR
  if descriptor.GetOptions().map_entry:
    # Fields of map entry should always be serialized.
    size = descriptor.fields_by_name['key']._sizer(self.key)
    size += descriptor.fields_by_name['value']._sizer(self.value)
  else:
    for field_descriptor, field_value in self.ListFields():
      size += field_descriptor._sizer(field_value)
    for tag_bytes, value_bytes in self._unknown_fields:
      size += len(tag_bytes) + len(value_bytes)

  self._cached_byte_size = size
  self._cached_byte_size_dirty = False
  self._listener_for_children.dirty = False
  return size

def Clear(self)

Expand source code

def _Clear(self):
  # Clear fields.
  self._fields = {}
  self._unknown_fields = ()
  # pylint: disable=protected-access
  if self._unknown_field_set is not None:
    self._unknown_field_set._clear()
    self._unknown_field_set = None

  self._oneofs = {}
  self._Modified()

def ClearField(self, field_name)

Expand source code

def ClearField(self, field_name):
  try:
    field = message_descriptor.fields_by_name[field_name]
  except KeyError:
    try:
      field = message_descriptor.oneofs_by_name[field_name]
      if field in self._oneofs:
        field = self._oneofs[field]
      else:
        return
    except KeyError:
      raise ValueError('Protocol message %s has no "%s" field.' %
                       (message_descriptor.name, field_name))

  if field in self._fields:
    # To match the C++ implementation, we need to invalidate iterators
    # for map fields when ClearField() happens.
    if hasattr(self._fields[field], 'InvalidateIterators'):
      self._fields[field].InvalidateIterators()

    # Note:  If the field is a sub-message, its listener will still point
    #   at us.  That's fine, because the worst than can happen is that it
    #   will call _Modified() and invalidate our byte size.  Big deal.
    del self._fields[field]

    if self._oneofs.get(field.containing_oneof, None) is field:
      del self._oneofs[field.containing_oneof]

  # Always call _Modified() -- even if nothing was changed, this is
  # a mutating method, and thus calling it should cause the field to become
  # present in the parent message.
  self._Modified()

def DiscardUnknownFields(self)

Expand source code

def _DiscardUnknownFields(self):
  self._unknown_fields = []
  self._unknown_field_set = None      # pylint: disable=protected-access
  for field, value in self.ListFields():
    if field.cpp_type == _FieldDescriptor.CPPTYPE_MESSAGE:
      if _IsMapField(field):
        if _IsMessageMapField(field):
          for key in value:
            value[key].DiscardUnknownFields()
      elif field.label == _FieldDescriptor.LABEL_REPEATED:
        for sub_message in value:
          sub_message.DiscardUnknownFields()
      else:
        value.DiscardUnknownFields()

def FindInitializationErrors(self)

Finds required fields which are not initialized.

Returns

A list of strings. Each string is a path to an uninitialized field from the top-level message, e.g. "foo.bar[5].baz".

Expand source code

def FindInitializationErrors(self):
  """Finds required fields which are not initialized.

  Returns:
    A list of strings.  Each string is a path to an uninitialized field from
    the top-level message, e.g. "foo.bar[5].baz".
  """

  errors = []  # simplify things

  for field in required_fields:
    if not self.HasField(field.name):
      errors.append(field.name)

  for field, value in self.ListFields():
    if field.cpp_type == _FieldDescriptor.CPPTYPE_MESSAGE:
      if field.is_extension:
        name = '(%s)' % field.full_name
      else:
        name = field.name

      if _IsMapField(field):
        if _IsMessageMapField(field):
          for key in value:
            element = value[key]
            prefix = '%s[%s].' % (name, key)
            sub_errors = element.FindInitializationErrors()
            errors += [prefix + error for error in sub_errors]
        else:
          # ScalarMaps can't have any initialization errors.
          pass
      elif field.label == _FieldDescriptor.LABEL_REPEATED:
        for i in range(len(value)):
          element = value[i]
          prefix = '%s[%d].' % (name, i)
          sub_errors = element.FindInitializationErrors()
          errors += [prefix + error for error in sub_errors]
      else:
        prefix = name + '.'
        sub_errors = value.FindInitializationErrors()
        errors += [prefix + error for error in sub_errors]

  return errors

def HasField(self, field_name)

Expand source code

def HasField(self, field_name):
  try:
    field = hassable_fields[field_name]
  except KeyError:
    raise ValueError(error_msg % (message_descriptor.full_name, field_name))

  if isinstance(field, descriptor_mod.OneofDescriptor):
    try:
      return HasField(self, self._oneofs[field].name)
    except KeyError:
      return False
  else:
    if field.cpp_type == _FieldDescriptor.CPPTYPE_MESSAGE:
      value = self._fields.get(field)
      return value is not None and value._is_present_in_parent
    else:
      return field in self._fields

def IsInitialized(self, errors=None)

Checks if all required fields of a message are set.

Args

errors

A list which, if provided, will be populated with the field paths of all missing required fields.

Returns

True iff the specified message has all required fields set.

Expand source code

def IsInitialized(self, errors=None):
  """Checks if all required fields of a message are set.

  Args:
    errors:  A list which, if provided, will be populated with the field
             paths of all missing required fields.

  Returns:
    True iff the specified message has all required fields set.
  """

  # Performance is critical so we avoid HasField() and ListFields().

  for field in required_fields:
    if (field not in self._fields or
        (field.cpp_type == _FieldDescriptor.CPPTYPE_MESSAGE and
         not self._fields[field]._is_present_in_parent)):
      if errors is not None:
        errors.extend(self.FindInitializationErrors())
      return False

  for field, value in list(self._fields.items()):  # dict can change size!
    if field.cpp_type == _FieldDescriptor.CPPTYPE_MESSAGE:
      if field.label == _FieldDescriptor.LABEL_REPEATED:
        if (field.message_type.has_options and
            field.message_type.GetOptions().map_entry):
          continue
        for element in value:
          if not element.IsInitialized():
            if errors is not None:
              errors.extend(self.FindInitializationErrors())
            return False
      elif value._is_present_in_parent and not value.IsInitialized():
        if errors is not None:
          errors.extend(self.FindInitializationErrors())
        return False

  return True

def ListFields(self)

Expand source code

def ListFields(self):
  all_fields = [item for item in self._fields.items() if _IsPresent(item)]
  all_fields.sort(key = lambda item: item[0].number)
  return all_fields

def MergeFrom(self, msg)

Expand source code

def MergeFrom(self, msg):
  if not isinstance(msg, cls):
    raise TypeError(
        'Parameter to MergeFrom() must be instance of same class: '
        'expected %s got %s.' % (_FullyQualifiedClassName(cls),
                                 _FullyQualifiedClassName(msg.__class__)))

  assert msg is not self
  self._Modified()

  fields = self._fields

  for field, value in msg._fields.items():
    if field.label == LABEL_REPEATED:
      field_value = fields.get(field)
      if field_value is None:
        # Construct a new object to represent this field.
        field_value = field._default_constructor(self)
        fields[field] = field_value
      field_value.MergeFrom(value)
    elif field.cpp_type == CPPTYPE_MESSAGE:
      if value._is_present_in_parent:
        field_value = fields.get(field)
        if field_value is None:
          # Construct a new object to represent this field.
          field_value = field._default_constructor(self)
          fields[field] = field_value
        field_value.MergeFrom(value)
    else:
      self._fields[field] = value
      if field.containing_oneof:
        self._UpdateOneofState(field)

  if msg._unknown_fields:
    if not self._unknown_fields:
      self._unknown_fields = []
    self._unknown_fields.extend(msg._unknown_fields)
    # pylint: disable=protected-access
    if self._unknown_field_set is None:
      self._unknown_field_set = containers.UnknownFieldSet()
    self._unknown_field_set._extend(msg._unknown_field_set)

def MergeFromString(self, serialized)

Expand source code

def MergeFromString(self, serialized):
  serialized = memoryview(serialized)
  length = len(serialized)
  try:
    if self._InternalParse(serialized, 0, length) != length:
      # The only reason _InternalParse would return early is if it
      # encountered an end-group tag.
      raise message_mod.DecodeError('Unexpected end-group tag.')
  except (IndexError, TypeError):
    # Now ord(buf[p:p+1]) == ord('') gets TypeError.
    raise message_mod.DecodeError('Truncated message.')
  except struct.error as e:
    raise message_mod.DecodeError(e)
  return length   # Return this for legacy reasons.

def SerializePartialToString(self, **kwargs)

Expand source code

def SerializePartialToString(self, **kwargs):
  out = BytesIO()
  self._InternalSerialize(out.write, **kwargs)
  return out.getvalue()

def SerializeToString(self, **kwargs)

Expand source code

def SerializeToString(self, **kwargs):
  # Check if the message has all of its required fields set.
  if not self.IsInitialized():
    raise message_mod.EncodeError(
        'Message %s is missing required fields: %s' % (
        self.DESCRIPTOR.full_name, ','.join(self.FindInitializationErrors())))
  return self.SerializePartialToString(**kwargs)

def SetInParent(self)

Sets the _cached_byte_size_dirty bit to true, and propagates this to our listener iff this was a state change.

Expand source code

def Modified(self):
  """Sets the _cached_byte_size_dirty bit to true,
  and propagates this to our listener iff this was a state change.
  """

  # Note:  Some callers check _cached_byte_size_dirty before calling
  #   _Modified() as an extra optimization.  So, if this method is ever
  #   changed such that it does stuff even when _cached_byte_size_dirty is
  #   already true, the callers need to be updated.
  if not self._cached_byte_size_dirty:
    self._cached_byte_size_dirty = True
    self._listener_for_children.dirty = True
    self._is_present_in_parent = True
    self._listener.Modified()

def UnknownFields(self)

Expand source code

def _UnknownFields(self):
  if self._unknown_field_set is None:  # pylint: disable=protected-access
    # pylint: disable=protected-access
    self._unknown_field_set = containers.UnknownFieldSet()
  return self._unknown_field_set    # pylint: disable=protected-access

def WhichOneof(self, oneof_name)

Returns the name of the currently set field inside a oneof, or None.

Expand source code

def WhichOneof(self, oneof_name):
  """Returns the name of the currently set field inside a oneof, or None."""
  try:
    field = message_descriptor.oneofs_by_name[oneof_name]
  except KeyError:
    raise ValueError(
        'Protocol message has no oneof "%s" field.' % oneof_name)

  nested_field = self._oneofs.get(field, None)
  if nested_field is not None and self.HasField(nested_field.name):
    return nested_field.name
  else:
    return None