Module meshtastic.tests.test_tunnel

Meshtastic unit tests for tunnel.py

"""Meshtastic unit tests for tunnel.py"""
import logging
import re
import sys
from unittest.mock import MagicMock, patch

import pytest

from meshtastic import mt_config

from ..tcp_interface import TCPInterface
from ..tunnel import Tunnel, onTunnelReceive


@pytest.mark.unit
@patch("platform.system")
def test_Tunnel_on_non_linux_system(mock_platform_system):
    """Test that we cannot instantiate a Tunnel on a non Linux system"""
    a_mock = MagicMock()
    a_mock.return_value = "notLinux"
    mock_platform_system.side_effect = a_mock
    with patch("socket.socket") as mock_socket:
        with pytest.raises(Tunnel.TunnelError) as pytest_wrapped_e:
            iface = TCPInterface(hostname="localhost", noProto=True)
            Tunnel(iface)
        assert pytest_wrapped_e.type == Tunnel.TunnelError
        assert mock_socket.called


@pytest.mark.unit
@patch("platform.system")
def test_Tunnel_without_interface(mock_platform_system):
    """Test that we can not instantiate a Tunnel without a valid interface"""
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with pytest.raises(Tunnel.TunnelError) as pytest_wrapped_e:
        Tunnel(None)
    assert pytest_wrapped_e.type == Tunnel.TunnelError


@pytest.mark.unitslow
@patch("platform.system")
def test_Tunnel_with_interface(mock_platform_system, caplog, iface_with_nodes):
    """Test that we can not instantiate a Tunnel without a valid interface"""
    iface = iface_with_nodes
    iface.myInfo.my_node_num = 2475227164
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.WARNING):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            assert tun == mt_config.tunnelInstance
            iface.close()
    assert re.search(r"Not creating a TapDevice()", caplog.text, re.MULTILINE)
    assert re.search(r"Not starting TUN reader", caplog.text, re.MULTILINE)
    assert re.search(r"Not sending packet", caplog.text, re.MULTILINE)


@pytest.mark.unitslow
@patch("platform.system")
def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with_nodes):
    """Test onTunnelReceive"""
    iface = iface_with_nodes
    iface.myInfo.my_node_num = 2475227164
    sys.argv = [""]
    mt_config.args = sys.argv
    packet = {"decoded": {"payload": "foo"}, "from": 2475227164}
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            mt_config.tunnelInstance = tun
            onTunnelReceive(packet, iface)
    assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)
    assert re.search(r"Ignoring message we sent", caplog.text, re.MULTILINE)


