build.tests.unit.test_main.TestMain.test_handle_link_maintenance_end() - Code Metrics - Inspection of "Maintenance listeners" - kytos/topology - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#97)

by Antonio

created 2020-05-05 14:45 UTC

TestMain.test_handle_link_maintenance_end() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	15
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	14
nop	2
dl	0
loc	15
rs	9.7
c	0
b	0
f	0

"""Module to test the main napp file."""
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)


class TestMain(TestCase):
    """Test the Main class."""
    # pylint: disable=too-many-public-methods

    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',
                           'kytos/maintenance.start_link',
                           'kytos/maintenance.end_link',
                           'kytos/maintenance.start_switch',
                           'kytos/maintenance.end_switch',
                           '.*.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.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/'),
         ({'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_id': '[interface_id]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/<interface_id>/disable'),
         ({'interface_id': '[interface_id]'}, {'POST', 'OPTIONS'},
          '/api/kytos/topology/v3/interfaces/<interface_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)

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

    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
    def test_handle_link_maintenance_start(self, status_change_mock):
        """Test handle_link_maintenance_start."""
        link1 = MagicMock()
        link1.id = 2
        link2 = MagicMock()
        link2.id = 3
        link3 = MagicMock()
        link3.id = 4
        content = {'links': [link1, link2]}
        event = MagicMock()
        event.content = content
        self.napp.links = {2: link1, 4: link3}
        self.napp.handle_link_maintenance_start(event)
        status_change_mock.assert_called_once_with(link1)

    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
    def test_handle_link_maintenance_end(self, status_change_mock):
        """Test handle_link_maintenance_end."""
        link1 = MagicMock()
        link1.id = 2
        link2 = MagicMock()
        link2.id = 3
        link3 = MagicMock()
        link3.id = 4
        content = {'links': [link1, link2]}
        event = MagicMock()
        event.content = content
        self.napp.links = {2: link1, 4: link3}
        self.napp.handle_link_maintenance_end(event)
        status_change_mock.assert_called_once_with(link1)

    @patch('napps.kytos.topology.main.Main.handle_link_down')

    def test_handle_switch_maintenance_start(self, handle_link_down_mock):
        """Test handle_switch_maintenance_start."""
        switch1 = MagicMock()
        interface1 = MagicMock()
        interface1.is_active.return_value = True
        interface2 = MagicMock()
        interface2.is_active.return_value = False
        interface3 = MagicMock()
        interface3.is_active.return_value = True
        switch1.interfaces = {1: interface1, 2: interface2, 3: interface3}
        switch2 = MagicMock()
        interface4 = MagicMock()
        interface4.is_active.return_value = False
        interface5 = MagicMock()
        interface5.is_active.return_value = True
        switch2.interfaces = {1: interface4, 2: interface5}
        content = {'switches': [switch1, switch2]}
        event = MagicMock()
        event.content = content
        self.napp.handle_switch_maintenance_start(event)
        self.assertEqual(handle_link_down_mock.call_count, 3)

    @patch('napps.kytos.topology.main.Main.handle_link_up')

    def test_handle_switch_maintenance_end(self, handle_link_up_mock):
        """Test handle_switch_maintenance_end."""
        switch1 = MagicMock()
        interface1 = MagicMock()
        interface1.is_active.return_value = True
        interface2 = MagicMock()
        interface2.is_active.return_value = False
        interface3 = MagicMock()
        interface3.is_active.return_value = True
        switch1.interfaces = {1: interface1, 2: interface2, 3: interface3}
        switch2 = MagicMock()
        interface4 = MagicMock()
        interface4.is_active.return_value = False
        interface5 = MagicMock()
        interface5.is_active.return_value = True
        switch2.interfaces = {1: interface4, 2: interface5}
        content = {'switches': [switch1, switch2]}
        event = MagicMock()
        event.content = content
        self.napp.handle_switch_maintenance_end(event)
        self.assertEqual(handle_link_up_mock.call_count, 5)


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

kytos / topology

Pull Request — master (#97)

TestMain.test_handle_link_maintenance_end() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like