build.tests.unit.test_main.TestMain.test_build_lldp_flow_mod() - Code Metrics - Inspection of "Create unit tests" - kytos/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#42)

by Gleyberson

created 2020-06-02 18:54 UTC

TestMain.test_build_lldp_flow_mod() B

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	55
Code Lines	45

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	45
nop	2
dl	0
loc	55
rs	8.8
c	0
b	0
f	0

How to fix Long Method

"""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


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()
        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_lldp, mock_dpid, mock_ethernet, mock_vlan, 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('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_lldp.main.KytosEvent')
    @patch('napps.kytos.of_lldp.main.Main._build_lldp_flow_mod')
    def test_install_lldp_flow(self, *args):
        """Test install_lldp_flow method."""
        (mock_build_lldp_packet_out, mock_kytos_event, mock_buffer_put) = args

        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        event = get_kytos_event_mock(name='kytos/of_core.handshake.completed',
                                     content={'switch': switch})

        mock_build_lldp_packet_out.side_effect = ['flow_mod', None]
        mock_kytos_event.return_value = 'ofpt_flow_mod'

        self.napp.install_lldp_flow(event)

        switch.connection = None

        self.napp.install_lldp_flow(event)

        mock_buffer_put.assert_called_once_with('ofpt_flow_mod')

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

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

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

    @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.InstructionApplyAction')
    @patch('napps.kytos.of_lldp.main.OxmTLV')
    @patch('napps.kytos.of_lldp.main.FM13')
    @patch('napps.kytos.of_lldp.main.FM10')
    @patch('napps.kytos.of_lldp.main.AO13')
    @patch('napps.kytos.of_lldp.main.AO10')
    def test_build_lldp_flow_mod(self, *args):
        """Test _build_lldp_flow_mod method."""
        (mock_ao10, mock_ao13, mock_fm10, mock_fm13, mock_oxm_tlv,
         mock_instruction) = args

        ao10 = MagicMock()
        fm10 = MagicMock()
        fm10.actions = []
        ao13 = MagicMock()
        fm13 = MagicMock()
        fm13.match.oxm_match_fields = []
        fm13.instructions = []
        instruction = MagicMock()
        instruction.actions = []

        match_lldp = MagicMock()
        match_lldp.oxm_field = 5
        match_lldp.oxm_value = 'ether_type'

        match_vlan = MagicMock()
        match_vlan.oxm_field = 6
        match_vlan.oxm_value = 'vlan_value'

        oxm_tlvs = [match_lldp, match_vlan]

        mock_ao10.return_value = ao10
        mock_fm10.return_value = fm10
        mock_ao13.return_value = ao13
        mock_fm13.return_value = fm13
        mock_instruction.return_value = instruction
        mock_oxm_tlv.side_effect = oxm_tlvs

        flow_mod10 = self.napp._build_lldp_flow_mod(0x01)
        flow_mod13 = self.napp._build_lldp_flow_mod(0x04)
        flow_mod14 = self.napp._build_lldp_flow_mod(0x05)

        self.assertEqual(flow_mod10.command, 0)
        self.assertEqual(flow_mod10.priority, 1000)
        self.assertEqual(flow_mod10.match.dl_type, 0x88CC)
        self.assertEqual(flow_mod10.match.dl_vlan, 3799)
        self.assertEqual(flow_mod10.actions, [ao10])

        self.assertEqual(flow_mod13.command, 0)
        self.assertEqual(flow_mod13.priority, 1000)
        self.assertEqual(flow_mod13.match.oxm_match_fields, oxm_tlvs)
        self.assertEqual(flow_mod13.instructions[0], instruction)
        self.assertEqual(flow_mod13.instructions[0].actions, [ao13])

        self.assertIsNone(flow_mod14)

    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, *args):
        """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, *args):
        """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, *args):
        """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)


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			from tests.helpers import get_topology_mock
8
9
10			class TestMain(TestCase):
11			"""Tests for the Main class."""
12
13			def setUp(self):
14			"""Execute steps before each tests."""
15			self.server_name_url = 'http://127.0.0.1:8181/api/kytos/of_lldp'
16
17			patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
18			from napps.kytos.of_lldp.main import Main
19			self.addCleanup(patch.stopall)
20
21			self.topology = get_topology_mock()
22			controller = get_controller_mock()
23			controller.switches = self.topology.switches
24
25			self.napp = Main(controller)
26
27			def get_topology_interfaces(self):
28			"""Return interfaces present in topology."""
29			interfaces = []
30			for switch in list(self.topology.switches.values()):
31			interfaces += list(switch.interfaces.values())
32			return interfaces
33
34			@patch('kytos.core.buffers.KytosEventBuffer.put')
35			@patch('napps.kytos.of_lldp.main.Main._build_lldp_packet_out')
36			@patch('napps.kytos.of_lldp.main.KytosEvent')
37			@patch('napps.kytos.of_lldp.main.VLAN')
38			@patch('napps.kytos.of_lldp.main.Ethernet')
39			@patch('napps.kytos.of_lldp.main.DPID')
40			@patch('napps.kytos.of_lldp.main.LLDP')
41			def test_execute(self, *args):
42			"""Test execute method."""
43			(mock_lldp, mock_dpid, mock_ethernet, mock_vlan, mock_kytos_event,
44			mock_build_lldp_packet_out, mock_buffer_put) = args
45
46			ethernet = MagicMock()
47			ethernet.pack.return_value = 'pack'
48			interfaces = self.get_topology_interfaces()
49			po_args = [(interface.switch.connection.protocol.version,
50			interface.port_number, 'pack') for interface in interfaces]
51
52			mock_ethernet.return_value = ethernet
53			mock_kytos_event.side_effect = po_args
54
55			self.napp.execute()
56
57			mock_build_lldp_packet_out.assert_has_calls([call(*(arg))
58			for arg in po_args])
59			mock_buffer_put.assert_has_calls([call(arg)
60			for arg in po_args])
61
62			@patch('kytos.core.buffers.KytosEventBuffer.put')
63			@patch('napps.kytos.of_lldp.main.KytosEvent')
64			@patch('napps.kytos.of_lldp.main.Main._build_lldp_flow_mod')
65			def test_install_lldp_flow(self, *args):
66			"""Test install_lldp_flow method."""
67			(mock_build_lldp_packet_out, mock_kytos_event, mock_buffer_put) = args
68
69			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
70			event = get_kytos_event_mock(name='kytos/of_core.handshake.completed',
71			content={'switch': switch})
72
73			mock_build_lldp_packet_out.side_effect = ['flow_mod', None]
74			mock_kytos_event.return_value = 'ofpt_flow_mod'
75
76			self.napp.install_lldp_flow(event)
77
78			switch.connection = None
79
80			self.napp.install_lldp_flow(event)
81
82			mock_buffer_put.assert_called_once_with('ofpt_flow_mod')
83
84			@patch('kytos.core.buffers.KytosEventBuffer.put')
85			@patch('napps.kytos.of_lldp.main.KytosEvent')
86			@patch('kytos.core.controller.Controller.get_switch_by_dpid')
87			@patch('napps.kytos.of_lldp.main.Main._unpack_non_empty')
88			@patch('napps.kytos.of_lldp.main.UBInt32')
89			@patch('napps.kytos.of_lldp.main.DPID')
90			@patch('napps.kytos.of_lldp.main.LLDP')
91			@patch('napps.kytos.of_lldp.main.Ethernet')
92			def test_notify_uplink_detected(self, *args):
93			"""Test notify_uplink_detected method."""
94			(mock_ethernet, mock_lldp, mock_dpid, mock_ubint32,
95			mock_unpack_non_empty, mock_get_switch_by_dpid, mock_kytos_event,
96			mock_buffer_put) = args
97
98			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
99			message = MagicMock()
100			message.in_port = 1
101			message.data = 'data'
102			event = get_kytos_event_mock(name='kytos/of_core.v0x0[14].messages.in.'
103			'ofpt_packet_in',
104			content={'source': switch.connection,
105			'message': message})
106
107			ethernet = MagicMock()
108			ethernet.ether_type = 0x88CC
109			ethernet.data = 'eth_data'
110			lldp = MagicMock()
111			lldp.chassis_id.sub_value = 'chassis_id'
112			lldp.port_id.sub_value = 'port_id'
113			dpid = MagicMock()
114			dpid.value = "00:00:00:00:00:00:00:02"
115			port_b = MagicMock()
116
117			mock_unpack_non_empty.side_effect = [ethernet, lldp, dpid, port_b]
118			mock_get_switch_by_dpid.return_value = get_switch_mock(dpid.value,
119			0x04)
120			mock_kytos_event.return_value = 'nni'
121
122			self.napp.notify_uplink_detected(event)
123
124			calls = [call(mock_ethernet, message.data),
125			call(mock_lldp, ethernet.data),
126			call(mock_dpid, lldp.chassis_id.sub_value),
127			call(mock_ubint32, lldp.port_id.sub_value)]
128			mock_unpack_non_empty.assert_has_calls(calls)
129			mock_buffer_put.assert_called_with('nni')
130
131			@patch('napps.kytos.of_lldp.main.PO13')
132			@patch('napps.kytos.of_lldp.main.PO10')
133			@patch('napps.kytos.of_lldp.main.AO13')
134			@patch('napps.kytos.of_lldp.main.AO10')
135			def test_build_lldp_packet_out(self, *args):
136			"""Test _build_lldp_packet_out method."""
137			(mock_ao10, mock_ao13, mock_po10, mock_po13) = args
138
139			ao10 = MagicMock()
140			ao13 = MagicMock()
141			po10 = MagicMock()
142			po10.actions = []
143			po13 = MagicMock()
144			po13.actions = []
145
146			mock_ao10.return_value = ao10
147			mock_ao13.return_value = ao13
148			mock_po10.return_value = po10
149			mock_po13.return_value = po13
150
151			packet_out10 = self.napp._build_lldp_packet_out(0x01, 1, 'data1')
152			packet_out13 = self.napp._build_lldp_packet_out(0x04, 2, 'data2')
153			packet_out14 = self.napp._build_lldp_packet_out(0x05, 3, 'data3')
154
155			self.assertEqual(packet_out10.data, 'data1')
156			self.assertEqual(packet_out10.actions, [ao10])
157			self.assertEqual(packet_out10.actions[0].port, 1)
158
159			self.assertEqual(packet_out13.data, 'data2')
160			self.assertEqual(packet_out13.actions, [ao13])
161			self.assertEqual(packet_out13.actions[0].port, 2)
162
163			self.assertIsNone(packet_out14)
164
165			@patch('napps.kytos.of_lldp.main.InstructionApplyAction')
166			@patch('napps.kytos.of_lldp.main.OxmTLV')
167			@patch('napps.kytos.of_lldp.main.FM13')
168			@patch('napps.kytos.of_lldp.main.FM10')
169			@patch('napps.kytos.of_lldp.main.AO13')
170			@patch('napps.kytos.of_lldp.main.AO10')
171			def test_build_lldp_flow_mod(self, *args):
172			"""Test _build_lldp_flow_mod method."""
173			(mock_ao10, mock_ao13, mock_fm10, mock_fm13, mock_oxm_tlv,
174			mock_instruction) = args
175
176			ao10 = MagicMock()
177			fm10 = MagicMock()
178			fm10.actions = []
179			ao13 = MagicMock()
180			fm13 = MagicMock()
181			fm13.match.oxm_match_fields = []
182			fm13.instructions = []
183			instruction = MagicMock()
184			instruction.actions = []
185
186			match_lldp = MagicMock()
187			match_lldp.oxm_field = 5
188			match_lldp.oxm_value = 'ether_type'
189
190			match_vlan = MagicMock()
191			match_vlan.oxm_field = 6
192			match_vlan.oxm_value = 'vlan_value'
193
194			oxm_tlvs = [match_lldp, match_vlan]
195
196			mock_ao10.return_value = ao10
197			mock_fm10.return_value = fm10
198			mock_ao13.return_value = ao13
199			mock_fm13.return_value = fm13
200			mock_instruction.return_value = instruction
201			mock_oxm_tlv.side_effect = oxm_tlvs
202
203			flow_mod10 = self.napp._build_lldp_flow_mod(0x01)
204			flow_mod13 = self.napp._build_lldp_flow_mod(0x04)
205			flow_mod14 = self.napp._build_lldp_flow_mod(0x05)
206
207			self.assertEqual(flow_mod10.command, 0)
208			self.assertEqual(flow_mod10.priority, 1000)
209			self.assertEqual(flow_mod10.match.dl_type, 0x88CC)
210			self.assertEqual(flow_mod10.match.dl_vlan, 3799)
211			self.assertEqual(flow_mod10.actions, [ao10])
212
213			self.assertEqual(flow_mod13.command, 0)
214			self.assertEqual(flow_mod13.priority, 1000)
215			self.assertEqual(flow_mod13.match.oxm_match_fields, oxm_tlvs)
216			self.assertEqual(flow_mod13.instructions[0], instruction)
217			self.assertEqual(flow_mod13.instructions[0].actions, [ao13])
218
219			self.assertIsNone(flow_mod14)
220
221			def test_unpack_non_empty(self):
222			"""Test _unpack_non_empty method."""
223			desired_class = MagicMock()
224			data = MagicMock()
225			data.value = 'data'
226
227			obj = self.napp._unpack_non_empty(desired_class, data)
228
229			obj.unpack.assert_called_with('data')
230
231			def test_get_data(self):
232			"""Test _get_data method."""
233			req = MagicMock()
234			interfaces = ['00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:01:2']
235			req.get_json.return_value = {'interfaces': interfaces}
236
237			data = self.napp._get_data(req)
238
239			self.assertEqual(data, interfaces)
240
241			def test_get_interfaces(self):
242			"""Test _get_interfaces method."""
243			expected_interfaces = self.get_topology_interfaces()
244
245			interfaces = self.napp._get_interfaces()
246
247			self.assertEqual(interfaces, expected_interfaces)
248
249			def test_get_interfaces_dict(self):
250			"""Test _get_interfaces_dict method."""
251			interfaces = self.napp._get_interfaces()
252			expected_interfaces = {inter.id: inter for inter in interfaces}
253
254			interfaces_dict = self.napp._get_interfaces_dict(interfaces)
255
256			self.assertEqual(interfaces_dict, expected_interfaces)
257
258			def test_get_lldp_interfaces(self):
259			"""Test _get_lldp_interfaces method."""
260			lldp_interfaces = self.napp._get_lldp_interfaces()
261
262			expected_interfaces = ['00:00:00:00:00:00:00:01:1',
263			'00:00:00:00:00:00:00:01:2',
264			'00:00:00:00:00:00:00:02:1',
265			'00:00:00:00:00:00:00:02:2',
266			'00:00:00:00:00:00:00:03:1',
267			'00:00:00:00:00:00:00:03:2']
268
269			self.assertEqual(lldp_interfaces, expected_interfaces)
270
271			def test_rest_get_lldp_interfaces(self):
272			"""Test get_lldp_interfaces method."""
273			api = get_test_client(self.napp.controller, self.napp)
274			url = f'{self.server_name_url}/v1/interfaces'
275			response = api.open(url, method='GET')
276
277			expected_data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
278			'00:00:00:00:00:00:00:01:2',
279			'00:00:00:00:00:00:00:02:1',
280			'00:00:00:00:00:00:00:02:2',
281			'00:00:00:00:00:00:00:03:1',
282			'00:00:00:00:00:00:00:03:2']}
283			self.assertEqual(response.json, expected_data)
284			self.assertEqual(response.status_code, 200)
285
286			def test_enable_disable_lldp_200(self, *args):
287			"""Test 200 response for enable_lldp and disable_lldp methods."""
288			data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
289			'00:00:00:00:00:00:00:01:2',
290			'00:00:00:00:00:00:00:02:1',
291			'00:00:00:00:00:00:00:02:2',
292			'00:00:00:00:00:00:00:03:1',
293			'00:00:00:00:00:00:00:03:2']}
294
295			api = get_test_client(self.napp.controller, self.napp)
296
297			url = f'{self.server_name_url}/v1/interfaces/disable'
298			disable_response = api.open(url, method='POST', json=data)
299
300			url = f'{self.server_name_url}/v1/interfaces/enable'
301			enable_response = api.open(url, method='POST', json=data)
302
303			self.assertEqual(disable_response.status_code, 200)
304			self.assertEqual(enable_response.status_code, 200)
305
306			def test_enable_disable_lldp_404(self, *args):
307			"""Test 404 response for enable_lldp and disable_lldp methods."""
308			data = {"interfaces": []}
309
310			self.napp.controller.switches = {}
311			api = get_test_client(self.napp.controller, self.napp)
312
313			url = f'{self.server_name_url}/v1/interfaces/disable'
314			disable_response = api.open(url, method='POST', json=data)
315
316			url = f'{self.server_name_url}/v1/interfaces/enable'
317			enable_response = api.open(url, method='POST', json=data)
318
319			self.assertEqual(disable_response.status_code, 404)
320			self.assertEqual(enable_response.status_code, 404)
321
322			def test_enable_disable_lldp_400(self, *args):
323			"""Test 400 response for enable_lldp and disable_lldp methods."""
324			data = {"interfaces": ['00:00:00:00:00:00:00:01:1',
325			'00:00:00:00:00:00:00:01:2',
326			'00:00:00:00:00:00:00:02:1',
327			'00:00:00:00:00:00:00:02:2',
328			'00:00:00:00:00:00:00:03:1',
329			'00:00:00:00:00:00:00:03:2',
330			'00:00:00:00:00:00:00:04:1']}
331
332			api = get_test_client(self.napp.controller, self.napp)
333
334			url = f'{self.server_name_url}/v1/interfaces/disable'
335			disable_response = api.open(url, method='POST', json=data)
336
337			url = f'{self.server_name_url}/v1/interfaces/enable'
338			enable_response = api.open(url, method='POST', json=data)
339
340			self.assertEqual(disable_response.status_code, 400)
341			self.assertEqual(enable_response.status_code, 400)
342

kytos / of_lldp

Pull Request — master (#42)

TestMain.test_build_lldp_flow_mod() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like