build.tests.unit.test_main.TestMain.test_handle_link_maintenance_end() - Code Metrics - Inspection of "Merge pull request #97 from ajoaoff/maintenance" - kytos/topology - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 4fca5c...e35304 )

by Humberto

created 2020-05-19 20:03 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."""
import time
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."""
    # 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'),
         ({}, {'POST', 'OPTIONS'}, '/api/kytos/topology/v3/restore'),
         ({}, {'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_1 = create_autospec(Interface)
        mock_interface_2 = create_autospec(Interface)
        mock_switch.interfaces = {1: mock_interface_1, 2: mock_interface_2}
        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))
        self.assertEqual(mock_interface_1.enable.call_count, 1)
        self.assertEqual(mock_interface_2.enable.call_count, 0)

        dpid = '00:00:00:00:00:00:00:01'
        mock_interface_1.enable.call_count = 0
        mock_interface_2.enable.call_count = 0
        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))
        self.assertEqual(mock_interface_1.enable.call_count, 1)
        self.assertEqual(mock_interface_2.enable.call_count, 1)

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

        # 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))
        self.assertEqual(mock_interface_1.enable.call_count, 0)
        self.assertEqual(mock_interface_2.enable.call_count, 0)

    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_1 = create_autospec(Interface)
        mock_interface_2 = create_autospec(Interface)
        mock_switch.interfaces = {1: mock_interface_1, 2: mock_interface_2}
        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))
        self.assertEqual(mock_interface_1.disable.call_count, 1)
        self.assertEqual(mock_interface_2.disable.call_count, 0)

        mock_interface_1.disable.call_count = 0
        mock_interface_2.disable.call_count = 0
        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))
        self.assertEqual(mock_interface_1.disable.call_count, 1)
        self.assertEqual(mock_interface_2.disable.call_count, 1)

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

        # 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))
        self.assertEqual(mock_interface_1.disable.call_count, 0)
        self.assertEqual(mock_interface_2.disable.call_count, 0)

    @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

        now = time.time()
        mock_event = MagicMock()
        mock_interface_a = create_autospec(Interface)
        mock_interface_a.is_active.return_value = False
        mock_interface_b = create_autospec(Interface)
        mock_interface_b.is_active.return_value = True
        mock_link = create_autospec(Link)
        mock_link.get_metadata.return_value = now
        mock_link.is_active.side_effect = [False, True]
        mock_link.endpoint_a = mock_interface_a
        mock_link.endpoint_b = mock_interface_b
        mock_link_from_interface.return_value = mock_link
        content = {'interface': mock_interface_a}
        mock_event.content = content
        self.napp.link_up_timer = 1
        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		import time
