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

Passed

Pull Request — master (#80)

unknown

created 2020-05-14 22:24 UTC

TestMain.test_disable_interfaces() B

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	48
Code Lines	41

Duplication

Lines	48
Ratio	100 %

Importance

Changes

Metric	Value
cc	1
eloc	41
nop	1
dl	48
loc	48
rs	8.896
c	0
b	0
f	0

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

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


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


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

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

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

        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        from napps.kytos.topology.main import Main
        self.addCleanup(patch.stopall)

        self.napp = Main(get_controller_mock())

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

    def test_verify_api_urls(self):

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

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

    def test_enable_interfaces(self):

        """Test enable_interfaces."""
        mock_switch = create_autospec(Switch)
        mock_interface_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

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

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

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

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

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

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

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

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

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


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

kytos / topology

Pull Request — master (#80)

TestMain.test_disable_interfaces() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like