
# Copyright (c) 2013 eBay Inc.
# Copyright (c) 2013 OpenStack Foundation
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.
"""Unit Tests for qos specs internal API."""

import time
from unittest import mock

from oslo_db import exception as db_exc
from oslo_utils import timeutils

from cinder import context
from cinder import db
from cinder import exception
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import test
from cinder.tests.unit import utils as test_utils
from cinder.volume import qos_specs
from cinder.volume import volume_types


def fake_db_qos_specs_create(context, values):
    if values['name'] == 'DupQoSName':
        raise exception.QoSSpecsExists(specs_id=values['name'])
    elif values['name'] == 'FailQoSName':
        raise db_exc.DBError()

    pass


def fake_db_get_vol_type(vol_type_number=1):
    return {'name': 'type-' + str(vol_type_number),
            'id': fake.QOS_SPEC_ID,
            'updated_at': None,
            'created_at': None,
            'deleted_at': None,
            'description': 'desc',
            'deleted': False,
            'is_public': True,
            'projects': [],
            'qos_specs_id': fake.QOS_SPEC_ID,
            'extra_specs': None}


class QoSSpecsTestCase(test.TestCase):
    """Test cases for qos specs code."""
    def setUp(self):
        super(QoSSpecsTestCase, self).setUp()
        self.ctxt = context.get_admin_context()

    def _create_qos_specs(self, name, consumer='back-end', values=None):
        """Create a transfer object."""
        if values is None:
            values = {'key1': 'value1', 'key2': 'value2'}

        specs = {'name': name,
                 'consumer': consumer,
                 'specs': values}
        return db.qos_specs_create(self.ctxt, specs)['id']

    def test_create(self):
        input = {'key1': 'value1',
                 'key2': 'value2',
                 'key3': 'value3'}
        ref = qos_specs.create(self.ctxt, 'FakeName', input)
        specs_obj = qos_specs.get_qos_specs(self.ctxt, ref['id'])
        specs_obj_dic = {'consumer': specs_obj['consumer'],
                         'id': specs_obj['id'],
                         'name': specs_obj['name'],
                         'specs': specs_obj['specs']}
        expected = {'consumer': 'back-end',
                    'id': ref['id'],
                    'name': 'FakeName',
                    'specs': input}
        self.assertDictEqual(expected,
                             specs_obj_dic)

        # qos specs must have unique name
        self.assertRaises(exception.QoSSpecsExists,
                          qos_specs.create, self.ctxt, 'FakeName', input)

        # consumer must be one of: front-end, back-end, both
        input['consumer'] = 'fake'
        self.assertRaises(exception.InvalidQoSSpecs,
                          qos_specs.create, self.ctxt, 'QoSName', input)

        del input['consumer']

        self.mock_object(db, 'qos_specs_create',
                         fake_db_qos_specs_create)
        # able to catch DBError
        self.assertRaises(exception.QoSSpecsCreateFailed,
                          qos_specs.create, self.ctxt, 'FailQoSName', input)

    def test_update(self):
        def fake_db_update(context, specs_id, values):
            raise db_exc.DBError()

        qos = {'consumer': 'back-end',
               'specs': {'key1': 'value1'}}

        # qos specs must exists
        self.assertRaises(exception.QoSSpecsNotFound,
                          qos_specs.update, self.ctxt, 'fake_id', qos['specs'])

        specs_id = self._create_qos_specs('Name',
                                          qos['consumer'],
                                          qos['specs'])

        qos_specs.update(self.ctxt, specs_id,
                         {'key1': 'newvalue1', 'key2': 'value2'})

        specs = qos_specs.get_qos_specs(self.ctxt, specs_id)
        self.assertEqual('newvalue1', specs['specs']['key1'])
        self.assertEqual('value2', specs['specs']['key2'])

        # consumer must be one of: front-end, back-end, both
        self.assertRaises(exception.InvalidQoSSpecs,
                          qos_specs.update, self.ctxt, specs_id,
                          {'consumer': 'not-real'})

        self.mock_object(db, 'qos_specs_update', fake_db_update)
        self.assertRaises(exception.QoSSpecsUpdateFailed,
                          qos_specs.update, self.ctxt, specs_id, {'key':
                                                                  'new_key'})

    def test_delete(self):
        qos_id = self._create_qos_specs('my_qos')

        def fake_db_associations_get(context, id):
            vol_types = []
            if id == qos_id:
                vol_types = [fake_db_get_vol_type(id)]
            return vol_types

        def fake_db_delete(context, id):
            return {'deleted': True,
                    'deleted_at': timeutils.utcnow()}

        def fake_disassociate_all(context, id):
            pass

        self.mock_object(db, 'qos_specs_associations_get',
                         fake_db_associations_get)
        self.mock_object(qos_specs, 'disassociate_all',
                         fake_disassociate_all)
        self.mock_object(db, 'qos_specs_delete', fake_db_delete)
        self.assertRaises(exception.InvalidQoSSpecs,
                          qos_specs.delete, self.ctxt, None)
        self.assertRaises(exception.QoSSpecsNotFound,
                          qos_specs.delete, self.ctxt, 'NotFound')
        self.assertRaises(exception.QoSSpecsInUse,
                          qos_specs.delete, self.ctxt, qos_id)
        # able to delete in-use qos specs if force=True
        qos_specs.delete(self.ctxt, qos_id, force=True)

        # Can delete without forcing when no volume types
        qos_id_with_no_vol_types = self._create_qos_specs('no_vol_types')
        qos_specs.delete(self.ctxt, qos_id_with_no_vol_types, force=False)

    def test_delete_keys(self):
        def fake_db_qos_delete_key(context, id, key):
            if key == 'NotFound':
                raise exception.QoSSpecsKeyNotFound(specs_id=id,
                                                    specs_key=key)
            else:
                pass

        value = {'foo': 'Foo', 'bar': 'Bar', 'zoo': 'tiger'}
        name = 'QoSName'
        consumer = 'front-end'
        specs_id = self._create_qos_specs(name, consumer, value)
        qos_specs.delete_keys(self.ctxt, specs_id, ['foo', 'bar'])

        del value['foo']
        del value['bar']
        expected = {'name': name,
                    'id': specs_id,
                    'consumer': consumer,
                    'specs': value}
        specs = qos_specs.get_qos_specs(self.ctxt, specs_id)
        specs_dic = {'consumer': specs['consumer'],