build.tests.unit.test_main.test_on_table_enabled() - Code Metrics - Inspection of "Added enable_table event" - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#79)

unknown

created 2023-04-27 14:53 UTC

build.tests.unit.test_main.test_on_table_enabled() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	22
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	16
nop	0
dl	0
loc	22
rs	9.6
c	0
b	0
f	0

"""Test Main methods."""
from unittest import TestCase
from unittest.mock import AsyncMock, MagicMock, call, patch

from kytos.lib.helpers import (get_controller_mock, get_kytos_event_mock,
                               get_switch_mock, get_test_client)

from kytos.core.events import KytosEvent
from napps.kytos.of_lldp.utils import get_cookie
from tests.helpers import get_topology_mock


@patch('kytos.core.controller.Controller.get_switch_by_dpid')

@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
@patch('napps.kytos.of_lldp.main.UBInt32')
@patch('napps.kytos.of_lldp.main.DPID')
@patch('napps.kytos.of_lldp.main.LLDP')
@patch('napps.kytos.of_lldp.main.Ethernet')
async def test_on_ofpt_packet_in(*args):
    """Test on_ofpt_packet_in."""
    (mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
     mock_unpack_non_empty, mock_get_switch_by_dpid) = args

    # pylint: disable=bad-option-value, import-outside-toplevel
    from napps.kytos.of_lldp.main import Main
    Main.get_liveness_controller = MagicMock()
    topology = get_topology_mock()
    controller = get_controller_mock()
    controller.buffers.app.aput = AsyncMock()
    controller.switches = topology.switches
    napp = Main(controller)
    napp.loop_manager.process_if_looped = AsyncMock()
    napp.liveness_manager.consume_hello_if_enabled = AsyncMock()

    switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
    message = MagicMock(in_port=1, data='data')
    event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
                       'message': message})

    mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
    mocked.value = 1
    mock_ubint32.return_value = mocked
    ethernet.ether_type = 0x88CC
    ethernet.data = 'eth_data'
    lldp.chassis_id.sub_value = 'chassis_id'
    lldp.port_id.sub_value = 'port_id'
    dpid.value = "00:00:00:00:00:00:00:02"
    port_b.value = 2

    mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
    mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
                                                           0x04)
    await napp.on_ofpt_packet_in(event)

    calls = [call(mock_ethernet, message.data),
             call(mock_lldp, ethernet.data),
             call(mock_dpid, lldp.chassis_id.sub_value),
             call(mock_ubint32, lldp.port_id.sub_value)]
    mock_unpack_non_empty.assert_has_calls(calls)
    assert napp.loop_manager.process_if_looped.call_count == 1
    assert napp.liveness_manager.consume_hello_if_enabled.call_count == 1
    assert controller.buffers.app.aput.call_count == 1


@patch('kytos.core.controller.Controller.get_switch_by_dpid')

@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
@patch('napps.kytos.of_lldp.main.UBInt32')
@patch('napps.kytos.of_lldp.main.DPID')
@patch('napps.kytos.of_lldp.main.LLDP')
@patch('napps.kytos.of_lldp.main.Ethernet')
async def test_on_ofpt_packet_in_early_intf(*args):
    """Test on_ofpt_packet_in early intf return."""
    (mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
     mock_unpack_non_empty, mock_get_switch_by_dpid) = args

    # pylint: disable=bad-option-value, import-outside-toplevel
    from napps.kytos.of_lldp.main import Main
    Main.get_liveness_controller = MagicMock()
    topology = get_topology_mock()
    controller = get_controller_mock()
    controller.buffers.app.aput = AsyncMock()
    controller.switches = topology.switches
    napp = Main(controller)
    napp.loop_manager.process_if_looped = AsyncMock()
    napp.liveness_manager.consume_hello_if_enabled = AsyncMock()

    switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
    message = MagicMock(in_port=1, data='data')
    event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
                       'message': message})

    mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
    mocked.value = 1
    mock_ubint32.return_value = mocked
    ethernet.ether_type = 0x88CC
    ethernet.data = 'eth_data'
    lldp.chassis_id.sub_value = 'chassis_id'
    lldp.port_id.sub_value = 'port_id'
    dpid.value = "00:00:00:00:00:00:00:02"
    port_b.value = 2

    mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
    mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
                                                           0x04)
    switch.get_interface_by_port_no = MagicMock(return_value=None)
    await napp.on_ofpt_packet_in(event)

    calls = [call(mock_ethernet, message.data),
             call(mock_lldp, ethernet.data),
             call(mock_dpid, lldp.chassis_id.sub_value),
             call(mock_ubint32, lldp.port_id.sub_value)]
    mock_unpack_non_empty.assert_has_calls(calls)
    switch.get_interface_by_port_no.assert_called()
    # early return shouldn't allow these to get called
    assert napp.loop_manager.process_if_looped.call_count == 0
    assert napp.liveness_manager.consume_hello_if_enabled.call_count == 0
    assert controller.buffers.app.aput.call_count == 0


