build.tests.test_main.TestMain.test_update_schedule() - Code Metrics - Inspection of "New API REST services to EVC Schedules" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#140)

by Rogerio

created 2019-08-09 18:41 UTC

TestMain.test_update_schedule() B

↳ Parent: build.tests.test_main

Complexity

Conditions

Size

Total Lines	70
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	52
nop	2
dl	0
loc	70
rs	8.5709
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):  # pylint: disable=too-many-public-methods
    """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 = [
            ({}, {'POST', 'OPTIONS'},
             '/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>'),

            ({}, {'OPTIONS', 'GET', 'HEAD'},
             '/api/kytos/mef_eline/v2/evc/schedule'),

            ({'circuit_id': '[circuit_id]'}, {'POST', 'OPTIONS'},
             '/api/kytos/mef_eline/v2/evc/<circuit_id>/schedule/'),

            ({'schedule_id': '[schedule_id]', 'circuit_id': '[circuit_id]'},
             {'OPTIONS', 'DELETE'},
             '/api/kytos/mef_eline/v2/'
             'evc/<circuit_id>/schedule/<schedule_id>'),

            ({'schedule_id': '[schedule_id]', 'circuit_id': '[circuit_id]'},
             {'OPTIONS', 'PATCH'},
             '/api/kytos/mef_eline/v2/'
             'evc/<circuit_id>/schedule/<schedule_id>'),

            ({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'GET', 'HEAD'},
             '/api/kytos/mef_eline/v2/evc/<circuit_id>/schedule/')
            ]

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

    @patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
    @patch('napps.kytos.mef_eline.models.EVCBase._validate')
    def test_evc_from_dict(self, _validate_mock, uni_from_dict_mock):
        """
        Test the helper method that create an EVN from dict.

        Verify object creation with circuit data and schedule data.
        """
        _validate_mock.return_value = True
        uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
        payload = {
            "name": "my evc1",
            "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
                }
            },
            "circuit_scheduler": {
                "frequency": "* * * * *",
                "action": "create"
            }
        }
        evc_response = self.napp.evc_from_dict(payload)
        self.assertIsNotNone(evc_response)

    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",
            "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_list_schedules__no_data(self):
        """Test list of schedules."""
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/schedule'
        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_schedules__no_data_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/schedule'

        response = api.get(url)
        expected_result = circuits

        self.assertEqual(response.status_code, 200)
        self.assertEqual(json.loads(response.data), expected_result)

    # pylint: disable=no-self-use
    def _add_storehouse_schedule_data(self, storehouse_data_mock):
        """Add schedule data to storehouse mock object."""
        circuits = {}
        payload_1 = {
            "id": "aa:aa:aa",
            "name": "my evc1",
            "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
                }
            },
            "circuit_scheduler": {
                "frequency": "* * * * *",
                "action": "create"
            }
        }
        circuits.update({'1': payload_1})
        payload_2 = {
            "id": "bb:bb:bb",
            "name": "my second evc1",
            "uni_a": {
                "interface_id": "00:00:00:00:00:00:00:01:2",
                "tag": {
                    "tag_type": 1,
                    "value": 90
                }
            },
            "uni_z": {
                "interface_id": "00:00:00:00:00:00:00:03:2",
                "tag": {
                    "tag_type": 1,
                    "value": 100
                }
            },
            "circuit_scheduler": {
                "frequency": "1 * * * *",
                "action": "create"
            }
        }
        circuits.update({'2': payload_2})
        # Add one circuit to the storehouse.
        storehouse_data_mock.return_value = circuits

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    def test_list_schedules_from_storehouse(self, storehouse_data_mock):
        """Test if list circuits return specific circuits stored."""
        self._add_storehouse_schedule_data(storehouse_data_mock)

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

        # Call URL
        response = api.get(url)

        # Expected JSON data from response
        expected = [{'aa:aa:aa':
                    {'action': 'create', 'frequency': '* * * * *'}},
                    {'bb:bb:bb':
                    {'action': 'create', 'frequency': '1 * * * *'}}]

        self.assertEqual(response.status_code, 200)
        self.assertEqual(expected, json.loads(response.data))

    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    def test_get_specific_schedule_from_storehouse(self, storehouse_data_mock):
        """Test get schedule byt ID. Return specific circuits stored."""
        self._add_storehouse_schedule_data(storehouse_data_mock)

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

        # Call URL
        response = api.get(url)

        # Expected JSON data from response
        expected = {'action': 'create', 'frequency': '1 * * * *'}

        self.assertEqual(response.status_code, 200)
        self.assertEqual(expected, json.loads(response.data))

    def test_get_specific_schedules_from_storehouse_not_found(self):
        """Test get specific schedule ID that does not exist."""
        requested_id = "blah"
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/{requested_id}/schedule/'

        # Call URL
        response = api.get(url)

        expected = {'response': 'circuit_id blah not found'}
        # Assert response not found
        self.assertEqual(response.status_code, 404)
        self.assertEqual(expected, json.loads(response.data))

    @patch('apscheduler.schedulers.background.BackgroundScheduler.add_job')
    @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_schedule(self, *args):  # pylint: disable=too-many-locals
        """Test create a circuit schedule."""
        (validate_mock, evc_as_dict_mock, save_evc_mock,
         uni_from_dict_mock, sched_add_mock, storehouse_data_mock,
         scheduler_add_job_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 = {}

        self._add_storehouse_schedule_data(storehouse_data_mock)

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

        payload = {
            "frequency": "1 * * * *",
            "action": "create"
        }

        # Call URL
        response = api.post(url, data=json.dumps(payload),
                            content_type='application/json')

        response_json = json.loads(response.data)

        scheduler_add_job_mock.assert_called_once()
        save_evc_mock.assert_called_once()
        self.assertEqual(response.status_code, 201)
        self.assertEqual(payload["frequency"], response_json["frequency"])
        self.assertEqual(payload["action"], response_json["action"])
        self.assertIsNotNone(response_json["id"])

    @patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job')
    @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_update_schedule(self, *args):  # pylint: disable=too-many-locals
        """Test create a circuit schedule."""
        (validate_mock, evc_as_dict_mock, save_evc_mock,
         uni_from_dict_mock, sched_add_mock, storehouse_data_mock,
         scheduler_remove_job_mock) = args

        storehouse_payload_1 = {
            "2": {
                "id": "2",
                "name": "my evc1",
                "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
                    }
                },
                "circuit_scheduler": {
                    "id": "1",
                    "frequency": "* * * * *",
                    "action": "create"
                }
            }
        }

        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']
        evc_as_dict_mock.return_value = {}
        storehouse_data_mock.return_value = storehouse_payload_1
        scheduler_remove_job_mock.return_value = True

        requested_circuit_id = "2"
        requested_schedule_id = "1"
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/{requested_circuit_id}/'\
              f'schedule/{requested_schedule_id}'

        payload = {
            "frequency": "1 * * * *",
            "action": "create"
        }

        # Call URL
        response = api.patch(url, data=json.dumps(payload),
                             content_type='application/json')

        response_json = json.loads(response.data)

        scheduler_remove_job_mock.assert_called_once()
        save_evc_mock.assert_called_once()
        self.assertEqual(response.status_code, 200)
        self.assertEqual(payload["frequency"], response_json["frequency"])
        self.assertEqual(payload["action"], response_json["action"])
        self.assertIsNotNone(response_json["id"])

    @patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job')
    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
    @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_delete_schedule(self, *args):
        """Test create a circuit schedule."""
        (validate_mock, evc_as_dict_mock, save_evc_mock,
         uni_from_dict_mock, storehouse_data_mock,
         scheduler_remove_job_mock) = args

        storehouse_payload_1 = {
            "2": {
                "id": "2",
                "name": "my evc1",
                "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
                    }
                },
                "circuit_scheduler": {
                    "id": "1",
                    "frequency": "* * * * *",
                    "action": "create"
                }
            }
        }

        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 = {}
        storehouse_data_mock.return_value = storehouse_payload_1
        scheduler_remove_job_mock.return_value = True

        requested_circuit_id = "2"
        requested_schedule_id = "1"
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/{requested_circuit_id}/'\
              f'schedule/{requested_schedule_id}'

        # Call URL
        response = api.delete(url)

        scheduler_remove_job_mock.assert_called_once()
        save_evc_mock.assert_called_once()
        self.assertEqual(response.status_code, 200)
        self.assertIn("Schedule removed", f"{response.data}")


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): # pylint: disable=too-many-public-methods
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		({}, {'POST', 'OPTIONS'},
36		'/api/kytos/mef_eline/v2/evc/'),
37
38		({}, {'OPTIONS', 'HEAD', 'GET'},
39		'/api/kytos/mef_eline/v2/evc/'),
40
41		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'DELETE'},
42		'/api/kytos/mef_eline/v2/evc/<circuit_id>'),
43
44		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'HEAD', 'GET'},
45		'/api/kytos/mef_eline/v2/evc/<circuit_id>'),
46
47		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'PATCH'},
48		'/api/kytos/mef_eline/v2/evc/<circuit_id>'),
49
50		({}, {'OPTIONS', 'GET', 'HEAD'},
51		'/api/kytos/mef_eline/v2/evc/schedule'),
52
53		({'circuit_id': '[circuit_id]'}, {'POST', 'OPTIONS'},
54		'/api/kytos/mef_eline/v2/evc/<circuit_id>/schedule/'),
55
56		({'schedule_id': '[schedule_id]', 'circuit_id': '[circuit_id]'},
57		{'OPTIONS', 'DELETE'},
58		'/api/kytos/mef_eline/v2/'
59		'evc/<circuit_id>/schedule/<schedule_id>'),
60
61		({'schedule_id': '[schedule_id]', 'circuit_id': '[circuit_id]'},
62		{'OPTIONS', 'PATCH'},
63		'/api/kytos/mef_eline/v2/'
64		'evc/<circuit_id>/schedule/<schedule_id>'),
65
66		({'circuit_id': '[circuit_id]'}, {'OPTIONS', 'GET', 'HEAD'},
67		'/api/kytos/mef_eline/v2/evc/<circuit_id>/schedule/')
68		]
69
70		urls = self.get_napp_urls(self.napp)
71		self.assertCountEqual(expected_urls, urls)
72
73		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
74		@patch('napps.kytos.mef_eline.models.EVCBase._validate')
75		def test_evc_from_dict(self, _validate_mock, uni_from_dict_mock):
76		"""
77		Test the helper method that create an EVN from dict.
78
79		Verify object creation with circuit data and schedule data.
80		"""
81		_validate_mock.return_value = True
82		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
83		payload = {
84		"name": "my evc1",
85		"uni_a": {
86		"interface_id": "00:00:00:00:00:00:00:01:1",
87		"tag": {
88		"tag_type": 1,
89		"value": 80
90		}
91		},
92		"uni_z": {
93		"interface_id": "00:00:00:00:00:00:00:02:2",
94		"tag": {
95		"tag_type": 1,
96		"value": 1
97		}
98		},
99		"circuit_scheduler": {
100		"frequency": "* * * * *",
101		"action": "create"
102		}
103		}
104		evc_response = self.napp.evc_from_dict(payload)
105		self.assertIsNotNone(evc_response)
106
107		def test_list_without_circuits(self):
108		"""Test if list circuits return 'no circuit stored.'."""
109		api = self.get_app_test_client(self.napp)
110		url = f'{self.server_name_url}/v2/evc/'
111		response = api.get(url)
112		self.assertEqual(response.status_code, 200)
113		self.assertEqual(json.loads(response.data.decode()), {})
114
115		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
116		def test_list_no_circuits_stored(self, storehouse_data_mock):
117		"""Test if list circuits return all circuits stored."""
118		circuits = {}
119		storehouse_data_mock.return_value = circuits
120
121		api = self.get_app_test_client(self.napp)
122		url = f'{self.server_name_url}/v2/evc/'
123
124		response = api.get(url)
125		expected_result = circuits
126		self.assertEqual(json.loads(response.data), expected_result)
127
128		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
129		def test_list_with_circuits_stored(self, storehouse_data_mock):
130		"""Test if list circuits return all circuits stored."""
131		circuits = {'1': {'name': 'circuit_1'},
132		'2': {'name': 'circuit_2'}}
133		storehouse_data_mock.return_value = circuits
134
135		api = self.get_app_test_client(self.napp)
136		url = f'{self.server_name_url}/v2/evc/'
137
138		response = api.get(url)
139		expected_result = circuits
140		self.assertEqual(json.loads(response.data), expected_result)
141
142	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...
143		def test_circuit_with_valid_id(self, storehouse_data_mock):
144		"""Test if get_circuit return the circuit attributes."""
145		circuits = {'1': {'name': 'circuit_1'},
146		'2': {'name': 'circuit_2'}}
147		storehouse_data_mock.return_value = circuits
148
149		api = self.get_app_test_client(self.napp)
150		url = f'{self.server_name_url}/v2/evc/1'
151		response = api.get(url)
152		expected_result = circuits['1']
153		self.assertEqual(json.loads(response.data), expected_result)
154
155	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...
156		def test_circuit_with_invalid_id(self, storehouse_data_mock):
157		"""Test if get_circuit return invalid circuit_id."""
158		circuits = {'1': {'name': 'circuit_1'},
159		'2': {'name': 'circuit_2'}}
160		storehouse_data_mock.return_value = circuits
161
162		api = self.get_app_test_client(self.napp)
163		url = f'{self.server_name_url}/v2/evc/3'
164		response = api.get(url)
165		expected_result = {'response': 'circuit_id 3 not found'}
166		self.assertEqual(json.loads(response.data), expected_result)
167
168		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
169		@patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
170		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
171		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
172		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
173		@patch('napps.kytos.mef_eline.models.EVC._validate')
174		def test_create_a_circuit_case_1(self, *args):
175		"""Test create a new circuit."""
176		(validate_mock, evc_as_dict_mock, save_evc_mock,
177		uni_from_dict_mock, sched_add_mock, storehouse_data_mock) = args
178
179		validate_mock.return_value = True
180		save_evc_mock.return_value = True
181		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
182		evc_as_dict_mock.return_value = {}
183		sched_add_mock.return_value = True
184		storehouse_data_mock.return_value = {}
185
186		api = self.get_app_test_client(self.napp)
187		url = f'{self.server_name_url}/v2/evc/'
188		payload = {
189		"name": "my evc1",
190		"frequency": "* * * * *",
191		"uni_a": {
192		"interface_id": "00:00:00:00:00:00:00:01:1",
193		"tag": {
194		"tag_type": 1,
195		"value": 80
196		}
197		},
198		"uni_z": {
199		"interface_id": "00:00:00:00:00:00:00:02:2",
200		"tag": {
201		"tag_type": 1,
202		"value": 1
203		}
204		}
205		}
206
207		response = api.post(url, data=json.dumps(payload),
208		content_type='application/json')
209		current_data = json.loads(response.data)
210
211		# verify expected result from request
212		self.assertEqual(201, response.status_code, response.data)
213		self.assertIn('circuit_id', current_data)
214
215		# verify uni called
216		uni_from_dict_mock.called_twice()
217		uni_from_dict_mock.assert_any_call(payload['uni_z'])
218		uni_from_dict_mock.assert_any_call(payload['uni_a'])
219
220		# verify validation called
221		validate_mock.assert_called_once()
222		validate_mock.assert_called_with(frequency='* * * * *',
223		name='my evc1',
224		uni_a='uni_a',
225		uni_z='uni_z')
226		# verify save method is called
227		save_evc_mock.assert_called_once()
228
229		# verify evc as dict is called to save in the box
230		evc_as_dict_mock.assert_called_once()
231		# verify add circuit in sched
232		sched_add_mock.assert_called_once()
233
234		@staticmethod
235		def get_napp_urls(napp):
236		"""Return the kytos/mef_eline urls.
237
238		The urls will be like:
239
240		urls = [
241		(options, methods, url)
242		]
243
244		"""
245		controller = napp.controller
246		controller.api_server.register_napp_endpoints(napp)
247
248		urls = []
249		for rule in controller.api_server.app.url_map.iter_rules():
250		options = {}
251		for arg in rule.arguments:
252		options[arg] = "[{0}]".format(arg)
253
254		if f'{napp.username}/{napp.name}' in str(rule):
255		urls.append((options, rule.methods, f'{str(rule)}'))
256
257		return urls
258
259		@staticmethod
260		def get_app_test_client(napp):
261		"""Return a flask api test client."""
262		napp.controller.api_server.register_napp_endpoints(napp)
263		return napp.controller.api_server.app.test_client()
264
265		def test_create_a_circuit_case_2(self):
266		"""Test create a new circuit trying to send request without a json."""
267		api = self.get_app_test_client(self.napp)
268		url = f'{self.server_name_url}/v2/evc/'
269
270		response = api.post(url)
271		current_data = json.loads(response.data)
272		expected_data = 'Bad request: The request do not have a json.'
273
274		self.assertEqual(400, response.status_code)
275		self.assertEqual(current_data, expected_data)
276
277		@patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
278		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
279		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
280		@patch('napps.kytos.mef_eline.models.EVC._validate')
281		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
282		def test_create_circuit_already_enabled(self, *args):
283		"""Test create an already created circuit."""
284		(evc_as_dict_mock, validate_mock, save_evc_mock,
285		uni_from_dict_mock, sched_add_mock) = args
286
287		validate_mock.return_value = True
288		save_evc_mock.return_value = True
289		sched_add_mock.return_value = True
290		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z', 'uni_a', 'uni_z']
291		payload1 = {'name': 'circuit_1'}
292
293		api = self.get_app_test_client(self.napp)
294		payload2 = {
295		"name": "my evc1",
296		"uni_a": {
297		"interface_id": "00:00:00:00:00:00:00:01:1",
298		"tag": {
299		"tag_type": 1,
300		"value": 80
301		}
302		},
303		"uni_z": {
304		"interface_id": "00:00:00:00:00:00:00:02:2",
305		"tag": {
306		"tag_type": 1,
307		"value": 1
308		}
309		}
310		}
311
312		evc_as_dict_mock.return_value = payload1
313		response = api.post(f'{self.server_name_url}/v2/evc/',
314		data=json.dumps(payload1),
315		content_type='application/json')
316		self.assertEqual(201, response.status_code)
317
318		evc_as_dict_mock.return_value = payload2
319		response = api.post(f'{self.server_name_url}/v2/evc/',
320		data=json.dumps(payload2),
321		content_type='application/json')
322		self.assertEqual(201, response.status_code)
323
324		response = api.post(f'{self.server_name_url}/v2/evc/',
325		data=json.dumps(payload2),
326		content_type='application/json')
327		current_data = json.loads(response.data)
328		expected_data = 'Not Acceptable: This evc already exists.'
329		self.assertEqual(current_data, expected_data)
330		self.assertEqual(409, response.status_code)
331
332		def test_list_schedules__no_data(self):
333		"""Test list of schedules."""
334		api = self.get_app_test_client(self.napp)
335		url = f'{self.server_name_url}/v2/evc/schedule'
336		response = api.get(url)
337		self.assertEqual(response.status_code, 200)
338		self.assertEqual(json.loads(response.data.decode()), {})
339
340		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
341		def test_list_schedules__no_data_stored(self, storehouse_data_mock):
342		"""Test if list circuits return all circuits stored."""
343		circuits = {}
344		storehouse_data_mock.return_value = circuits
345
346		api = self.get_app_test_client(self.napp)
347		url = f'{self.server_name_url}/v2/evc/schedule'
348
349		response = api.get(url)
350		expected_result = circuits
351
352		self.assertEqual(response.status_code, 200)
353		self.assertEqual(json.loads(response.data), expected_result)
354
355		# pylint: disable=no-self-use
356		def _add_storehouse_schedule_data(self, storehouse_data_mock):
357		"""Add schedule data to storehouse mock object."""
358		circuits = {}
359		payload_1 = {
360		"id": "aa:aa:aa",
361		"name": "my evc1",
362		"uni_a": {
363		"interface_id": "00:00:00:00:00:00:00:01:1",
364		"tag": {
365		"tag_type": 1,
366		"value": 80
367		}
368		},
369		"uni_z": {
370		"interface_id": "00:00:00:00:00:00:00:02:2",
371		"tag": {
372		"tag_type": 1,
373		"value": 1
374		}
375		},
376		"circuit_scheduler": {
377		"frequency": "* * * * *",
378		"action": "create"
379		}
380		}
381		circuits.update({'1': payload_1})
382		payload_2 = {
383		"id": "bb:bb:bb",
384		"name": "my second evc1",
385		"uni_a": {
386		"interface_id": "00:00:00:00:00:00:00:01:2",
387		"tag": {
388		"tag_type": 1,
389		"value": 90
390		}
391		},
392		"uni_z": {
393		"interface_id": "00:00:00:00:00:00:00:03:2",
394		"tag": {
395		"tag_type": 1,
396		"value": 100
397		}
398		},
399		"circuit_scheduler": {
400		"frequency": "1 * * * *",
401		"action": "create"
402		}
403		}
404		circuits.update({'2': payload_2})
405		# Add one circuit to the storehouse.
406		storehouse_data_mock.return_value = circuits
407
408		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
409		def test_list_schedules_from_storehouse(self, storehouse_data_mock):
410		"""Test if list circuits return specific circuits stored."""
411		self._add_storehouse_schedule_data(storehouse_data_mock)
412
413		api = self.get_app_test_client(self.napp)
414		url = f'{self.server_name_url}/v2/evc/schedule'
415
416		# Call URL
417		response = api.get(url)
418
419		# Expected JSON data from response
420		expected = [{'aa:aa:aa':
421		{'action': 'create', 'frequency': '* * * * *'}},
422		{'bb:bb:bb':
423		{'action': 'create', 'frequency': '1 * * * *'}}]
424
425		self.assertEqual(response.status_code, 200)
426		self.assertEqual(expected, json.loads(response.data))
427
428		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
429		def test_get_specific_schedule_from_storehouse(self, storehouse_data_mock):
430		"""Test get schedule byt ID. Return specific circuits stored."""
431		self._add_storehouse_schedule_data(storehouse_data_mock)
432
433		requested_id = "2"
434		api = self.get_app_test_client(self.napp)
435		url = f'{self.server_name_url}/v2/evc/{requested_id}/schedule/'
436
437		# Call URL
438		response = api.get(url)
439
440		# Expected JSON data from response
441		expected = {'action': 'create', 'frequency': '1 * * * *'}
442
443		self.assertEqual(response.status_code, 200)
444		self.assertEqual(expected, json.loads(response.data))
445
446		def test_get_specific_schedules_from_storehouse_not_found(self):
447		"""Test get specific schedule ID that does not exist."""
448		requested_id = "blah"
449		api = self.get_app_test_client(self.napp)
450		url = f'{self.server_name_url}/v2/evc/{requested_id}/schedule/'
451
452		# Call URL
453		response = api.get(url)
454
455		expected = {'response': 'circuit_id blah not found'}
456		# Assert response not found
457		self.assertEqual(response.status_code, 404)
458		self.assertEqual(expected, json.loads(response.data))
459
460		@patch('apscheduler.schedulers.background.BackgroundScheduler.add_job')
461		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
462		@patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
463		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
464		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
465		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
466		@patch('napps.kytos.mef_eline.models.EVC._validate')
467		def test_create_schedule(self, *args): # pylint: disable=too-many-locals
468		"""Test create a circuit schedule."""
469		(validate_mock, evc_as_dict_mock, save_evc_mock,
470		uni_from_dict_mock, sched_add_mock, storehouse_data_mock,
471		scheduler_add_job_mock) = args
472
473		validate_mock.return_value = True
474		save_evc_mock.return_value = True
475		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
476		evc_as_dict_mock.return_value = {}
477		sched_add_mock.return_value = True
478		storehouse_data_mock.return_value = {}
479
480		self._add_storehouse_schedule_data(storehouse_data_mock)
481
482		requested_id = "2"
483		api = self.get_app_test_client(self.napp)
484		url = f'{self.server_name_url}/v2/evc/{requested_id}/schedule/'
485
486		payload = {
487		"frequency": "1 * * * *",
488		"action": "create"
489		}
490
491		# Call URL
492		response = api.post(url, data=json.dumps(payload),
493		content_type='application/json')
494
495		response_json = json.loads(response.data)
496
497		scheduler_add_job_mock.assert_called_once()
498		save_evc_mock.assert_called_once()
499		self.assertEqual(response.status_code, 201)
500		self.assertEqual(payload["frequency"], response_json["frequency"])
501		self.assertEqual(payload["action"], response_json["action"])
502		self.assertIsNotNone(response_json["id"])
503
504		@patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job')
505		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
506		@patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
507		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
508		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
509		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
510		@patch('napps.kytos.mef_eline.models.EVC._validate')
511		def test_update_schedule(self, *args): # pylint: disable=too-many-locals
512		"""Test create a circuit schedule."""
513		(validate_mock, evc_as_dict_mock, save_evc_mock,
514		uni_from_dict_mock, sched_add_mock, storehouse_data_mock,
515		scheduler_remove_job_mock) = args
516
517		storehouse_payload_1 = {
518		"2": {
519		"id": "2",
520		"name": "my evc1",
521		"uni_a": {
522		"interface_id": "00:00:00:00:00:00:00:01:1",
523		"tag": {
524		"tag_type": 1,
525		"value": 80
526		}
527		},
528		"uni_z": {
529		"interface_id": "00:00:00:00:00:00:00:02:2",
530		"tag": {
531		"tag_type": 1,
532		"value": 1
533		}
534		},
535		"circuit_scheduler": {
536		"id": "1",
537		"frequency": "* * * * *",
538		"action": "create"
539		}
540		}
541		}
542
543		validate_mock.return_value = True
544		save_evc_mock.return_value = True
545		sched_add_mock.return_value = True
546		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
547		evc_as_dict_mock.return_value = {}
548		storehouse_data_mock.return_value = storehouse_payload_1
549		scheduler_remove_job_mock.return_value = True
550
551		requested_circuit_id = "2"
552		requested_schedule_id = "1"
553		api = self.get_app_test_client(self.napp)
554		url = f'{self.server_name_url}/v2/evc/{requested_circuit_id}/'\
555		f'schedule/{requested_schedule_id}'
556
557		payload = {
558		"frequency": "1 * * * *",
559		"action": "create"
560		}
561
562		# Call URL
563		response = api.patch(url, data=json.dumps(payload),
564		content_type='application/json')
565
566		response_json = json.loads(response.data)
567
568		scheduler_remove_job_mock.assert_called_once()
569		save_evc_mock.assert_called_once()
570		self.assertEqual(response.status_code, 200)
571		self.assertEqual(payload["frequency"], response_json["frequency"])
572		self.assertEqual(payload["action"], response_json["action"])
573		self.assertIsNotNone(response_json["id"])
574
575		@patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job')
576		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.get_data')
577		@patch('napps.kytos.mef_eline.main.Main.uni_from_dict')
578		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
579		@patch('napps.kytos.mef_eline.main.EVC.as_dict')
580		@patch('napps.kytos.mef_eline.models.EVC._validate')
581		def test_delete_schedule(self, *args):
582		"""Test create a circuit schedule."""
583		(validate_mock, evc_as_dict_mock, save_evc_mock,
584		uni_from_dict_mock, storehouse_data_mock,
585		scheduler_remove_job_mock) = args
586
587		storehouse_payload_1 = {
588		"2": {
589		"id": "2",
590		"name": "my evc1",
591		"uni_a": {
592		"interface_id": "00:00:00:00:00:00:00:01:1",
593		"tag": {
594		"tag_type": 1,
595		"value": 80
596		}
597		},
598		"uni_z": {
599		"interface_id": "00:00:00:00:00:00:00:02:2",
600		"tag": {
601		"tag_type": 1,
602		"value": 1
603		}
604		},
605		"circuit_scheduler": {
606		"id": "1",
607		"frequency": "* * * * *",
608		"action": "create"
609		}
610		}
611		}
612
613		validate_mock.return_value = True
614		save_evc_mock.return_value = True
615		uni_from_dict_mock.side_effect = ['uni_a', 'uni_z']
616		evc_as_dict_mock.return_value = {}
617		storehouse_data_mock.return_value = storehouse_payload_1
618		scheduler_remove_job_mock.return_value = True
619
620		requested_circuit_id = "2"
621		requested_schedule_id = "1"
622		api = self.get_app_test_client(self.napp)
623		url = f'{self.server_name_url}/v2/evc/{requested_circuit_id}/'\
624		f'schedule/{requested_schedule_id}'
625
626		# Call URL
627		response = api.delete(url)
628
629		scheduler_remove_job_mock.assert_called_once()
630		save_evc_mock.assert_called_once()
631		self.assertEqual(response.status_code, 200)
632		self.assertIn("Schedule removed", f"{response.data}")
633

kytos / mef_eline

Pull Request — master (#140)

TestMain.test_update_schedule() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like