build.tests.unit.test_main.TestMain.test_rest_get_lldp_interfaces() - Code Metrics - Inspection of "[Feature] Loop detection via LLDP `ofpt_packet_in`" - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#36)

by Vinicius

created 2022-01-07 22:32 UTC

TestMain.test_rest_get_lldp_interfaces() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	14
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	12
nop	1
dl	0
loc	14
rs	9.8
c	0
b	0
f	0

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

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

from tests.helpers import get_topology_mock


# 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)

        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('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_lldp.main.Main._build_lldp_packet_out')
    @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_build_lldp_packet_out,
         mock_buffer_put) = 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

        self.napp.execute()

        mock_build_lldp_packet_out.assert_has_calls([call(*(arg))
                                                     for arg in po_args])
        mock_buffer_put.assert_has_calls([call(arg)
                                          for arg in po_args])

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

        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.Main._is_port_looped')
    @patch('napps.kytos.of_lldp.main.Main._is_loop_ignored')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_lldp.main.KytosEvent')
    @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')
    def test_notify_uplink_detected(self, *args):
        """Test notify_uplink_detected method."""
        (mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
         mock_unpack_non_empty, mock_get_switch_by_dpid, mock_kytos_event,
         mock_buffer_put, mock_loop_ignored, mock_port_looped) = args

        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        message = MagicMock()
        message.in_port = 1
        message.data = 'data'
        event = get_kytos_event_mock(name='kytos/of_core.v0x0[14].messages.in.'
                                          'ofpt_packet_in',
                                     content={'source': switch.connection,
                                              'message': message})

        mocked = MagicMock()
        mocked.value = 1
        mock_ubint32.return_value = mocked
        ethernet = MagicMock()
        ethernet.ether_type = 0x88CC
        ethernet.data = 'eth_data'
        lldp = MagicMock()
        lldp.chassis_id.sub_value = 'chassis_id'
        lldp.port_id.sub_value = 'port_id'
        dpid = MagicMock()
        dpid.value = "00:00:00:00:00:00:00:02"
        port_b = MagicMock()
        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)
        mock_kytos_event.return_value = 'nni'

        self.napp.notify_uplink_detected(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)
        mock_buffer_put.assert_called_with('nni')
        mock_loop_ignored.assert_called()
        mock_port_looped.assert_called()

    def test_is_port_looped(self):
        """Test is_port_looped cases."""

        dpid_a = "00:00:00:00:00:00:00:01"
        dpid_b = "00:00:00:00:00:00:00:02"
        values = [
            (dpid_a, 7, dpid_a, 6, True),
            (dpid_a, 1, dpid_a, 2, True),
            (dpid_a, 7, dpid_a, 7, False),
            (dpid_a, 1, dpid_b, 2, False),
            (dpid_a, 2, dpid_b, 1, False),
        ]
        for dpid_a, port_a, dpid_b, port_b, looped in values:
            with self.subTest(dpid_a=dpid_a, port_a=port_a, port_b=port_b,
                              looped=looped):
                assert (
                    self.napp._is_port_looped(dpid_a, port_a, dpid_b, port_b)
                    == looped
                )

    def test_is_loop_ignored(self):
        """Test is_loop_ignored."""

        dpid = "00:00:00:00:00:00:00:01"
        port_a = 1
        port_b = 2
        self.napp.ignored_loops[dpid] = {(port_a, port_b)}

        assert self.napp._is_loop_ignored(dpid, port_a=port_a, port_b=port_b)
        assert self.napp._is_loop_ignored(dpid, port_a=port_b, port_b=port_a)

        assert not self.napp._is_loop_ignored(dpid, port_a + 20, port_b)

        dpid = "00:00:00:00:00:00:00:02"
        assert not self.napp._is_loop_ignored(dpid, port_a, port_b)

    @patch('napps.kytos.of_lldp.main.log')
    def test_lldp_loop_handler_log_actio(self, mock_log):
        """Test lldp_loop_handler log action."""

        switch = MagicMock()
        dpid = "00:00:00:00:00:00:00:01"
        switch.id = dpid
        intf_a = MagicMock()
        intf_a.port_number = 1
        intf_b = MagicMock()
        intf_b.port_number = 2
        self.napp.lldp_loop_handler(switch, intf_a, intf_b, action="log")
        mock_log.warning.assert_called()

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

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

        mock_ao10.return_value = ao10
        mock_ao13.return_value = ao13
        mock_po10.return_value = po10
        mock_po13.return_value = po13

        packet_out10 = self.napp._build_lldp_packet_out(0x01, 1, 'data1')
        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_out10.data, 'data1')
        self.assertEqual(packet_out10.actions, [ao10])
        self.assertEqual(packet_out10.actions[0].port, 1)

        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')
    @patch('napps.kytos.of_lldp.main.Port10')
    def test_build_lldp_flow(self, *args):
        """Test _build_lldp_flow method."""
        (mock_v0x01_port, mock_v0x04_port, mock_ethertype,
         mock_settings) = args
        self.napp.vlan_id = None
        mock_v0x01_port.OFPP_CONTROLLER = 123
        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

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

        flow['match'] = match
        expected_flow_v0x01 = flow.copy()
        expected_flow_v0x04 = flow.copy()

        expected_flow_v0x01['actions'] = [{'action_type': 'output',
                                           'port': 123}]

        expected_flow_v0x04['actions'] = [{'action_type': 'output',
                                           'port': 1234}]

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

        self.assertDictEqual(flow_mod10, expected_flow_v0x01)
        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_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',
                               '00:00:00:00:00:00:00:03:1',
                               '00:00:00:00:00:00:00:03: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',
                                        '00:00:00:00:00:00:00:03:1',
                                        '00:00:00:00:00:00:00:03: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',
                               '00:00:00:00:00:00:00:03:1',
                               '00:00:00:00:00:00:00:03:2']}

        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, 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)

        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, 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)


