build.tests.integration.test_main.TestMain.test_get_interfaces() - Code Metrics - Inspection of "Migrated from storehouse to MongoDB" - kytos-ng/topology - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#84)

by Vinicius

created 2022-04-08 20:50 UTC

TestMain.test_get_interfaces() A

↳ Parent: build.tests.integration.test_main

Complexity

Conditions

Size

Total Lines	18
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	17
nop	1
dl	0
loc	18
rs	9.55
c	0
b	0
f	0

"""Module to test the main napp file."""
from unittest import TestCase
from unittest.mock import Mock, patch, MagicMock

# pylint: disable=import-error,no-name-in-module
from kytos.core.events import KytosEvent
from tests.integration.helpers import (get_controller_mock, get_interface_mock,
                                       get_switch_mock)

LINK_DATA = {
    "active": False,
    "enabled": True,
    "endpoint_a": {
        "id": "00:00:00:00:00:00:00:01:1",
        "link": "26927949-df3c-4c25-874b-3da30d8ae983",
        "mac": "26:fb:42:20:b8:b1",
        "name": "s1-eth1",
        "nni": False,
        "port_number": 1,
        "speed": "10 Gbps",
        "switch": "00:00:00:00:00:00:00:01",
        "type": "interface",
        "uni": True
    },
    "endpoint_b": {
        "id": "00:00:00:00:00:00:00:01:1",
        "link": "26927949-df3c-4c25-874b-3da30d8ae983",
        "mac": "26:fb:42:20:b8:b1",
        "name": "s1-eth1",
        "nni": False,
        "port_number": 1,
        "speed": "10 Gbps",
        "switch": "00:00:00:00:00:00:00:01",
        "type": "interface",
        "uni": True
    }
}

SWITCH_DATA = {
    "id": "00:00:00:00:00:00:00:01",
    "name": "my-beautiful-switch",
    "serial": "string",
    "software": "Version 2.3.4",
    "ofp_version": "0x01",
    "connection": "127.0.0.1:49330",
    "data_path": "string",
    "manufacturer": "Unkown Manufactor",
    "hardware": "Hardware version 2.0",
    "type": "switch",
    "active": True,
    "enabled": False,
    "dpid": "00:00:00:00:00:00:00:01",
    "metadata": {},
    "interfaces": {
        "additionalProp1": {
            "id": "00:00:00:00:00:00:00:01:1",
            "link": "26927949-df3c-4c25-874b-3da30d8ae983",
            "mac": "26:fb:42:20:b8:b1",
            "name": "s1-eth1",
            "nni": False,
            "port_number": 1,
            "speed": "10 Gbps",
            "switch": "00:00:00:00:00:00:00:01",
            "type": "interface",
            "uni": True
        },
        "additionalProp2": {
            "id": "00:00:00:00:00:00:00:01:1",
            "link": "26927949-df3c-4c25-874b-3da30d8ae983",
            "mac": "26:fb:42:20:b8:b1",
            "name": "s1-eth1",
            "nni": False,
            "port_number": 1,
            "speed": "10 Gbps",
            "switch": "00:00:00:00:00:00:00:01",
            "type": "interface",
            "uni": True
        },
        "additionalProp3": {
            "id": "00:00:00:00:00:00:00:01:1",
            "link": "26927949-df3c-4c25-874b-3da30d8ae983",
            "mac": "26:fb:42:20:b8:b1",
            "name": "s1-eth1",
            "nni": False,
            "port_number": 1,
            "speed": "10 Gbps",
            "switch": "00:00:00:00:00:00:00:01",
            "type": "interface",
            "uni": True
        }
    }
}


class FakeBox:
    """Simulate a Storehouse Box."""

    def __init__(self, data):
        """Initizalize default values to FakeBox."""
        self.data = data
        self.namespace = None
        self.name = None
        self.box_id = None
        self.created_at = None
        self.owner = None


