build.tests.test_main.TestMain.test_create_circuit_already_enabled() - Code Metrics - Inspection of "Merge pull request #137 from ajoaoff/fix_enabled" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 087782...9ad0e5 )

by Antonio

created 2019-06-13 17:13 UTC

TestMain.test_create_circuit_already_enabled() B

↳ Parent: build.tests.test_main

Complexity

Conditions

Size

Total Lines	49
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	38
nop	2
dl	0
loc	49
rs	8.968
c	0
b	0
f	0

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

from napps.kytos.mef_eline.main import Main
from tests.helpers import get_controller_mock


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

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

        Set the server_name_url_url from kytos/mef_eline
        """
        self.server_name_url = 'http://localhost:8181/api/kytos/mef_eline'
        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/mef_eline',
                           'kytos/topology.link_up',
                           'kytos/topology.link_down']
        actual_events = self.napp.listeners()

        for _event in expected_events:
            self.assertIn(_event, actual_events, '%s' % _event)

    def test_verify_api_urls(self):
        """Verify all APIs registered."""
        expected_urls = [
            ({}, {'OPTIONS', 'POST'},
             '/api/kytos/mef_eline/v2/evc/'),
            ({}, {'OPTIONS', 'HEAD', 'GET'},
             '/api/kytos/mef_eline/v2/evc/'),
            ({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'DELETE'},
             '/api/kytos/mef_eline/v2/evc/<circuit_id>'),
            ({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'HEAD', 'GET'},
             '/api/kytos/mef_eline/v2/evc/<circuit_id>'),
            ({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'PATCH'},
             '/api/kytos/mef_eline/v2/evc/<circuit_id>')]
        urls = self.get_napp_urls(self.napp)
        self.assertEqual(expected_urls, urls)

    def test_list_without_circuits(self):
        """Test if list circuits return 'no circuit stored.'."""
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/'
        response = api.get(url)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(json.loads(response.data.decode()), {})

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    def test_list_no_circuits_stored(self, storehouse_data_mock):
        """Test if list circuits return all circuits stored."""
        circuits = {}
        storehouse_data_mock.return_value = circuits

        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/'

        response = api.get(url)
        expected_result = circuits
        self.assertEqual(json.loads(response.data), expected_result)

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    def test_list_with_circuits_stored(self, storehouse_data_mock):
        """Test if list circuits return all circuits stored."""
        circuits = {'1': {'name': 'circuit_1'},
                    '2': {'name': 'circuit_2'}}
        storehouse_data_mock.return_value = circuits

        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/'

        response = api.get(url)
        expected_result = circuits
        self.assertEqual(json.loads(response.data), expected_result)

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')

    def test_circuit_with_valid_id(self, storehouse_data_mock):
        """Test if get_circuit return the circuit attributes."""
        circuits = {'1': {'name': 'circuit_1'},
                    '2': {'name': 'circuit_2'}}
        storehouse_data_mock.return_value = circuits

        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/1'
        response = api.get(url)
        expected_result = circuits['1']
        self.assertEqual(json.loads(response.data), expected_result)

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')

    def test_circuit_with_invalid_id(self, storehouse_data_mock):
        """Test if get_circuit return invalid circuit_id."""
        circuits = {'1': {'name': 'circuit_1'},
                    '2': {'name': 'circuit_2'}}
        storehouse_data_mock.return_value = circuits

        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/3'
        response = api.get(url)
        expected_result = {'response': 'circuit_id 3 not found'}
        self.assertEqual(json.loads(response.data), expected_result)

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    @patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
    @patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
    @patch('napps.kytos.mef_eline.main.EVC.as_dict')
    @patch('napps.kytos.mef_eline.models.EVC._validate')
    def test_create_a_circuit_case_1(self, *args):
        """Test create a new circuit."""
        (validate_mock, evc_as_dict_mock, save_evc_mock,
         uni_from_dict_mock, sched_add_mock, storehouse_data_mock) = args

        validate_mock.return_value = True
        save_evc_mock.return_value = True
        uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
        evc_as_dict_mock.return_value = {}
        sched_add_mock.return_value = True
        storehouse_data_mock.return_value = {}

        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/'
        payload = {
                   "name": "my evc1",
                   "frequency": "* * * * *",
                   "uni_a": {
                     "interface_id": "00:00:00:00:00:00:00:01:1",
                     "tag": {
                       "tag_type": 1,
                       "value": 80
                     }
                   },
                   "uni_z": {
                     "interface_id": "00:00:00:00:00:00:00:02:2",
                     "tag": {
                       "tag_type": 1,
                       "value": 1
                     }
                   }
                 }

        response = api.post(url, data=json.dumps(payload),
                            content_type='application/json')
        current_data = json.loads(response.data)

        # verify expected result from request
        self.assertEqual(201, response.status_code, response.data)
        self.assertIn('circuit_id', current_data)

        # verify uni called
        uni_from_dict_mock.called_twice()
        uni_from_dict_mock.assert_any_call(payload['uni_z'])
        uni_from_dict_mock.assert_any_call(payload['uni_a'])

        # verify validation called
        validate_mock.assert_called_once()
        validate_mock.assert_called_with(frequency='* * * * *',
                                         name='my evc1',
                                         uni_a='uni_a',
                                         uni_z='uni_z')
        # verify save method is called
        save_evc_mock.assert_called_once()

        # verify evc as dict is called to save in the box
        evc_as_dict_mock.assert_called_once()
        # verify add circuit in sched
        sched_add_mock.assert_called_once()

    @staticmethod
    def get_napp_urls(napp):
        """Return the kytos/mef_eline urls.

        The urls will be like:

        urls = [
            (options, methods, url)
        ]

        """
        controller = napp.controller
        controller.api_server.register_napp_endpoints(napp)

        urls = []
        for rule in controller.api_server.app.url_map.iter_rules():
            options = {}
            for arg in rule.arguments:
                options[arg] = "[{0}]".format(arg)

            if f'{napp.username}/{napp.name}' in str(rule):
                urls.append((options, rule.methods, f'{str(rule)}'))

        return urls

    @staticmethod
    def get_app_test_client(napp):
        """Return a flask api test client."""
        napp.controller.api_server.register_napp_endpoints(napp)
        return napp.controller.api_server.app.test_client()

    def test_create_a_circuit_case_2(self):
        """Test create a new circuit trying to send request without a json."""
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/'

        response = api.post(url)
        current_data = json.loads(response.data)
        expected_data = 'Bad request: The request do not have a json.'

        self.assertEqual(400, response.status_code)
        self.assertEqual(current_data, expected_data)

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    @patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
    @patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
    @patch('napps.kytos.mef_eline.models.EVC._validate')
    @patch('napps.kytos.mef_eline.main.EVC.as_dict')
    def test_create_circuit_already_enabled(self, *args):
        """Test create an already created circuit."""
        (evc_as_dict_mock, validate_mock, save_evc_mock,
         uni_from_dict_mock, sched_add_mock, storehouse_data_mock) = args

        validate_mock.return_value = True
        save_evc_mock.return_value = True
        sched_add_mock.return_value = True
        uni_from_dict_mock.side_effect = ['uni_a', 'uni_z', 'uni_a', 'uni_z']
        circuits = {'1': {'name': 'circuit_1'}}

        api = self.get_app_test_client(self.napp)
        payload = {
            "name": "my evc1",
            "frequency": "* * * * *",
            "uni_a": {
                "interface_id": "00:00:00:00:00:00:00:01:1",
                "tag": {
                    "tag_type": 1,
                    "value": 80
                }
            },
            "uni_z": {
                "interface_id": "00:00:00:00:00:00:00:02:2",
                "tag": {
                    "tag_type": 1,
                    "value": 1
                }
            }
        }

        circuits.update({2: payload})
        storehouse_data_mock.return_value = circuits
        evc_as_dict_mock.return_value = payload

        response = api.post(f'{self.server_name_url}/v2/evc/',
                            data=json.dumps(payload),
                            content_type='application/json')
        current_data = json.loads(response.data)
        expected_data = 'Not Acceptable: This evc already exists.'

        self.assertEqual(current_data, expected_data)
        self.assertEqual(409, response.status_code)


1		"""Module to test the main napp file."""
2		import json
3		from unittest import TestCase
4		from unittest.mock import patch
5
6		from napps.kytos.mef_eline.main import Main
7		from tests.helpers import get_controller_mock
8
9
10		class TestMain(TestCase):
11		"""Test the Main class."""
12
13		def setUp(self):
14		"""Execute steps before each tests.
15
16		Set the server_name_url_url from kytos/mef_eline
17		"""
18		self.server_name_url = 'http://localhost:8181/api/kytos/mef_eline'
19		self.napp = Main(get_controller_mock())
20
21		def test_get_event_listeners(self):
22		"""Verify all event listeners registered."""
23		expected_events = ['kytos/core.shutdown',
24		'kytos/core.shutdown.kytos/mef_eline',
25		'kytos/topology.link_up',
26		'kytos/topology.link_down']
27		actual_events = self.napp.listeners()
28
29		for _event in expected_events:
30		self.assertIn(_event, actual_events, '%s' % _event)
31
32		def test_verify_api_urls(self):
33		"""Verify all APIs registered."""
34		expected_urls = [
35		({}, {'OPTIONS', 'POST'},
36		'/api/kytos/mef_eline/v2/evc/'),
37		({}, {'OPTIONS', 'HEAD', 'GET'},
38		'/api/kytos/mef_eline/v2/evc/'),
39		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'DELETE'},
40		'/api/kytos/mef_eline/v2/evc/<circuit_id>'),
41		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'HEAD', 'GET'},
42		'/api/kytos/mef_eline/v2/evc/<circuit_id>'),
43		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'PATCH'},
44		'/api/kytos/mef_eline/v2/evc/<circuit_id>')]
45		urls = self.get_napp_urls(self.napp)
46		self.assertEqual(expected_urls, urls)
47
48		def test_list_without_circuits(self):
49		"""Test if list circuits return 'no circuit stored.'."""
50		api = self.get_app_test_client(self.napp)
51		url = f'{self.server_name_url}/v2/evc/'
52		response = api.get(url)
53		self.assertEqual(response.status_code, 200)
54		self.assertEqual(json.loads(response.data.decode()), {})
55
56		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
57		def test_list_no_circuits_stored(self, storehouse_data_mock):
58		"""Test if list circuits return all circuits stored."""
59		circuits = {}
60		storehouse_data_mock.return_value = circuits
61
62		api = self.get_app_test_client(self.napp)
63		url = f'{self.server_name_url}/v2/evc/'
64
65		response = api.get(url)
66		expected_result = circuits
67		self.assertEqual(json.loads(response.data), expected_result)
68
69		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
70		def test_list_with_circuits_stored(self, storehouse_data_mock):
71		"""Test if list circuits return all circuits stored."""
72		circuits = {'1': {'name': 'circuit_1'},
73		'2': {'name': 'circuit_2'}}
74		storehouse_data_mock.return_value = circuits
75
76		api = self.get_app_test_client(self.napp)
77		url = f'{self.server_name_url}/v2/evc/'
78
79		response = api.get(url)
80		expected_result = circuits
81		self.assertEqual(json.loads(response.data), expected_result)
82
83	View Code Duplication	@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
		0 ignored issues – show Duplication introduced 2018-09-04 18:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
84		def test_circuit_with_valid_id(self, storehouse_data_mock):
85		"""Test if get_circuit return the circuit attributes."""
86		circuits = {'1': {'name': 'circuit_1'},
87		'2': {'name': 'circuit_2'}}
88		storehouse_data_mock.return_value = circuits
89
90		api = self.get_app_test_client(self.napp)
91		url = f'{self.server_name_url}/v2/evc/1'
92		response = api.get(url)
93		expected_result = circuits['1']
94		self.assertEqual(json.loads(response.data), expected_result)
95
96	View Code Duplication	@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
		0 ignored issues – show Duplication introduced 2018-08-28 17:42 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
97		def test_circuit_with_invalid_id(self, storehouse_data_mock):
98		"""Test if get_circuit return invalid circuit_id."""
99		circuits = {'1': {'name': 'circuit_1'},
100		'2': {'name': 'circuit_2'}}
101		storehouse_data_mock.return_value = circuits
102
103		api = self.get_app_test_client(self.napp)
104		url = f'{self.server_name_url}/v2/evc/3'
105		response = api.get(url)
106		expected_result = {'response': 'circuit_id 3 not found'}
107		self.assertEqual(json.loads(response.data), expected_result)
108
109		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
110		@patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
111		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
112		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
113		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
114		@patch('napps.kytos.mef_eline.models.EVC._validate')
115		def test_create_a_circuit_case_1(self, *args):
116		"""Test create a new circuit."""
117		(validate_mock, evc_as_dict_mock, save_evc_mock,
118		uni_from_dict_mock, sched_add_mock, storehouse_data_mock) = args
119
120		validate_mock.return_value = True
121		save_evc_mock.return_value = True
122		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
123		evc_as_dict_mock.return_value = {}
124		sched_add_mock.return_value = True
125		storehouse_data_mock.return_value = {}
126
127		api = self.get_app_test_client(self.napp)
128		url = f'{self.server_name_url}/v2/evc/'
129		payload = {
130		"name": "my evc1",
131		"frequency": "* * * * *",
132		"uni_a": {
133		"interface_id": "00:00:00:00:00:00:00:01:1",
134		"tag": {
135		"tag_type": 1,
136		"value": 80
137		}
138		},
139		"uni_z": {
140		"interface_id": "00:00:00:00:00:00:00:02:2",
141		"tag": {
142		"tag_type": 1,
143		"value": 1
144		}
145		}
146		}
147
148		response = api.post(url, data=json.dumps(payload),
149		content_type='application/json')
150		current_data = json.loads(response.data)
151
152		# verify expected result from request
153		self.assertEqual(201, response.status_code, response.data)
154		self.assertIn('circuit_id', current_data)
155
156		# verify uni called
157		uni_from_dict_mock.called_twice()
158		uni_from_dict_mock.assert_any_call(payload['uni_z'])
159		uni_from_dict_mock.assert_any_call(payload['uni_a'])
160
161		# verify validation called
162		validate_mock.assert_called_once()
163		validate_mock.assert_called_with(frequency='* * * * *',
164		name='my evc1',
165		uni_a='uni_a',
166		uni_z='uni_z')
167		# verify save method is called
168		save_evc_mock.assert_called_once()
169
170		# verify evc as dict is called to save in the box
171		evc_as_dict_mock.assert_called_once()
172		# verify add circuit in sched
173		sched_add_mock.assert_called_once()
174
175		@staticmethod
176		def get_napp_urls(napp):
177		"""Return the kytos/mef_eline urls.
178
179		The urls will be like:
180
181		urls = [
182		(options, methods, url)
183		]
184
185		"""
186		controller = napp.controller
187		controller.api_server.register_napp_endpoints(napp)
188
189		urls = []
190		for rule in controller.api_server.app.url_map.iter_rules():
191		options = {}
192		for arg in rule.arguments:
193		options[arg] = "[{0}]".format(arg)
194
195		if f'{napp.username}/{napp.name}' in str(rule):
196		urls.append((options, rule.methods, f'{str(rule)}'))
197
198		return urls
199
200		@staticmethod
201		def get_app_test_client(napp):
202		"""Return a flask api test client."""
203		napp.controller.api_server.register_napp_endpoints(napp)
204		return napp.controller.api_server.app.test_client()
205
206		def test_create_a_circuit_case_2(self):
207		"""Test create a new circuit trying to send request without a json."""
208		api = self.get_app_test_client(self.napp)
209		url = f'{self.server_name_url}/v2/evc/'
210
211		response = api.post(url)
212		current_data = json.loads(response.data)
213		expected_data = 'Bad request: The request do not have a json.'
214
215		self.assertEqual(400, response.status_code)
216		self.assertEqual(current_data, expected_data)
217
218		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
219		@patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
220		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
221		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
222		@patch('napps.kytos.mef_eline.models.EVC._validate')
223		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
224		def test_create_circuit_already_enabled(self, *args):
225		"""Test create an already created circuit."""
226		(evc_as_dict_mock, validate_mock, save_evc_mock,
227		uni_from_dict_mock, sched_add_mock, storehouse_data_mock) = args
228
229		validate_mock.return_value = True
230		save_evc_mock.return_value = True
231		sched_add_mock.return_value = True
232		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z', 'uni_a', 'uni_z']
233		circuits = {'1': {'name': 'circuit_1'}}
234
235		api = self.get_app_test_client(self.napp)
236		payload = {
237		"name": "my evc1",
238		"frequency": "* * * * *",
239		"uni_a": {
240		"interface_id": "00:00:00:00:00:00:00:01:1",
241		"tag": {
242		"tag_type": 1,
243		"value": 80
244		}
245		},
246		"uni_z": {
247		"interface_id": "00:00:00:00:00:00:00:02:2",
248		"tag": {
249		"tag_type": 1,
250		"value": 1
251		}
252		}
253		}
254
255		circuits.update({2: payload})
256		storehouse_data_mock.return_value = circuits
257		evc_as_dict_mock.return_value = payload
258
259		response = api.post(f'{self.server_name_url}/v2/evc/',
260		data=json.dumps(payload),
261		content_type='application/json')
262		current_data = json.loads(response.data)
263		expected_data = 'Not Acceptable: This evc already exists.'
264
265		self.assertEqual(current_data, expected_data)
266		self.assertEqual(409, response.status_code)
267

kytos / mef_eline

Push — master ( 087782...9ad0e5 )

TestMain.test_create_circuit_already_enabled() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like