1			"""Test Main methods."""
2			from unittest import TestCase
3			from unittest.mock import 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 tests.helpers import get_topology_mock
9
10
11			# pylint: disable=protected-access,too-many-public-methods
12			class TestMain(TestCase):
13			"""Tests for the Main class."""
14
15			def setUp(self):
16			"""Execute steps before each tests."""
17			self.server_name_url = 'http://127.0.0.1:8181/api/kytos/of_lldp'
18
19			patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
20			# pylint: disable=bad-option-value, import-outside-toplevel
21			from napps.kytos.of_lldp.main import Main
22			self.addCleanup(patch.stopall)
23
24			self.topology = get_topology_mock()
25			controller = get_controller_mock()
26			controller.switches = self.topology.switches
27
28			self.napp = Main(controller)
29
30			def get_topology_interfaces(self):
31			"""Return interfaces present in topology."""
32			interfaces = []
33			for switch in list(self.topology.switches.values()):
34			interfaces += list(switch.interfaces.values())
35			return interfaces
36
37			@patch('kytos.core.buffers.KytosEventBuffer.put')
38			@patch('napps.kytos.of_lldp.main.Main._build_lldp_packet_out')
39			@patch('napps.kytos.of_lldp.main.KytosEvent')
40			@patch('napps.kytos.of_lldp.main.VLAN')
41			@patch('napps.kytos.of_lldp.main.Ethernet')
42			@patch('napps.kytos.of_lldp.main.DPID')
43			@patch('napps.kytos.of_lldp.main.LLDP')
44			def test_execute(self, *args):
45			"""Test execute method."""
46			(_, _, mock_ethernet, _, mock_kytos_event, mock_build_lldp_packet_out,
47			mock_buffer_put) = args
48
49			ethernet = MagicMock()
50			ethernet.pack.return_value = 'pack'
51			interfaces = self.get_topology_interfaces()
52			po_args = [(interface.switch.connection.protocol.version,
53			interface.port_number, 'pack') for interface in interfaces]
54
55			mock_ethernet.return_value = ethernet
56			mock_kytos_event.side_effect = po_args
57
58			self.napp.execute()
59
60			mock_build_lldp_packet_out.assert_has_calls([call(*(arg))
61			for arg in po_args])
62			mock_buffer_put.assert_has_calls([call(arg)
63			for arg in po_args])
64
65			@patch('requests.delete')
66			@patch('requests.post')
67			def test_handle_lldp_flows(self, mock_post, mock_delete):
68			"""Test handle_lldp_flow method."""
69			dpid = "00:00:00:00:00:00:00:01"
70			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
71			self.napp.controller.switches = {dpid: switch}
72			event_post = get_kytos_event_mock(name='kytos/topology.switch.enabled',
73			content={'dpid': dpid})
74
75			event_del = get_kytos_event_mock(name='kytos/topology.switch.disabled',
76			content={'dpid': dpid})
77
78			self.napp.handle_lldp_flows(event_post)
79			mock_post.assert_called()
80
81			self.napp.handle_lldp_flows(event_del)
82			mock_delete.assert_called()
83
84			@patch("time.sleep")
85			@patch("requests.post")
86			def test_handle_lldp_flows_retries(self, mock_post, _):
87			"""Test handle_lldp_flow method retries."""
88			dpid = "00:00:00:00:00:00:00:01"
89			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
90			self.napp.controller.switches = {dpid: switch}
91			event_post = get_kytos_event_mock(name="kytos/topology.switch.enabled",
92			content={"dpid": dpid})
93
94			mock = MagicMock()
95			mock.request.method = "POST"
96			mock.status_code = 500
97			mock.text = "some_err"
98			mock_post.return_value = mock
99			self.napp._handle_lldp_flows(event_post)
100			self.assertTrue(mock_post.call_count, 3)
101
102			@patch('napps.kytos.of_lldp.main.Main._is_port_looped')
103			@patch('napps.kytos.of_lldp.main.Main._is_loop_ignored')
104			@patch('kytos.core.buffers.KytosEventBuffer.put')
105			@patch('napps.kytos.of_lldp.main.KytosEvent')
106			@patch('kytos.core.controller.Controller.get_switch_by_dpid')
107			@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
108			@patch('napps.kytos.of_lldp.main.UBInt32')
109			@patch('napps.kytos.of_lldp.main.DPID')
110			@patch('napps.kytos.of_lldp.main.LLDP')
111			@patch('napps.kytos.of_lldp.main.Ethernet')
112			def test_notify_uplink_detected(self, *args):
113			"""Test notify_uplink_detected method."""
114			(mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
115			mock_unpack_non_empty, mock_get_switch_by_dpid, mock_kytos_event,
116			mock_buffer_put, mock_loop_ignored, mock_port_looped) = args
117
118			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
119			message = MagicMock()
120			message.in_port = 1
121			message.data = 'data'
122			event = get_kytos_event_mock(name='kytos/of_core.v0x0[14].messages.in.'
123			'ofpt_packet_in',
124			content={'source': switch.connection,
125			'message': message})
126
127			mocked = MagicMock()
128			mocked.value = 1
129			mock_ubint32.return_value = mocked
130			ethernet = MagicMock()
131			ethernet.ether_type = 0x88CC
132			ethernet.data = 'eth_data'
133			lldp = MagicMock()
134			lldp.chassis_id.sub_value = 'chassis_id'
135			lldp.port_id.sub_value = 'port_id'
136			dpid = MagicMock()
137			dpid.value = "00:00:00:00:00:00:00:02"
138			port_b = MagicMock()
139			port_b.value = 2
140
141			mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
142			mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
143			0x04)
144			mock_kytos_event.return_value = 'nni'
145
146			self.napp.notify_uplink_detected(event)
147
148			calls = [call(mock_ethernet, message.data),
149			call(mock_lldp, ethernet.data),
150			call(mock_dpid, lldp.chassis_id.sub_value),
151			call(mock_ubint32, lldp.port_id.sub_value)]
152			mock_unpack_non_empty.assert_has_calls(calls)
153			mock_buffer_put.assert_called_with('nni')
154			mock_loop_ignored.assert_called()
155			mock_port_looped.assert_called()
156
157			def test_is_port_looped(self):
158			"""Test is_port_looped cases."""
159
160			dpid_a = "00:00:00:00:00:00:00:01"
161			dpid_b = "00:00:00:00:00:00:00:02"
162			values = [
163			(dpid_a, 7, dpid_a, 6, True),
164			(dpid_a, 1, dpid_a, 2, True),
165			(dpid_a, 7, dpid_a, 7, False),
166			(dpid_a, 1, dpid_b, 2, False),
167			(dpid_a, 2, dpid_b, 1, False),
168			]
169			for dpid_a, port_a, dpid_b, port_b, looped in values:
170			with self.subTest(dpid_a=dpid_a, port_a=port_a, port_b=port_b,
171			looped=looped):
172			assert (
173			self.napp._is_port_looped(dpid_a, port_a, dpid_b, port_b)
174			== looped
175			)
176
177			def test_is_loop_ignored(self):
178			"""Test is_loop_ignored."""
179
180			dpid = "00:00:00:00:00:00:00:01"
181			port_a = 1
182			port_b = 2
183			self.napp.ignored_loops[dpid] = {(port_a, port_b)}
184
185			assert self.napp._is_loop_ignored(dpid, port_a=port_a, port_b=port_b)
186			assert self.napp._is_loop_ignored(dpid, port_a=port_b, port_b=port_a)
187
188			assert not self.napp._is_loop_ignored(dpid, port_a + 20, port_b)
189
190			dpid = "00:00:00:00:00:00:00:02"
191			assert not self.napp._is_loop_ignored(dpid, port_a, port_b)
192
193			@patch('napps.kytos.of_lldp.main.log')
194			def test_lldp_loop_handler_log_actio(self, mock_log):
195			"""Test lldp_loop_handler log action."""
196
197			switch = MagicMock()
198			dpid = "00:00:00:00:00:00:00:01"
199			switch.id = dpid
200			intf_a = MagicMock()
201			intf_a.port_number = 1
202			intf_b = MagicMock()
203			intf_b.port_number = 2
204			self.napp.lldp_loop_handler(switch, intf_a, intf_b, action="log")
205			mock_log.warning.assert_called()
206
207			@patch('napps.kytos.of_lldp.main.PO13')
208			@patch('napps.kytos.of_lldp.main.PO10')
209			@patch('napps.kytos.of_lldp.main.AO13')
210			@patch('napps.kytos.of_lldp.main.AO10')
211			def test_build_lldp_packet_out(self, *args):
212			"""Test _build_lldp_packet_out method."""
213			(mock_ao10, mock_ao13, mock_po10, mock_po13) = args
214
215			ao10 = MagicMock()
216			ao13 = MagicMock()
217			po10 = MagicMock()
218			po10.actions = []
219			po13 = MagicMock()
220			po13.actions = []
221
222			mock_ao10.return_value = ao10
223			mock_ao13.return_value = ao13
224			mock_po10.return_value = po10
225			mock_po13.return_value = po13
226
227			packet_out10 = self.napp._build_lldp_packet_out(0x01, 1, 'data1')
228			packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
229			packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')
230
231			self.assertEqual(packet_out10.data, 'data1')
232			self.assertEqual(packet_out10.actions, [ao10])
233			self.assertEqual(packet_out10.actions[0].port, 1)
234
235			self.assertEqual(packet_out13.data, 'data2')
236			self.assertEqual(packet_out13.actions, [ao13])
237			self.assertEqual(packet_out13.actions[0].port, 2)
238
239			self.assertIsNone(packet_out14)
240
241			@patch('napps.kytos.of_lldp.main.settings')
242			@patch('napps.kytos.of_lldp.main.EtherType')
243			@patch('napps.kytos.of_lldp.main.Port13')
244			@patch('napps.kytos.of_lldp.main.Port10')
245			def test_build_lldp_flow(self, *args):
246			"""Test _build_lldp_flow method."""
247			(mock_v0x01_port, mock_v0x04_port, mock_ethertype,
248			mock_settings) = args
249			self.napp.vlan_id = None
250			mock_v0x01_port.OFPP_CONTROLLER = 123
251			mock_v0x04_port.OFPP_CONTROLLER = 1234
252
253			mock_ethertype.LLDP = 10
254			mock_settings.FLOW_VLAN_VID = None
255			mock_settings.FLOW_PRIORITY = 1500
256			mock_settings.TABLE_ID = 0
257
258			flow = {}
259			match = {}
260			flow['priority'] = 1500
261			flow['table_id'] = 0
262			match['dl_type'] = 10
263
264			flow['match'] = match
265			expected_flow_v0x01 = flow.copy()
266			expected_flow_v0x04 = flow.copy()
267
268			expected_flow_v0x01['actions'] = [{'action_type': 'output',
269			'port': 123}]
270
271			expected_flow_v0x04['actions'] = [{'action_type': 'output',
272			'port': 1234}]
273
274			flow_mod10 = self.napp._build_lldp_flow(0x01)
275			flow_mod13 = self.napp._build_lldp_flow(0x04)
276
277			self.assertDictEqual(flow_mod10, expected_flow_v0x01)
278			self.assertDictEqual(flow_mod13, expected_flow_v0x04)
279
280			def test_unpack_non_empty(self):
281			"""Test _unpack_non_empty method."""
282			desired_class = MagicMock()
283			data = MagicMock()
284			data.value = 'data'
285
286			obj = self.napp._unpack_non_empty(desired_class, data)
287
288			obj.unpack.assert_called_with('data')
289
290			def test_get_data(self):
291			"""Test _get_data method."""
292			req = MagicMock()
293			interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
294			req.get_json.return_value = {'interfaces': interfaces}
295
296			data = self.napp._get_data(req)
297
298			self.assertEqual(data, interfaces)
299
300			def test_get_interfaces(self):
301			"""Test _get_interfaces method."""
302			expected_interfaces = self.get_topology_interfaces()
303
304			interfaces = self.napp._get_interfaces()
305
306			self.assertEqual(interfaces, expected_interfaces)
307
308			def test_get_interfaces_dict(self):
309			"""Test _get_interfaces_dict method."""
310			interfaces = self.napp._get_interfaces()
311			expected_interfaces = {inter.id: inter for inter in interfaces}
312
313			interfaces_dict = self.napp._get_interfaces_dict(interfaces)
314
315			self.assertEqual(interfaces_dict, expected_interfaces)
316
317			def test_get_lldp_interfaces(self):
318			"""Test _get_lldp_interfaces method."""
319			lldp_interfaces = self.napp._get_lldp_interfaces()
320
321			expected_interfaces = ['00:00:00:00:00:00:00:01:1',
322			'00:00:00:00:00:00:00:01:2',
323			'00:00:00:00:00:00:00:02:1',
324			'00:00:00:00:00:00:00:02:2',
325			'00:00:00:00:00:00:00:03:1',
326			'00:00:00:00:00:00:00:03:2']
327
328			self.assertEqual(lldp_interfaces, expected_interfaces)
329
330			def test_rest_get_lldp_interfaces(self):
331			"""Test get_lldp_interfaces method."""
332			api = get_test_client(self.napp.controller, self.napp)
333			url = f'{self.server_name_url}/v1/interfaces'
334			response = api.open(url, method='GET')
335
336			expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
337			'00:00:00:00:00:00:00:01:2',
338			'00:00:00:00:00:00:00:02:1',
339			'00:00:00:00:00:00:00:02:2',
340			'00:00:00:00:00:00:00:03:1',
341			'00:00:00:00:00:00:00:03:2']}
342			self.assertEqual(response.json, expected_data)
343			self.assertEqual(response.status_code, 200)
344
345			def test_enable_disable_lldp_200(self):
346			"""Test 200 response for enable_lldp and disable_lldp methods."""
347			data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
348			'00:00:00:00:00:00:00:01:2',
349			'00:00:00:00:00:00:00:02:1',
350			'00:00:00:00:00:00:00:02:2',
351			'00:00:00:00:00:00:00:03:1',
352			'00:00:00:00:00:00:00:03:2']}
353
354			api = get_test_client(self.napp.controller, self.napp)
355
356			url = f'{self.server_name_url}/v1/interfaces/disable'
357			disable_response = api.open(url, method='POST', json=data)
358
359			url = f'{self.server_name_url}/v1/interfaces/enable'
360			enable_response = api.open(url, method='POST', json=data)
361
362			self.assertEqual(disable_response.status_code, 200)
363			self.assertEqual(enable_response.status_code, 200)
364
365			def test_enable_disable_lldp_404(self):
366			"""Test 404 response for enable_lldp and disable_lldp methods."""
367			data = {"interfaces": []}
368
369			self.napp.controller.switches = {}
370			api = get_test_client(self.napp.controller, self.napp)
371
372			url = f'{self.server_name_url}/v1/interfaces/disable'
373			disable_response = api.open(url, method='POST', json=data)
374
375			url = f'{self.server_name_url}/v1/interfaces/enable'
376			enable_response = api.open(url, method='POST', json=data)
377
378			self.assertEqual(disable_response.status_code, 404)
379			self.assertEqual(enable_response.status_code, 404)
380
381			def test_enable_disable_lldp_400(self):
382			"""Test 400 response for enable_lldp and disable_lldp methods."""
383			data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
384			'00:00:00:00:00:00:00:01:2',
385			'00:00:00:00:00:00:00:02:1',
386			'00:00:00:00:00:00:00:02:2',
387			'00:00:00:00:00:00:00:03:1',
388			'00:00:00:00:00:00:00:03:2',
389			'00:00:00:00:00:00:00:04:1']}
390
391			api = get_test_client(self.napp.controller, self.napp)
392
393			url = f'{self.server_name_url}/v1/interfaces/disable'
394			disable_response = api.open(url, method='POST', json=data)
395
396			url = f'{self.server_name_url}/v1/interfaces/enable'
397			enable_response = api.open(url, method='POST', json=data)
398
399			self.assertEqual(disable_response.status_code, 400)
400			self.assertEqual(enable_response.status_code, 400)
401
402			def test_get_time(self):
403			"""Test get polling time."""
404			api = get_test_client(self.napp.controller, self.napp)
405
406			url = f'{self.server_name_url}/v1/polling_time'
407			response = api.open(url, method='GET')
408
409			self.assertEqual(response.status_code, 200)
410
411			def test_set_time(self):
412			"""Test update polling time."""
413			data = {"polling_time": 5}
414
415			api = get_test_client(self.napp.controller, self.napp)
416
417			url = f'{self.server_name_url}/v1/polling_time'
418			response = api.open(url, method='POST', json=data)
419
420			self.assertEqual(response.status_code, 200)
421			self.assertEqual(self.napp.polling_time, data['polling_time'])
422
423			def test_set_time_400(self):
424			"""Test fail case the update polling time."""
425			api = get_test_client(self.napp.controller, self.napp)
426
427			url = f'{self.server_name_url}/v1/polling_time'
428
429			data = {'polling_time': 'A'}
430			response = api.open(url, method='POST', json=data)
431			self.assertEqual(response.status_code, 400)
432

kytos-ng / of_lldp

Pull Request — master (#36)

TestMain.test_rest_get_lldp_interfaces() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like