From d5d282dd1ffab96e85332bea580689c3297a25f8 Mon Sep 17 00:00:00 2001 From: spisarski Date: Wed, 18 Oct 2017 11:52:51 -0600 Subject: Second patch for volume support. * Added support for volume types * Created tests for volume types, QoS Spec, and encryption JIRA: SNAPS-196 Change-Id: I9154fc20772191cecf4f2f9feb7e8d8634167a9c Signed-off-by: spisarski --- snaps/domain/test/volume_tests.py | 63 ++++- snaps/domain/volume.py | 59 ++++ snaps/openstack/create_volume_type.py | 235 ++++++++++++++++ snaps/openstack/tests/create_volume_type_tests.py | 323 ++++++++++++++++++++++ snaps/openstack/utils/cinder_utils.py | 200 ++++++++++++-- snaps/openstack/utils/tests/cinder_utils_tests.py | 320 ++++++++++++++++++++- snaps/test_suite_builder.py | 44 ++- 7 files changed, 1209 insertions(+), 35 deletions(-) create mode 100644 snaps/openstack/create_volume_type.py create mode 100644 snaps/openstack/tests/create_volume_type_tests.py diff --git a/snaps/domain/test/volume_tests.py b/snaps/domain/test/volume_tests.py index f105e38..ec5f7b7 100644 --- a/snaps/domain/test/volume_tests.py +++ b/snaps/domain/test/volume_tests.py @@ -14,7 +14,68 @@ # limitations under the License. import unittest -from snaps.domain.volume import QoSSpec +from snaps.domain.volume import QoSSpec, VolumeType, VolumeTypeEncryption + + +class VolumeTypeDomainObjectTests(unittest.TestCase): + """ + Tests the construction of the snaps.domain.volume.VolumeType class + """ + + def test_construction_positional(self): + encryption = VolumeTypeEncryption( + 'id-encrypt1', 'id-vol-type1', 'loc1', 'provider1', 'cipher1', 99) + qos_spec = QoSSpec('name', 'id', 'consumer') + + volume_type = VolumeType('name', 'id', True, encryption, qos_spec) + self.assertEqual('name', volume_type.name) + self.assertEqual('id', volume_type.id) + self.assertTrue(volume_type.public) + self.assertEqual(encryption, volume_type.encryption) + self.assertEqual(qos_spec, volume_type.qos_spec) + + def test_construction_named(self): + encryption = VolumeTypeEncryption( + 'id-encrypt1', 'id-vol-type1', 'loc1', 'provider1', 'cipher1', 99) + qos_spec = QoSSpec('name', 'id', 'consumer') + + volume_type = VolumeType( + qos_spec=qos_spec, encryption=encryption, volume_type_id='id', + name='name', public='true') + self.assertEqual('name', volume_type.name) + self.assertEqual('id', volume_type.id) + self.assertTrue(volume_type.public) + self.assertEqual(encryption, volume_type.encryption) + self.assertEqual(qos_spec, volume_type.qos_spec) + + +class VolumeTypeEncryptionObjectTests(unittest.TestCase): + """ + Tests the construction of the snaps.domain.volume.VolumeTypeEncryption + class + """ + + def test_construction_positional(self): + encryption = VolumeTypeEncryption( + 'id-encrypt1', 'id-vol-type1', 'loc1', 'provider1', 'cipher1', 99) + self.assertEqual('id-encrypt1', encryption.id) + self.assertEqual('id-vol-type1', encryption.volume_type_id) + self.assertEqual('loc1', encryption.control_location) + self.assertEqual('provider1', encryption.provider) + self.assertEqual('cipher1', encryption.cipher) + self.assertEqual(99, encryption.key_size) + + def test_construction_named(self): + encryption = VolumeTypeEncryption( + key_size=89, cipher='cipher2', provider='provider2', + control_location='loc2', volume_type_id='id-vol-type2', + volume_encryption_id='id-encrypt2') + self.assertEqual('id-encrypt2', encryption.id) + self.assertEqual('id-vol-type2', encryption.volume_type_id) + self.assertEqual('loc2', encryption.control_location) + self.assertEqual('provider2', encryption.provider) + self.assertEqual('cipher2', encryption.cipher) + self.assertEqual(89, encryption.key_size) class QoSSpecDomainObjectTests(unittest.TestCase): diff --git a/snaps/domain/volume.py b/snaps/domain/volume.py index 9b35c9b..e82a60a 100644 --- a/snaps/domain/volume.py +++ b/snaps/domain/volume.py @@ -14,6 +14,65 @@ # limitations under the License. +class VolumeType: + """ + SNAPS domain object for Volume Types. Should contain attributes that + are shared amongst cloud providers + """ + def __init__(self, name, volume_type_id, public, encryption, qos_spec): + """ + Constructor + :param name: the volume's name + :param volume_type_id: the volume type's id + :param public: True if public + :param encryption: instance of a VolumeTypeEncryption domain object + :param qos_spec: instance of a QoSSpec domain object + """ + self.name = name + self.id = volume_type_id + self.public = public + self.encryption = encryption + self.qos_spec = qos_spec + + def __eq__(self, other): + return (self.name == other.name and self.id == other.id + and self.public == other.public + and self.encryption == other.encryption + and self.qos_spec == other.qos_spec) + + +class VolumeTypeEncryption: + """ + SNAPS domain object for Volume Types. Should contain attributes that + are shared amongst cloud providers + """ + def __init__(self, volume_encryption_id, volume_type_id, + control_location, provider, cipher, key_size): + """ + Constructor + :param volume_encryption_id: the encryption id + :param volume_type_id: the associated volume type's id + :param control_location: front-end | back-end + :param provider: the encryption provider class + :param cipher: the encryption cipher + :param key_size: the encryption key size + """ + self.id = volume_encryption_id + self.volume_type_id = volume_type_id + self.control_location = control_location + self.provider = provider + self.cipher = cipher + self.key_size = key_size + + def __eq__(self, other): + return (self.id == other.id + and self.volume_type_id == other.volume_type_id + and self.control_location == other.control_location + and self.provider == other.provider + and self.cipher == other.cipher + and self.key_size == other.key_size) + + class QoSSpec: """ SNAPS domain object for Volume Types. Should contain attributes that diff --git a/snaps/openstack/create_volume_type.py b/snaps/openstack/create_volume_type.py new file mode 100644 index 0000000..a60bb1e --- /dev/null +++ b/snaps/openstack/create_volume_type.py @@ -0,0 +1,235 @@ +# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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 logging + +import enum +from cinderclient.exceptions import NotFound +from neutronclient.common.utils import str2bool + +from snaps.openstack.openstack_creator import OpenStackVolumeObject +from snaps.openstack.utils import cinder_utils + +__author__ = 'spisarski' + +logger = logging.getLogger('create_volume_type') + + +class OpenStackVolumeType(OpenStackVolumeObject): + """ + Class responsible for managing an volume in OpenStack + """ + + def __init__(self, os_creds, volume_type_settings): + """ + Constructor + :param os_creds: The OpenStack connection credentials + :param volume_type_settings: The volume type settings + :return: + """ + super(self.__class__, self).__init__(os_creds) + + self.volume_type_settings = volume_type_settings + self.__volume_type = None + + def initialize(self): + """ + Loads the existing Volume + :return: The Volume domain object or None + """ + super(self.__class__, self).initialize() + + self.__volume_type = cinder_utils.get_volume_type( + self._cinder, volume_type_settings=self.volume_type_settings) + + return self.__volume_type + + def create(self, block=False): + """ + Creates the volume in OpenStack if it does not already exist and + returns the domain Volume object + :return: The Volume domain object or None + """ + self.initialize() + + if not self.__volume_type: + self.__volume_type = cinder_utils.create_volume_type( + self._cinder, self.volume_type_settings) + logger.info( + 'Created volume type with name - %s', + self.volume_type_settings.name) + + return self.__volume_type + + def clean(self): + """ + Cleanse environment of all artifacts + :return: void + """ + if self.__volume_type: + try: + cinder_utils.delete_volume_type(self._cinder, + self.__volume_type) + except NotFound: + pass + + self.__volume_type = None + + def get_volume_type(self): + """ + Returns the domain Volume object as it was populated when create() was + called + :return: the object + """ + return self.__volume_type + + +class VolumeTypeSettings: + def __init__(self, **kwargs): + """ + Constructor + :param name: the volume's name (required) + :param description: the volume's name (optional) + :param encryption: VolumeTypeEncryptionSettings (optional) + :param qos_spec_name: name of the QoS Spec to associate (optional) + :param public: When True, an image will be created with public + visibility (default - False) + + TODO - Implement project_access parameter that will associate this + VolumeType to a list of project names + """ + + self.name = kwargs.get('name') + self.description = kwargs.get('description') + self.qos_spec_name = kwargs.get('qos_spec_name') + + if 'encryption' in kwargs: + if isinstance(kwargs['encryption'], dict): + self.encryption = VolumeTypeEncryptionSettings( + **kwargs['encryption']) + elif isinstance(kwargs['encryption'], + VolumeTypeEncryptionSettings): + self.encryption = kwargs['encryption'] + else: + self.encryption = None + + if 'public' in kwargs: + if isinstance(kwargs['public'], str): + self.public = str2bool(kwargs['public']) + else: + self.public = kwargs['public'] + else: + self.public = False + + if not self.name: + raise VolumeTypeSettingsError("The attribute name is required") + + def __eq__(self, other): + return (self.name == other.name + and self.description == other.description + and self.qos_spec_name == other.qos_spec_name + and self.encryption == other.encryption + and self.public == other.public) + + +class ControlLocation(enum.Enum): + """ + QoS Specification consumer types + """ + front_end = 'front-end' + back_end = 'back-end' + + +class VolumeTypeEncryptionSettings: + def __init__(self, **kwargs): + """ + Constructor + :param name: the volume's name (required) + :param provider_class: the volume's provider class (e.g. LuksEncryptor) + :param control_location: the notional service where encryption is + performed (e.g., front-end=Nova). The default + value is 'front-end.' + :param cipher: the encryption algorithm/mode to use + (e.g., aes-xts-plain64). If the field is left empty, + the provider default will be used + :param key_size: the size of the encryption key, in bits + (e.g., 128, 256). If the field is left empty, the + provider default will be used + """ + + self.name = kwargs.get('name') + self.provider_class = kwargs.get('provider_class') + self.control_location = kwargs.get('control_location') + if kwargs.get('control_location'): + self.control_location = map_control_location( + kwargs['control_location']) + else: + self.control_location = None + + self.cipher = kwargs.get('cipher') + self.key_size = kwargs.get('key_size') + + if (not self.name or not self.provider_class + or not self.control_location): + raise VolumeTypeSettingsError( + 'The attributes name, provider_class, and control_location ' + 'are required') + + def __eq__(self, other): + return (self.name == other.name + and self.provider_class == other.provider_class + and self.control_location == other.control_location + and self.cipher == other.cipher + and self.key_size == other.key_size) + + +def map_control_location(control_location): + """ + Takes a the protocol value maps it to the Consumer enum. When None return + None + :param control_location: the value to map to the Enum + :return: a ControlLocation enum object + :raise: Exception if control_location parameter is invalid + """ + if not control_location: + return None + elif isinstance(control_location, ControlLocation): + return control_location + else: + proto_str = str(control_location) + if proto_str == 'front-end': + return ControlLocation.front_end + elif proto_str == 'back-end': + return ControlLocation.back_end + else: + raise VolumeTypeSettingsError('Invalid Consumer - ' + proto_str) + + +class VolumeTypeSettingsError(Exception): + """ + Exception to be thrown when an volume settings are incorrect + """ + + def __init__(self, message): + Exception.__init__(self, message) + + +class VolumeTypeCreationError(Exception): + """ + Exception to be thrown when an volume cannot be created + """ + + def __init__(self, message): + Exception.__init__(self, message) diff --git a/snaps/openstack/tests/create_volume_type_tests.py b/snaps/openstack/tests/create_volume_type_tests.py new file mode 100644 index 0000000..93e9351 --- /dev/null +++ b/snaps/openstack/tests/create_volume_type_tests.py @@ -0,0 +1,323 @@ +# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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. +from snaps.openstack.create_qos import QoSSettings, Consumer, OpenStackQoS + +try: + from urllib.request import URLError +except ImportError: + from urllib2 import URLError + +import logging +import unittest +import uuid + +from snaps.openstack import create_volume_type +from snaps.openstack.create_volume_type import ( + VolumeTypeSettings, VolumeTypeSettingsError, VolumeTypeEncryptionSettings, + ControlLocation) +from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase +from snaps.openstack.utils import cinder_utils + +__author__ = 'spisarski' + +logger = logging.getLogger('create_volume_type_tests') + + +class VolumeTypeSettingsUnitTests(unittest.TestCase): + """ + Tests the construction of the VolumeTypeSettings class + """ + + def test_no_params(self): + with self.assertRaises(VolumeTypeSettingsError): + VolumeTypeSettings() + + def test_empty_config(self): + with self.assertRaises(VolumeTypeSettingsError): + VolumeTypeSettings(**dict()) + + def test_name_only(self): + settings = VolumeTypeSettings(name='foo') + self.assertEqual('foo', settings.name) + self.assertIsNone(settings.description) + self.assertIsNone(settings.qos_spec_name) + self.assertIsNone(settings.encryption) + self.assertFalse(settings.public) + + def test_config_with_name_only(self): + settings = VolumeTypeSettings(**{'name': 'foo'}) + self.assertEqual('foo', settings.name) + self.assertIsNone(settings.description) + self.assertIsNone(settings.qos_spec_name) + self.assertIsNone(settings.encryption) + self.assertFalse(settings.public) + + def test_all(self): + encryption_settings = VolumeTypeEncryptionSettings( + name='foo', provider_class='bar', + control_location=ControlLocation.back_end) + settings = VolumeTypeSettings( + name='foo', description='desc', encryption=encryption_settings, + qos_spec_name='spec_name', public=True) + self.assertEqual('foo', settings.name) + self.assertEqual('desc', settings.description) + self.assertEqual('spec_name', settings.qos_spec_name) + self.assertEqual(encryption_settings, settings.encryption) + self.assertTrue(True, settings.public) + + def test_all_string(self): + encryption_settings = { + 'name': 'foo', 'provider_class': 'bar', + 'control_location': 'back-end'} + settings = VolumeTypeSettings( + name='foo', description='desc', encryption=encryption_settings, + qos_spec_name='spec_name', public='true') + self.assertEqual('foo', settings.name) + self.assertEqual('desc', settings.description) + self.assertEqual('spec_name', settings.qos_spec_name) + self.assertEqual(VolumeTypeEncryptionSettings(**encryption_settings), + settings.encryption) + self.assertTrue(settings.public) + + def test_config_all(self): + encryption_settings = { + 'name': 'foo', 'provider_class': 'bar', + 'control_location': 'back-end'} + settings = VolumeTypeSettings( + **{'name': 'foo', 'description': 'desc', + 'encryption': encryption_settings, + 'qos_spec_name': 'spec_name', 'public': 'false'}) + self.assertEqual('foo', settings.name) + self.assertEqual('desc', settings.description) + self.assertEqual('spec_name', settings.qos_spec_name) + self.assertEqual(VolumeTypeEncryptionSettings(**encryption_settings), + settings.encryption) + self.assertFalse(settings.public) + + +class CreateSimpleVolumeTypeSuccessTests(OSIntegrationTestCase): + """ + Test for the OpenStackVolumeType class defined in create_volume_type.py + without any QoS Specs or Encryption + """ + + def setUp(self): + """ + Instantiates the CreateVolumeType object that is responsible for + downloading and creating an OS volume type file within OpenStack + """ + super(self.__class__, self).__start__() + + guid = uuid.uuid4() + self.volume_type_settings = VolumeTypeSettings( + name=self.__class__.__name__ + '-' + str(guid)) + + self.cinder = cinder_utils.cinder_client(self.os_creds) + self.volume_type_creator = None + + def tearDown(self): + """ + Cleans the volume type + """ + if self.volume_type_creator: + self.volume_type_creator.clean() + + super(self.__class__, self).__clean__() + + def test_create_volume_type(self): + """ + Tests the creation of an OpenStack volume. + """ + # Create VolumeType + self.volume_type_creator = create_volume_type.OpenStackVolumeType( + self.os_creds, self.volume_type_settings) + created_volume_type = self.volume_type_creator.create() + self.assertIsNotNone(created_volume_type) + self.assertEqual(self.volume_type_settings.name, + created_volume_type.name) + + retrieved_volume_type1 = cinder_utils.get_volume_type( + self.cinder, volume_type_settings=self.volume_type_settings) + self.assertIsNotNone(retrieved_volume_type1) + self.assertEqual(created_volume_type, retrieved_volume_type1) + + retrieved_volume_type2 = cinder_utils.get_volume_type_by_id( + self.cinder, created_volume_type.id) + self.assertEqual(created_volume_type, retrieved_volume_type2) + + def test_create_delete_volume_type(self): + """ + Tests the creation then deletion of an OpenStack volume type to ensure + clean() does not raise an Exception. + """ + # Create VolumeType + self.volume_type_creator = create_volume_type.OpenStackVolumeType( + self.os_creds, self.volume_type_settings) + created_volume_type = self.volume_type_creator.create() + self.assertIsNotNone(created_volume_type) + + retrieved_volume_type = cinder_utils.get_volume_type( + self.cinder, volume_type_settings=self.volume_type_settings) + self.assertIsNotNone(retrieved_volume_type) + self.assertEqual(created_volume_type, retrieved_volume_type) + + # Delete VolumeType manually + cinder_utils.delete_volume_type(self.cinder, created_volume_type) + + self.assertIsNone(cinder_utils.get_volume_type( + self.cinder, volume_type_settings=self.volume_type_settings)) + + # Must not throw an exception when attempting to cleanup non-existent + # volume_type + self.volume_type_creator.clean() + self.assertIsNone(self.volume_type_creator.get_volume_type()) + + def test_create_same_volume_type(self): + """ + Tests the creation of an OpenStack volume_type when one already exists. + """ + # Create VolumeType + self.volume_type_creator = create_volume_type.OpenStackVolumeType( + self.os_creds, self.volume_type_settings) + volume_type1 = self.volume_type_creator.create() + + retrieved_volume_type = cinder_utils.get_volume_type( + self.cinder, volume_type_settings=self.volume_type_settings) + self.assertEqual(volume_type1, retrieved_volume_type) + + # Should be retrieving the instance data + os_volume_type_2 = create_volume_type.OpenStackVolumeType( + self.os_creds, self.volume_type_settings) + volume_type2 = os_volume_type_2.create() + self.assertEqual(volume_type2, volume_type2) + + +class CreateVolumeTypeComplexTests(OSIntegrationTestCase): + """ + Test cases for the CreateVolumeType class that include QoS Specs and/or + encryption + """ + + def setUp(self): + super(self.__class__, self).__start__() + + self.cinder = cinder_utils.cinder_client(self.os_creds) + + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + + self.volume_type_name = guid + '-vol_type' + self.volume_type_creator = None + + qos_settings = QoSSettings( + name=guid + '-qos-spec', consumer=Consumer.both) + self.qos_creator = OpenStackQoS(self.os_creds, qos_settings) + self.qos_creator.create() + + def tearDown(self): + if self.volume_type_creator: + self.volume_type_creator.clean() + + if self.qos_creator: + self.qos_creator.clean() + + super(self.__class__, self).__clean__() + + def test_volume_type_with_qos(self): + """ + Creates a Volume Type object with an associated QoS Spec + """ + self.volume_type_creator = create_volume_type.OpenStackVolumeType( + self.os_creds, + VolumeTypeSettings( + name=self.volume_type_name, + qos_spec_name=self.qos_creator.qos_settings.name)) + + vol_type = self.volume_type_creator.create() + self.assertEqual(self.volume_type_creator.volume_type_settings.name, + vol_type.name) + self.assertEqual(self.volume_type_creator.volume_type_settings.name, + vol_type.name) + self.assertIsNotNone(vol_type.qos_spec) + self.assertEqual( + self.volume_type_creator.volume_type_settings.qos_spec_name, + vol_type.qos_spec.name) + self.assertIsNone(vol_type.encryption) + + vol_type_query = cinder_utils.get_volume_type_by_id( + self.cinder, vol_type.id) + self.assertIsNotNone(vol_type_query) + self.assertEqual(vol_type, vol_type_query) + + def test_volume_type_with_encryption(self): + """ + Creates a Volume Type object with encryption + """ + encryption_settings = VolumeTypeEncryptionSettings( + name='foo', provider_class='bar', + control_location=ControlLocation.back_end) + self.volume_type_creator = create_volume_type.OpenStackVolumeType( + self.os_creds, + VolumeTypeSettings( + name=self.volume_type_name, + encryption=encryption_settings)) + + vol_type = self.volume_type_creator.create() + self.assertEqual(self.volume_type_creator.volume_type_settings.name, + vol_type.name) + self.assertEqual(self.volume_type_creator.volume_type_settings.name, + vol_type.name) + self.assertIsNone(vol_type.qos_spec) + self.assertIsNotNone(vol_type.encryption) + + self.assertEqual(encryption_settings.control_location.value, + vol_type.encryption.control_location) + + vol_type_query = cinder_utils.get_volume_type_by_id( + self.cinder, vol_type.id) + self.assertIsNotNone(vol_type_query) + self.assertEqual(vol_type, vol_type_query) + + def test_volume_type_with_qos_and_encryption(self): + """ + Creates a Volume Type object with encryption and an associated QoS Spec + """ + encryption_settings = VolumeTypeEncryptionSettings( + name='foo', provider_class='bar', + control_location=ControlLocation.back_end) + self.volume_type_creator = create_volume_type.OpenStackVolumeType( + self.os_creds, + VolumeTypeSettings( + name=self.volume_type_name, + encryption=encryption_settings, + qos_spec_name=self.qos_creator.qos_settings.name)) + + vol_type = self.volume_type_creator.create() + self.assertEqual(self.volume_type_creator.volume_type_settings.name, + vol_type.name) + self.assertEqual(self.volume_type_creator.volume_type_settings.name, + vol_type.name) + self.assertIsNotNone(vol_type.qos_spec) + self.assertEqual( + self.volume_type_creator.volume_type_settings.qos_spec_name, + vol_type.qos_spec.name) + self.assertIsNotNone(vol_type.encryption) + + self.assertEqual(encryption_settings.control_location.value, + vol_type.encryption.control_location) + + vol_type_query = cinder_utils.get_volume_type_by_id( + self.cinder, vol_type.id) + self.assertIsNotNone(vol_type_query) + self.assertEqual(vol_type, vol_type_query) diff --git a/snaps/openstack/utils/cinder_utils.py b/snaps/openstack/utils/cinder_utils.py index 5f847a1..d13277d 100644 --- a/snaps/openstack/utils/cinder_utils.py +++ b/snaps/openstack/utils/cinder_utils.py @@ -15,15 +15,15 @@ import logging from cinderclient.client import Client +from cinderclient.exceptions import NotFound -from snaps.domain.volume import QoSSpec +from snaps.domain.volume import QoSSpec, VolumeType, VolumeTypeEncryption from snaps.openstack.utils import keystone_utils __author__ = 'spisarski' logger = logging.getLogger('cinder_utils') -VERSION_1 = 1 VERSION_2 = 2 VERSION_3 = 3 @@ -42,7 +42,172 @@ def cinder_client(os_creds): region_name=os_creds.region_name) -def get_qos(cinder, qos_name=None, qos_settings=None): +def get_volume_type(cinder, volume_type_name=None, volume_type_settings=None): + """ + Returns an OpenStack volume type object for a given name + :param cinder: the Cinder client + :param volume_type_name: the volume type name to lookup + :param volume_type_settings: the volume type settings used for lookups + :return: the volume type object or None + """ + if not volume_type_name and not volume_type_settings: + return None + + if volume_type_settings: + volume_type_name = volume_type_settings.name + + volume_types = cinder.volume_types.list() + for vol_type in volume_types: + if vol_type.name == volume_type_name: + encryption = get_volume_encryption_by_type(cinder, vol_type) + return VolumeType(vol_type.name, vol_type.id, vol_type.is_public, + encryption, None) + + +def __get_os_volume_type_by_id(cinder, volume_type_id): + """ + Returns an OpenStack volume type object for a given name + :param cinder: the Cinder client + :param volume_type_id: the volume_type ID to lookup + :return: the SNAPS-OO Domain Volume object or None + """ + try: + return cinder.volume_types.get(volume_type_id) + except NotFound: + logger.info('Volume with ID [%s] does not exist', + volume_type_id) + + +def get_volume_type_by_id(cinder, volume_type_id): + """ + Returns an OpenStack volume type object for a given name + :param cinder: the Cinder client + :param volume_type_id: the volume_type ID to lookup + :return: the SNAPS-OO Domain Volume object or None + """ + os_vol_type = __get_os_volume_type_by_id(cinder, volume_type_id) + if os_vol_type: + temp_vol_type = VolumeType(os_vol_type.name, os_vol_type.id, + os_vol_type.is_public, None, None) + encryption = get_volume_encryption_by_type(cinder, temp_vol_type) + + qos_spec = None + if os_vol_type.qos_specs_id: + qos_spec = get_qos_by_id(cinder, os_vol_type.qos_specs_id) + + return VolumeType(os_vol_type.name, os_vol_type.id, + os_vol_type.is_public, encryption, qos_spec) + + +def create_volume_type(cinder, type_settings): + """ + Creates and returns OpenStack volume type object with an external URL + :param cinder: the cinder client + :param type_settings: the volume type settings object + :return: the volume type domain object + :raise Exception if using a file and it cannot be found + """ + vol_type = cinder.volume_types.create( + type_settings.name, type_settings.description, + type_settings.public) + + vol_encryption = None + if type_settings.encryption: + try: + vol_encryption = create_volume_encryption( + cinder, vol_type, type_settings.encryption) + except Exception as e: + logger.warn('Error creating volume encryption - %s', e) + + qos_spec = None + if type_settings.qos_spec_name: + try: + qos_spec = get_qos(cinder, qos_name=type_settings.qos_spec_name) + cinder.qos_specs.associate(qos_spec, vol_type.id) + except NotFound as e: + logger.warn('Unable to locate qos_spec named %s - %s', + type_settings.qos_spec_name, e) + + return VolumeType(vol_type.name, vol_type.id, vol_type.is_public, + vol_encryption, qos_spec) + + +def delete_volume_type(cinder, vol_type): + """ + Deletes an volume from OpenStack + :param cinder: the cinder client + :param vol_type: the VolumeType domain object + """ + logger.info('Deleting volume named - %s', vol_type.name) + cinder.volume_types.delete(vol_type.id) + + +def get_volume_encryption_by_type(cinder, volume_type): + """ + Returns an OpenStack volume type object for a given name + :param cinder: the Cinder client + :param volume_type: the VolumeType domain object + :return: the VolumeEncryption domain object or None + """ + os_vol_type = __get_os_volume_type_by_id(cinder, volume_type.id) + encryption = cinder.volume_encryption_types.get(os_vol_type) + if hasattr(encryption, 'encryption_id'): + cipher = None + if hasattr(encryption, 'cipher'): + cipher = encryption.cipher + key_size = None + if hasattr(encryption, 'key_size'): + key_size = encryption.key_size + return VolumeTypeEncryption( + encryption.encryption_id, encryption.volume_type_id, + encryption.control_location, encryption.provider, cipher, key_size) + + +def create_volume_encryption(cinder, volume_type, encryption_settings): + """ + Creates and returns OpenStack volume type object with an external URL + :param cinder: the cinder client + :param volume_type: the VolumeType object to associate the encryption + :param encryption_settings: the volume type encryption settings object + :return: the VolumeTypeEncryption domain object + """ + specs = {'name': encryption_settings.name, + 'provider': encryption_settings.provider_class} + if encryption_settings.key_size: + specs['key_size'] = encryption_settings.key_size + if encryption_settings.provider_class: + specs['provider_class'] = encryption_settings.provider_class + if encryption_settings.control_location: + specs['control_location'] = encryption_settings.control_location.value + if encryption_settings.cipher: + specs['cipher'] = encryption_settings.cipher + + encryption = cinder.volume_encryption_types.create(volume_type.id, specs) + + cipher = None + if hasattr(encryption, 'cipher'): + cipher = encryption.cipher + key_size = None + if hasattr(encryption, 'key_size'): + key_size = encryption.key_size + return VolumeTypeEncryption( + encryption.encryption_id, encryption.volume_type_id, + encryption.control_location, encryption.provider, cipher, key_size) + + +def delete_volume_type_encryption(cinder, vol_type): + """ + Deletes an volume from OpenStack + :param cinder: the cinder client + :param vol_type: the associated VolumeType domain object + """ + logger.info('Deleting volume encryption for volume type - %s', + vol_type.name) + os_vol_type = __get_os_volume_type_by_id(cinder, vol_type.id) + cinder.volume_encryption_types.delete(os_vol_type) + + +def __get_os_qos(cinder, qos_name=None, qos_settings=None): """ Returns an OpenStack QoS object for a given name :param cinder: the Cinder client @@ -53,20 +218,27 @@ def get_qos(cinder, qos_name=None, qos_settings=None): if not qos_name and not qos_settings: return None - qos_name = qos_name if qos_settings: qos_name = qos_settings.name qoss = cinder.qos_specs.list() for qos in qoss: if qos.name == qos_name: - if qos_settings: - if qos_settings.consumer.value == qos.consumer: - return QoSSpec(name=qos.name, spec_id=qos.id, - consumer=qos.consumer) - else: - return QoSSpec(name=qos.name, spec_id=qos.id, - consumer=qos.consumer) + return qos + + +def get_qos(cinder, qos_name=None, qos_settings=None): + """ + Returns an OpenStack QoS object for a given name + :param cinder: the Cinder client + :param qos_name: the qos name to lookup + :param qos_settings: the qos settings used for lookups + :return: the qos object or None + """ + os_qos = __get_os_qos(cinder, qos_name, qos_settings) + if os_qos: + return QoSSpec(name=os_qos.name, spec_id=os_qos.id, + consumer=os_qos.consumer) def get_qos_by_id(cinder, qos_id): @@ -102,9 +274,3 @@ def delete_qos(cinder, qos): """ logger.info('Deleting QoS named - %s', qos.name) cinder.qos_specs.delete(qos.id) - - -class CinderException(Exception): - """ - Exception when calls to the Cinder client cannot be served properly - """ diff --git a/snaps/openstack/utils/tests/cinder_utils_tests.py b/snaps/openstack/utils/tests/cinder_utils_tests.py index e6ad2a0..a45167e 100644 --- a/snaps/openstack/utils/tests/cinder_utils_tests.py +++ b/snaps/openstack/utils/tests/cinder_utils_tests.py @@ -15,9 +15,11 @@ import logging import uuid -from cinderclient.exceptions import NotFound +from cinderclient.exceptions import NotFound, BadRequest from snaps.openstack.create_qos import QoSSettings, Consumer +from snaps.openstack.create_volume_type import ( + VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation) from snaps.openstack.tests import validation_utils from snaps.openstack.tests.os_source_file_test import OSComponentTestCase from snaps.openstack.utils import cinder_utils @@ -86,15 +88,14 @@ class CinderUtilsQoSTests(OSComponentTestCase): """ qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, consumer=Consumer.both) - self.qos = cinder_utils.create_qos( - self.cinder, qos_settings) + self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings) self.assertIsNotNone(qos1) validation_utils.objects_equivalent(self.qos, qos1) - qos2 = cinder_utils.get_qos(self.cinder, qos_name=qos_settings.name) + qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id) self.assertIsNotNone(qos2) validation_utils.objects_equivalent(self.qos, qos2) @@ -104,15 +105,14 @@ class CinderUtilsQoSTests(OSComponentTestCase): """ qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, consumer=Consumer.front_end) - self.qos = cinder_utils.create_qos( - self.cinder, qos_settings) + self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings) self.assertIsNotNone(qos1) validation_utils.objects_equivalent(self.qos, qos1) - qos2 = cinder_utils.get_qos(self.cinder, qos_name=qos_settings.name) + qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id) self.assertIsNotNone(qos2) validation_utils.objects_equivalent(self.qos, qos2) @@ -122,15 +122,14 @@ class CinderUtilsQoSTests(OSComponentTestCase): """ qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, consumer=Consumer.back_end) - self.qos = cinder_utils.create_qos( - self.cinder, qos_settings) + self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings) self.assertIsNotNone(qos1) validation_utils.objects_equivalent(self.qos, qos1) - qos2 = cinder_utils.get_qos(self.cinder, qos_name=qos_settings.name) + qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id) self.assertIsNotNone(qos2) validation_utils.objects_equivalent(self.qos, qos2) @@ -139,8 +138,7 @@ class CinderUtilsQoSTests(OSComponentTestCase): Tests the cinder_utils.create_qos() """ qos_settings = QoSSettings(name=self.qos_name, consumer=Consumer.both) - self.qos = cinder_utils.create_qos( - self.cinder, qos_settings) + self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) self.assertEqual(self.qos_name, self.qos.name) @@ -152,3 +150,301 @@ class CinderUtilsQoSTests(OSComponentTestCase): cinder_utils.delete_qos(self.cinder, self.qos) self.assertIsNone(cinder_utils.get_qos( self.cinder, qos_settings=qos_settings)) + + +class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase): + """ + Tests the creation of a Volume Type without any external settings such as + QoS Specs or encryption + """ + + def setUp(self): + """ + Instantiates the CreateVolume object that is responsible for + downloading and creating an OS volume file within OpenStack + """ + guid = uuid.uuid4() + volume_type_name = self.__class__.__name__ + '-' + str(guid) + self.volume_type_settings = VolumeTypeSettings(name=volume_type_name) + self.volume_type = None + self.cinder = cinder_utils.cinder_client(self.os_creds) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.volume_type: + try: + cinder_utils.delete_volume_type(self.cinder, self.volume_type) + except NotFound: + pass + + def test_create_simple_volume_type(self): + """ + Tests the cinder_utils.create_volume_type(), get_volume_type(), and + get_volume_type_by_id() + """ + self.volume_type = cinder_utils.create_volume_type( + self.cinder, self.volume_type_settings) + self.assertIsNotNone(self.volume_type) + self.assertEqual(self.volume_type_settings.name, self.volume_type.name) + + volume_type1 = cinder_utils.get_volume_type( + self.cinder, volume_type_settings=self.volume_type_settings) + self.assertEquals(self.volume_type, volume_type1) + self.assertEquals(self.volume_type_settings.public, + volume_type1.public) + + volume_type2 = cinder_utils.get_volume_type_by_id( + self.cinder, volume_type1.id) + self.assertEquals(self.volume_type, volume_type2) + self.assertIsNone(self.volume_type.encryption) + + def test_create_delete_volume_type(self): + """ + Primarily tests the cinder_utils.delete_volume() + """ + self.volume_type = cinder_utils.create_volume_type( + self.cinder, self.volume_type_settings) + self.assertIsNotNone(self.volume_type) + self.assertEqual(self.volume_type_settings.name, self.volume_type.name) + + volume_type = cinder_utils.get_volume_type( + self.cinder, volume_type_settings=self.volume_type_settings) + self.assertIsNotNone(volume_type) + validation_utils.objects_equivalent(self.volume_type, volume_type) + self.assertIsNone(self.volume_type.encryption) + + cinder_utils.delete_volume_type(self.cinder, self.volume_type) + self.assertIsNone(cinder_utils.get_volume_type_by_id( + self.cinder, self.volume_type.id)) + + +class CinderUtilsAddEncryptionTests(OSComponentTestCase): + """ + Tests the creation of an encryption and association to and existing + VolumeType object + """ + + def setUp(self): + """ + Instantiates the CreateVolume object that is responsible for + downloading and creating an OS volume file within OpenStack + """ + guid = uuid.uuid4() + self.encryption_name = self.__class__.__name__ + '-' + str(guid) + self.encryption = None + + self.cinder = cinder_utils.cinder_client(self.os_creds) + + volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type' + self.volume_type = cinder_utils.create_volume_type( + self.cinder, VolumeTypeSettings(name=volume_type_name)) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.encryption: + try: + cinder_utils.delete_volume_type_encryption( + self.cinder, self.volume_type) + except NotFound: + pass + + if self.volume_type: + try: + cinder_utils.delete_volume_type(self.cinder, self.volume_type) + except NotFound: + pass + + def test_create_simple_encryption(self): + """ + Tests the cinder_utils.create_volume_encryption(), + get_volume_encryption(), and get_volume_encryption_by_id() + """ + encryption_settings = VolumeTypeEncryptionSettings( + name=self.encryption_name, provider_class='foo', + control_location=ControlLocation.front_end) + self.encryption = cinder_utils.create_volume_encryption( + self.cinder, self.volume_type, encryption_settings) + self.assertIsNotNone(self.encryption) + self.assertEqual('foo', self.encryption.provider) + self.assertEqual(ControlLocation.front_end.value, + self.encryption.control_location) + + encryption1 = cinder_utils.get_volume_encryption_by_type( + self.cinder, self.volume_type) + self.assertEquals(self.encryption, encryption1) + + def test_create_delete_encryption(self): + """ + Primarily tests the cinder_utils.delete_volume_type_encryption() + """ + encryption_settings = VolumeTypeEncryptionSettings( + name=self.encryption_name, provider_class='LuksEncryptor', + control_location=ControlLocation.back_end) + self.encryption = cinder_utils.create_volume_encryption( + self.cinder, self.volume_type, encryption_settings) + self.assertIsNotNone(self.encryption) + self.assertEqual('LuksEncryptor', self.encryption.provider) + self.assertEqual(ControlLocation.back_end.value, + self.encryption.control_location) + + encryption1 = cinder_utils.get_volume_encryption_by_type( + self.cinder, self.volume_type) + self.assertEquals(self.encryption, encryption1) + + cinder_utils.delete_volume_type_encryption( + self.cinder, self.volume_type) + + encryption2 = cinder_utils.get_volume_encryption_by_type( + self.cinder, self.volume_type) + self.assertIsNone(encryption2) + + def test_create_with_all_attrs(self): + """ + Tests the cinder_utils.create_volume_encryption() with all valid + settings + """ + encryption_settings = VolumeTypeEncryptionSettings( + name=self.encryption_name, provider_class='foo', + cipher='bar', control_location=ControlLocation.back_end, + key_size=1) + self.encryption = cinder_utils.create_volume_encryption( + self.cinder, self.volume_type, encryption_settings) + self.assertIsNotNone(self.encryption) + self.assertEqual('foo', self.encryption.provider) + self.assertEqual('bar', self.encryption.cipher) + self.assertEqual(1, self.encryption.key_size) + self.assertEqual(ControlLocation.back_end.value, + self.encryption.control_location) + + encryption1 = cinder_utils.get_volume_encryption_by_type( + self.cinder, self.volume_type) + self.assertEquals(self.encryption, encryption1) + + def test_create_bad_key_size(self): + """ + Tests the cinder_utils.create_volume_encryption() raises an exception + when the provider class does not exist + """ + encryption_settings = VolumeTypeEncryptionSettings( + name=self.encryption_name, provider_class='foo', + cipher='bar', control_location=ControlLocation.back_end, + key_size=-1) + + with self.assertRaises(BadRequest): + self.encryption = cinder_utils.create_volume_encryption( + self.cinder, self.volume_type, encryption_settings) + + +class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase): + """ + Tests to ensure that a volume type can have a QoS Spec added to it + """ + + def setUp(self): + """ + Creates objects for testing cinder_utils.py + """ + guid = uuid.uuid4() + self.qos_name = self.__class__.__name__ + '-' + str(guid) + '-qos' + self.vol_type_name = self.__class__.__name__ + '-' + str(guid) + self.specs = {'foo': 'bar'} + self.cinder = cinder_utils.cinder_client(self.os_creds) + qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, + consumer=Consumer.both) + self.qos = cinder_utils.create_qos(self.cinder, qos_settings) + self.volume_type = None + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.volume_type: + if self.volume_type.encryption: + try: + cinder_utils.delete_volume_type_encryption( + self.cinder, self.volume_type) + except NotFound: + pass + try: + cinder_utils.delete_volume_type(self.cinder, self.volume_type) + except NotFound: + pass + + if self.qos: + try: + cinder_utils.delete_qos(self.cinder, self.qos) + except NotFound: + pass + + def test_create_with_encryption(self): + """ + Tests the cinder_utils.create_volume_type() where encryption has been + configured + """ + encryption_settings = VolumeTypeEncryptionSettings( + name='foo', provider_class='bar', + control_location=ControlLocation.back_end) + volume_type_settings = VolumeTypeSettings( + name=self.vol_type_name, encryption=encryption_settings) + self.volume_type = cinder_utils.create_volume_type( + self.cinder, volume_type_settings) + + vol_encrypt = cinder_utils.get_volume_encryption_by_type( + self.cinder, self.volume_type) + self.assertIsNotNone(vol_encrypt) + self.assertIsNone(self.volume_type.qos_spec) + self.assertEqual(self.volume_type.encryption, vol_encrypt) + self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id) + + def test_create_with_qos(self): + """ + Tests the cinder_utils.create_volume_type() with an associated QoS Spec + """ + volume_type_settings = VolumeTypeSettings( + name=self.vol_type_name, qos_spec_name=self.qos_name) + self.volume_type = cinder_utils.create_volume_type( + self.cinder, volume_type_settings) + + self.assertIsNotNone(self.volume_type) + self.assertIsNone(self.volume_type.encryption) + self.assertIsNotNone(self.volume_type.qos_spec) + self.assertEqual(self.qos.id, self.volume_type.qos_spec.id) + + def test_create_with_invalid_qos(self): + """ + Tests the cinder_utils.create_volume_type() when the QoS Spec name + does not exist + """ + volume_type_settings = VolumeTypeSettings( + name=self.vol_type_name, qos_spec_name='foo') + + self.volume_type = cinder_utils.create_volume_type( + self.cinder, volume_type_settings) + + self.assertIsNone(self.volume_type.qos_spec) + + def test_create_with_qos_and_encryption(self): + """ + Tests the cinder_utils.create_volume_type() with encryption and an + associated QoS Spec + """ + encryption_settings = VolumeTypeEncryptionSettings( + name='foo', provider_class='bar', + control_location=ControlLocation.back_end) + volume_type_settings = VolumeTypeSettings( + name=self.vol_type_name, qos_spec_name=self.qos_name, + encryption=encryption_settings) + self.volume_type = cinder_utils.create_volume_type( + self.cinder, volume_type_settings) + + self.assertIsNotNone(self.volume_type) + vol_encrypt = cinder_utils.get_volume_encryption_by_type( + self.cinder, self.volume_type) + self.assertIsNotNone(vol_encrypt) + self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id) + self.assertIsNotNone(self.volume_type.qos_spec) + self.assertEqual(self.qos.id, self.volume_type.qos_spec.id) diff --git a/snaps/test_suite_builder.py b/snaps/test_suite_builder.py index b71fdf1..a1b72aa 100644 --- a/snaps/test_suite_builder.py +++ b/snaps/test_suite_builder.py @@ -32,7 +32,9 @@ from snaps.domain.test.stack_tests import ( from snaps.domain.test.user_tests import UserDomainObjectTests from snaps.domain.test.vm_inst_tests import ( VmInstDomainObjectTests, FloatingIpDomainObjectTests) -from snaps.domain.test.volume_tests import QoSSpecDomainObjectTests +from snaps.domain.test.volume_tests import ( + QoSSpecDomainObjectTests, VolumeTypeDomainObjectTests, + VolumeTypeEncryptionObjectTests) from snaps.openstack.tests.conf.os_credentials_tests import ( ProxySettingsUnitTests, OSCredsUnitTests) from snaps.openstack.tests.create_flavor_tests import ( @@ -55,8 +57,8 @@ from snaps.openstack.tests.create_network_tests import ( from snaps.openstack.tests.create_project_tests import ( CreateProjectSuccessTests, ProjectSettingsUnitTests, CreateProjectUserTests) -from snaps.openstack.tests.create_qos_tests import (QoSSettingsUnitTests, - CreateQoSTests) +from snaps.openstack.tests.create_qos_tests import ( + QoSSettingsUnitTests, CreateQoSTests) from snaps.openstack.tests.create_router_tests import ( CreateRouterSuccessTests, CreateRouterNegativeTests, RouterSettingsUnitTests) @@ -68,10 +70,14 @@ from snaps.openstack.tests.create_stack_tests import ( CreateComplexStackTests) from snaps.openstack.tests.create_user_tests import ( UserSettingsUnitTests, CreateUserSuccessTests) +from snaps.openstack.tests.create_volume_type_tests import ( + VolumeTypeSettingsUnitTests, CreateSimpleVolumeTypeSuccessTests, + CreateVolumeTypeComplexTests) from snaps.openstack.tests.os_source_file_test import ( OSComponentTestCase, OSIntegrationTestCase) -from snaps.openstack.utils.tests.cinder_utils_tests import (CinderSmokeTests, - CinderUtilsQoSTests) +from snaps.openstack.utils.tests.cinder_utils_tests import ( + CinderSmokeTests, CinderUtilsQoSTests, CinderUtilsSimpleVolumeTypeTests, + CinderUtilsAddEncryptionTests, CinderUtilsVolumeTypeCompleteTests) from snaps.openstack.utils.tests.glance_utils_tests import ( GlanceSmokeTests, GlanceUtilsTests) from snaps.openstack.utils.tests.heat_utils_tests import ( @@ -168,6 +174,10 @@ def add_unit_tests(suite): ResourceDomainObjectTests)) suite.addTest(unittest.TestLoader().loadTestsFromTestCase( StackSettingsUnitTests)) + suite.addTest(unittest.TestLoader().loadTestsFromTestCase( + VolumeTypeDomainObjectTests)) + suite.addTest(unittest.TestLoader().loadTestsFromTestCase( + VolumeTypeEncryptionObjectTests)) suite.addTest(unittest.TestLoader().loadTestsFromTestCase( QoSSpecDomainObjectTests)) suite.addTest(unittest.TestLoader().loadTestsFromTestCase( @@ -176,6 +186,8 @@ def add_unit_tests(suite): FloatingIpDomainObjectTests)) suite.addTest(unittest.TestLoader().loadTestsFromTestCase( QoSSettingsUnitTests)) + suite.addTest(unittest.TestLoader().loadTestsFromTestCase( + VolumeTypeSettingsUnitTests)) def add_openstack_client_tests(suite, os_creds, ext_net_name, @@ -299,6 +311,18 @@ def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True, CinderUtilsQoSTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level, image_metadata=image_metadata)) + suite.addTest(OSComponentTestCase.parameterize( + CinderUtilsSimpleVolumeTypeTests, os_creds=os_creds, + ext_net_name=ext_net_name, log_level=log_level, + image_metadata=image_metadata)) + suite.addTest(OSComponentTestCase.parameterize( + CinderUtilsAddEncryptionTests, os_creds=os_creds, + ext_net_name=ext_net_name, log_level=log_level, + image_metadata=image_metadata)) + suite.addTest(OSComponentTestCase.parameterize( + CinderUtilsVolumeTypeCompleteTests, os_creds=os_creds, + ext_net_name=ext_net_name, log_level=log_level, + image_metadata=image_metadata)) def add_openstack_integration_tests(suite, os_creds, ext_net_name, @@ -383,6 +407,16 @@ def add_openstack_integration_tests(suite, os_creds, ext_net_name, ext_net_name=ext_net_name, use_keystone=use_keystone, flavor_metadata=flavor_metadata, image_metadata=image_metadata, log_level=log_level)) + suite.addTest(OSIntegrationTestCase.parameterize( + CreateSimpleVolumeTypeSuccessTests, os_creds=os_creds, + ext_net_name=ext_net_name, use_keystone=use_keystone, + flavor_metadata=flavor_metadata, image_metadata=image_metadata, + log_level=log_level)) + suite.addTest(OSIntegrationTestCase.parameterize( + CreateVolumeTypeComplexTests, os_creds=os_creds, + ext_net_name=ext_net_name, use_keystone=use_keystone, + flavor_metadata=flavor_metadata, image_metadata=image_metadata, + log_level=log_level)) # VM Instances suite.addTest(OSIntegrationTestCase.parameterize( -- cgit 1.2.3-korg