build.tests.unit.test_main.TestMain.test_disable_interfaces() - Code Metrics - Inspection of "Enable/disable all interfaces from a switch" - kytos/topology - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#80)

unknown

created 2020-05-14 15:24 UTC

TestMain.test_disable_interfaces() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	35
Code Lines	28

Duplication

Lines	35
Ratio	100 %

Importance

Changes

Metric	Value
cc	1
eloc	28
nop	1
dl	35
loc	35
rs	9.208
c	0
b	0
f	0

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

from kytos.core.switch import Switch
from kytos.core.interface import Interface
from kytos.core.link import Link


from tests.unit.helpers import (get_controller_mock, get_napp_urls,
                                get_app_test_client)


class TestMain(TestCase):
    """Test the Main class."""

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

        Set the server_name_url_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
        self.addCleanup(patch.stopall)

        self.napp = Main(get_controller_mock())

    def test_get_event_listeners(self):
        """Verify all event listeners registered."""
        expected_events = ['kytos/core.shutdown',
                           'kytos/core.shutdown.kytos/topology',
                           '.*.interface.is.nni',
                           '.*.connection.lost',
                           '.*.switch.interface.created',
                           '.*.switch.interface.deleted',
                           '.*.switch.interface.link_down',
                           '.*.switch.interface.link_up',
                           '.*.switch.(new|reconnected)',
                           '.*.switch.port.created',
                           'kytos/topology.*.metadata.*']
        actual_events = self.napp.listeners()
        self.assertEqual(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 = get_napp_urls(self.napp)
        self.assertEqual(expected_urls, urls)

    def test_enable_interfaces(self):

        """Test enable_interfaces."""
        mock_switch = create_autospec(Switch)
        mock_interface = create_autospec(Interface)
        mock_switch.interfaces = {1: mock_interface}
        self.napp.controller.switches = {'00:00:00:00:00:00:00:01':
                                         mock_switch}
        api = get_app_test_client(self.napp)
        expected_success = 'Operation successful'

        interface_id = '00:00:00:00:00:00:00:01:1'
        url = f'{self.server_name_url}/v3/interfaces/{interface_id}/enable'
        response = api.post(url)
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(expected_success, json.loads(response.data))

        dpid = '00:00:00:00:00:00:00:01'
        url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/enable'
        response = api.post(url)
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(expected_success, json.loads(response.data))

        # test interface not found
        interface_id = '00:00:00:00:00:00:00:01:2'
        url = f'{self.server_name_url}/v3/interfaces/{interface_id}/enable'
        response = api.post(url)
        self.assertEqual(response.status_code, 409, response.data)

        # test switch not found
        dpid = '00:00:00:00:00:00:00:02'
        expected_fail = f"Switch not found: '{dpid}'"
        url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/enable'
        response = api.post(url)
        self.assertEqual(response.status_code, 404, response.data)
        self.assertEqual(expected_fail, json.loads(response.data))

    def test_disable_interfaces(self):

        """Test disable_interfaces."""
        interface_id = '00:00:00:00:00:00:00:01:1'
        dpid = '00:00:00:00:00:00:00:01'
        expected = 'Operation successful'
        mock_switch = create_autospec(Switch)
        mock_interface = create_autospec(Interface)
        mock_switch.interfaces = {1: mock_interface}
        self.napp.controller.switches = {'00:00:00:00:00:00:00:01':
                                         mock_switch}
        api = get_app_test_client(self.napp)

        url = f'{self.server_name_url}/v3/interfaces/{interface_id}/disable'
        response = api.post(url)
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(expected, json.loads(response.data))

        url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/disable'
        response = api.post(url)
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(expected, json.loads(response.data))

        # test interface not found
        interface_id = '00:00:00:00:00:00:00:01:2'
        url = f'{self.server_name_url}/v3/interfaces/{interface_id}/disable'
        response = api.post(url)
        self.assertEqual(response.status_code, 409, response.data)

        # test switch not found
        dpid = '00:00:00:00:00:00:00:02'
        expected_fail = f"Switch not found: '{dpid}'"
        url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/disable'
        response = api.post(url)
        self.assertEqual(response.status_code, 404, response.data)
        self.assertEqual(expected_fail, json.loads(response.data))

    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    @patch('napps.kytos.topology.main.Main.update_instance_metadata')
    def test_handle_new_switch(self, *args):
        """Test handle_new_switch."""
        (mock_instance_metadata, mock_notify_topology_update) = args
        mock_event = MagicMock()
        mock_switch = create_autospec(Switch)
        mock_event.content['switch'] = mock_switch
        self.napp.handle_new_switch(mock_event)
        mock_notify_topology_update.assert_called()
        mock_instance_metadata.assert_called()

    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    def test_handle_connection_lost(self, mock_notify_topology_update):
        """Test handle connection_lost."""
        mock_event = MagicMock()
        mock_switch = create_autospec(Switch)
        mock_switch.return_value = True
        mock_event.content['source'] = mock_switch
        self.napp.handle_connection_lost(mock_event)
        mock_notify_topology_update.assert_called()

    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    @patch('napps.kytos.topology.main.Main.update_instance_metadata')
    def test_handle_interface_up(self, *args):
        """Test handle_interface_up."""
        (mock_instance_metadata, mock_notify_topology_update) = args
        mock_event = MagicMock()
        mock_interface = create_autospec(Interface)
        mock_event.content['interface'] = mock_interface
        self.napp.handle_interface_up(mock_event)
        mock_notify_topology_update.assert_called()
        mock_instance_metadata.assert_called()

    @patch('napps.kytos.topology.main.Main.handle_interface_up')
    def test_handle_interface_created(self, mock_handle_interface_up):
        """Test handle interface created."""
        mock_event = MagicMock()
        self.napp.handle_interface_created(mock_event)
        mock_handle_interface_up.assert_called()

    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    @patch('napps.kytos.topology.main.Main.handle_interface_link_down')
    def test_handle_interface_down(self, *args):
        """Test handle interface down."""
        (mock_handle_interface_link_down, mock_notify_topology_update) = args
        mock_event = MagicMock()
        mock_interface = create_autospec(Interface)
        mock_event.content['interface'] = mock_interface
        self.napp.handle_interface_down(mock_event)
        mock_handle_interface_link_down.assert_called()
        mock_notify_topology_update.assert_called()

    @patch('napps.kytos.topology.main.Main.handle_interface_down')
    def test_interface_deleted(self, mock_handle_interface_link_down):
        """Test interface deleted."""
        mock_event = MagicMock()
        self.napp.handle_interface_deleted(mock_event)
        mock_handle_interface_link_down.assert_called()

    @patch('napps.kytos.topology.main.Main._get_link_from_interface')
    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    @patch('napps.kytos.topology.main.Main.update_instance_metadata')
    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
    def test_interface_link_up(self, *args):
        """Test interface link_up."""
        (mock_status_change, mock_instance_metadata, mock_topology_update,
         mock_link_from_interface) = args

        mock_event = MagicMock()
        mock_interface = create_autospec(Interface)
        mock_link = create_autospec(Link)
        mock_link.is_active.return_value = False
        mock_link_from_interface.return_value = mock_link
        mock_event.content['interface'] = mock_interface
        self.napp.handle_interface_link_up(mock_event)
        mock_topology_update.assert_called()
        mock_instance_metadata.assert_called()
        mock_status_change.assert_called()

    @patch('napps.kytos.topology.main.Main._get_link_from_interface')
    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
    def test_interface_link_down(self, *args):
        """Test interface link down."""
        (mock_status_change, mock_topology_update,
         mock_link_from_interface) = args

        mock_event = MagicMock()
        mock_interface = create_autospec(Interface)
        mock_link = create_autospec(Link)
        mock_link.is_active.return_value = True
        mock_link_from_interface.return_value = mock_link
        mock_event.content['interface'] = mock_interface
        self.napp.handle_interface_link_down(mock_event)
        mock_topology_update.assert_called()
        mock_status_change.assert_called()

    @patch('napps.kytos.topology.main.Main._get_link_or_create')
    @patch('napps.kytos.topology.main.Main.notify_topology_update')
    def test_add_links(self, *args):
        """Test add_links."""
        (mock_notify_topology_update, mock_get_link_or_create) = args
        mock_event = MagicMock()
        self.napp.add_links(mock_event)
        mock_get_link_or_create.assert_called()
        mock_notify_topology_update.assert_called()

    @patch('napps.kytos.topology.main.KytosEvent')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_notify_topology_update(self, *args):
        """Test notify_topology_update."""
        (mock_buffers_put, mock_event) = args
        self.napp.notify_topology_update()
        mock_event.assert_called()
        mock_buffers_put.assert_called()

    @patch('napps.kytos.topology.main.KytosEvent')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_notify_link_status_change(self, *args):
        """Test notify link status change."""
        (mock_buffers_put, mock_event) = args
        mock_link = create_autospec(Link)
        self.napp.notify_link_status_change(mock_link)
        mock_event.assert_called()
        mock_buffers_put.assert_called()

    @patch('napps.kytos.topology.main.KytosEvent')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.topology.main.isinstance')
    def test_notify_metadata_changes(self, *args):
        """Test notify metadata changes."""
        (mock_isinstance, mock_buffers_put, mock_event) = args
        mock_isinstance.return_value = True
        mock_obj = MagicMock()
        mock_action = create_autospec(Switch)
        self.napp.notify_metadata_changes(mock_obj, mock_action)
        mock_event.assert_called()
        mock_isinstance.assert_called()
        mock_buffers_put.assert_called()

    @patch('napps.kytos.topology.main.KytosEvent')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_notify_port_created(self, *args):
        """Test notify port created."""
        (mock_buffers_put, mock_kytos_event) = args
        mock_event = MagicMock()
        self.napp.notify_port_created(mock_event)
        mock_kytos_event.assert_called()
        mock_buffers_put.assert_called()

    @patch('napps.kytos.topology.main.KytosEvent')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_verify_storehouse(self, *args):
        """Test verify_storehouse."""
        (mock_buffers_put, mock_kytos_event) = args
        mock_entities = MagicMock()
        self.napp.verify_storehouse(mock_entities)
        mock_buffers_put.assert_called()
        mock_kytos_event.assert_called()


1		"""Module to test the main napp file."""
2		import json
3		from unittest import TestCase
4		from unittest.mock import MagicMock, create_autospec, patch
5
6		from kytos.core.switch import Switch
7		from kytos.core.interface import Interface
8		from kytos.core.link import Link
9
10
11		from tests.unit.helpers import (get_controller_mock, get_napp_urls,
12		get_app_test_client)
13
14
15		class TestMain(TestCase):
16		"""Test the Main class."""
17
18		def setUp(self):
19		"""Execute steps before each tests.
20
21		Set the server_name_url_url from kytos/topology
22		"""
23		self.server_name_url = 'http://localhost:8181/api/kytos/topology'
24
25		patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
26		from napps.kytos.topology.main import Main
27		self.addCleanup(patch.stopall)
28
29		self.napp = Main(get_controller_mock())
30
31		def test_get_event_listeners(self):
32		"""Verify all event listeners registered."""
33		expected_events = ['kytos/core.shutdown',
34		'kytos/core.shutdown.kytos/topology',
35		'.*.interface.is.nni',
36		'.*.connection.lost',
37		'.*.switch.interface.created',
38		'.*.switch.interface.deleted',
39		'.*.switch.interface.link_down',
40		'.*.switch.interface.link_up',
41		'.*.switch.(new\|reconnected)',
42		'.*.switch.port.created',
43		'kytos/topology..metadata.']
44		actual_events = self.napp.listeners()
45		self.assertEqual(expected_events, actual_events)
46
47	View Code Duplication	def test_verify_api_urls(self):
		0 ignored issues – show Duplication introduced 2020-04-02 12:34 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
48		"""Verify all APIs registered."""
49		expected_urls = [
50		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/interfaces'),
51		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/switches'),
52		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/links'),
53		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/'),
54		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
55		'/api/kytos/topology/v3/interfaces/switch/<dpid>/disable'),
56		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
57		'/api/kytos/topology/v3/interfaces/switch/<dpid>/enable'),
58		({'key': '[key]', 'interface_id': '[interface_id]'},
59		{'OPTIONS', 'DELETE'},
60		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata/<key>'),
61		({'interface_id': '[interface_id]'}, {'POST', 'OPTIONS'},
62		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
63		({'interface_id': '[interface_id]'}, {'GET', 'OPTIONS', 'HEAD'},
64		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
65		({'interface_disable_id': '[interface_disable_id]'},
66		{'POST', 'OPTIONS'},
67		'/api/kytos/topology/v3/interfaces/<interface_disable_id>/disable'),
68		({'interface_enable_id': '[interface_enable_id]'},
69		{'POST', 'OPTIONS'},
70		'/api/kytos/topology/v3/interfaces/<interface_enable_id>/enable'),
71		({'dpid': '[dpid]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
72		'/api/kytos/topology/v3/switches/<dpid>/metadata/<key>'),
73		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
74		'/api/kytos/topology/v3/switches/<dpid>/metadata'),
75		({'dpid': '[dpid]'}, {'GET', 'OPTIONS', 'HEAD'},
76		'/api/kytos/topology/v3/switches/<dpid>/metadata'),
77		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
78		'/api/kytos/topology/v3/switches/<dpid>/disable'),
79		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
80		'/api/kytos/topology/v3/switches/<dpid>/enable'),
81		({'link_id': '[link_id]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
82		'/api/kytos/topology/v3/links/<link_id>/metadata/<key>'),
83		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
84		'/api/kytos/topology/v3/links/<link_id>/metadata'),
85		({'link_id': '[link_id]'}, {'GET', 'OPTIONS', 'HEAD'},
86		'/api/kytos/topology/v3/links/<link_id>/metadata'),
87		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
88		'/api/kytos/topology/v3/links/<link_id>/disable'),
89		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
90		'/api/kytos/topology/v3/links/<link_id>/enable')]
91
92		urls = get_napp_urls(self.napp)
93		self.assertEqual(expected_urls, urls)
94
95	View Code Duplication	def test_enable_interfaces(self):
		0 ignored issues – show Duplication introduced 2020-05-14 15:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
