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

Passed

Pull Request — master (#42)

by Gleyberson

created 2020-04-23 04:12 UTC

test_main.TestMain.test_execute() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	34
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	27
nop	2
dl	0
loc	34
rs	9.232
c	0
b	0
f	0

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

from tests.helpers import (get_connection_mock, get_controller_mock,
                           get_kytos_event_mock, get_switch_mock,
                           get_test_client, get_topology_mock)


class TestMain(TestCase):
    """Tests for the Main class."""

    def setUp(self):
        """Execute steps before each tests."""
        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        from napps.kytos.of_lldp.main import Main
        self.addCleanup(patch.stopall)

        self.napp = Main(get_controller_mock())

    @staticmethod
    def get_topology_interfaces(topology):
        """Return interfaces present in topology."""
        interfaces = []
        for switch in list(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

        topology = get_topology_mock()
        interfaces = self.get_topology_interfaces(topology)
        self.napp.controller.switches = topology.switches

        ethernet = MagicMock()
        ethernet.pack.return_value = 'pack'

        po_args = [(interface.switch.connection.protocol.version,
                    interface.port_number, 'pack') for interface in interfaces]

        mock_lldp.return_value = MagicMock()
        mock_dpid.return_value = MagicMock()
        mock_ethernet.return_value = ethernet
        mock_vlan.return_value = MagicMock()
        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 _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()
        req.get_json.return_value = {'interfaces': ['i1', 'i2']}

        data = self.napp._get_data(req)

        self.assertEqual(data, ['i1', 'i2'])

    def test_get_interfaces(self):
        """Test _get_interfaces method."""
        topology = get_topology_mock()
        self.napp.controller.switches = topology.switches
        expected_interfaces = self.get_topology_interfaces(topology)

        interfaces = self.napp._get_interfaces()

        self.assertEqual(interfaces, expected_interfaces)

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

        interfaces_dict = self.napp._get_interfaces_dict(interfaces)

        self.assertEqual(interfaces_dict, expected_interfaces)

    @patch('napps.kytos.of_lldp.main.Main._get_interfaces')
    def test_get_lldp_interfaces(self, mock_get_interfaces):
        """Test _get_lldp_interfaces method."""
        topology = get_topology_mock()
        interfaces = self.get_topology_interfaces(topology)
        interfaces[0].lldp = False
        expected_interfaces = [inter.id for inter in interfaces if inter.lldp]

        mock_get_interfaces.return_value = interfaces

        lldp_interfaces = self.napp._get_lldp_interfaces()

        self.assertEqual(len(lldp_interfaces), len(expected_interfaces))
        self.assertEqual(lldp_interfaces, expected_interfaces)

    @patch('napps.kytos.of_lldp.main.Main._get_lldp_interfaces')
    def test_rest_get_lldp_interfaces(self, mock_get_lldp_interfaces):
        """Test get_lldp_interfaces method."""
        topology = get_topology_mock()
        interfaces = self.get_topology_interfaces(topology)
        interfaces_ids = [inter.id for inter in interfaces]

        mock_get_lldp_interfaces.return_value = interfaces_ids

        api = get_test_client(self.napp.controller, self.napp)
        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces"
        response = api.open(url, method='GET')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json, {'interfaces': interfaces_ids})

    @patch('napps.kytos.of_lldp.main.Main._get_interfaces_dict')

    @patch('napps.kytos.of_lldp.main.Main._get_interfaces')
    @patch('napps.kytos.of_lldp.main.Main._get_data')
    def test_enable_disable_lldp_200(self, *args):
        """Test 200 response for enable_lldp and disable_lldp methods."""
        (mock_get_data, mock_get_interfaces, mock_get_interfaces_dict) = args

        topology = get_topology_mock()
        interfaces = self.get_topology_interfaces(topology)
        interfaces_dict = {inter.id: inter for inter in interfaces}
        interfaces_ids = [inter.id for inter in interfaces]

        mock_get_data.return_value = interfaces_ids
        mock_get_interfaces.return_value = interfaces
        mock_get_interfaces_dict.return_value = interfaces_dict

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

        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces/disable"
        disable_response = api.open(url, method='POST')

        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces/enable"
        enable_response = api.open(url, method='POST')

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

    @patch('napps.kytos.of_lldp.main.Main._get_interfaces')
    @patch('napps.kytos.of_lldp.main.Main._get_data')
    def test_enable_disable_lldp_404(self, *args):
        """Test 404 response for enable_lldp and disable_lldp methods."""
        (mock_get_data, mock_get_interfaces) = args

        mock_get_data.return_value = []
        mock_get_interfaces.return_value = None

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

        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces/disable"
        disable_response = api.open(url, method='POST')

        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces/enable"
        enable_response = api.open(url, method='POST')

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

    @patch('napps.kytos.of_lldp.main.Main._get_interfaces_dict')

    @patch('napps.kytos.of_lldp.main.Main._get_interfaces')
    @patch('napps.kytos.of_lldp.main.Main._get_data')
    def test_enable_disable_lldp_400(self, *args):
        """Test 400 response for enable_lldp and disable_lldp methods."""
        (mock_get_data, mock_get_interfaces, mock_get_interfaces_dict) = args

        topology = get_topology_mock()
        interfaces = self.get_topology_interfaces(topology)
        interfaces_dict = {inter.id: inter for inter in interfaces}
        interfaces_ids = [inter.id for inter in interfaces]
        interfaces_ids.append("00:00:00:00:00:00:00:01:3")

        mock_get_data.return_value = interfaces_ids
        mock_get_interfaces.return_value = interfaces
        mock_get_interfaces_dict.return_value = interfaces_dict

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

        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces/disable"
        disable_response = api.open(url, method='POST')

        url = "http://127.0.0.1:8181/api/kytos/of_lldp/v1/interfaces/enable"
        enable_response = api.open(url, method='POST')

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

kytos / of_lldp

Pull Request — master (#42)

test_main.TestMain.test_execute() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like