build.tests.unit.test_main.TestMain.test_load_circuits_by_interface()

Complexity

Conditions

1

Size

Total Lines	187
Code Lines	153

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

"""Module to test the main napp file."""

        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 = 'The EVC already exists.'
        self.assertEqual(current_data['description'], expected_data)
        self.assertEqual(409, response.status_code)

    @patch('napps.kytos.mef_eline.main.Main._uni_from_dict')
    def test_create_circuit_case_5(self, uni_from_dict_mock):
        """Test when neither primary path nor dynamic_backup_path is set."""
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/'
        uni1 = create_autospec(UNI)
        uni2 = create_autospec(UNI)
        uni1.interface = create_autospec(Interface)
        uni2.interface = create_autospec(Interface)
        uni1.interface.switch = '00:00:00:00:00:00:00:01'
        uni2.interface.switch = '00:00:00:00:00:00:00:02'
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]

        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)
        expected = 'The EVC must have a primary path or allow dynamic paths.'

        self.assertEqual(400, response.status_code, response.data)
        self.assertEqual(current_data['description'], expected)

    def test_redeploy_evc(self):
        """Test endpoint to redeploy an EVC."""
        evc1 = MagicMock()
        evc1.is_enabled.return_value = True
        self.napp.circuits = {'1': evc1,
                              '2': MagicMock()}
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/1/redeploy'
        response = api.patch(url)
        self.assertEqual(response.status_code, 202, response.data)

    def test_redeploy_evc_disabled(self):
        """Test endpoint to redeploy an EVC."""
        evc1 = MagicMock()
        evc1.is_enabled.return_value = False
        self.napp.circuits = {'1': evc1,
                              '2': MagicMock()}
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/1/redeploy'
        response = api.patch(url)
        self.assertEqual(response.status_code, 409, response.data)

    def test_redeploy_evc_deleted(self):
        """Test endpoint to redeploy an EVC."""
        evc1 = MagicMock()
        evc1.is_enabled.return_value = True
        self.napp.circuits = {'1': evc1,
                              '2': MagicMock()}
        api = self.get_app_test_client(self.napp)
        url = f'{self.server_name_url}/v2/evc/3/redeploy'
        response = api.patch(url)
        self.assertEqual(response.status_code, 404, response.data)

    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, response.data)
        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, response.data)
        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": [
                {
                    "id": "1",
                    "frequency": "* * * * *",
                    "action": "create"
                },
                {
                    "id": "2",
                    "frequency": "1 * * * *",
                    "action": "remove"
                }
            ]
        }
        circuits.update({"aa:aa:aa": payload_1})
        payload_2 = {
            "id": "bb:bb:bb",
            "name": "my second evc2",
            "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": [
                {
                    "id": "3",
                    "frequency": "1 * * * *",
                    "action": "create"
                },
                {
                    "id": "4",
                    "frequency": "2 * * * *",
                    "action": "remove"
                }
            ]
        }
        circuits.update({"bb:bb:bb": payload_2})
        payload_3 = {
            "id": "cc:cc:cc",
            "name": "my third evc3",
            "uni_a": {
                "interface_id": "00:00:00:00:00:00:00:03:1",
                "tag": {
                    "tag_type": 1,
                    "value": 90
                }
            },
            "uni_z": {
                "interface_id": "00:00:00:00:00:00:00:04:2",
                "tag": {
                    "tag_type": 1,
                    "value": 100
                }
            }