@pytest.mark.unit
@patch("platform.system")
def test_onTunnelReceive_from_someone_else(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test onTunnelReceive"""
    iface = iface_with_nodes
    iface.myInfo.my_node_num = 2475227164
    sys.argv = [""]
    mt_config.args = sys.argv
    packet = {"decoded": {"payload": "foo"}, "from": 123}
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            mt_config.tunnelInstance = tun
            onTunnelReceive(packet, iface)
    assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)


@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_random(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # random packet
    packet = b"1234567890123456789012345678901234567890"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert not ignore


@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_in_blacklist(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked IGMP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert ignore


@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_icmp(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked ICMP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"forwarding ICMP message", caplog.text, re.MULTILINE)
            assert not ignore


@pytest.mark.unit
@patch("platform.system")
def test_shouldFilterPacket_udp(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked UDP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"forwarding udp", caplog.text, re.MULTILINE)
            assert not ignore


@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_udp_blacklisted(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked UDP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07\x6c\x07\x6c\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    # Note: custom logging level
    LOG_TRACE = 5
    with caplog.at_level(LOG_TRACE):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"ignoring blacklisted UDP", caplog.text, re.MULTILINE)
            assert ignore


@pytest.mark.unit
@patch("platform.system")
def test_shouldFilterPacket_tcp(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked TCP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"forwarding tcp", caplog.text, re.MULTILINE)
            assert not ignore


@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_tcp_blacklisted(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked TCP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x17\x0c\x17\x0c\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    # Note: custom logging level
    LOG_TRACE = 5
    with caplog.at_level(LOG_TRACE):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"ignoring blacklisted TCP", caplog.text, re.MULTILINE)
            assert ignore


@pytest.mark.unitslow
@patch("platform.system")
def test_ipToNodeId_none(mock_platform_system, caplog, iface_with_nodes):
    """Test _ipToNodeId()"""
    iface = iface_with_nodes
    iface.noProto = True
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            nodeid = tun._ipToNodeId("something not useful")
            assert nodeid is None


@pytest.mark.unitslow
@patch("platform.system")
def test_ipToNodeId_all(mock_platform_system, caplog, iface_with_nodes):
    """Test _ipToNodeId()"""
    iface = iface_with_nodes
    iface.noProto = True
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            nodeid = tun._ipToNodeId(b"\x00\x00\xff\xff")
            assert nodeid == "^all"

Functions

def test_Tunnel_on_non_linux_system(mock_platform_system)

Test that we cannot instantiate a Tunnel on a non Linux system

Expand source code

@pytest.mark.unit
@patch("platform.system")
def test_Tunnel_on_non_linux_system(mock_platform_system):
    """Test that we cannot instantiate a Tunnel on a non Linux system"""
    a_mock = MagicMock()
    a_mock.return_value = "notLinux"
    mock_platform_system.side_effect = a_mock
    with patch("socket.socket") as mock_socket:
        with pytest.raises(Tunnel.TunnelError) as pytest_wrapped_e:
            iface = TCPInterface(hostname="localhost", noProto=True)
            Tunnel(iface)
        assert pytest_wrapped_e.type == Tunnel.TunnelError
        assert mock_socket.called

def test_Tunnel_with_interface(mock_platform_system, caplog, iface_with_nodes)

Test that we can not instantiate a Tunnel without a valid interface

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_Tunnel_with_interface(mock_platform_system, caplog, iface_with_nodes):
    """Test that we can not instantiate a Tunnel without a valid interface"""
    iface = iface_with_nodes
    iface.myInfo.my_node_num = 2475227164
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.WARNING):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            assert tun == mt_config.tunnelInstance
            iface.close()
    assert re.search(r"Not creating a TapDevice()", caplog.text, re.MULTILINE)
    assert re.search(r"Not starting TUN reader", caplog.text, re.MULTILINE)
    assert re.search(r"Not sending packet", caplog.text, re.MULTILINE)

def test_Tunnel_without_interface(mock_platform_system)

Test that we can not instantiate a Tunnel without a valid interface

Expand source code

@pytest.mark.unit
@patch("platform.system")
def test_Tunnel_without_interface(mock_platform_system):
    """Test that we can not instantiate a Tunnel without a valid interface"""
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with pytest.raises(Tunnel.TunnelError) as pytest_wrapped_e:
        Tunnel(None)
    assert pytest_wrapped_e.type == Tunnel.TunnelError

def test_ipToNodeId_all(mock_platform_system, caplog, iface_with_nodes)

Test _ipToNodeId()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_ipToNodeId_all(mock_platform_system, caplog, iface_with_nodes):
    """Test _ipToNodeId()"""
    iface = iface_with_nodes
    iface.noProto = True
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            nodeid = tun._ipToNodeId(b"\x00\x00\xff\xff")
            assert nodeid == "^all"

def test_ipToNodeId_none(mock_platform_system, caplog, iface_with_nodes)

Test _ipToNodeId()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_ipToNodeId_none(mock_platform_system, caplog, iface_with_nodes):
    """Test _ipToNodeId()"""
    iface = iface_with_nodes
    iface.noProto = True
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            nodeid = tun._ipToNodeId("something not useful")
            assert nodeid is None

def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with_nodes)

Test onTunnelReceive

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with_nodes):
    """Test onTunnelReceive"""
    iface = iface_with_nodes
    iface.myInfo.my_node_num = 2475227164
    sys.argv = [""]
    mt_config.args = sys.argv
    packet = {"decoded": {"payload": "foo"}, "from": 2475227164}
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            mt_config.tunnelInstance = tun
            onTunnelReceive(packet, iface)
    assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)
    assert re.search(r"Ignoring message we sent", caplog.text, re.MULTILINE)

def test_onTunnelReceive_from_someone_else(mock_platform_system, caplog, iface_with_nodes)

Test onTunnelReceive

Expand source code

@pytest.mark.unit
@patch("platform.system")
def test_onTunnelReceive_from_someone_else(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test onTunnelReceive"""
    iface = iface_with_nodes
    iface.myInfo.my_node_num = 2475227164
    sys.argv = [""]
    mt_config.args = sys.argv
    packet = {"decoded": {"payload": "foo"}, "from": 123}
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            mt_config.tunnelInstance = tun
            onTunnelReceive(packet, iface)
    assert re.search(r"in onTunnelReceive", caplog.text, re.MULTILINE)

def test_shouldFilterPacket_icmp(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_icmp(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked ICMP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"forwarding ICMP message", caplog.text, re.MULTILINE)
            assert not ignore

def test_shouldFilterPacket_in_blacklist(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_in_blacklist(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked IGMP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert ignore

def test_shouldFilterPacket_random(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_random(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # random packet
    packet = b"1234567890123456789012345678901234567890"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert not ignore

def test_shouldFilterPacket_tcp(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unit
@patch("platform.system")
def test_shouldFilterPacket_tcp(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked TCP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"forwarding tcp", caplog.text, re.MULTILINE)
            assert not ignore

def test_shouldFilterPacket_tcp_blacklisted(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_tcp_blacklisted(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked TCP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x17\x0c\x17\x0c\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    # Note: custom logging level
    LOG_TRACE = 5
    with caplog.at_level(LOG_TRACE):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"ignoring blacklisted TCP", caplog.text, re.MULTILINE)
            assert ignore

def test_shouldFilterPacket_udp(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unit
@patch("platform.system")
def test_shouldFilterPacket_udp(mock_platform_system, caplog, iface_with_nodes):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked UDP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    with caplog.at_level(logging.DEBUG):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"forwarding udp", caplog.text, re.MULTILINE)
            assert not ignore

def test_shouldFilterPacket_udp_blacklisted(mock_platform_system, caplog, iface_with_nodes)

Test _shouldFilterPacket()

Expand source code

@pytest.mark.unitslow
@patch("platform.system")
def test_shouldFilterPacket_udp_blacklisted(
    mock_platform_system, caplog, iface_with_nodes
):
    """Test _shouldFilterPacket()"""
    iface = iface_with_nodes
    iface.noProto = True
    # faked UDP
    packet = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x11\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07\x6c\x07\x6c\x00\x00\x00"
    a_mock = MagicMock()
    a_mock.return_value = "Linux"
    mock_platform_system.side_effect = a_mock
    # Note: custom logging level
    LOG_TRACE = 5
    with caplog.at_level(LOG_TRACE):
        with patch("socket.socket"):
            tun = Tunnel(iface)
            ignore = tun._shouldFilterPacket(packet)
            assert re.search(r"ignoring blacklisted UDP", caplog.text, re.MULTILINE)
            assert ignore