96		"""Test enable_interfaces."""
97		mock_switch = create_autospec(Switch)
98		mock_interface = create_autospec(Interface)
99		mock_switch.interfaces = {1: mock_interface}
100		self.napp.controller.switches = {'00:00:00:00:00:00:00:01':
101		mock_switch}
102		api = get_app_test_client(self.napp)
103		expected_success = 'Operation successful'
104
105		interface_id = '00:00:00:00:00:00:00:01:1'
106		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/enable'
107		response = api.post(url)
108		self.assertEqual(response.status_code, 200, response.data)
109		self.assertEqual(expected_success, json.loads(response.data))
110
111		dpid = '00:00:00:00:00:00:00:01'
112		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/enable'
113		response = api.post(url)
114		self.assertEqual(response.status_code, 200, response.data)
115		self.assertEqual(expected_success, json.loads(response.data))
116
117		# test interface not found
118		interface_id = '00:00:00:00:00:00:00:01:2'
119		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/enable'
120		response = api.post(url)
121		self.assertEqual(response.status_code, 409, response.data)
122
123		# test switch not found
124		dpid = '00:00:00:00:00:00:00:02'
125		expected_fail = f"Switch not found: '{dpid}'"
126		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/enable'
127		response = api.post(url)
128		self.assertEqual(response.status_code, 404, response.data)
129		self.assertEqual(expected_fail, json.loads(response.data))
130
131	View Code Duplication	def test_disable_interfaces(self):
		0 ignored issues – show Duplication introduced 2020-05-14 15:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
