* Added support for volumes integrated with QoS and encryption.
* Created tests for volumes at an API and state machine level.
JIRA: SNAPS-197
Change-Id: I07326875b9f1a30e50389531d0d2571ee648675f
Signed-off-by: spisarski <s.pisarski@cablelabs.com>
# limitations under the License.
import unittest
-from snaps.domain.volume import QoSSpec, VolumeType, VolumeTypeEncryption
+from snaps.domain.volume import (
+ QoSSpec, VolumeType, VolumeTypeEncryption, Volume)
+
+
+class VolumeDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.volume.Volume class
+ """
+
+ def test_construction_positional(self):
+ volume = Volume('name1', 'id1', 'desc_val1', 2, 'type_val1',
+ 'avail_zone1', False)
+ self.assertEqual('name1', volume.name)
+ self.assertEqual('id1', volume.id)
+ self.assertEqual('desc_val1', volume.description)
+ self.assertEqual(2, volume.size)
+ self.assertEqual('type_val1', volume.type)
+ self.assertEqual('avail_zone1', volume.availability_zone)
+ self.assertFalse(volume.multi_attach)
+
+ def test_construction_named(self):
+ volume = Volume(multi_attach=True, availability_zone='avail_zone2',
+ vol_type='type_val2', size=3, description='desc_val2',
+ volume_id='id2', name='name2')
+ self.assertEqual('name2', volume.name)
+ self.assertEqual('id2', volume.id)
+ self.assertEqual('desc_val2', volume.description)
+ self.assertEqual(3, volume.size)
+ self.assertEqual('type_val2', volume.type)
+ self.assertEqual('avail_zone2', volume.availability_zone)
+ self.assertTrue(volume.multi_attach)
class VolumeTypeDomainObjectTests(unittest.TestCase):
# limitations under the License.
+class Volume:
+ """
+ SNAPS domain object for Volumes. Should contain attributes that
+ are shared amongst cloud providers
+ """
+ def __init__(self, name, volume_id, description, size, vol_type,
+ availability_zone, multi_attach):
+ """
+ Constructor
+ :param name: the volume's name
+ :param volume_id: the volume's id
+ :param description: the volume's description
+ :param size: the volume's size in GB
+ :param vol_type: the volume's type
+ :param availability_zone: the volume's availability zone
+ :param multi_attach: When true, volume can be attached to multiple VMs
+ """
+ self.name = name
+ self.id = volume_id
+ self.description = description
+ self.size = size
+ self.type = vol_type
+ self.availability_zone = availability_zone
+ self.multi_attach = multi_attach
+
+ def __eq__(self, other):
+ return (self.name == other.name and self.id == other.id
+ and self.description == other.description
+ and self.size == other.size
+ and self.type == other.type
+ and self.availability_zone == other.availability_zone
+ and self.multi_attach == other.multi_attach)
+
+
class VolumeType:
"""
SNAPS domain object for Volume Types. Should contain attributes that
auth_url=self._os_creds.auth_url,
project_name=project_name,
identity_api_version=self._os_creds.identity_api_version,
+ image_api_version=self._os_creds.image_api_version,
+ network_api_version=self._os_creds.network_api_version,
+ compute_api_version=self._os_creds.compute_api_version,
+ heat_api_version=self._os_creds.heat_api_version,
+ volume_api_version=self._os_creds.volume_api_version,
user_domain_name=self._os_creds.user_domain_name,
user_domain_id=self._os_creds.user_domain_id,
project_domain_name=self._os_creds.project_domain_name,
--- /dev/null
+# 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 time
+
+from cinderclient.exceptions import NotFound
+
+from snaps.openstack.openstack_creator import OpenStackVolumeObject
+from snaps.openstack.utils import cinder_utils
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('create_volume')
+
+VOLUME_ACTIVE_TIMEOUT = 300
+VOLUME_DELETE_TIMEOUT = 60
+POLL_INTERVAL = 3
+STATUS_ACTIVE = 'available'
+STATUS_FAILED = 'failed'
+STATUS_DELETED = 'deleted'
+
+
+class OpenStackVolume(OpenStackVolumeObject):
+ """
+ Class responsible for managing an volume in OpenStack
+ """
+
+ def __init__(self, os_creds, volume_settings):
+ """
+ Constructor
+ :param os_creds: The OpenStack connection credentials
+ :param volume_settings: The volume settings
+ :return:
+ """
+ super(self.__class__, self).__init__(os_creds)
+
+ self.volume_settings = volume_settings
+ self.__volume = None
+
+ def initialize(self):
+ """
+ Loads the existing Volume
+ :return: The Volume domain object or None
+ """
+ super(self.__class__, self).initialize()
+
+ self.__volume = cinder_utils.get_volume(
+ self._cinder, volume_settings=self.volume_settings)
+ return self.__volume
+
+ 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:
+ self.__volume = cinder_utils.create_volume(
+ self._cinder, self.volume_settings)
+
+ logger.info(
+ 'Created volume with name - %s', self.volume_settings.name)
+ if self.__volume:
+ if block:
+ if self.volume_active(block=True):
+ logger.info('Volume is now active with name - %s',
+ self.volume_settings.name)
+ return self.__volume
+ else:
+ raise VolumeCreationError(
+ 'Volume was not created or activated in the '
+ 'alloted amount of time')
+ else:
+ logger.info('Did not create volume due to cleanup mode')
+
+ return self.__volume
+
+ def clean(self):
+ """
+ Cleanse environment of all artifacts
+ :return: void
+ """
+ if self.__volume:
+ try:
+ if self.volume_active(block=True):
+ cinder_utils.delete_volume(self._cinder, self.__volume)
+ else:
+ logger.warn('Timeout waiting to delete volume %s',
+ self.__volume.name)
+ except NotFound:
+ pass
+
+ try:
+ if self.volume_deleted(block=True):
+ logger.info(
+ 'Volume has been properly deleted with name - %s',
+ self.volume_settings.name)
+ self.__vm = None
+ else:
+ logger.error(
+ 'Volume not deleted within the timeout period of %s '
+ 'seconds', VOLUME_DELETE_TIMEOUT)
+ except Exception as e:
+ logger.error(
+ 'Unexpected error while checking VM instance status - %s',
+ e)
+
+ self.__volume = None
+
+ def get_volume(self):
+ """
+ Returns the domain Volume object as it was populated when create() was
+ called
+ :return: the object
+ """
+ return self.__volume
+
+ def volume_active(self, block=False, timeout=VOLUME_ACTIVE_TIMEOUT,
+ poll_interval=POLL_INTERVAL):
+ """
+ Returns true when the volume status returns the value of
+ expected_status_code
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
+ :param timeout: The timeout value
+ :param poll_interval: The polling interval in seconds
+ :return: T/F
+ """
+ return self._volume_status_check(STATUS_ACTIVE, block, timeout,
+ poll_interval)
+
+ def volume_deleted(self, block=False, poll_interval=POLL_INTERVAL):
+ """
+ Returns true when the VM status returns the value of
+ expected_status_code or instance retrieval throws a NotFound exception.
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
+ :param poll_interval: The polling interval in seconds
+ :return: T/F
+ """
+ try:
+ return self._volume_status_check(
+ STATUS_DELETED, block, VOLUME_DELETE_TIMEOUT, poll_interval)
+ except NotFound as e:
+ logger.debug(
+ "Volume not found when querying status for %s with message "
+ "%s", STATUS_DELETED, e)
+ return True
+
+ def _volume_status_check(self, expected_status_code, block, timeout,
+ poll_interval):
+ """
+ Returns true when the volume status returns the value of
+ expected_status_code
+ :param expected_status_code: instance status evaluated with this string
+ value
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
+ :param timeout: The timeout value
+ :param poll_interval: The polling interval in seconds
+ :return: T/F
+ """
+ # sleep and wait for volume status change
+ if block:
+ start = time.time()
+ else:
+ start = time.time() - timeout + 10
+
+ while timeout > time.time() - start:
+ status = self._status(expected_status_code)
+ if status:
+ logger.debug('Volume is active with name - %s',
+ self.volume_settings.name)
+ return True
+
+ logger.debug('Retry querying volume status in %s seconds',
+ str(poll_interval))
+ time.sleep(poll_interval)
+ logger.debug('Volume status query timeout in %s',
+ str(timeout - (time.time() - start)))
+
+ logger.error(
+ 'Timeout checking for volume status for ' + expected_status_code)
+ return False
+
+ def _status(self, expected_status_code):
+ """
+ Returns True when active else False
+ :param expected_status_code: instance status evaluated with this string
+ value
+ :return: T/F
+ """
+ status = cinder_utils.get_volume_status(self._cinder, self.__volume)
+ if not status:
+ logger.warning(
+ 'Cannot volume status for volume with ID - %s',
+ self.__volume.id)
+ return False
+
+ if status == 'ERROR':
+ raise VolumeCreationError(
+ 'Instance had an error during deployment')
+ logger.debug('Instance status is - ' + status)
+ return status == expected_status_code
+
+
+class VolumeSettings:
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param name: the volume's name (required)
+ :param description: the volume's name (required)
+ :param size: the volume's size in GB (default 1)
+ :param image_name: when a glance image is used for the image source
+ (optional)
+ :param type_name: the associated volume's type name (optional)
+ :param availability_zone: the name of the compute server on which to
+ deploy the volume (optional)
+ :param multi_attach: when true, volume can be attached to more than one
+ server (default False)
+ """
+
+ self.name = kwargs.get('name')
+ self.description = kwargs.get('description')
+ self.size = int(kwargs.get('size', 1))
+ self.image_name = kwargs.get('image_name')
+ self.type_name = kwargs.get('type_name')
+ self.availability_zone = kwargs.get('availability_zone')
+
+ if kwargs.get('availability_zone'):
+ self.multi_attach = bool(kwargs.get('availability_zone'))
+ else:
+ self.multi_attach = False
+
+ if not self.name:
+ raise VolumeSettingsError("The attribute name is required")
+
+
+class VolumeSettingsError(Exception):
+ """
+ Exception to be thrown when an volume settings are incorrect
+ """
+
+ def __init__(self, message):
+ Exception.__init__(self, message)
+
+
+class VolumeCreationError(Exception):
+ """
+ Exception to be thrown when an volume cannot be created
+ """
+
+ def __init__(self, message):
+ Exception.__init__(self, message)
return self.__volume_type
- def create(self, block=False):
+ def create(self):
"""
Creates the volume in OpenStack if it does not already exist and
returns the domain Volume object
--- /dev/null
+# 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 cinderclient.exceptions import NotFound, BadRequest
+
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_volume_type import (
+ VolumeTypeSettings, OpenStackVolumeType)
+from snaps.openstack.tests import openstack_tests
+
+try:
+ from urllib.request import URLError
+except ImportError:
+ from urllib2 import URLError
+
+import logging
+import unittest
+import uuid
+
+from snaps.openstack import create_volume
+from snaps.openstack.create_volume import (
+ VolumeSettings, VolumeSettingsError, OpenStackVolume)
+from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
+from snaps.openstack.utils import cinder_utils
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('create_volume_tests')
+
+
+class VolumeSettingsUnitTests(unittest.TestCase):
+ """
+ Tests the construction of the VolumeSettings class
+ """
+
+ def test_no_params(self):
+ with self.assertRaises(VolumeSettingsError):
+ VolumeSettings()
+
+ def test_empty_config(self):
+ with self.assertRaises(VolumeSettingsError):
+ VolumeSettings(**dict())
+
+ def test_name_only(self):
+ settings = VolumeSettings(name='foo')
+ self.assertEqual('foo', settings.name)
+ self.assertIsNone(settings.description)
+ self.assertEquals(1, settings.size)
+ self.assertIsNone(settings.image_name)
+ self.assertIsNone(settings.type_name)
+ self.assertIsNone(settings.availability_zone)
+ self.assertFalse(settings.multi_attach)
+
+ def test_config_with_name_only(self):
+ settings = VolumeSettings(**{'name': 'foo'})
+ self.assertEqual('foo', settings.name)
+ self.assertIsNone(settings.description)
+ self.assertEquals(1, settings.size)
+ self.assertIsNone(settings.image_name)
+ self.assertIsNone(settings.type_name)
+ self.assertIsNone(settings.availability_zone)
+ self.assertFalse(settings.multi_attach)
+
+ def test_all_strings(self):
+ settings = VolumeSettings(
+ name='foo', description='desc', size='2', image_name='image',
+ type_name='type', availability_zone='zone1', multi_attach='true')
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('desc', settings.description)
+ self.assertEqual(2, settings.size)
+ self.assertEqual('image', settings.image_name)
+ self.assertEqual('type', settings.type_name)
+ self.assertEqual('zone1', settings.availability_zone)
+ self.assertTrue(settings.multi_attach)
+
+ def test_all_correct_type(self):
+ settings = VolumeSettings(
+ name='foo', description='desc', size=2, image_name='image',
+ type_name='bar', availability_zone='zone1', multi_attach=True)
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('desc', settings.description)
+ self.assertEqual(2, settings.size)
+ self.assertEqual('image', settings.image_name)
+ self.assertEqual('bar', settings.type_name)
+ self.assertEqual('zone1', settings.availability_zone)
+ self.assertTrue(settings.multi_attach)
+
+ def test_config_all(self):
+ settings = VolumeSettings(
+ **{'name': 'foo', 'description': 'desc', 'size': '2',
+ 'image_name': 'foo', 'type_name': 'bar',
+ 'availability_zone': 'zone1', 'multi_attach': 'true'})
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('desc', settings.description)
+ self.assertEqual(2, settings.size)
+ self.assertEqual('foo', settings.image_name)
+ self.assertEqual('bar', settings.type_name)
+ self.assertEqual('zone1', settings.availability_zone)
+ self.assertTrue(settings.multi_attach)
+
+
+class CreateSimpleVolumeSuccessTests(OSIntegrationTestCase):
+ """
+ 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
+ """
+ super(self.__class__, self).__start__()
+
+ guid = uuid.uuid4()
+ self.volume_settings = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(guid))
+
+ self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.volume_creator = None
+
+ def tearDown(self):
+ """
+ Cleans the volume and downloaded volume file
+ """
+ if self.volume_creator:
+ self.volume_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_create_volume_simple(self):
+ """
+ Tests the creation of an OpenStack volume from a URL.
+ """
+ # Create Volume
+ self.volume_creator = create_volume.OpenStackVolume(
+ self.os_creds, self.volume_settings)
+ created_volume = self.volume_creator.create(block=True)
+ self.assertIsNotNone(created_volume)
+
+ retrieved_volume = cinder_utils.get_volume(
+ self.cinder, volume_settings=self.volume_settings)
+
+ self.assertIsNotNone(retrieved_volume)
+ self.assertEqual(created_volume.id, retrieved_volume.id)
+ self.assertTrue(created_volume == retrieved_volume)
+
+ def test_create_delete_volume(self):
+ """
+ Tests the creation then deletion of an OpenStack volume to ensure
+ clean() does not raise an Exception.
+ """
+ # Create Volume
+ self.volume_creator = create_volume.OpenStackVolume(
+ self.os_creds, self.volume_settings)
+ created_volume = self.volume_creator.create(block=True)
+ self.assertIsNotNone(created_volume)
+
+ retrieved_volume = cinder_utils.get_volume(
+ self.cinder, volume_settings=self.volume_settings)
+ self.assertIsNotNone(retrieved_volume)
+ self.assertEqual(created_volume, retrieved_volume)
+
+ # Delete Volume manually
+ self.volume_creator.clean()
+
+ self.assertIsNone(cinder_utils.get_volume(
+ self.cinder, volume_settings=self.volume_settings))
+
+ # Must not throw an exception when attempting to cleanup non-existent
+ # volume
+ self.volume_creator.clean()
+ self.assertIsNone(self.volume_creator.get_volume())
+
+ def test_create_same_volume(self):
+ """
+ Tests the creation of an OpenStack volume when one already exists.
+ """
+ # Create Volume
+ self.volume_creator = OpenStackVolume(
+ self.os_creds, self.volume_settings)
+ volume1 = self.volume_creator.create(block=True)
+
+ retrieved_volume = cinder_utils.get_volume(
+ self.cinder, volume_settings=self.volume_settings)
+ self.assertEqual(volume1, retrieved_volume)
+
+ # Should be retrieving the instance data
+ os_volume_2 = OpenStackVolume(
+ self.os_creds, self.volume_settings)
+ volume2 = os_volume_2.create(block=True)
+ self.assertEqual(volume1, volume2)
+
+
+class CreateVolumeWithTypeTests(OSIntegrationTestCase):
+ """
+ Test cases for the CreateVolume when attempting to associate it to a
+ Volume Type
+ """
+
+ def setUp(self):
+ super(self.__class__, self).__start__()
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.volume_name = guid + '-vol'
+ self.volume_type_name = guid + '-vol-type'
+
+ self.volume_type_creator = OpenStackVolumeType(
+ self.os_creds, VolumeTypeSettings(name=self.volume_type_name))
+ self.volume_type_creator.create()
+ self.volume_creator = None
+
+ def tearDown(self):
+ if self.volume_creator:
+ self.volume_creator.clean()
+ if self.volume_type_creator:
+ self.volume_type_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_bad_volume_type(self):
+ """
+ Expect a NotFound to be raised when the volume type does not exist
+ """
+ self.volume_creator = OpenStackVolume(
+ self.os_creds,
+ VolumeSettings(name=self.volume_name, type_name='foo'))
+
+ with self.assertRaises(NotFound):
+ self.volume_creator.create()
+
+ def test_valid_volume_type(self):
+ """
+ Expect a NotFound to be raised when the volume type does not exist
+ """
+ self.volume_creator = OpenStackVolume(
+ self.os_creds,
+ VolumeSettings(name=self.volume_name,
+ type_name=self.volume_type_name))
+
+ created_volume = self.volume_creator.create()
+ self.assertIsNotNone(created_volume)
+ self.assertEqual(self.volume_type_name, created_volume.type)
+
+
+class CreateVolumeWithImageTests(OSIntegrationTestCase):
+ """
+ Test cases for the CreateVolume when attempting to associate it to an Image
+ """
+
+ def setUp(self):
+ super(self.__class__, self).__start__()
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.volume_name = guid + '-vol'
+ self.image_name = guid + '-image'
+
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=self.image_metadata)
+ # Create Image
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
+ self.image_creator.create()
+ self.volume_creator = None
+
+ def tearDown(self):
+ if self.volume_creator:
+ try:
+ self.volume_creator.clean()
+ except:
+ pass
+ if self.image_creator:
+ try:
+ self.image_creator.clean()
+ except:
+ pass
+
+ super(self.__class__, self).__clean__()
+
+ def test_bad_image_name(self):
+ """
+ Expect a NotFound to be raised when the volume type does not exist
+ """
+ self.volume_creator = OpenStackVolume(
+ self.os_creds,
+ VolumeSettings(name=self.volume_name, image_name='foo'))
+
+ with self.assertRaises(BadRequest):
+ self.volume_creator.create(block=True)
+
+ def test_valid_volume_image(self):
+ """
+ Expect a NotFound to be raised when the volume type does not exist
+ """
+ self.volume_creator = OpenStackVolume(
+ self.os_creds,
+ VolumeSettings(name=self.volume_name, image_name=self.image_name))
+
+ created_volume = self.volume_creator.create(block=True)
+ self.assertIsNotNone(created_volume)
+ self.assertIsNone(created_volume.type)
+ self.assertTrue(self.volume_creator.volume_active())
from cinderclient.client import Client
from cinderclient.exceptions import NotFound
-from snaps.domain.volume import QoSSpec, VolumeType, VolumeTypeEncryption
+from snaps.domain.volume import (
+ QoSSpec, VolumeType, VolumeTypeEncryption, Volume)
from snaps.openstack.utils import keystone_utils
__author__ = 'spisarski'
region_name=os_creds.region_name)
+def get_volume(cinder, volume_name=None, volume_settings=None):
+ """
+ Returns an OpenStack volume object for a given name
+ :param cinder: the Cinder client
+ :param volume_name: the volume name to lookup
+ :param volume_settings: the volume settings used for lookups
+ :return: the volume object or None
+ """
+ if volume_settings:
+ volume_name = volume_settings.name
+
+ volumes = cinder.volumes.list()
+ for volume in volumes:
+ if volume.name == volume_name:
+ return Volume(
+ name=volume.name, volume_id=volume.id,
+ description=volume.description, size=volume.size,
+ vol_type=volume.volume_type,
+ availability_zone=volume.availability_zone,
+ multi_attach=volume.multiattach)
+
+
+def get_volume_by_id(cinder, volume_id):
+ """
+ Returns an OpenStack volume object for a given name
+ :param cinder: the Cinder client
+ :param volume_id: the volume ID to lookup
+ :return: the SNAPS-OO Domain Volume object or None
+ """
+ volume = cinder.volumes.get(volume_id)
+ return Volume(
+ name=volume.name, volume_id=volume.id, description=volume.description,
+ size=volume.size, vol_type=volume.volume_type,
+ availability_zone=volume.availability_zone,
+ multi_attach=volume.multiattach)
+
+
+def get_volume_status(cinder, volume):
+ """
+ Returns a new OpenStack Volume object for a given OpenStack volume object
+ :param cinder: the Cinder client
+ :param volume: the domain Volume object
+ :return: the OpenStack Volume object
+ """
+ os_volume = cinder.volumes.get(volume.id)
+ return os_volume.status
+
+
+def create_volume(cinder, volume_settings):
+ """
+ Creates and returns OpenStack volume object with an external URL
+ :param cinder: the cinder client
+ :param volume_settings: the volume settings object
+ :return: the OpenStack volume object
+ :raise Exception if using a file and it cannot be found
+ """
+ created_volume = cinder.volumes.create(
+ name=volume_settings.name, description=volume_settings.description,
+ size=volume_settings.size, imageRef=volume_settings.image_name,
+ volume_type=volume_settings.type_name,
+ availability_zone=volume_settings.availability_zone,
+ multiattach=volume_settings.multi_attach)
+
+ return Volume(
+ name=created_volume.name, volume_id=created_volume.id,
+ description=created_volume.description,
+ size=created_volume.size, vol_type=created_volume.volume_type,
+ availability_zone=created_volume.availability_zone,
+ multi_attach=created_volume.multiattach)
+
+
+def delete_volume(cinder, volume):
+ """
+ Deletes an volume from OpenStack
+ :param cinder: the cinder client
+ :param volume: the volume to delete
+ """
+ logger.info('Deleting volume named - %s', volume.name)
+ cinder.volumes.delete(volume.id)
+
+
def get_volume_type(cinder, volume_type_name=None, volume_type_settings=None):
"""
Returns an OpenStack volume type object for a given name
import logging
import uuid
+import time
from cinderclient.exceptions import NotFound, BadRequest
+from snaps.openstack import create_volume
from snaps.openstack.create_qos import QoSSettings, Consumer
+from snaps.openstack.create_volume import VolumeSettings
from snaps.openstack.create_volume_type import (
VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation)
from snaps.openstack.tests import validation_utils
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)
+
+ 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 = VolumeSettings(name=self.volume_name)
+ self.volume = cinder_utils.create_volume(
+ self.cinder, 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, volume_settings=volume_settings)
+ self.assertIsNotNone(volume)
+ validation_utils.objects_equivalent(self.volume, volume)
+
+ def test_create_delete_volume(self):
+ """
+ Tests the cinder_utils.create_volume()
+ """
+ volume_settings = VolumeSettings(name=self.volume_name)
+ self.volume = cinder_utils.create_volume(
+ self.cinder, 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, volume_settings=volume_settings)
+ 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, volume_settings))
+
+
+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
VmInstDomainObjectTests, FloatingIpDomainObjectTests)
from snaps.domain.test.volume_tests import (
QoSSpecDomainObjectTests, VolumeTypeDomainObjectTests,
- VolumeTypeEncryptionObjectTests)
+ VolumeTypeEncryptionObjectTests, VolumeDomainObjectTests)
from snaps.openstack.tests.conf.os_credentials_tests import (
ProxySettingsUnitTests, OSCredsUnitTests)
from snaps.openstack.tests.create_flavor_tests import (
CreateComplexStackTests)
from snaps.openstack.tests.create_user_tests import (
UserSettingsUnitTests, CreateUserSuccessTests)
+from snaps.openstack.tests.create_volume_tests import (
+ VolumeSettingsUnitTests, CreateSimpleVolumeSuccessTests,
+ CreateVolumeWithTypeTests, CreateVolumeWithImageTests)
from snaps.openstack.tests.create_volume_type_tests import (
VolumeTypeSettingsUnitTests, CreateSimpleVolumeTypeSuccessTests,
CreateVolumeTypeComplexTests)
OSComponentTestCase, OSIntegrationTestCase)
from snaps.openstack.utils.tests.cinder_utils_tests import (
CinderSmokeTests, CinderUtilsQoSTests, CinderUtilsSimpleVolumeTypeTests,
- CinderUtilsAddEncryptionTests, CinderUtilsVolumeTypeCompleteTests)
+ CinderUtilsAddEncryptionTests, CinderUtilsVolumeTypeCompleteTests,
+ CinderUtilsVolumeTests)
from snaps.openstack.utils.tests.glance_utils_tests import (
GlanceSmokeTests, GlanceUtilsTests)
from snaps.openstack.utils.tests.heat_utils_tests import (
VolumeTypeDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
VolumeTypeEncryptionObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ VolumeDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
QoSSpecDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
QoSSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
VolumeTypeSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ VolumeSettingsUnitTests))
def add_openstack_client_tests(suite, os_creds, ext_net_name,
CinderUtilsQoSTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level,
image_metadata=image_metadata))
+ suite.addTest(OSComponentTestCase.parameterize(
+ CinderUtilsVolumeTests, 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,
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(
+ CreateSimpleVolumeSuccessTests, 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(
+ CreateVolumeWithTypeTests, 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(
+ CreateVolumeWithImageTests, 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(
Code Review / snaps.git / commitdiff