¡@

Home 

OpenStack Study: test_types_manage.py

OpenStack Index

**** CubicPower OpenStack Study ****

# Copyright 2011 OpenStack Foundation

# All Rights Reserved.

#

# 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.

import webob

from cinder.api.contrib import types_manage

from cinder import exception

from cinder import test

from cinder.tests.api import fakes

from cinder.tests import fake_notifier

from cinder.volume import volume_types

**** CubicPower OpenStack Study ****

def stub_volume_type(id):

    specs = {"key1": "value1",

             "key2": "value2",

             "key3": "value3",

             "key4": "value4",

             "key5": "value5"}

    return dict(id=id, name='vol_type_%s' % str(id), extra_specs=specs)

**** CubicPower OpenStack Study ****

def return_volume_types_get_volume_type(context, id):

    if id == "777":

        raise exception.VolumeTypeNotFound(volume_type_id=id)

    return stub_volume_type(int(id))

**** CubicPower OpenStack Study ****

def return_volume_types_destroy(context, name):

    if name == "777":

        raise exception.VolumeTypeNotFoundByName(volume_type_name=name)

    pass

**** CubicPower OpenStack Study ****

def return_volume_types_with_volumes_destroy(context, id):

    if id == "1":

        raise exception.VolumeTypeInUse(volume_type_id=id)

    pass

**** CubicPower OpenStack Study ****

def return_volume_types_create(context, name, specs):

    pass

**** CubicPower OpenStack Study ****

def return_volume_types_create_duplicate_type(context, name, specs):

    raise exception.VolumeTypeExists(id=name)

**** CubicPower OpenStack Study ****

def return_volume_types_get_by_name(context, name):

    if name == "777":

        raise exception.VolumeTypeNotFoundByName(volume_type_name=name)

    return stub_volume_type(int(name.split("_")[2]))

**** CubicPower OpenStack Study ****

class VolumeTypesManageApiTest(test.TestCase):

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(VolumeTypesManageApiTest, self).setUp()

        self.flags(host='fake')

        self.controller = types_manage.VolumeTypesManageController()

        """to reset notifier drivers left over from other api/contrib tests"""

        fake_notifier.reset()

        self.addCleanup(fake_notifier.reset)

**** CubicPower OpenStack Study ****

    def tearDown(self):

        super(VolumeTypesManageApiTest, self).tearDown()

**** CubicPower OpenStack Study ****

    def test_volume_types_delete(self):

        self.stubs.Set(volume_types, 'get_volume_type',

                       return_volume_types_get_volume_type)

        self.stubs.Set(volume_types, 'destroy',

                       return_volume_types_destroy)

        req = fakes.HTTPRequest.blank('/v2/fake/types/1')

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 0)

        self.controller._delete(req, 1)

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)

**** CubicPower OpenStack Study ****

    def test_volume_types_delete_not_found(self):

        self.stubs.Set(volume_types, 'get_volume_type',

                       return_volume_types_get_volume_type)

        self.stubs.Set(volume_types, 'destroy',

                       return_volume_types_destroy)

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 0)

        req = fakes.HTTPRequest.blank('/v2/fake/types/777')

        self.assertRaises(webob.exc.HTTPNotFound, self.controller._delete,

                          req, '777')

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)

**** CubicPower OpenStack Study ****

    def test_volume_types_with_volumes_destroy(self):

        self.stubs.Set(volume_types, 'get_volume_type',

                       return_volume_types_get_volume_type)

        self.stubs.Set(volume_types, 'destroy',

                       return_volume_types_with_volumes_destroy)

        req = fakes.HTTPRequest.blank('/v2/fake/types/1')

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 0)

        self.controller._delete(req, 1)

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)

**** CubicPower OpenStack Study ****

    def test_create(self):

        self.stubs.Set(volume_types, 'create',

                       return_volume_types_create)

        self.stubs.Set(volume_types, 'get_volume_type_by_name',

                       return_volume_types_get_by_name)

        body = {"volume_type": {"name": "vol_type_1",

                                "extra_specs": {"key1": "value1"}}}

        req = fakes.HTTPRequest.blank('/v2/fake/types')

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 0)

        res_dict = self.controller._create(req, body)

        self.assertEqual(len(fake_notifier.NOTIFICATIONS), 1)

        self.assertEqual(1, len(res_dict))

        self.assertEqual('vol_type_1', res_dict['volume_type']['name'])

**** CubicPower OpenStack Study ****

    def test_create_duplicate_type_fail(self):

        self.stubs.Set(volume_types, 'create',

                       return_volume_types_create_duplicate_type)

        self.stubs.Set(volume_types, 'get_volume_type_by_name',

                       return_volume_types_get_by_name)

        body = {"volume_type": {"name": "vol_type_1",

                                "extra_specs": {"key1": "value1"}}}

        req = fakes.HTTPRequest.blank('/v2/fake/types')

        self.assertRaises(webob.exc.HTTPConflict,

                          self.controller._create, req, body)

**** CubicPower OpenStack Study ****

    def _create_volume_type_bad_body(self, body):

        req = fakes.HTTPRequest.blank('/v2/fake/types')

        req.method = 'POST'

        self.assertRaises(webob.exc.HTTPBadRequest,

                          self.controller._create, req, body)

**** CubicPower OpenStack Study ****

    def test_create_no_body(self):

        self._create_volume_type_bad_body(body=None)

**** CubicPower OpenStack Study ****

    def test_create_missing_volume(self):

        body = {'foo': {'a': 'b'}}

        self._create_volume_type_bad_body(body=body)

**** CubicPower OpenStack Study ****

    def test_create_malformed_entity(self):

        body = {'volume_type': 'string'}

        self._create_volume_type_bad_body(body=body)