async def test_on_table_enabled():
    """Test on_table_enabled"""
    # pylint: disable=bad-option-value, import-outside-toplevel
    from napps.kytos.of_lldp.main import Main
    controller = get_controller_mock()
    controller.buffers.app.aput = AsyncMock()
    napp = Main(controller)

    # Succesfully setting table groups
    content = {"of_lldp": {"base": 123}}
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
                       content=content)
    await napp.on_table_enabled(event)
    assert napp.table_group == content["of_lldp"]
    assert controller.buffers.app.aput.call_count == 1

    # Failure at setting table groups
    content = {"of_lldp": {"unknown": 123}}
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
                       content=content)
    await napp.on_table_enabled(event)
    assert controller.buffers.app.aput.call_count == 1


# pylint: disable=protected-access,too-many-public-methods
class TestMain(TestCase):
    """Tests for the Main class."""

    def setUp(self):
        """Execute steps before each tests."""
        self.server_name_url = 'http://127.0.0.1:8181/api/kytos/of_lldp'

        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        # pylint: disable=bad-option-value, import-outside-toplevel
        from napps.kytos.of_lldp.main import Main
        self.addCleanup(patch.stopall)
        Main.get_liveness_controller = MagicMock()
        self.topology = get_topology_mock()
        controller = get_controller_mock()
        controller.switches = self.topology.switches

        self.napp = Main(controller)

    def get_topology_interfaces(self):
        """Return interfaces present in topology."""
        interfaces = []
        for switch in list(self.topology.switches.values()):
            interfaces += list(switch.interfaces.values())
        return interfaces

    @patch('napps.kytos.of_lldp.main.of_msg_prio')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_lldp.main.KytosEvent')
    @patch('napps.kytos.of_lldp.main.VLAN')
    @patch('napps.kytos.of_lldp.main.Ethernet')
    @patch('napps.kytos.of_lldp.main.DPID')
    @patch('napps.kytos.of_lldp.main.LLDP')
    def test_execute(self, *args):
        """Test execute method."""
        (_, _, mock_ethernet, _, mock_kytos_event,
         mock_buffer_put, mock_of_msg_prio) = args

        ethernet = MagicMock()
        ethernet.pack.return_value = 'pack'
        interfaces = self.get_topology_interfaces()
        po_args = [(interface.switch.connection.protocol.version,
                    interface.port_number, 'pack') for interface in interfaces]

        mock_ethernet.return_value = ethernet
        mock_kytos_event.side_effect = po_args

        mock_publish_stopped = MagicMock()
        self.napp.try_to_publish_stopped_loops = mock_publish_stopped
        self.napp.execute()

        mock_of_msg_prio.assert_called()
        mock_buffer_put.assert_has_calls([call(arg)
                                          for arg in po_args])
        mock_publish_stopped.assert_called()

    @patch('requests.delete')
    @patch('requests.post')
    def test_handle_lldp_flows(self, mock_post, mock_delete):
        """Test handle_lldp_flow method."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        self.napp.controller.switches = {dpid: switch}
        event_post = get_kytos_event_mock(name='kytos/topology.switch.enabled',
                                          content={'dpid': dpid})

        event_del = get_kytos_event_mock(name='kytos/topology.switch.disabled',
                                         content={'dpid': dpid})

        mock_post.return_value = MagicMock(status_code=202)
        mock_delete.return_value = MagicMock(status_code=202)

        self.napp._handle_lldp_flows(event_post)
        mock_post.assert_called()

        self.napp._handle_lldp_flows(event_del)
        mock_delete.assert_called()

    @patch("time.sleep")
    @patch("requests.post")
    def test_handle_lldp_flows_retries(self, mock_post, _):
        """Test handle_lldp_flow method retries."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        self.napp.controller.switches = {dpid: switch}
        event_post = get_kytos_event_mock(name="kytos/topology.switch.enabled",
                                          content={"dpid": dpid})

        mock = MagicMock()
        mock.request.method = "POST"
        mock.status_code = 500
        mock.text = "some_err"
        mock_post.return_value = mock
        self.napp._handle_lldp_flows(event_post)
        self.assertTrue(mock_post.call_count, 3)

    @patch('napps.kytos.of_lldp.main.PO13')
    @patch('napps.kytos.of_lldp.main.AO13')
    def test_build_lldp_packet_out(self, *args):
        """Test _build_lldp_packet_out method."""
        (mock_ao13, mock_po13) = args

        ao13 = MagicMock()
        po13 = MagicMock()
        po13.actions = []

        mock_ao13.return_value = ao13
        mock_po13.return_value = po13

        packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
        packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')

        self.assertEqual(packet_out13.data, 'data2')
        self.assertEqual(packet_out13.actions, [ao13])
        self.assertEqual(packet_out13.actions[0].port, 2)

        self.assertIsNone(packet_out14)

    @patch('napps.kytos.of_lldp.main.settings')
    @patch('napps.kytos.of_lldp.main.EtherType')
    @patch('napps.kytos.of_lldp.main.Port13')
    def test_build_lldp_flow(self, *args):
        """Test _build_lldp_flow method."""
        (mock_v0x04_port, mock_ethertype,
         mock_settings) = args
        self.napp.vlan_id = None
        mock_v0x04_port.OFPP_CONTROLLER = 1234

        mock_ethertype.LLDP = 10
        mock_settings.FLOW_VLAN_VID = None
        mock_settings.FLOW_PRIORITY = 1500
        mock_settings.TABLE_ID = 0
        dpid = "00:00:00:00:00:00:00:01"

        flow = {}
        match = {}
        flow['priority'] = 1500
        flow['table_id'] = 0
        match['dl_type'] = 10

        flow['match'] = match
        expected_flow_v0x04 = flow.copy()
        expected_flow_v0x04['cookie'] = get_cookie(dpid)
        expected_flow_v0x04['cookie_mask'] = 0xffffffffffffffff

        expected_flow_v0x04['actions'] = [{'action_type': 'output',
                                           'port': 1234}]
        expected_flow_v0x04['table_group'] = 'base'
        expected_flow_v0x04['owner'] = 'of_lldp'

        flow_mod10 = self.napp._build_lldp_flow(0x01, get_cookie(dpid))
        flow_mod13 = self.napp._build_lldp_flow(0x04, get_cookie(dpid))

        self.assertIsNone(flow_mod10)
        self.assertDictEqual(flow_mod13, expected_flow_v0x04)

    def test_unpack_non_empty(self):
        """Test _unpack_non_empty method."""
        desired_class = MagicMock()
        data = MagicMock()
        data.value = 'data'

        obj = self.napp._unpack_non_empty(desired_class, data)

        obj.unpack.assert_called_with('data')

    def test_get_data(self):
        """Test _get_data method."""
        req = MagicMock()
        interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
        req.get_json.return_value = {'interfaces': interfaces}

        data = self.napp._get_data(req)

        self.assertEqual(data, interfaces)

    def test_load_liveness(self) -> None:
        """Test load_liveness."""
        self.napp.load_liveness()
        count = self.napp.liveness_controller.get_enabled_interfaces.call_count
        assert count == 1

    def test_handle_topology_loaded(self) -> None:
        """Test handle_topology_loaded."""
        event = KytosEvent("kytos/topology.topology_loaded",
                           content={"topology": {}})
        self.napp.load_liveness = MagicMock()
        self.napp.loop_manager.handle_topology_loaded = MagicMock()
        self.napp.handle_topology_loaded(event)
        assert self.napp.loop_manager.handle_topology_loaded.call_count == 1
        assert self.napp.load_liveness.call_count == 1

    def test_publish_liveness_status(self) -> None:
        """Test publish_liveness_status."""
        self.napp.controller.buffers.app.put = MagicMock()
        event_suffix, interfaces = "up", [MagicMock(id=1), MagicMock(id=2)]
        self.napp.publish_liveness_status(event_suffix, interfaces)
        assert self.napp.controller.buffers.app.put.call_count == 1
        event = self.napp.controller.buffers.app.put.call_args[0][0]
        assert event.name == f"kytos/of_lldp.liveness.{event_suffix}"
        assert event.content["interfaces"] == interfaces

    def test_get_interfaces(self):
        """Test _get_interfaces method."""
        expected_interfaces = self.get_topology_interfaces()

        interfaces = self.napp._get_interfaces()

        self.assertEqual(interfaces, expected_interfaces)

    def test_get_interfaces_dict(self):
        """Test _get_interfaces_dict method."""
        interfaces = self.napp._get_interfaces()
        expected_interfaces = {inter.id: inter for inter in interfaces}

        interfaces_dict = self.napp._get_interfaces_dict(interfaces)

        self.assertEqual(interfaces_dict, expected_interfaces)

    def test_get_lldp_interfaces(self):
        """Test _get_lldp_interfaces method."""
        lldp_interfaces = self.napp._get_lldp_interfaces()

        expected_interfaces = ['00:00:00:00:00:00:00:01:1',
                               '00:00:00:00:00:00:00:01:2',
                               '00:00:00:00:00:00:00:02:1',
                               '00:00:00:00:00:00:00:02:2']

        self.assertEqual(lldp_interfaces, expected_interfaces)

    def test_rest_get_lldp_interfaces(self):
        """Test get_lldp_interfaces method."""
        api = get_test_client(self.napp.controller, self.napp)
        url = f'{self.server_name_url}/v1/interfaces'
        response = api.open(url, method='GET')

        expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
                                        '00:00:00:00:00:00:00:01:2',
                                        '00:00:00:00:00:00:00:02:1',
                                        '00:00:00:00:00:00:00:02:2']}
        self.assertEqual(response.json, expected_data)
        self.assertEqual(response.status_code, 200)

    def test_enable_disable_lldp_200(self):
        """Test 200 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
                               '00:00:00:00:00:00:00:01:2',
                               '00:00:00:00:00:00:00:02:1',
                               '00:00:00:00:00:00:00:02:2']}

        api = get_test_client(self.napp.controller, self.napp)
        self.napp.publish_liveness_status = MagicMock()

        url = f'{self.server_name_url}/v1/interfaces/disable'
        disable_response = api.open(url, method='POST', json=data)
        assert self.napp.liveness_controller.disable_interfaces.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1

        url = f'{self.server_name_url}/v1/interfaces/enable'
        enable_response = api.open(url, method='POST', json=data)

        self.assertEqual(disable_response.status_code, 200)
        self.assertEqual(enable_response.status_code, 200)

    def test_enable_disable_lldp_404(self):
        """Test 404 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": []}

        self.napp.controller.switches = {}
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/interfaces/disable'
        disable_response = api.open(url, method='POST', json=data)

        url = f'{self.server_name_url}/v1/interfaces/enable'
        enable_response = api.open(url, method='POST', json=data)

        self.assertEqual(disable_response.status_code, 404)
        self.assertEqual(enable_response.status_code, 404)

    def test_enable_disable_lldp_400(self):
        """Test 400 response for enable_lldp and disable_lldp methods."""
        data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
                               '00:00:00:00:00:00:00:01:2',
                               '00:00:00:00:00:00:00:02:1',
                               '00:00:00:00:00:00:00:02:2',
                               '00:00:00:00:00:00:00:03:1',
                               '00:00:00:00:00:00:00:03:2',
                               '00:00:00:00:00:00:00:04:1']}

        api = get_test_client(self.napp.controller, self.napp)
        self.napp.publish_liveness_status = MagicMock()

        url = f'{self.server_name_url}/v1/interfaces/disable'
        disable_response = api.open(url, method='POST', json=data)
        assert self.napp.publish_liveness_status.call_count == 1

        url = f'{self.server_name_url}/v1/interfaces/enable'
        enable_response = api.open(url, method='POST', json=data)

        self.assertEqual(disable_response.status_code, 400)
        self.assertEqual(enable_response.status_code, 400)

    def test_get_time(self):
        """Test get polling time."""
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/polling_time'
        response = api.open(url, method='GET')

        self.assertEqual(response.status_code, 200)

    def test_set_time(self):
        """Test update polling time."""
        data = {"polling_time": 5}

        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/polling_time'
        response = api.open(url, method='POST', json=data)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(self.napp.polling_time, data['polling_time'])

    def test_set_time_400(self):
        """Test fail case the update polling time."""
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v1/polling_time'

        data = {'polling_time': 'A'}
        response = api.open(url, method='POST', json=data)
        self.assertEqual(response.status_code, 400)

    def test_endpoint_enable_liveness(self):
        """Test POST v1/liveness/enable."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/enable"
        data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
        response = api.open(url, method="POST", json=data)
        assert self.napp.liveness_controller.enable_interfaces.call_count == 1
        assert self.napp.liveness_manager.enable.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1
        assert response.json == {}
        assert response.status_code == 200

    def test_endpoint_disable_liveness(self):
        """Test POST v1/liveness/disable."""
        self.napp.liveness_manager.disable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/disable"
        data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
        response = api.open(url, method='POST', json=data)
        assert self.napp.liveness_controller.disable_interfaces.call_count == 1
        assert self.napp.liveness_manager.disable.call_count == 1
        assert self.napp.publish_liveness_status.call_count == 1
        assert response.json == {}
        assert response.status_code == 200

    def test_endpoint_get_liveness(self):
        """Test GET v1/liveness/."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/"
        response = api.open(url, method="GET")
        assert response.json == {"interfaces": []}
        assert response.status_code == 200

    def test_endpoint_get_pair_liveness(self):
        """Test GET v1/liveness//pair."""
        self.napp.liveness_manager.enable = MagicMock()
        self.napp.publish_liveness_status = MagicMock()
        api = get_test_client(self.napp.controller, self.napp)
        url = f"{self.server_name_url}/v1/liveness/pair"
        response = api.open(url, method="GET")
        assert response.json == {"pairs": []}
        assert response.status_code == 200

    def test_set_flow_table_group_owner(self):
        """Test set_flow_table_group_owner"""
        self.napp.table_group = {"base": 2}
        flow = {}
        self.napp.set_flow_table_group_owner(flow, "base")
        assert "table_group" in flow
        assert "owner" in flow
        assert flow["table_id"] == 2