132		"""Test disable_interfaces."""
133		interface_id = '00:00:00:00:00:00:00:01:1'
134		dpid = '00:00:00:00:00:00:00:01'
135		expected = 'Operation successful'
136		mock_switch = create_autospec(Switch)
137		mock_interface = create_autospec(Interface)
138		mock_switch.interfaces = {1: mock_interface}
139		self.napp.controller.switches = {'00:00:00:00:00:00:00:01':
140		mock_switch}
141		api = get_app_test_client(self.napp)
142
143		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/disable'
144		response = api.post(url)
145		self.assertEqual(response.status_code, 200, response.data)
146		self.assertEqual(expected, json.loads(response.data))
147
148		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/disable'
149		response = api.post(url)
150		self.assertEqual(response.status_code, 200, response.data)
151		self.assertEqual(expected, json.loads(response.data))
152
153		# test interface not found
154		interface_id = '00:00:00:00:00:00:00:01:2'
155		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/disable'
156		response = api.post(url)
157		self.assertEqual(response.status_code, 409, response.data)
158
159		# test switch not found
160		dpid = '00:00:00:00:00:00:00:02'
161		expected_fail = f"Switch not found: '{dpid}'"
162		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/disable'
163		response = api.post(url)
164		self.assertEqual(response.status_code, 404, response.data)
165		self.assertEqual(expected_fail, json.loads(response.data))
166
167		@patch('napps.kytos.topology.main.Main.notify_topology_update')
168		@patch('napps.kytos.topology.main.Main.update_instance_metadata')
169		def test_handle_new_switch(self, *args):
170		"""Test handle_new_switch."""
171		(mock_instance_metadata, mock_notify_topology_update) = args
172		mock_event = MagicMock()
173		mock_switch = create_autospec(Switch)
174		mock_event.content['switch'] = mock_switch
175		self.napp.handle_new_switch(mock_event)
176		mock_notify_topology_update.assert_called()
177		mock_instance_metadata.assert_called()
178
179		@patch('napps.kytos.topology.main.Main.notify_topology_update')
180		def test_handle_connection_lost(self, mock_notify_topology_update):
181		"""Test handle connection_lost."""
182		mock_event = MagicMock()
183		mock_switch = create_autospec(Switch)
184		mock_switch.return_value = True
185		mock_event.content['source'] = mock_switch
186		self.napp.handle_connection_lost(mock_event)
187		mock_notify_topology_update.assert_called()
188
189		@patch('napps.kytos.topology.main.Main.notify_topology_update')
190		@patch('napps.kytos.topology.main.Main.update_instance_metadata')
191		def test_handle_interface_up(self, *args):
192		"""Test handle_interface_up."""
193		(mock_instance_metadata, mock_notify_topology_update) = args
194		mock_event = MagicMock()
195		mock_interface = create_autospec(Interface)
196		mock_event.content['interface'] = mock_interface
197		self.napp.handle_interface_up(mock_event)
198		mock_notify_topology_update.assert_called()
199		mock_instance_metadata.assert_called()
200
201		@patch('napps.kytos.topology.main.Main.handle_interface_up')
202		def test_handle_interface_created(self, mock_handle_interface_up):
203		"""Test handle interface created."""
204		mock_event = MagicMock()
205		self.napp.handle_interface_created(mock_event)
206		mock_handle_interface_up.assert_called()
207
208		@patch('napps.kytos.topology.main.Main.notify_topology_update')
209		@patch('napps.kytos.topology.main.Main.handle_interface_link_down')
210		def test_handle_interface_down(self, *args):
211		"""Test handle interface down."""
212		(mock_handle_interface_link_down, mock_notify_topology_update) = args
213		mock_event = MagicMock()
214		mock_interface = create_autospec(Interface)
215		mock_event.content['interface'] = mock_interface
216		self.napp.handle_interface_down(mock_event)
217		mock_handle_interface_link_down.assert_called()
218		mock_notify_topology_update.assert_called()
219
220		@patch('napps.kytos.topology.main.Main.handle_interface_down')
221		def test_interface_deleted(self, mock_handle_interface_link_down):
222		"""Test interface deleted."""
223		mock_event = MagicMock()
224		self.napp.handle_interface_deleted(mock_event)
225		mock_handle_interface_link_down.assert_called()
226
227		@patch('napps.kytos.topology.main.Main._get_link_from_interface')
228		@patch('napps.kytos.topology.main.Main.notify_topology_update')
229		@patch('napps.kytos.topology.main.Main.update_instance_metadata')
230		@patch('napps.kytos.topology.main.Main.notify_link_status_change')
231		def test_interface_link_up(self, *args):
232		"""Test interface link_up."""
233		(mock_status_change, mock_instance_metadata, mock_topology_update,
234		mock_link_from_interface) = args
235
236		mock_event = MagicMock()
237		mock_interface = create_autospec(Interface)
238		mock_link = create_autospec(Link)
239		mock_link.is_active.return_value = False
240		mock_link_from_interface.return_value = mock_link
241		mock_event.content['interface'] = mock_interface
242		self.napp.handle_interface_link_up(mock_event)
243		mock_topology_update.assert_called()
244		mock_instance_metadata.assert_called()
245		mock_status_change.assert_called()
246
247		@patch('napps.kytos.topology.main.Main._get_link_from_interface')
248		@patch('napps.kytos.topology.main.Main.notify_topology_update')
249		@patch('napps.kytos.topology.main.Main.notify_link_status_change')
250		def test_interface_link_down(self, *args):
251		"""Test interface link down."""
252		(mock_status_change, mock_topology_update,
253		mock_link_from_interface) = args
254
255		mock_event = MagicMock()
256		mock_interface = create_autospec(Interface)
257		mock_link = create_autospec(Link)
258		mock_link.is_active.return_value = True
259		mock_link_from_interface.return_value = mock_link
260		mock_event.content['interface'] = mock_interface
261		self.napp.handle_interface_link_down(mock_event)
262		mock_topology_update.assert_called()
263		mock_status_change.assert_called()
264
265		@patch('napps.kytos.topology.main.Main._get_link_or_create')
266		@patch('napps.kytos.topology.main.Main.notify_topology_update')
267		def test_add_links(self, *args):
268		"""Test add_links."""
269		(mock_notify_topology_update, mock_get_link_or_create) = args
270		mock_event = MagicMock()
271		self.napp.add_links(mock_event)
272		mock_get_link_or_create.assert_called()
273		mock_notify_topology_update.assert_called()
274
275		@patch('napps.kytos.topology.main.KytosEvent')
276		@patch('kytos.core.buffers.KytosEventBuffer.put')
277		def test_notify_topology_update(self, *args):
278		"""Test notify_topology_update."""
279		(mock_buffers_put, mock_event) = args
280		self.napp.notify_topology_update()
281		mock_event.assert_called()
282		mock_buffers_put.assert_called()
283
284		@patch('napps.kytos.topology.main.KytosEvent')
285		@patch('kytos.core.buffers.KytosEventBuffer.put')
286		def test_notify_link_status_change(self, *args):
287		"""Test notify link status change."""
288		(mock_buffers_put, mock_event) = args
289		mock_link = create_autospec(Link)
290		self.napp.notify_link_status_change(mock_link)
291		mock_event.assert_called()
292		mock_buffers_put.assert_called()
293
294		@patch('napps.kytos.topology.main.KytosEvent')
295		@patch('kytos.core.buffers.KytosEventBuffer.put')
296		@patch('napps.kytos.topology.main.isinstance')
297		def test_notify_metadata_changes(self, *args):
298		"""Test notify metadata changes."""
299		(mock_isinstance, mock_buffers_put, mock_event) = args
300		mock_isinstance.return_value = True
301		mock_obj = MagicMock()
302		mock_action = create_autospec(Switch)
303		self.napp.notify_metadata_changes(mock_obj, mock_action)
304		mock_event.assert_called()
305		mock_isinstance.assert_called()
306		mock_buffers_put.assert_called()
307
308		@patch('napps.kytos.topology.main.KytosEvent')
309		@patch('kytos.core.buffers.KytosEventBuffer.put')
310		def test_notify_port_created(self, *args):
311		"""Test notify port created."""
312		(mock_buffers_put, mock_kytos_event) = args
313		mock_event = MagicMock()
314		self.napp.notify_port_created(mock_event)
315		mock_kytos_event.assert_called()
316		mock_buffers_put.assert_called()
317
318		@patch('napps.kytos.topology.main.KytosEvent')
319		@patch('kytos.core.buffers.KytosEventBuffer.put')
320		def test_verify_storehouse(self, *args):
321		"""Test verify_storehouse."""
322		(mock_buffers_put, mock_kytos_event) = args
323		mock_entities = MagicMock()
324		self.napp.verify_storehouse(mock_entities)
325		mock_buffers_put.assert_called()
326		mock_kytos_event.assert_called()
327

kytos / topology

Pull Request — master (#80)

TestMain.test_disable_interfaces() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like