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

Test Failed

Push — master ( 98eee7...9ba056 )

by Antonio

created 2019-09-03 18:06 UTC

TestMain.test_load_circuits_by_interface() B

↳ Parent: build.tests.test_main

Complexity

Conditions

Size

Total Lines	189
Code Lines	155

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	155
nop	1
dl	0
loc	189
rs	7
c	0
b	0
f	0

How to fix Long Method

"""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.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) = 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']
        payload1 = {'name': 'circuit_1'}

        api = self.get_app_test_client(self.napp)
        payload2 = {
            "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
                }
            }
        }

        evc_as_dict_mock.return_value = payload1
        response = api.post(f'{self.server_name_url}/v2/evc/',
                            data=json.dumps(payload1),
                            content_type='application/json')
        self.assertEqual(201, response.status_code)

        evc_as_dict_mock.return_value = payload2
        response = api.post(f'{self.server_name_url}/v2/evc/',
                            data=json.dumps(payload2),
                            content_type='application/json')
        self.assertEqual(201, response.status_code)

        response = api.post(f'{self.server_name_url}/v2/evc/',
                            data=json.dumps(payload2),
                            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)

    def test_load_circuits_by_interface(self):
        """Test if existing circuits are correctly loaded to the cache."""
        stored_circuits = {
            "182f5bac84074017a262a2321195dbb4": {
                "active": False,
                "archived": True,
                "backup_links": [],
                "backup_path": [],
                "bandwidth": 0,
                "circuit_scheduler": [
                    {
                        "action": "create",
                        "frequency": "*/3 * * * *",
                        "id": "db7f8a301e2b4ff69a2ad9a6267430e2"
                    },
                    {
                        "action": "remove",
                        "frequency": "2-59/3 * * * *",
                        "id": "b8a8bbe85bc144b0afc65181e4c069a1"
                    }
                ],
                "creation_time": "2019-08-09T19:25:06",
                "current_path": [],
                "dynamic_backup_path": True,
                "enabled": False,
                "end_date": "2018-12-29T15:16:50",
                "id": "182f5bac84074017a262a2321195dbb4",
                "name": "Teste2",
                "owner": None,
                "primary_links": [],
                "primary_path": [],
                "priority": 0,
                "request_time": "2019-08-09T19:25:06",
                "start_date": "2019-08-09T19:25:06",
                "uni_a": {
                    "interface_id": "00:00:00:00:00:00:00:03:12",
                    "tag": {
                        "tag_type": 1,
                        "value": 321
                    }
                },
                "uni_z": {
                    "interface_id": "00:00:00:00:00:00:00:06:11",
                    "tag": {
                        "tag_type": 1,
                        "value": 612
                    }
                }
            },
            "65c4582cc8f249c2a5947ef500c19e37": {
                "active": False,
                "archived": False,
                "backup_links": [],
                "backup_path": [],
                "bandwidth": 0,
                "circuit_scheduler": [
                    {
                        "action": "create",
                        "frequency": "*/3 * * * *",
                        "id": "0939dedf66ce431f85beb53daf578d73"
                    },
                    {
                        "action": "remove",
                        "frequency": "2-59/3 * * * *",
                        "id": "6cdcab31a11f44708e23776b4dad7893"
                    }
                ],
                "creation_time": "2019-07-22T16:01:24",
                "current_path": [],
                "dynamic_backup_path": True,
                "enabled": False,
                "end_date": "2018-12-29T15:16:50",
                "id": "65c4582cc8f249c2a5947ef500c19e37",
                "name": "Teste2",
                "owner": None,
                "primary_links": [],
                "primary_path": [
                    {
                        "active": False,
                        "enabled": True,
                        "endpoint_a": {
                            "active": False,
                            "enabled": True,
                            "id": "00:00:00:00:00:00:00:03:3",
                            "link": "0e2b5d7bc858b9f38db11b69",
                            "mac": "ae:6e:d3:96:83:5a",
                            "metadata": {},
                            "name": "s3-eth3",
                            "nni": True,
                            "port_number": 3,
                            "speed": 1250000000.0,
                            "switch": "00:00:00:00:00:00:00:03",
                            "type": "interface",
                            "uni": False
                        },
                        "endpoint_b": {
                            "active": False,
                            "enabled": True,
                            "id": "00:00:00:00:00:00:00:05:2",
                            "link": "0e2b5d7bc858b9f38db11b69",
                            "mac": "de:eb:d0:b0:14:cf",
                            "metadata": {},
                            "name": "s5-eth2",
                            "nni": True,
                            "port_number": 2,
                            "speed": 1250000000.0,
                            "switch": "00:00:00:00:00:00:00:05",
                            "type": "interface",
                            "uni": False
                        },
                        "id": "0e2b5d7bc858b9f38db11b69",
                        "metadata": {}
                    },
                    {
                        "active": False,
                        "enabled": True,
                        "endpoint_a": {
                            "active": False,
                            "enabled": True,
                            "id": "00:00:00:00:00:00:00:05:4",
                            "link": "53bd36ff55a5aa2029bd5d50",
                            "mac": "6e:c2:ea:c4:18:12",
                            "metadata": {},
                            "name": "s5-eth4",
                            "nni": True,
                            "port_number": 4,
                            "speed": 1250000000.0,
                            "switch": "00:00:00:00:00:00:00:05",
                            "type": "interface",
                            "uni": False
                        },
                        "endpoint_b": {
                            "active": False,
                            "enabled": True,
                            "id": "00:00:00:00:00:00:00:06:2",
                            "link": "53bd36ff55a5aa2029bd5d50",
                            "mac": "5a:25:7b:7c:0d:ac",
                            "metadata": {},
                            "name": "s6-eth2",
                            "nni": True,
                            "port_number": 2,
                            "speed": 1250000000.0,
                            "switch": "00:00:00:00:00:00:00:06",
                            "type": "interface",
                            "uni": False
                        },
                        "id": "53bd36ff55a5aa2029bd5d50",
                        "metadata": {}
                    }
                ],
                "priority": 0,
                "request_time": "2019-07-22T16:01:24",
                "start_date": "2019-07-22T16:01:24",
                "uni_a": {
                    "interface_id": "00:00:00:00:00:00:00:03:12",
                    "tag": {
                        "tag_type": 1,
                        "value": 321
                    }
                },
                "uni_z": {
                    "interface_id": "00:00:00:00:00:00:00:06:11",
                    "tag": {
                        "tag_type": 1,
                        "value": 612
                    }
                }
            }
        }

        expected_result = {
            '00:00:00:00:00:00:00:03:12':
                {'182f5bac84074017a262a2321195dbb4',
                 '65c4582cc8f249c2a5947ef500c19e37'},
            '00:00:00:00:00:00:00:06:11':
                {'182f5bac84074017a262a2321195dbb4',
                 '65c4582cc8f249c2a5947ef500c19e37'},
            '00:00:00:00:00:00:00:03:3':
                {'65c4582cc8f249c2a5947ef500c19e37'},
            '00:00:00:00:00:00:00:05:2':
                {'65c4582cc8f249c2a5947ef500c19e37'},
            '00:00:00:00:00:00:00:05:4':
                {'65c4582cc8f249c2a5947ef500c19e37'},
            '00:00:00:00:00:00:00:06:2':
                {'65c4582cc8f249c2a5947ef500c19e37'}
        }
        self.napp.load_circuits_by_interface(stored_circuits)
        # pylint: disable=protected-access
        self.assertEqual(self.napp._circuits_by_interface, expected_result)


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.scheduler.Scheduler.add')
219		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
220		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
221		@patch('napps.kytos.mef_eline.models.EVC._validate')
222		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
223		def test_create_circuit_already_enabled(self, *args):
224		"""Test create an already created circuit."""
225		(evc_as_dict_mock, validate_mock, save_evc_mock,
226		uni_from_dict_mock, sched_add_mock) = args
227
228		validate_mock.return_value = True
229		save_evc_mock.return_value = True
230		sched_add_mock.return_value = True
231		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z', 'uni_a', 'uni_z']
232		payload1 = {'name': 'circuit_1'}
233
234		api = self.get_app_test_client(self.napp)
235		payload2 = {
236		"name": "my evc1",
237		"frequency": "* * * * *",
238		"uni_a": {
239		"interface_id": "00:00:00:00:00:00:00:01:1",
240		"tag": {
241		"tag_type": 1,
242		"value": 80
243		}
244		},
245		"uni_z": {
246		"interface_id": "00:00:00:00:00:00:00:02:2",
247		"tag": {
248		"tag_type": 1,
249		"value": 1
250		}
251		}
252		}
253
254		evc_as_dict_mock.return_value = payload1
255		response = api.post(f'{self.server_name_url}/v2/evc/',
256		data=json.dumps(payload1),
257		content_type='application/json')
258		self.assertEqual(201, response.status_code)
259
260		evc_as_dict_mock.return_value = payload2
261		response = api.post(f'{self.server_name_url}/v2/evc/',
262		data=json.dumps(payload2),
263		content_type='application/json')
264		self.assertEqual(201, response.status_code)
265
266		response = api.post(f'{self.server_name_url}/v2/evc/',
267		data=json.dumps(payload2),
268		content_type='application/json')
269		current_data = json.loads(response.data)
270		expected_data = 'Not Acceptable: This evc already exists.'
271		self.assertEqual(current_data, expected_data)
272		self.assertEqual(409, response.status_code)
273
274		def test_load_circuits_by_interface(self):
275		"""Test if existing circuits are correctly loaded to the cache."""
276		stored_circuits = {
277		"182f5bac84074017a262a2321195dbb4": {
278		"active": False,
279		"archived": True,
280		"backup_links": [],
281		"backup_path": [],
282		"bandwidth": 0,
283		"circuit_scheduler": [
284		{
285		"action": "create",
286		"frequency": "/3 * * *",
287		"id": "db7f8a301e2b4ff69a2ad9a6267430e2"
288		},
289		{
290		"action": "remove",
291		"frequency": "2-59/3 * * * *",
292		"id": "b8a8bbe85bc144b0afc65181e4c069a1"
293		}
294		],
295		"creation_time": "2019-08-09T19:25:06",
296		"current_path": [],
297		"dynamic_backup_path": True,
298		"enabled": False,
299		"end_date": "2018-12-29T15:16:50",
300		"id": "182f5bac84074017a262a2321195dbb4",
301		"name": "Teste2",
302		"owner": None,
303		"primary_links": [],
304		"primary_path": [],
305		"priority": 0,
306		"request_time": "2019-08-09T19:25:06",
307		"start_date": "2019-08-09T19:25:06",
308		"uni_a": {
309		"interface_id": "00:00:00:00:00:00:00:03:12",
310		"tag": {
311		"tag_type": 1,
312		"value": 321
313		}
314		},
315		"uni_z": {
316		"interface_id": "00:00:00:00:00:00:00:06:11",
317		"tag": {
318		"tag_type": 1,
319		"value": 612
320		}
321		}
322		},
323		"65c4582cc8f249c2a5947ef500c19e37": {
324		"active": False,
325		"archived": False,
326		"backup_links": [],
327		"backup_path": [],
328		"bandwidth": 0,
329		"circuit_scheduler": [
330		{
331		"action": "create",
332		"frequency": "/3 * * *",
333		"id": "0939dedf66ce431f85beb53daf578d73"
334		},
335		{
336		"action": "remove",
337		"frequency": "2-59/3 * * * *",
338		"id": "6cdcab31a11f44708e23776b4dad7893"
339		}
340		],
341		"creation_time": "2019-07-22T16:01:24",
342		"current_path": [],
343		"dynamic_backup_path": True,
344		"enabled": False,
345		"end_date": "2018-12-29T15:16:50",
346		"id": "65c4582cc8f249c2a5947ef500c19e37",
347		"name": "Teste2",
348		"owner": None,
349		"primary_links": [],
350		"primary_path": [
351		{
352		"active": False,
353		"enabled": True,
354		"endpoint_a": {
355		"active": False,
356		"enabled": True,
357		"id": "00:00:00:00:00:00:00:03:3",
358		"link": "0e2b5d7bc858b9f38db11b69",
359		"mac": "ae:6e:d3:96:83:5a",
360		"metadata": {},
361		"name": "s3-eth3",
362		"nni": True,
363		"port_number": 3,
364		"speed": 1250000000.0,
365		"switch": "00:00:00:00:00:00:00:03",
366		"type": "interface",
367		"uni": False
368		},
369		"endpoint_b": {
370		"active": False,
371		"enabled": True,
372		"id": "00:00:00:00:00:00:00:05:2",
373		"link": "0e2b5d7bc858b9f38db11b69",
374		"mac": "de:eb:d0:b0:14:cf",
375		"metadata": {},
376		"name": "s5-eth2",
377		"nni": True,
378		"port_number": 2,
379		"speed": 1250000000.0,
380		"switch": "00:00:00:00:00:00:00:05",
381		"type": "interface",
382		"uni": False
383		},
384		"id": "0e2b5d7bc858b9f38db11b69",
385		"metadata": {}
386		},
387		{
388		"active": False,
389		"enabled": True,
390		"endpoint_a": {
391		"active": False,
392		"enabled": True,
393		"id": "00:00:00:00:00:00:00:05:4",
394		"link": "53bd36ff55a5aa2029bd5d50",
395		"mac": "6e:c2:ea:c4:18:12",
396		"metadata": {},
397		"name": "s5-eth4",
398		"nni": True,
399		"port_number": 4,
400		"speed": 1250000000.0,
401		"switch": "00:00:00:00:00:00:00:05",
402		"type": "interface",
403		"uni": False
404		},
405		"endpoint_b": {
406		"active": False,
407		"enabled": True,
408		"id": "00:00:00:00:00:00:00:06:2",
409		"link": "53bd36ff55a5aa2029bd5d50",
410		"mac": "5a:25:7b:7c:0d:ac",
411		"metadata": {},
412		"name": "s6-eth2",
413		"nni": True,
414		"port_number": 2,
415		"speed": 1250000000.0,
416		"switch": "00:00:00:00:00:00:00:06",
417		"type": "interface",
418		"uni": False
419		},
420		"id": "53bd36ff55a5aa2029bd5d50",
421		"metadata": {}
422		}
423		],
424		"priority": 0,
425		"request_time": "2019-07-22T16:01:24",
426		"start_date": "2019-07-22T16:01:24",
427		"uni_a": {
428		"interface_id": "00:00:00:00:00:00:00:03:12",
429		"tag": {
430		"tag_type": 1,
431		"value": 321
432		}
433		},
434		"uni_z": {
435		"interface_id": "00:00:00:00:00:00:00:06:11",
436		"tag": {
437		"tag_type": 1,
438		"value": 612
439		}
440		}
441		}
442		}
443
444		expected_result = {
445		'00:00:00:00:00:00:00:03:12':
446		{'182f5bac84074017a262a2321195dbb4',
447		'65c4582cc8f249c2a5947ef500c19e37'},
448		'00:00:00:00:00:00:00:06:11':
449		{'182f5bac84074017a262a2321195dbb4',
450		'65c4582cc8f249c2a5947ef500c19e37'},
451		'00:00:00:00:00:00:00:03:3':
452		{'65c4582cc8f249c2a5947ef500c19e37'},
453		'00:00:00:00:00:00:00:05:2':
454		{'65c4582cc8f249c2a5947ef500c19e37'},
455		'00:00:00:00:00:00:00:05:4':
456		{'65c4582cc8f249c2a5947ef500c19e37'},
457		'00:00:00:00:00:00:00:06:2':
458		{'65c4582cc8f249c2a5947ef500c19e37'}
459		}
460		self.napp.load_circuits_by_interface(stored_circuits)
461		# pylint: disable=protected-access
462		self.assertEqual(self.napp._circuits_by_interface, expected_result)
463

kytos / mef_eline

Push — master ( 98eee7...9ba056 )

TestMain.test_load_circuits_by_interface() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like