1		"""Test Main methods."""
2		from unittest import TestCase
3		from unittest.mock import AsyncMock, MagicMock, call, patch
4
5		from kytos.lib.helpers import (get_controller_mock, get_kytos_event_mock,
6		get_switch_mock, get_test_client)
7
8		from kytos.core.events import KytosEvent
9		from napps.kytos.of_lldp.utils import get_cookie
10		from tests.helpers import get_topology_mock
11
12
13	View Code Duplication	@patch('kytos.core.controller.Controller.get_switch_by_dpid')
		0 ignored issues – show Duplication introduced 2022-11-18 17:10 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
14		@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
15		@patch('napps.kytos.of_lldp.main.UBInt32')
16		@patch('napps.kytos.of_lldp.main.DPID')
17		@patch('napps.kytos.of_lldp.main.LLDP')
18		@patch('napps.kytos.of_lldp.main.Ethernet')
19		async def test_on_ofpt_packet_in(*args):
20		"""Test on_ofpt_packet_in."""
21		(mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
22		mock_unpack_non_empty, mock_get_switch_by_dpid) = args
23
24		# pylint: disable=bad-option-value, import-outside-toplevel
25		from napps.kytos.of_lldp.main import Main
26		Main.get_liveness_controller = MagicMock()
27		topology = get_topology_mock()
28		controller = get_controller_mock()
29		controller.buffers.app.aput = AsyncMock()
30		controller.switches = topology.switches
31		napp = Main(controller)
32		napp.loop_manager.process_if_looped = AsyncMock()
33		napp.liveness_manager.consume_hello_if_enabled = AsyncMock()
34
35		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
36		message = MagicMock(in_port=1, data='data')
37		event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
38		'message': message})
39
40		mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
41		mocked.value = 1
42		mock_ubint32.return_value = mocked
43		ethernet.ether_type = 0x88CC
44		ethernet.data = 'eth_data'
45		lldp.chassis_id.sub_value = 'chassis_id'
46		lldp.port_id.sub_value = 'port_id'
47		dpid.value = "00:00:00:00:00:00:00:02"
48		port_b.value = 2
49
50		mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
51		mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
52		0x04)
53		await napp.on_ofpt_packet_in(event)
54
55		calls = [call(mock_ethernet, message.data),
56		call(mock_lldp, ethernet.data),
57		call(mock_dpid, lldp.chassis_id.sub_value),
58		call(mock_ubint32, lldp.port_id.sub_value)]
59		mock_unpack_non_empty.assert_has_calls(calls)
60		assert napp.loop_manager.process_if_looped.call_count == 1
61		assert napp.liveness_manager.consume_hello_if_enabled.call_count == 1
62		assert controller.buffers.app.aput.call_count == 1
63
64
65	View Code Duplication	@patch('kytos.core.controller.Controller.get_switch_by_dpid')
		0 ignored issues – show Duplication introduced 2022-11-18 17:10 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