3		import json
4
5		from unittest import TestCase
6		from unittest.mock import MagicMock, create_autospec, patch
7
8		from kytos.core.switch import Switch
9		from kytos.core.interface import Interface
10		from kytos.core.link import Link
11
12
13		from tests.unit.helpers import (get_controller_mock, get_napp_urls,
14		get_app_test_client)
15
16
17		class TestMain(TestCase):
18		"""Test the Main class."""
19		# pylint: disable=too-many-public-methods
20
21		def setUp(self):
22		"""Execute steps before each tests.
23
24		Set the server_name_url_url from kytos/topology
25		"""
26		self.server_name_url = 'http://localhost:8181/api/kytos/topology'
27
28		patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
29		from napps.kytos.topology.main import Main
30		self.addCleanup(patch.stopall)
31
32		self.napp = Main(get_controller_mock())
33
34		def test_get_event_listeners(self):
35		"""Verify all event listeners registered."""
36		expected_events = ['kytos/core.shutdown',
37		'kytos/core.shutdown.kytos/topology',
38		'kytos/maintenance.start_link',
39		'kytos/maintenance.end_link',
40		'kytos/maintenance.start_switch',
41		'kytos/maintenance.end_switch',
42		'.*.interface.is.nni',
43		'.*.connection.lost',
44		'.*.switch.interface.created',
45		'.*.switch.interface.deleted',
46		'.*.switch.interface.link_down',
47		'.*.switch.interface.link_up',
48		'.*.switch.(new\|reconnected)',
49		'.*.switch.port.created',
50		'kytos/topology..metadata.']
51		actual_events = self.napp.listeners()
52		self.assertCountEqual(expected_events, actual_events)
53
54	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...
55		"""Verify all APIs registered."""
56		expected_urls = [
57		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/interfaces'),
58		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/switches'),
59		({}, {'POST', 'OPTIONS'}, '/api/kytos/topology/v3/restore'),
60		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/links'),
61		({}, {'GET', 'OPTIONS', 'HEAD'}, '/api/kytos/topology/v3/'),
62		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
63		'/api/kytos/topology/v3/interfaces/switch/<dpid>/disable'),
64		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
65		'/api/kytos/topology/v3/interfaces/switch/<dpid>/enable'),
66		({'key': '[key]', 'interface_id': '[interface_id]'},
67		{'OPTIONS', 'DELETE'},
68		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata/<key>'),
69		({'interface_id': '[interface_id]'}, {'POST', 'OPTIONS'},
70		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
71		({'interface_id': '[interface_id]'}, {'GET', 'OPTIONS', 'HEAD'},
72		'/api/kytos/topology/v3/interfaces/<interface_id>/metadata'),
73		({'interface_disable_id': '[interface_disable_id]'},
74		{'POST', 'OPTIONS'},
75		'/api/kytos/topology/v3/interfaces/<interface_disable_id>/disable'),
76		({'interface_enable_id': '[interface_enable_id]'},
77		{'POST', 'OPTIONS'},
78		'/api/kytos/topology/v3/interfaces/<interface_enable_id>/enable'),
79		({'dpid': '[dpid]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
80		'/api/kytos/topology/v3/switches/<dpid>/metadata/<key>'),
81		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
82		'/api/kytos/topology/v3/switches/<dpid>/metadata'),
83		({'dpid': '[dpid]'}, {'GET', 'OPTIONS', 'HEAD'},
84		'/api/kytos/topology/v3/switches/<dpid>/metadata'),
85		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
86		'/api/kytos/topology/v3/switches/<dpid>/disable'),
87		({'dpid': '[dpid]'}, {'POST', 'OPTIONS'},
88		'/api/kytos/topology/v3/switches/<dpid>/enable'),
89		({'link_id': '[link_id]', 'key': '[key]'}, {'OPTIONS', 'DELETE'},
90		'/api/kytos/topology/v3/links/<link_id>/metadata/<key>'),
91		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
92		'/api/kytos/topology/v3/links/<link_id>/metadata'),
93		({'link_id': '[link_id]'}, {'GET', 'OPTIONS', 'HEAD'},
94		'/api/kytos/topology/v3/links/<link_id>/metadata'),
95		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
96		'/api/kytos/topology/v3/links/<link_id>/disable'),
97		({'link_id': '[link_id]'}, {'POST', 'OPTIONS'},
98		'/api/kytos/topology/v3/links/<link_id>/enable')]
99
100		urls = get_napp_urls(self.napp)
101		self.assertEqual(expected_urls, urls)
102
103	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...
104		"""Test enable_interfaces."""
105		mock_switch = create_autospec(Switch)
106		mock_interface_1 = create_autospec(Interface)
107		mock_interface_2 = create_autospec(Interface)
108		mock_switch.interfaces = {1: mock_interface_1, 2: mock_interface_2}
109		self.napp.controller.switches = {'00:00:00:00:00:00:00:01':
110		mock_switch}
111		api = get_app_test_client(self.napp)
112		expected_success = 'Operation successful'
113
114		interface_id = '00:00:00:00:00:00:00:01:1'
115		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/enable'
116		response = api.post(url)
117		self.assertEqual(response.status_code, 200, response.data)
118		self.assertEqual(expected_success, json.loads(response.data))
119		self.assertEqual(mock_interface_1.enable.call_count, 1)
120		self.assertEqual(mock_interface_2.enable.call_count, 0)
121
122		dpid = '00:00:00:00:00:00:00:01'
123		mock_interface_1.enable.call_count = 0
124		mock_interface_2.enable.call_count = 0
125		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/enable'
126		response = api.post(url)
127		self.assertEqual(response.status_code, 200, response.data)
128		self.assertEqual(expected_success, json.loads(response.data))
129		self.assertEqual(mock_interface_1.enable.call_count, 1)
130		self.assertEqual(mock_interface_2.enable.call_count, 1)
131
132		# test interface not found
133		interface_id = '00:00:00:00:00:00:00:01:3'
134		mock_interface_1.enable.call_count = 0
135		mock_interface_2.enable.call_count = 0
136		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/enable'
137		response = api.post(url)
138		self.assertEqual(response.status_code, 409, response.data)
139		self.assertEqual(mock_interface_1.enable.call_count, 0)
140		self.assertEqual(mock_interface_2.enable.call_count, 0)
141
142		# test switch not found
143		dpid = '00:00:00:00:00:00:00:02'
144		expected_fail = f"Switch not found: '{dpid}'"
145		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/enable'
146		response = api.post(url)
147		self.assertEqual(response.status_code, 404, response.data)
148		self.assertEqual(expected_fail, json.loads(response.data))
149		self.assertEqual(mock_interface_1.enable.call_count, 0)
150		self.assertEqual(mock_interface_2.enable.call_count, 0)
151
152	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...
153		"""Test disable_interfaces."""
154		interface_id = '00:00:00:00:00:00:00:01:1'
155		dpid = '00:00:00:00:00:00:00:01'
156		expected = 'Operation successful'
157		mock_switch = create_autospec(Switch)
158		mock_interface_1 = create_autospec(Interface)
159		mock_interface_2 = create_autospec(Interface)
160		mock_switch.interfaces = {1: mock_interface_1, 2: mock_interface_2}
161		self.napp.controller.switches = {'00:00:00:00:00:00:00:01':
162		mock_switch}
163		api = get_app_test_client(self.napp)
164
165		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/disable'
166		response = api.post(url)
167		self.assertEqual(response.status_code, 200, response.data)
168		self.assertEqual(expected, json.loads(response.data))
169		self.assertEqual(mock_interface_1.disable.call_count, 1)
170		self.assertEqual(mock_interface_2.disable.call_count, 0)
171
172		mock_interface_1.disable.call_count = 0
173		mock_interface_2.disable.call_count = 0
174		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/disable'
175		response = api.post(url)
176		self.assertEqual(response.status_code, 200, response.data)
177		self.assertEqual(expected, json.loads(response.data))
178		self.assertEqual(mock_interface_1.disable.call_count, 1)
179		self.assertEqual(mock_interface_2.disable.call_count, 1)
180
181		# test interface not found
182		interface_id = '00:00:00:00:00:00:00:01:3'
183		mock_interface_1.disable.call_count = 0
184		mock_interface_2.disable.call_count = 0
185		url = f'{self.server_name_url}/v3/interfaces/{interface_id}/disable'
186		response = api.post(url)
187		self.assertEqual(response.status_code, 409, response.data)
188		self.assertEqual(mock_interface_1.disable.call_count, 0)
189		self.assertEqual(mock_interface_2.disable.call_count, 0)
190
191		# test switch not found
192		dpid = '00:00:00:00:00:00:00:02'
193		expected_fail = f"Switch not found: '{dpid}'"
194		url = f'{self.server_name_url}/v3/interfaces/switch/{dpid}/disable'
195		response = api.post(url)
196		self.assertEqual(response.status_code, 404, response.data)
197		self.assertEqual(expected_fail, json.loads(response.data))
198		self.assertEqual(mock_interface_1.disable.call_count, 0)
199		self.assertEqual(mock_interface_2.disable.call_count, 0)
200
201		@patch('napps.kytos.topology.main.Main.notify_topology_update')
202		@patch('napps.kytos.topology.main.Main.update_instance_metadata')
203		def test_handle_new_switch(self, *args):
204		"""Test handle_new_switch."""
205		(mock_instance_metadata, mock_notify_topology_update) = args
206		mock_event = MagicMock()
207		mock_switch = create_autospec(Switch)
208		mock_event.content['switch'] = mock_switch
209		self.napp.handle_new_switch(mock_event)
210		mock_notify_topology_update.assert_called()
211		mock_instance_metadata.assert_called()
212
213		@patch('napps.kytos.topology.main.Main.notify_topology_update')
214		def test_handle_connection_lost(self, mock_notify_topology_update):
215		"""Test handle connection_lost."""
216		mock_event = MagicMock()
217		mock_switch = create_autospec(Switch)
218		mock_switch.return_value = True
219		mock_event.content['source'] = mock_switch
220		self.napp.handle_connection_lost(mock_event)
221		mock_notify_topology_update.assert_called()
222
223		@patch('napps.kytos.topology.main.Main.notify_topology_update')
224		@patch('napps.kytos.topology.main.Main.update_instance_metadata')
225		def test_handle_interface_up(self, *args):
226		"""Test handle_interface_up."""
227		(mock_instance_metadata, mock_notify_topology_update) = args
228		mock_event = MagicMock()
229		mock_interface = create_autospec(Interface)
230		mock_event.content['interface'] = mock_interface
231		self.napp.handle_interface_up(mock_event)
232		mock_notify_topology_update.assert_called()
233		mock_instance_metadata.assert_called()
234
235		@patch('napps.kytos.topology.main.Main.handle_interface_up')
236		def test_handle_interface_created(self, mock_handle_interface_up):
237		"""Test handle interface created."""
238		mock_event = MagicMock()
239		self.napp.handle_interface_created(mock_event)
240		mock_handle_interface_up.assert_called()
241
242		@patch('napps.kytos.topology.main.Main.notify_topology_update')
243		@patch('napps.kytos.topology.main.Main.handle_interface_link_down')
244		def test_handle_interface_down(self, *args):
245		"""Test handle interface down."""
246		(mock_handle_interface_link_down, mock_notify_topology_update) = args
247		mock_event = MagicMock()
248		mock_interface = create_autospec(Interface)
249		mock_event.content['interface'] = mock_interface
250		self.napp.handle_interface_down(mock_event)
251		mock_handle_interface_link_down.assert_called()
252		mock_notify_topology_update.assert_called()
253
254		@patch('napps.kytos.topology.main.Main.handle_interface_down')
255		def test_interface_deleted(self, mock_handle_interface_link_down):
256		"""Test interface deleted."""
257		mock_event = MagicMock()
258		self.napp.handle_interface_deleted(mock_event)
259		mock_handle_interface_link_down.assert_called()
260
261		@patch('napps.kytos.topology.main.Main._get_link_from_interface')
262		@patch('napps.kytos.topology.main.Main.notify_topology_update')
263		@patch('napps.kytos.topology.main.Main.update_instance_metadata')
264		@patch('napps.kytos.topology.main.Main.notify_link_status_change')
265		def test_interface_link_up(self, *args):
266		"""Test interface link_up."""
267		(mock_status_change, mock_instance_metadata, mock_topology_update,
268		mock_link_from_interface) = args
269
270		now = time.time()
271		mock_event = MagicMock()
272		mock_interface_a = create_autospec(Interface)
273		mock_interface_a.is_active.return_value = False
274		mock_interface_b = create_autospec(Interface)
275		mock_interface_b.is_active.return_value = True
276		mock_link = create_autospec(Link)
277		mock_link.get_metadata.return_value = now
278		mock_link.is_active.side_effect = [False, True]
279		mock_link.endpoint_a = mock_interface_a
280		mock_link.endpoint_b = mock_interface_b
281		mock_link_from_interface.return_value = mock_link
282		content = {'interface': mock_interface_a}
283		mock_event.content = content
284		self.napp.link_up_timer = 1
285		self.napp.handle_interface_link_up(mock_event)
286		mock_topology_update.assert_called()
287		mock_instance_metadata.assert_called()
288		mock_status_change.assert_called()
289
290		@patch('napps.kytos.topology.main.Main._get_link_from_interface')
291		@patch('napps.kytos.topology.main.Main.notify_topology_update')
292		@patch('napps.kytos.topology.main.Main.notify_link_status_change')
293		def test_interface_link_down(self, *args):
294		"""Test interface link down."""
295		(mock_status_change, mock_topology_update,
296		mock_link_from_interface) = args
297
298		mock_event = MagicMock()
299		mock_interface = create_autospec(Interface)
300		mock_link = create_autospec(Link)
301		mock_link.is_active.return_value = True
302		mock_link_from_interface.return_value = mock_link
303		mock_event.content['interface'] = mock_interface
304		self.napp.handle_interface_link_down(mock_event)
305		mock_topology_update.assert_called()
306		mock_status_change.assert_called()
307
308		@patch('napps.kytos.topology.main.Main._get_link_or_create')
309		@patch('napps.kytos.topology.main.Main.notify_topology_update')
310		def test_add_links(self, *args):
311		"""Test add_links."""
312		(mock_notify_topology_update, mock_get_link_or_create) = args
313		mock_event = MagicMock()
314		self.napp.add_links(mock_event)
315		mock_get_link_or_create.assert_called()
316		mock_notify_topology_update.assert_called()
317
318		@patch('napps.kytos.topology.main.KytosEvent')
319		@patch('kytos.core.buffers.KytosEventBuffer.put')
320		def test_notify_topology_update(self, *args):
321		"""Test notify_topology_update."""
322		(mock_buffers_put, mock_event) = args
323		self.napp.notify_topology_update()
324		mock_event.assert_called()
325		mock_buffers_put.assert_called()
326
327		@patch('napps.kytos.topology.main.KytosEvent')
328		@patch('kytos.core.buffers.KytosEventBuffer.put')
329		def test_notify_link_status_change(self, *args):
330		"""Test notify link status change."""
331		(mock_buffers_put, mock_event) = args
332		mock_link = create_autospec(Link)
333		self.napp.notify_link_status_change(mock_link)
334		mock_event.assert_called()
335		mock_buffers_put.assert_called()
336
337		@patch('napps.kytos.topology.main.KytosEvent')
338		@patch('kytos.core.buffers.KytosEventBuffer.put')
339		@patch('napps.kytos.topology.main.isinstance')
340		def test_notify_metadata_changes(self, *args):
341		"""Test notify metadata changes."""
342		(mock_isinstance, mock_buffers_put, mock_event) = args
343		mock_isinstance.return_value = True
344		mock_obj = MagicMock()
345		mock_action = create_autospec(Switch)
346		self.napp.notify_metadata_changes(mock_obj, mock_action)
347		mock_event.assert_called()
348		mock_isinstance.assert_called()
349		mock_buffers_put.assert_called()
350
351		@patch('napps.kytos.topology.main.KytosEvent')
352		@patch('kytos.core.buffers.KytosEventBuffer.put')
353		def test_notify_port_created(self, *args):
354		"""Test notify port created."""
355		(mock_buffers_put, mock_kytos_event) = args
356		mock_event = MagicMock()
357		self.napp.notify_port_created(mock_event)
358		mock_kytos_event.assert_called()
359		mock_buffers_put.assert_called()
360
361		@patch('napps.kytos.topology.main.KytosEvent')
362		@patch('kytos.core.buffers.KytosEventBuffer.put')
363		def test_verify_storehouse(self, *args):
364		"""Test verify_storehouse."""
365		(mock_buffers_put, mock_kytos_event) = args
366		mock_entities = MagicMock()
367		self.napp.verify_storehouse(mock_entities)
368		mock_buffers_put.assert_called()
369		mock_kytos_event.assert_called()
370
371		@patch('napps.kytos.topology.main.Main.notify_link_status_change')
372		def test_handle_link_maintenance_start(self, status_change_mock):
373		"""Test handle_link_maintenance_start."""
374		link1 = MagicMock()
375		link1.id = 2
376		link2 = MagicMock()
377		link2.id = 3
378		link3 = MagicMock()
379		link3.id = 4
380		content = {'links': [link1, link2]}
381		event = MagicMock()
382		event.content = content
383		self.napp.links = {2: link1, 4: link3}
384		self.napp.handle_link_maintenance_start(event)
385		status_change_mock.assert_called_once_with(link1)
386
387		@patch('napps.kytos.topology.main.Main.notify_link_status_change')
388		def test_handle_link_maintenance_end(self, status_change_mock):
389		"""Test handle_link_maintenance_end."""
390		link1 = MagicMock()
391		link1.id = 2
392		link2 = MagicMock()
393		link2.id = 3
394		link3 = MagicMock()
395		link3.id = 4
396		content = {'links': [link1, link2]}
397		event = MagicMock()
398		event.content = content
399		self.napp.links = {2: link1, 4: link3}
400		self.napp.handle_link_maintenance_end(event)
401		status_change_mock.assert_called_once_with(link1)
402
403	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...
404		def test_handle_switch_maintenance_start(self, handle_link_down_mock):
405		"""Test handle_switch_maintenance_start."""
406		switch1 = MagicMock()
407		interface1 = MagicMock()
408		interface1.is_active.return_value = True
409		interface2 = MagicMock()
410		interface2.is_active.return_value = False
411		interface3 = MagicMock()
412		interface3.is_active.return_value = True
413		switch1.interfaces = {1: interface1, 2: interface2, 3: interface3}
414		switch2 = MagicMock()
415		interface4 = MagicMock()
416		interface4.is_active.return_value = False
417		interface5 = MagicMock()
418		interface5.is_active.return_value = True
419		switch2.interfaces = {1: interface4, 2: interface5}
420		content = {'switches': [switch1, switch2]}
421		event = MagicMock()
422		event.content = content
423		self.napp.handle_switch_maintenance_start(event)
424		self.assertEqual(handle_link_down_mock.call_count, 3)
425
426	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...
427		def test_handle_switch_maintenance_end(self, handle_link_up_mock):
428		"""Test handle_switch_maintenance_end."""
429		switch1 = MagicMock()
430		interface1 = MagicMock()
431		interface1.is_active.return_value = True
432		interface2 = MagicMock()
433		interface2.is_active.return_value = False
434		interface3 = MagicMock()
435		interface3.is_active.return_value = True
436		switch1.interfaces = {1: interface1, 2: interface2, 3: interface3}
437		switch2 = MagicMock()
438		interface4 = MagicMock()
439		interface4.is_active.return_value = False
440		interface5 = MagicMock()
441		interface5.is_active.return_value = True
442		switch2.interfaces = {1: interface4, 2: interface5}
443		content = {'switches': [switch1, switch2]}
444		event = MagicMock()
445		event.content = content
446		self.napp.handle_switch_maintenance_end(event)
447		self.assertEqual(handle_link_up_mock.call_count, 5)
448

kytos / topology

Push — master ( 4fca5c...e35304 )

TestMain.test_handle_link_maintenance_end() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like