diff options
Diffstat (limited to 'snaps/openstack/utils/tests/cinder_utils_tests.py')
| -rw-r--r-- | snaps/openstack/utils/tests/cinder_utils_tests.py | 190 |
1 files changed, 160 insertions, 30 deletions
diff --git a/snaps/openstack/utils/tests/cinder_utils_tests.py b/snaps/openstack/utils/tests/cinder_utils_tests.py index a45167e..073d574 100644 --- a/snaps/openstack/utils/tests/cinder_utils_tests.py +++ b/snaps/openstack/utils/tests/cinder_utils_tests.py @@ -15,14 +15,18 @@ import logging import uuid +import time 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.config.volume import VolumeConfig +from snaps.config.volume_type import ( + VolumeTypeConfig, ControlLocation, VolumeTypeEncryptionConfig) +from snaps.config.qos import Consumer, QoSConfig +from snaps.openstack import create_volume +from snaps.openstack.create_qos import Consumer from snaps.openstack.tests import validation_utils from snaps.openstack.tests.os_source_file_test import OSComponentTestCase -from snaps.openstack.utils import cinder_utils +from snaps.openstack.utils import cinder_utils, keystone_utils __author__ = 'spisarski' @@ -39,7 +43,7 @@ class CinderSmokeTests(OSComponentTestCase): """ Tests to ensure that the proper credentials can connect. """ - cinder = cinder_utils.cinder_client(self.os_creds) + cinder = cinder_utils.cinder_client(self.os_creds, self.os_session) volumes = cinder.volumes.list() self.assertIsNotNone(volumes) self.assertTrue(isinstance(volumes, list)) @@ -57,6 +61,120 @@ class CinderSmokeTests(OSComponentTestCase): cinder.volumes.list() +class CinderUtilsVolumeTests(OSComponentTestCase): + """ + Test for the CreateVolume class defined in create_volume.py + """ + + def setUp(self): + """ + Instantiates the CreateVolume object that is responsible for + downloading and creating an OS volume file within OpenStack + """ + guid = uuid.uuid4() + self.volume_name = self.__class__.__name__ + '-' + str(guid) + self.volume = None + self.cinder = cinder_utils.cinder_client( + self.os_creds, self.os_session) + self.keystone = keystone_utils.keystone_client( + self.os_creds, self.os_session) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.volume: + try: + cinder_utils.delete_volume(self.cinder, self.volume) + except NotFound: + pass + + self.assertTrue(volume_deleted(self.cinder, self.volume)) + + def test_create_simple_volume(self): + """ + Tests the cinder_utils.create_volume() + """ + volume_settings = VolumeConfig(name=self.volume_name) + self.volume = cinder_utils.create_volume( + self.cinder, self.keystone, volume_settings) + self.assertIsNotNone(self.volume) + self.assertEqual(self.volume_name, self.volume.name) + + self.assertTrue(volume_active(self.cinder, self.volume)) + + volume = cinder_utils.get_volume( + self.cinder, self.keystone, volume_settings=volume_settings, + project_name=self.os_creds.project_name) + self.assertIsNotNone(volume) + validation_utils.objects_equivalent(self.volume, volume) + + def test_create_delete_volume(self): + """ + Tests the cinder_utils.create_volume() + """ + volume_settings = VolumeConfig(name=self.volume_name) + self.volume = cinder_utils.create_volume( + self.cinder, self.keystone, volume_settings) + self.assertIsNotNone(self.volume) + self.assertEqual(self.volume_name, self.volume.name) + + self.assertTrue(volume_active(self.cinder, self.volume)) + + volume = cinder_utils.get_volume( + self.cinder, self.keystone, volume_settings=volume_settings, + project_name=self.os_creds.project_name) + self.assertIsNotNone(volume) + validation_utils.objects_equivalent(self.volume, volume) + + cinder_utils.delete_volume(self.cinder, self.volume) + self.assertTrue(volume_deleted(self.cinder, self.volume)) + self.assertIsNone( + cinder_utils.get_volume( + self.cinder, self.keystone, volume_settings, + project_name=self.os_creds.project_name)) + + +def volume_active(cinder, volume): + """ + Returns true if volume becomes active + :param cinder: + :param volume: + :return: + """ + end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT + while time.time() < end_time: + status = cinder_utils.get_volume_status(cinder, volume) + if status == create_volume.STATUS_ACTIVE: + return True + elif status == create_volume.STATUS_FAILED: + return False + time.sleep(3) + + return False + + +def volume_deleted(cinder, volume): + """ + Returns true if volume becomes active + :param cinder: + :param volume: + :return: + """ + end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT + while time.time() < end_time: + try: + status = cinder_utils.get_volume_status(cinder, volume) + if status == create_volume.STATUS_DELETED: + return True + except NotFound: + return True + + time.sleep(3) + + return False + + class CinderUtilsQoSTests(OSComponentTestCase): """ Test for the CreateQos class defined in create_qos.py @@ -70,7 +188,8 @@ class CinderUtilsQoSTests(OSComponentTestCase): self.qos_name = self.__class__.__name__ + '-' + str(guid) self.specs = {'foo': 'bar '} self.qos = None - self.cinder = cinder_utils.cinder_client(self.os_creds) + self.cinder = cinder_utils.cinder_client( + self.os_creds, self.os_session) def tearDown(self): """ @@ -82,12 +201,14 @@ class CinderUtilsQoSTests(OSComponentTestCase): except NotFound: pass + super(self.__class__, self).__clean__() + def test_create_qos_both(self): """ Tests the cinder_utils.create_qos() """ - qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, - consumer=Consumer.both) + qos_settings = QoSConfig( + name=self.qos_name, specs=self.specs, consumer=Consumer.both) self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) @@ -103,8 +224,8 @@ class CinderUtilsQoSTests(OSComponentTestCase): """ Tests the cinder_utils.create_qos() """ - qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, - consumer=Consumer.front_end) + qos_settings = QoSConfig( + name=self.qos_name, specs=self.specs, consumer=Consumer.front_end) self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) @@ -120,8 +241,8 @@ class CinderUtilsQoSTests(OSComponentTestCase): """ Tests the cinder_utils.create_qos() """ - qos_settings = QoSSettings(name=self.qos_name, specs=self.specs, - consumer=Consumer.back_end) + qos_settings = QoSConfig( + name=self.qos_name, specs=self.specs, consumer=Consumer.back_end) self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) @@ -137,7 +258,7 @@ class CinderUtilsQoSTests(OSComponentTestCase): """ Tests the cinder_utils.create_qos() """ - qos_settings = QoSSettings(name=self.qos_name, consumer=Consumer.both) + qos_settings = QoSConfig(name=self.qos_name, consumer=Consumer.both) self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.assertIsNotNone(self.qos) self.assertEqual(self.qos_name, self.qos.name) @@ -165,9 +286,10 @@ class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase): """ guid = uuid.uuid4() volume_type_name = self.__class__.__name__ + '-' + str(guid) - self.volume_type_settings = VolumeTypeSettings(name=volume_type_name) + self.volume_type_settings = VolumeTypeConfig(name=volume_type_name) self.volume_type = None - self.cinder = cinder_utils.cinder_client(self.os_creds) + self.cinder = cinder_utils.cinder_client( + self.os_creds, self.os_session) def tearDown(self): """ @@ -179,6 +301,8 @@ class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase): except NotFound: pass + super(self.__class__, self).__clean__() + def test_create_simple_volume_type(self): """ Tests the cinder_utils.create_volume_type(), get_volume_type(), and @@ -235,11 +359,12 @@ class CinderUtilsAddEncryptionTests(OSComponentTestCase): self.encryption_name = self.__class__.__name__ + '-' + str(guid) self.encryption = None - self.cinder = cinder_utils.cinder_client(self.os_creds) + self.cinder = cinder_utils.cinder_client( + self.os_creds, self.os_session) volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type' self.volume_type = cinder_utils.create_volume_type( - self.cinder, VolumeTypeSettings(name=volume_type_name)) + self.cinder, VolumeTypeConfig(name=volume_type_name)) def tearDown(self): """ @@ -258,12 +383,14 @@ class CinderUtilsAddEncryptionTests(OSComponentTestCase): except NotFound: pass + super(self.__class__, self).__clean__() + 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( + encryption_settings = VolumeTypeEncryptionConfig( name=self.encryption_name, provider_class='foo', control_location=ControlLocation.front_end) self.encryption = cinder_utils.create_volume_encryption( @@ -281,7 +408,7 @@ class CinderUtilsAddEncryptionTests(OSComponentTestCase): """ Primarily tests the cinder_utils.delete_volume_type_encryption() """ - encryption_settings = VolumeTypeEncryptionSettings( + encryption_settings = VolumeTypeEncryptionConfig( name=self.encryption_name, provider_class='LuksEncryptor', control_location=ControlLocation.back_end) self.encryption = cinder_utils.create_volume_encryption( @@ -307,7 +434,7 @@ class CinderUtilsAddEncryptionTests(OSComponentTestCase): Tests the cinder_utils.create_volume_encryption() with all valid settings """ - encryption_settings = VolumeTypeEncryptionSettings( + encryption_settings = VolumeTypeEncryptionConfig( name=self.encryption_name, provider_class='foo', cipher='bar', control_location=ControlLocation.back_end, key_size=1) @@ -329,7 +456,7 @@ class CinderUtilsAddEncryptionTests(OSComponentTestCase): Tests the cinder_utils.create_volume_encryption() raises an exception when the provider class does not exist """ - encryption_settings = VolumeTypeEncryptionSettings( + encryption_settings = VolumeTypeEncryptionConfig( name=self.encryption_name, provider_class='foo', cipher='bar', control_location=ControlLocation.back_end, key_size=-1) @@ -352,9 +479,10 @@ class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase): 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.cinder = cinder_utils.cinder_client( + self.os_creds, self.os_session) + qos_settings = QoSConfig( + name=self.qos_name, specs=self.specs, consumer=Consumer.both) self.qos = cinder_utils.create_qos(self.cinder, qos_settings) self.volume_type = None @@ -380,15 +508,17 @@ class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase): except NotFound: pass + super(self.__class__, self).__clean__() + def test_create_with_encryption(self): """ Tests the cinder_utils.create_volume_type() where encryption has been configured """ - encryption_settings = VolumeTypeEncryptionSettings( + encryption_settings = VolumeTypeEncryptionConfig( name='foo', provider_class='bar', control_location=ControlLocation.back_end) - volume_type_settings = VolumeTypeSettings( + volume_type_settings = VolumeTypeConfig( name=self.vol_type_name, encryption=encryption_settings) self.volume_type = cinder_utils.create_volume_type( self.cinder, volume_type_settings) @@ -404,7 +534,7 @@ class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase): """ Tests the cinder_utils.create_volume_type() with an associated QoS Spec """ - volume_type_settings = VolumeTypeSettings( + volume_type_settings = VolumeTypeConfig( name=self.vol_type_name, qos_spec_name=self.qos_name) self.volume_type = cinder_utils.create_volume_type( self.cinder, volume_type_settings) @@ -419,7 +549,7 @@ class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase): Tests the cinder_utils.create_volume_type() when the QoS Spec name does not exist """ - volume_type_settings = VolumeTypeSettings( + volume_type_settings = VolumeTypeConfig( name=self.vol_type_name, qos_spec_name='foo') self.volume_type = cinder_utils.create_volume_type( @@ -432,10 +562,10 @@ class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase): Tests the cinder_utils.create_volume_type() with encryption and an associated QoS Spec """ - encryption_settings = VolumeTypeEncryptionSettings( + encryption_settings = VolumeTypeEncryptionConfig( name='foo', provider_class='bar', control_location=ControlLocation.back_end) - volume_type_settings = VolumeTypeSettings( + volume_type_settings = VolumeTypeConfig( name=self.vol_type_name, qos_spec_name=self.qos_name, encryption=encryption_settings) self.volume_type = cinder_utils.create_volume_type( |