66		@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
67		@patch('napps.kytos.of_lldp.main.UBInt32')
68		@patch('napps.kytos.of_lldp.main.DPID')
69		@patch('napps.kytos.of_lldp.main.LLDP')
70		@patch('napps.kytos.of_lldp.main.Ethernet')
71		async def test_on_ofpt_packet_in_early_intf(*args):
72		"""Test on_ofpt_packet_in early intf return."""
73		(mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
74		mock_unpack_non_empty, mock_get_switch_by_dpid) = args
75
76		# pylint: disable=bad-option-value, import-outside-toplevel
77		from napps.kytos.of_lldp.main import Main
78		Main.get_liveness_controller = MagicMock()
79		topology = get_topology_mock()
80		controller = get_controller_mock()
81		controller.buffers.app.aput = AsyncMock()
82		controller.switches = topology.switches
83		napp = Main(controller)
84		napp.loop_manager.process_if_looped = AsyncMock()
85		napp.liveness_manager.consume_hello_if_enabled = AsyncMock()
86
87		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
88		message = MagicMock(in_port=1, data='data')
89		event = KytosEvent('ofpt_packet_in', content={'source': switch.connection,
90		'message': message})
91
92		mocked, ethernet, lldp, dpid, port_b = [MagicMock() for _ in range(5)]
93		mocked.value = 1
94		mock_ubint32.return_value = mocked
95		ethernet.ether_type = 0x88CC
96		ethernet.data = 'eth_data'
97		lldp.chassis_id.sub_value = 'chassis_id'
98		lldp.port_id.sub_value = 'port_id'
99		dpid.value = "00:00:00:00:00:00:00:02"
100		port_b.value = 2
101
102		mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
103		mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
104		0x04)
105		switch.get_interface_by_port_no = MagicMock(return_value=None)
106		await napp.on_ofpt_packet_in(event)
107
108		calls = [call(mock_ethernet, message.data),
109		call(mock_lldp, ethernet.data),
110		call(mock_dpid, lldp.chassis_id.sub_value),
111		call(mock_ubint32, lldp.port_id.sub_value)]
112		mock_unpack_non_empty.assert_has_calls(calls)
113		switch.get_interface_by_port_no.assert_called()
114		# early return shouldn't allow these to get called
115		assert napp.loop_manager.process_if_looped.call_count == 0
116		assert napp.liveness_manager.consume_hello_if_enabled.call_count == 0
117		assert controller.buffers.app.aput.call_count == 0
118
119
120		async def test_on_table_enabled():
121		"""Test on_table_enabled"""
122		# pylint: disable=bad-option-value, import-outside-toplevel
123		from napps.kytos.of_lldp.main import Main
124		controller = get_controller_mock()
125		controller.buffers.app.aput = AsyncMock()
126		napp = Main(controller)
127
128		# Succesfully setting table groups
129		content = {"of_lldp": {"base": 123}}
130		event = KytosEvent(name="kytos/of_multi_table.enable_table",
131		content=content)
132		await napp.on_table_enabled(event)
133		assert napp.table_group == content["of_lldp"]
134		assert controller.buffers.app.aput.call_count == 1
135
136		# Failure at setting table groups
137		content = {"of_lldp": {"unknown": 123}}
138		event = KytosEvent(name="kytos/of_multi_table.enable_table",
139		content=content)
140		await napp.on_table_enabled(event)
141		assert controller.buffers.app.aput.call_count == 1
142
143
144		# pylint: disable=protected-access,too-many-public-methods
145		class TestMain(TestCase):
146		"""Tests for the Main class."""
147
148		def setUp(self):
149		"""Execute steps before each tests."""
150		self.server_name_url = 'http://127.0.0.1:8181/api/kytos/of_lldp'
151
152		patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
153		# pylint: disable=bad-option-value, import-outside-toplevel
154		from napps.kytos.of_lldp.main import Main
155		self.addCleanup(patch.stopall)
156		Main.get_liveness_controller = MagicMock()
157		self.topology = get_topology_mock()
158		controller = get_controller_mock()
159		controller.switches = self.topology.switches
160
161		self.napp = Main(controller)
162
163		def get_topology_interfaces(self):
164		"""Return interfaces present in topology."""
165		interfaces = []
166		for switch in list(self.topology.switches.values()):
167		interfaces += list(switch.interfaces.values())
168		return interfaces
169
170		@patch('napps.kytos.of_lldp.main.of_msg_prio')
171		@patch('kytos.core.buffers.KytosEventBuffer.put')
172		@patch('napps.kytos.of_lldp.main.KytosEvent')
173		@patch('napps.kytos.of_lldp.main.VLAN')
174		@patch('napps.kytos.of_lldp.main.Ethernet')
175		@patch('napps.kytos.of_lldp.main.DPID')
176		@patch('napps.kytos.of_lldp.main.LLDP')
177		def test_execute(self, *args):
178		"""Test execute method."""
179		(_, _, mock_ethernet, _, mock_kytos_event,
180		mock_buffer_put, mock_of_msg_prio) = args
181
182		ethernet = MagicMock()
183		ethernet.pack.return_value = 'pack'
184		interfaces = self.get_topology_interfaces()
185		po_args = [(interface.switch.connection.protocol.version,
186		interface.port_number, 'pack') for interface in interfaces]
187
188		mock_ethernet.return_value = ethernet
189		mock_kytos_event.side_effect = po_args
190
191		mock_publish_stopped = MagicMock()
192		self.napp.try_to_publish_stopped_loops = mock_publish_stopped
193		self.napp.execute()
194
195		mock_of_msg_prio.assert_called()
196		mock_buffer_put.assert_has_calls([call(arg)
197		for arg in po_args])
198		mock_publish_stopped.assert_called()
199
200		@patch('requests.delete')
201		@patch('requests.post')
202		def test_handle_lldp_flows(self, mock_post, mock_delete):
203		"""Test handle_lldp_flow method."""
204		dpid = "00:00:00:00:00:00:00:01"
205		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
206		self.napp.controller.switches = {dpid: switch}
207		event_post = get_kytos_event_mock(name='kytos/topology.switch.enabled',
208		content={'dpid': dpid})
209
210		event_del = get_kytos_event_mock(name='kytos/topology.switch.disabled',
211		content={'dpid': dpid})
212
213		mock_post.return_value = MagicMock(status_code=202)
214		mock_delete.return_value = MagicMock(status_code=202)
215
216		self.napp._handle_lldp_flows(event_post)
217		mock_post.assert_called()
218
219		self.napp._handle_lldp_flows(event_del)
220		mock_delete.assert_called()
221
222		@patch("time.sleep")
223		@patch("requests.post")
224		def test_handle_lldp_flows_retries(self, mock_post, _):
225		"""Test handle_lldp_flow method retries."""
226		dpid = "00:00:00:00:00:00:00:01"
227		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
228		self.napp.controller.switches = {dpid: switch}
229		event_post = get_kytos_event_mock(name="kytos/topology.switch.enabled",
230		content={"dpid": dpid})
231
232		mock = MagicMock()
233		mock.request.method = "POST"
234		mock.status_code = 500
235		mock.text = "some_err"
236		mock_post.return_value = mock
237		self.napp._handle_lldp_flows(event_post)
238		self.assertTrue(mock_post.call_count, 3)
239
240		@patch('napps.kytos.of_lldp.main.PO13')
241		@patch('napps.kytos.of_lldp.main.AO13')
242		def test_build_lldp_packet_out(self, *args):
243		"""Test _build_lldp_packet_out method."""
244		(mock_ao13, mock_po13) = args
245
246		ao13 = MagicMock()
247		po13 = MagicMock()
248		po13.actions = []
249
250		mock_ao13.return_value = ao13
251		mock_po13.return_value = po13
252
253		packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
254		packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')
255
256		self.assertEqual(packet_out13.data, 'data2')
257		self.assertEqual(packet_out13.actions, [ao13])
258		self.assertEqual(packet_out13.actions[0].port, 2)
259
260		self.assertIsNone(packet_out14)
261
262		@patch('napps.kytos.of_lldp.main.settings')
263		@patch('napps.kytos.of_lldp.main.EtherType')
264		@patch('napps.kytos.of_lldp.main.Port13')
265		def test_build_lldp_flow(self, *args):
266		"""Test _build_lldp_flow method."""
267		(mock_v0x04_port, mock_ethertype,
268		mock_settings) = args
269		self.napp.vlan_id = None
270		mock_v0x04_port.OFPP_CONTROLLER = 1234
271
272		mock_ethertype.LLDP = 10
273		mock_settings.FLOW_VLAN_VID = None
274		mock_settings.FLOW_PRIORITY = 1500
275		mock_settings.TABLE_ID = 0
276		dpid = "00:00:00:00:00:00:00:01"
277
278		flow = {}
279		match = {}
280		flow['priority'] = 1500
281		flow['table_id'] = 0
282		match['dl_type'] = 10
283
284		flow['match'] = match
285		expected_flow_v0x04 = flow.copy()
286		expected_flow_v0x04['cookie'] = get_cookie(dpid)
287		expected_flow_v0x04['cookie_mask'] = 0xffffffffffffffff
288
289		expected_flow_v0x04['actions'] = [{'action_type': 'output',
290		'port': 1234}]
291		expected_flow_v0x04['table_group'] = 'base'
292		expected_flow_v0x04['owner'] = 'of_lldp'
293
294		flow_mod10 = self.napp._build_lldp_flow(0x01, get_cookie(dpid))
295		flow_mod13 = self.napp._build_lldp_flow(0x04, get_cookie(dpid))
296
297		self.assertIsNone(flow_mod10)
298		self.assertDictEqual(flow_mod13, expected_flow_v0x04)
299
300		def test_unpack_non_empty(self):
301		"""Test _unpack_non_empty method."""
302		desired_class = MagicMock()
303		data = MagicMock()
304		data.value = 'data'
305
306		obj = self.napp._unpack_non_empty(desired_class, data)
307
308		obj.unpack.assert_called_with('data')
309
310		def test_get_data(self):
311		"""Test _get_data method."""
312		req = MagicMock()
313		interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
314		req.get_json.return_value = {'interfaces': interfaces}
315
316		data = self.napp._get_data(req)
317
318		self.assertEqual(data, interfaces)
319
320		def test_load_liveness(self) -> None:
321		"""Test load_liveness."""
322		self.napp.load_liveness()
323		count = self.napp.liveness_controller.get_enabled_interfaces.call_count
324		assert count == 1
325
326		def test_handle_topology_loaded(self) -> None:
327		"""Test handle_topology_loaded."""
328		event = KytosEvent("kytos/topology.topology_loaded",
329		content={"topology": {}})
330		self.napp.load_liveness = MagicMock()
331		self.napp.loop_manager.handle_topology_loaded = MagicMock()
332		self.napp.handle_topology_loaded(event)
333		assert self.napp.loop_manager.handle_topology_loaded.call_count == 1
334		assert self.napp.load_liveness.call_count == 1
335
336		def test_publish_liveness_status(self) -> None:
337		"""Test publish_liveness_status."""
338		self.napp.controller.buffers.app.put = MagicMock()
339		event_suffix, interfaces = "up", [MagicMock(id=1), MagicMock(id=2)]
340		self.napp.publish_liveness_status(event_suffix, interfaces)
341		assert self.napp.controller.buffers.app.put.call_count == 1
342		event = self.napp.controller.buffers.app.put.call_args[0][0]
343		assert event.name == f"kytos/of_lldp.liveness.{event_suffix}"
344		assert event.content["interfaces"] == interfaces
345
346		def test_get_interfaces(self):
347		"""Test _get_interfaces method."""
348		expected_interfaces = self.get_topology_interfaces()
349
350		interfaces = self.napp._get_interfaces()
351
352		self.assertEqual(interfaces, expected_interfaces)
353
354		def test_get_interfaces_dict(self):
355		"""Test _get_interfaces_dict method."""
356		interfaces = self.napp._get_interfaces()
357		expected_interfaces = {inter.id: inter for inter in interfaces}
358
359		interfaces_dict = self.napp._get_interfaces_dict(interfaces)
360
361		self.assertEqual(interfaces_dict, expected_interfaces)
362
363		def test_get_lldp_interfaces(self):
364		"""Test _get_lldp_interfaces method."""
365		lldp_interfaces = self.napp._get_lldp_interfaces()
366
367		expected_interfaces = ['00:00:00:00:00:00:00:01:1',
368		'00:00:00:00:00:00:00:01:2',
369		'00:00:00:00:00:00:00:02:1',
370		'00:00:00:00:00:00:00:02:2']
371
372		self.assertEqual(lldp_interfaces, expected_interfaces)
373
374		def test_rest_get_lldp_interfaces(self):
375		"""Test get_lldp_interfaces method."""
376		api = get_test_client(self.napp.controller, self.napp)
377		url = f'{self.server_name_url}/v1/interfaces'
378		response = api.open(url, method='GET')
379
380		expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
381		'00:00:00:00:00:00:00:01:2',
382		'00:00:00:00:00:00:00:02:1',
383		'00:00:00:00:00:00:00:02:2']}
384		self.assertEqual(response.json, expected_data)
385		self.assertEqual(response.status_code, 200)
386
387		def test_enable_disable_lldp_200(self):
388		"""Test 200 response for enable_lldp and disable_lldp methods."""
389		data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
390		'00:00:00:00:00:00:00:01:2',
391		'00:00:00:00:00:00:00:02:1',
392		'00:00:00:00:00:00:00:02:2']}
393
394		api = get_test_client(self.napp.controller, self.napp)
395		self.napp.publish_liveness_status = MagicMock()
396
397		url = f'{self.server_name_url}/v1/interfaces/disable'
398		disable_response = api.open(url, method='POST', json=data)
399		assert self.napp.liveness_controller.disable_interfaces.call_count == 1
400		assert self.napp.publish_liveness_status.call_count == 1
401
402		url = f'{self.server_name_url}/v1/interfaces/enable'
403		enable_response = api.open(url, method='POST', json=data)
404
405		self.assertEqual(disable_response.status_code, 200)
406		self.assertEqual(enable_response.status_code, 200)
407
408		def test_enable_disable_lldp_404(self):
409		"""Test 404 response for enable_lldp and disable_lldp methods."""
410		data = {"interfaces": []}
411
412		self.napp.controller.switches = {}
413		api = get_test_client(self.napp.controller, self.napp)
414
415		url = f'{self.server_name_url}/v1/interfaces/disable'
416		disable_response = api.open(url, method='POST', json=data)
417
418		url = f'{self.server_name_url}/v1/interfaces/enable'
419		enable_response = api.open(url, method='POST', json=data)
420
421		self.assertEqual(disable_response.status_code, 404)
422		self.assertEqual(enable_response.status_code, 404)
423
424		def test_enable_disable_lldp_400(self):
425		"""Test 400 response for enable_lldp and disable_lldp methods."""
426		data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
427		'00:00:00:00:00:00:00:01:2',
428		'00:00:00:00:00:00:00:02:1',
429		'00:00:00:00:00:00:00:02:2',
430		'00:00:00:00:00:00:00:03:1',
431		'00:00:00:00:00:00:00:03:2',
432		'00:00:00:00:00:00:00:04:1']}
433
434		api = get_test_client(self.napp.controller, self.napp)
435		self.napp.publish_liveness_status = MagicMock()
436
437		url = f'{self.server_name_url}/v1/interfaces/disable'
438		disable_response = api.open(url, method='POST', json=data)
439		assert self.napp.publish_liveness_status.call_count == 1
440
441		url = f'{self.server_name_url}/v1/interfaces/enable'
442		enable_response = api.open(url, method='POST', json=data)
443
444		self.assertEqual(disable_response.status_code, 400)
445		self.assertEqual(enable_response.status_code, 400)
446
447		def test_get_time(self):
448		"""Test get polling time."""
449		api = get_test_client(self.napp.controller, self.napp)
450
451		url = f'{self.server_name_url}/v1/polling_time'
452		response = api.open(url, method='GET')
453
454		self.assertEqual(response.status_code, 200)
455
456		def test_set_time(self):
457		"""Test update polling time."""
458		data = {"polling_time": 5}
459
460		api = get_test_client(self.napp.controller, self.napp)
461
462		url = f'{self.server_name_url}/v1/polling_time'
463		response = api.open(url, method='POST', json=data)
464
465		self.assertEqual(response.status_code, 200)
466		self.assertEqual(self.napp.polling_time, data['polling_time'])
467
468		def test_set_time_400(self):
469		"""Test fail case the update polling time."""
470		api = get_test_client(self.napp.controller, self.napp)
471
472		url = f'{self.server_name_url}/v1/polling_time'
473
474		data = {'polling_time': 'A'}
475		response = api.open(url, method='POST', json=data)
476		self.assertEqual(response.status_code, 400)
477
478		def test_endpoint_enable_liveness(self):
479		"""Test POST v1/liveness/enable."""
480		self.napp.liveness_manager.enable = MagicMock()
481		self.napp.publish_liveness_status = MagicMock()
482		api = get_test_client(self.napp.controller, self.napp)
483		url = f"{self.server_name_url}/v1/liveness/enable"
484		data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
485		response = api.open(url, method="POST", json=data)
486		assert self.napp.liveness_controller.enable_interfaces.call_count == 1
487		assert self.napp.liveness_manager.enable.call_count == 1
488		assert self.napp.publish_liveness_status.call_count == 1
489		assert response.json == {}
490		assert response.status_code == 200
491
492		def test_endpoint_disable_liveness(self):
493		"""Test POST v1/liveness/disable."""
494		self.napp.liveness_manager.disable = MagicMock()
495		self.napp.publish_liveness_status = MagicMock()
496		api = get_test_client(self.napp.controller, self.napp)
497		url = f"{self.server_name_url}/v1/liveness/disable"
498		data = {"interfaces": ["00:00:00:00:00:00:00:01:1"]}
499		response = api.open(url, method='POST', json=data)
500		assert self.napp.liveness_controller.disable_interfaces.call_count == 1
501		assert self.napp.liveness_manager.disable.call_count == 1
502		assert self.napp.publish_liveness_status.call_count == 1
503		assert response.json == {}
504		assert response.status_code == 200
505
506		def test_endpoint_get_liveness(self):
507		"""Test GET v1/liveness/."""
508		self.napp.liveness_manager.enable = MagicMock()
509		self.napp.publish_liveness_status = MagicMock()
510		api = get_test_client(self.napp.controller, self.napp)
511		url = f"{self.server_name_url}/v1/liveness/"
512		response = api.open(url, method="GET")
513		assert response.json == {"interfaces": []}
514		assert response.status_code == 200
515
516		def test_endpoint_get_pair_liveness(self):
517		"""Test GET v1/liveness//pair."""
518		self.napp.liveness_manager.enable = MagicMock()
519		self.napp.publish_liveness_status = MagicMock()
520		api = get_test_client(self.napp.controller, self.napp)
521		url = f"{self.server_name_url}/v1/liveness/pair"
522		response = api.open(url, method="GET")
523		assert response.json == {"pairs": []}
524		assert response.status_code == 200
525
526		def test_set_flow_table_group_owner(self):
527		"""Test set_flow_table_group_owner"""
528		self.napp.table_group = {"base": 2}
529		flow = {}
530		self.napp.set_flow_table_group_owner(flow, "base")
531		assert "table_group" in flow
532		assert "owner" in flow
533		assert flow["table_id"] == 2
534

kytos-ng / of_lldp

Pull Request — master (#79)

build.tests.unit.test_main.test_on_table_enabled() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like