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

Passed

Pull Request — master (#99)

unknown

created 2020-05-19 02:57 UTC

TestMain.test_enable_switch() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	24
Code Lines	20

Duplication

Lines	25
Ratio	104.17 %

Importance

Changes

Metric	Value
cc	1
eloc	20
nop	1
dl	25
loc	24
rs	9.4
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 kytos.lib.helpers import get_switch_mock, get_test_client


from tests.unit.helpers import get_controller_mock, get_napp_urls


# pylint: disable=too-many-public-methods
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_switch(self):

        """Test enable_swicth."""
        dpid = "00:00:00:00:00:00:00:01"
        mock_switch = get_switch_mock(dpid)
        msg_success = "Operation successful"
        msg_fail = "Switch not found"
        self.napp.controller.switches = {"00:00:00:00:00:00:00:01":
                                         mock_switch}
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v3/switches/{dpid}/enable'
        response = api.post(url)
        self.assertEqual(response.status_code, 201, response.data)
        self.assertEqual(msg_success, json.loads(response.data))
        self.assertEqual(mock_switch.enable.call_count, 1)

        # fail case
        mock_switch.enable.call_count = 0
        dpid = "00:00:00:00:00:00:00:02"
        url = f'{self.server_name_url}/v3/switches/{dpid}/enable'
        response = api.post(url)
        self.assertEqual(response.status_code, 404, response.data)
        self.assertEqual(msg_fail, json.loads(response.data))
        self.assertEqual(mock_switch.enable.call_count, 0)

    def test_disable_switch(self):

        """Test disable_swicth."""
        dpid = "00:00:00:00:00:00:00:01"
        mock_switch = get_switch_mock(dpid)
        msg_success = "Operation successful"
        msg_fail = "Switch not found"
        self.napp.controller.switches = {dpid: mock_switch}
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v3/switches/{dpid}/disable'
        response = api.post(url)
        self.assertEqual(response.status_code, 201, response.data)
        self.assertEqual(msg_success, json.loads(response.data))
        self.assertEqual(mock_switch.disable.call_count, 1)

        # fail case
        mock_switch.disable.call_count = 0
        dpid = "00:00:00:00:00:00:00:02"
        url = f'{self.server_name_url}/v3/switches/{dpid}/disable'
        response = api.post(url)
        self.assertEqual(response.status_code, 404, response.data)
        self.assertEqual(msg_fail, json.loads(response.data))
        self.assertEqual(mock_switch.disable.call_count, 0)

    def test_get_switch_metadata(self):
        """Test get_switch_metadata."""
        dpid = "00:00:00:00:00:00:00:01"
        mock_switch = get_switch_mock(dpid)
        mock_switch.metadata = "A"
        self.napp.controller.switches = {dpid: mock_switch}
        api = get_test_client(self.napp.controller, self.napp)

        url = f'{self.server_name_url}/v3/switches/{dpid}/metadata'
        response = api.get(url)
        self.assertEqual(response.status_code, 200, response.data)

        # fail case
        dpid = "00:00:00:00:00:00:00:02"
        url = f'{self.server_name_url}/v3/switches/{dpid}/metadata'
        response = api.get(url)
        self.assertEqual(response.status_code, 404, response.data)

    @patch('napps.kytos.topology.main.Main.notify_metadata_changes')
    def test_add_switch_metadata(self, mock_metadata_changes):
        """Test add_switch_metadata."""
        dpid = "00:00:00:00:00:00:00:01"
        mock_switch = get_switch_mock(dpid)
        self.napp.controller.switches = {dpid: mock_switch}
        api = get_test_client(self.napp.controller, self.napp)
        payload = {"data": "A"}

        url = f'{self.server_name_url}/v3/switches/{dpid}/metadata'
        response = api.post(url, data=json.dumps(payload),
                            content_type='application/json')
        self.assertEqual(response.status_code, 201, response.data)
        mock_metadata_changes.assert_called()

        # fail case
        dpid = "00:00:00:00:00:00:00:02"
        url = f'{self.server_name_url}/v3/switches/{dpid}/metadata'
        response = api.post(url, data=json.dumps(payload),
                            content_type='application/json')
        self.assertEqual(response.status_code, 404, response.data)

    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_test_client(self.napp.controller, 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_test_client(self.napp.controller, 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

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

kytos / topology

Pull Request — master (#99)

TestMain.test_enable_switch() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like