# pylint: disable=import-outside-toplevel
class TestMain(TestCase):
    """Test the Main class."""

    def setUp(self):
        """Execute steps before each tests.

        Set the server_name_url from kytos/topology
        """
        self.server_name_url = 'http://localhost:8181/api/kytos/topology'

        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        from napps.kytos.topology.main import Main
        Main.ensure_db_or_core_shutdown = MagicMock(return_value=False)
        self.addCleanup(patch.stopall)
        self.napp = Main(get_controller_mock())
        self.napp.topo_controller = MagicMock()

    def test_get_switches_dict(self):
        """Basic test for switch listing."""
        # pylint: disable=protected-access,
        # pylint: disable=use-implicit-booleaness-not-comparison
        switches = self.napp._get_switches_dict()
        assert isinstance(switches['switches'], dict)
        assert switches['switches'] == {}

    def test_get_event_listeners(self):
        """Verify all event listeners registered."""
        actual_events = self.napp.listeners()
        expected_events = ['kytos/core.shutdown',
                           'kytos/core.shutdown.kytos/topology',
                           'kytos/maintenance.start_link',
                           'kytos/maintenance.end_link',
                           'kytos/maintenance.start_switch',
                           'kytos/maintenance.end_switch',
                           'kytos/.*.link_available_tags',
                           '.*.of_lldp.network_status.updated',
                           '.*.interface.is.nni',
                           '.*.connection.lost',
                           '.*.switch.interfaces.created',
                           '.*.topology.switch.interface.created',
                           '.*.switch.interface.deleted',
                           '.*.switch.interface.link_down',
                           '.*.switch.interface.link_up',
                           '.*.switch.(new|reconnected)',
                           '.*.switch.port.created']
        self.assertCountEqual(expected_events, actual_events)

    def test_verify_api_urls(self):

        """Verify all APIs registered."""
        expected_urls = [
         ({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/interfaces'),
         ({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/switches'),
         ({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/links'),
         ({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/'),
         ({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/switch/<dpid>/disable'),
         ({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/switch/<dpid>/enable'),
         ({'key': '[key]', 'interface_id': '[interface_id]'},
          {'OPTIONS', 'DELETE'},
          '/api/kytos/topology/v3/interfaces/<interface_id>/metadata/<key>'),
         ({'interface_id': '[interface_id]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
         ({'interface_id': '[interface_id]'}, {'GET', 'OPTIONS', 'HEAD'},
          '/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
         ({'interface_disable_id': '[interface_disable_id]'},
          {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/<interface_disable_id>/disable'),
         ({'interface_enable_id': '[interface_enable_id]'},
          {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/<interface_enable_id>/enable'),
         ({'dpid': '[dpid]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
          '/api/kytos/topology/v3/switches/<dpid>/metadata/<key>'),
         ({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/switches/<dpid>/metadata'),
         ({'dpid': '[dpid]'}, {'GET', 'OPTIONS', 'HEAD'},
          '/api/kytos/topology/v3/switches/<dpid>/metadata'),
         ({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/switches/<dpid>/disable'),
         ({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/switches/<dpid>/enable'),
         ({'link_id': '[link_id]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
          '/api/kytos/topology/v3/links/<link_id>/metadata/<key>'),
         ({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/links/<link_id>/metadata'),
         ({'link_id': '[link_id]'}, {'GET', 'OPTIONS', 'HEAD'},
          '/api/kytos/topology/v3/links/<link_id>/metadata'),
         ({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/links/<link_id>/disable'),
         ({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/links/<link_id>/enable')]

        urls = self.get_napp_urls(self.napp)
        self.assertEqual(expected_urls, urls)

    @staticmethod

    def get_napp_urls(napp):
        """Return the kytos/topology urls.

        The urls will be like:

        urls = [
            (options, methods, url)
        ]

        """
        controller = napp.controller
        controller.api_server.register_napp_endpoints(napp)

        urls = []
        for rule in controller.api_server.app.url_map.iter_rules():
            options = {}
            for arg in rule.arguments:
                options[arg] = f"[{arg}]"

            if f'{napp.username}/{napp.name}' in str(rule):
                urls.append((options, rule.methods, f'{str(rule)}'))

        return urls

    @staticmethod
    def get_app_test_client(napp):
        """Return a flask api test client."""
        napp.controller.api_server.register_napp_endpoints(napp)
        return napp.controller.api_server.app.test_client()

    def test_get_interfaces(self):
        """test_get_interfaces."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(0x04, dpid=dpid)
        switch.dpid = dpid
        switch.metadata = {"lat": 0, "lng": 0}
        switch.interfaces = {f"{dpid}:1": f"{dpid}:1"}
        switch.as_dict = lambda: {"dpid": dpid, "metadata": switch.metadata,
                                  "interfaces": switch.interfaces}
        self.napp.controller.switches = {dpid: switch}
        client = self.get_app_test_client(self.napp)
        url = "http://127.0.0.1:8181/api/kytos/topology/v3/interfaces"
        response = client.get(url)
        assert response.status_code == 200
        data = response.json
        assert "interfaces" in data
        assert len(data["interfaces"]) == 1
        assert data["interfaces"][f"{dpid}:1"]

    def test_handle_new_switch(self):
        """Test handle new switch."""
        event_name = '.*.switch.(new|reconnected)'
        switch = get_switch_mock(0x04)
        event = KytosEvent(name=event_name,
                           content={'switch': switch})
        self.napp.handle_new_switch(event)
        event_response = self.napp.controller.buffers.app.get()
        self.assertEqual(event_response.name,
                         'kytos/topology.updated')

    def test_handle_interface_deleted(self):
        """Test handle interface deleted."""
        event_name = '.*.switch.interface.deleted'
        interface = get_interface_mock("interface1", 7)
        interface.switch.dpid = "00:00:00:00:00:00:00:01"
        stats_event = KytosEvent(name=event_name,
                                 content={'interface': interface})
        self.napp.handle_interface_deleted(stats_event)
        event_updated_response = self.napp.controller.buffers.app.get()
        self.assertEqual(event_updated_response.name,
                         'kytos/topology.updated')

    def test_handle_connection_lost(self):
        """Test handle connection lost."""
        event_name = '.*.connection.lost'
        source = Mock()
        stats_event = KytosEvent(name=event_name,
                                 content={'source': source})
        self.napp.handle_connection_lost(stats_event)
        event_updated_response = self.napp.controller.buffers.app.get()
        self.assertEqual(event_updated_response.name,
                         'kytos/topology.updated')


1		"""Module to test the main napp file."""
2		from unittest import TestCase
3		from unittest.mock import Mock, patch, MagicMock
4
5		# pylint: disable=import-error,no-name-in-module
6		from kytos.core.events import KytosEvent
7		from tests.integration.helpers import (get_controller_mock, get_interface_mock,
8		get_switch_mock)
9
10		LINK_DATA = {
11		"active": False,
12		"enabled": True,
13		"endpoint_a": {
14		"id": "00:00:00:00:00:00:00:01:1",
15		"link": "26927949-df3c-4c25-874b-3da30d8ae983",
16		"mac": "26:fb:42:20:b8:b1",
17		"name": "s1-eth1",
18		"nni": False,
19		"port_number": 1,
20		"speed": "10 Gbps",
21		"switch": "00:00:00:00:00:00:00:01",
22		"type": "interface",
23		"uni": True
24		},
25		"endpoint_b": {
26		"id": "00:00:00:00:00:00:00:01:1",
27		"link": "26927949-df3c-4c25-874b-3da30d8ae983",
28		"mac": "26:fb:42:20:b8:b1",
29		"name": "s1-eth1",
30		"nni": False,
31		"port_number": 1,
32		"speed": "10 Gbps",
33		"switch": "00:00:00:00:00:00:00:01",
34		"type": "interface",
35		"uni": True
36		}
37		}
38
39		SWITCH_DATA = {
40		"id": "00:00:00:00:00:00:00:01",
41		"name": "my-beautiful-switch",
42		"serial": "string",
43		"software": "Version 2.3.4",
44		"ofp_version": "0x01",
45		"connection": "127.0.0.1:49330",
46		"data_path": "string",
47		"manufacturer": "Unkown Manufactor",
48		"hardware": "Hardware version 2.0",
49		"type": "switch",
50		"active": True,
51		"enabled": False,
52		"dpid": "00:00:00:00:00:00:00:01",
53		"metadata": {},
54		"interfaces": {
55		"additionalProp1": {
56		"id": "00:00:00:00:00:00:00:01:1",
57		"link": "26927949-df3c-4c25-874b-3da30d8ae983",
58		"mac": "26:fb:42:20:b8:b1",
59		"name": "s1-eth1",
60		"nni": False,
61		"port_number": 1,
62		"speed": "10 Gbps",
63		"switch": "00:00:00:00:00:00:00:01",
64		"type": "interface",
65		"uni": True
66		},
67		"additionalProp2": {
68		"id": "00:00:00:00:00:00:00:01:1",
69		"link": "26927949-df3c-4c25-874b-3da30d8ae983",
70		"mac": "26:fb:42:20:b8:b1",
71		"name": "s1-eth1",
72		"nni": False,
73		"port_number": 1,
74		"speed": "10 Gbps",
75		"switch": "00:00:00:00:00:00:00:01",
76		"type": "interface",
77		"uni": True
78		},
79		"additionalProp3": {
80		"id": "00:00:00:00:00:00:00:01:1",
81		"link": "26927949-df3c-4c25-874b-3da30d8ae983",
82		"mac": "26:fb:42:20:b8:b1",
83		"name": "s1-eth1",
84		"nni": False,
85		"port_number": 1,
86		"speed": "10 Gbps",
87		"switch": "00:00:00:00:00:00:00:01",
88		"type": "interface",
89		"uni": True
90		}
91		}
92		}
93
94
95		class FakeBox:
96		"""Simulate a Storehouse Box."""
97
98		def __init__(self, data):
99		"""Initizalize default values to FakeBox."""
100		self.data = data
101		self.namespace = None
102		self.name = None
103		self.box_id = None
104		self.created_at = None
105		self.owner = None
106
107
108		# pylint: disable=import-outside-toplevel
109		class TestMain(TestCase):
110		"""Test the Main class."""
111
112		def setUp(self):
113		"""Execute steps before each tests.
114
115		Set the server_name_url from kytos/topology
116		"""
117		self.server_name_url = 'http://localhost:8181/api/kytos/topology'
118
119		patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
120		from napps.kytos.topology.main import Main
121		Main.ensure_db_or_core_shutdown = MagicMock(return_value=False)
122		self.addCleanup(patch.stopall)
123		self.napp = Main(get_controller_mock())
124		self.napp.topo_controller = MagicMock()
125
126		def test_get_switches_dict(self):
127		"""Basic test for switch listing."""
128		# pylint: disable=protected-access,
129		# pylint: disable=use-implicit-booleaness-not-comparison
130		switches = self.napp._get_switches_dict()
131		assert isinstance(switches['switches'], dict)
132		assert switches['switches'] == {}
133
134		def test_get_event_listeners(self):
135		"""Verify all event listeners registered."""
136		actual_events = self.napp.listeners()
137		expected_events = ['kytos/core.shutdown',
138		'kytos/core.shutdown.kytos/topology',
139		'kytos/maintenance.start_link',
140		'kytos/maintenance.end_link',
141		'kytos/maintenance.start_switch',
142		'kytos/maintenance.end_switch',
143		'kytos/.*.link_available_tags',
144		'.*.of_lldp.network_status.updated',
145		'.*.interface.is.nni',
146		'.*.connection.lost',
147		'.*.switch.interfaces.created',
148		'.*.topology.switch.interface.created',
149		'.*.switch.interface.deleted',
150		'.*.switch.interface.link_down',
151		'.*.switch.interface.link_up',
152		'.*.switch.(new\|reconnected)',
153		'.*.switch.port.created']
154		self.assertCountEqual(expected_events, actual_events)
155
156	View Code Duplication	def test_verify_api_urls(self):
		0 ignored issues – show Duplication introduced 2021-11-30 20:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
157		"""Verify all APIs registered."""
158		expected_urls = [
159		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/interfaces'),
160		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/switches'),
161		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/links'),
162		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/'),
163		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
164		'/api/kytos/topology/v3/interfaces/switch/<dpid>/disable'),
165		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
166		'/api/kytos/topology/v3/interfaces/switch/<dpid>/enable'),
167		({'key': '[key]', 'interface_id': '[interface_id]'},
168		{'OPTIONS', 'DELETE'},
169		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata/<key>'),
170		({'interface_id': '[interface_id]'}, {'POST', 'OPTIONS'},
171		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
172		({'interface_id': '[interface_id]'}, {'GET', 'OPTIONS', 'HEAD'},
173		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
174		({'interface_disable_id': '[interface_disable_id]'},
175		{'POST', 'OPTIONS'},
176		'/api/kytos/topology/v3/interfaces/<interface_disable_id>/disable'),
177		({'interface_enable_id': '[interface_enable_id]'},
178		{'POST', 'OPTIONS'},
179		'/api/kytos/topology/v3/interfaces/<interface_enable_id>/enable'),
180		({'dpid': '[dpid]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
181		'/api/kytos/topology/v3/switches/<dpid>/metadata/<key>'),
182		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
183		'/api/kytos/topology/v3/switches/<dpid>/metadata'),
184		({'dpid': '[dpid]'}, {'GET', 'OPTIONS', 'HEAD'},
185		'/api/kytos/topology/v3/switches/<dpid>/metadata'),
186		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
187		'/api/kytos/topology/v3/switches/<dpid>/disable'),
188		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
189		'/api/kytos/topology/v3/switches/<dpid>/enable'),
190		({'link_id': '[link_id]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
191		'/api/kytos/topology/v3/links/<link_id>/metadata/<key>'),
192		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
193		'/api/kytos/topology/v3/links/<link_id>/metadata'),
194		({'link_id': '[link_id]'}, {'GET', 'OPTIONS', 'HEAD'},
195		'/api/kytos/topology/v3/links/<link_id>/metadata'),
196		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
197		'/api/kytos/topology/v3/links/<link_id>/disable'),
198		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
199		'/api/kytos/topology/v3/links/<link_id>/enable')]
200
201		urls = self.get_napp_urls(self.napp)
202		self.assertEqual(expected_urls, urls)
203
204	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2021-11-30 20:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
205		def get_napp_urls(napp):
206		"""Return the kytos/topology urls.
207
208		The urls will be like:
209
210		urls = [
211		(options, methods, url)
212		]
213
214		"""
215		controller = napp.controller
216		controller.api_server.register_napp_endpoints(napp)
217
218		urls = []
219		for rule in controller.api_server.app.url_map.iter_rules():
220		options = {}
221		for arg in rule.arguments:
222		options[arg] = f"[{arg}]"
223
224		if f'{napp.username}/{napp.name}' in str(rule):
225		urls.append((options, rule.methods, f'{str(rule)}'))
226
227		return urls
228
229		@staticmethod
230		def get_app_test_client(napp):
231		"""Return a flask api test client."""
232		napp.controller.api_server.register_napp_endpoints(napp)
233		return napp.controller.api_server.app.test_client()
234
235		def test_get_interfaces(self):
236		"""test_get_interfaces."""
237		dpid = "00:00:00:00:00:00:00:01"
238		switch = get_switch_mock(0x04, dpid=dpid)
239		switch.dpid = dpid
240		switch.metadata = {"lat": 0, "lng": 0}
241		switch.interfaces = {f"{dpid}:1": f"{dpid}:1"}
242		switch.as_dict = lambda: {"dpid": dpid, "metadata": switch.metadata,
243		"interfaces": switch.interfaces}
244		self.napp.controller.switches = {dpid: switch}
245		client = self.get_app_test_client(self.napp)
246		url = "http://127.0.0.1:8181/api/kytos/topology/v3/interfaces"
247		response = client.get(url)
248		assert response.status_code == 200
249		data = response.json
250		assert "interfaces" in data
251		assert len(data["interfaces"]) == 1
252		assert data["interfaces"][f"{dpid}:1"]
253
254		def test_handle_new_switch(self):
255		"""Test handle new switch."""
256		event_name = '.*.switch.(new\|reconnected)'
257		switch = get_switch_mock(0x04)
258		event = KytosEvent(name=event_name,
259		content={'switch': switch})
260		self.napp.handle_new_switch(event)
261		event_response = self.napp.controller.buffers.app.get()
262		self.assertEqual(event_response.name,
263		'kytos/topology.updated')
264
265		def test_handle_interface_deleted(self):
266		"""Test handle interface deleted."""
267		event_name = '.*.switch.interface.deleted'
268		interface = get_interface_mock("interface1", 7)
269		interface.switch.dpid = "00:00:00:00:00:00:00:01"
270		stats_event = KytosEvent(name=event_name,
271		content={'interface': interface})
272		self.napp.handle_interface_deleted(stats_event)
273		event_updated_response = self.napp.controller.buffers.app.get()
274		self.assertEqual(event_updated_response.name,
275		'kytos/topology.updated')
276
277		def test_handle_connection_lost(self):
278		"""Test handle connection lost."""
279		event_name = '.*.connection.lost'
280		source = Mock()
281		stats_event = KytosEvent(name=event_name,
282		content={'source': source})
283		self.napp.handle_connection_lost(stats_event)
284		event_updated_response = self.napp.controller.buffers.app.get()
285		self.assertEqual(event_updated_response.name,
286		'kytos/topology.updated')
287

kytos-ng / topology

Pull Request — master (#84)

TestMain.test_get_interfaces() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like