| | | volume when associating with a valid image |
+----------------------------------------+---------------+-----------------------------------------------------------+
+create_volume_tests.py - CreateVolMultipleCredsTests
+----------------------------------------------------
+
++----------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Cinder API | Description |
++========================================+===============+===========================================================+
+| test_create_by_admin_to_other_proj | 2 & 3 | Tests to ensure the creation of a Volume as a user with |
+| | | an 'admin' role can create a volume to another project |
+| | | and a creator with the credentails to that project will |
+| | | not create another with the same name |
+| | | Currently inactive due to |
+| | | https://bugs.launchpad.net/cinder/+bug/1641982 |
++----------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_two_vol_same_name_diff_proj| 2 & 3 | Tests to ensure the creation of a Volume with the same |
+| | | name by two different creators with different credentials |
+| | | will create two different volumes with the same name |
+| | | that are applied to each project in question |
++----------------------------------------+---------------+-----------------------------------------------------------+
+
create_stack_tests.py - CreateStackSuccessTests
-----------------------------------------------
def test_name_only(self):
settings = VolumeConfig(name='foo')
self.assertEqual('foo', settings.name)
+ self.assertIsNone(settings.project_name)
self.assertIsNone(settings.description)
self.assertEquals(1, settings.size)
self.assertIsNone(settings.image_name)
def test_config_with_name_only(self):
settings = VolumeConfig(**{'name': 'foo'})
self.assertEqual('foo', settings.name)
+ self.assertIsNone(settings.project_name)
self.assertIsNone(settings.description)
self.assertEquals(1, settings.size)
self.assertIsNone(settings.image_name)
def test_all_strings(self):
settings = VolumeConfig(
- name='foo', description='desc', size='2', image_name='image',
- type_name='type', availability_zone='zone1', multi_attach='true')
+ name='foo', project_name='proj-foo', description='desc', size='2',
+ image_name='image', type_name='type', availability_zone='zone1',
+ multi_attach='true')
self.assertEqual('foo', settings.name)
+ self.assertEqual('proj-foo', settings.project_name)
self.assertEqual('desc', settings.description)
self.assertEqual(2, settings.size)
self.assertEqual('image', settings.image_name)
def test_all_correct_type(self):
settings = VolumeConfig(
- name='foo', description='desc', size=2, image_name='image',
- type_name='bar', availability_zone='zone1', multi_attach=True)
+ name='foo', project_name='proj-foo', description='desc', size=2,
+ image_name='image', type_name='bar', availability_zone='zone1',
+ multi_attach=True)
self.assertEqual('foo', settings.name)
+ self.assertEqual('proj-foo', settings.project_name)
self.assertEqual('desc', settings.description)
self.assertEqual(2, settings.size)
self.assertEqual('image', settings.image_name)
def test_config_all(self):
settings = VolumeConfig(
- **{'name': 'foo', 'description': 'desc', 'size': '2',
+ **{'name': 'foo', 'project_name': 'proj-foo',
+ 'description': 'desc', 'size': '2',
'image_name': 'foo', 'type_name': 'bar',
'availability_zone': 'zone1', 'multi_attach': 'true'})
self.assertEqual('foo', settings.name)
+ self.assertEqual('proj-foo', settings.project_name)
self.assertEqual('desc', settings.description)
self.assertEqual(2, settings.size)
self.assertEqual('foo', settings.image_name)
"""
Constructor
:param name: the volume's name (required)
+ :param project_name: the name of the project to associate (optional)
+ note: due to a bug in the Cinder API, this functionality will not
+ work. see https://bugs.launchpad.net/cinder/+bug/1641982
:param description: the volume's name (optional)
:param size: the volume's size in GB (default 1)
:param image_name: when a glance image is used for the image source
"""
self.name = kwargs.get('name')
+ self.project_name = kwargs.get('project_name')
self.description = kwargs.get('description')
self.size = int(kwargs.get('size', 1))
self.image_name = kwargs.get('image_name')
"""
def test_construction_positional(self):
- volume = Volume('name1', 'id1', 'desc_val1', 2, 'type_val1',
- 'avail_zone1', False, [{'attached_at': 'foo'}])
+ volume = Volume('name1', 'id1', 'proj_id1', 'desc_val1', 2,
+ 'type_val1', 'avail_zone1', False,
+ [{'attached_at': 'foo'}])
self.assertEqual('name1', volume.name)
self.assertEqual('id1', volume.id)
+ self.assertEqual('proj_id1', volume.project_id)
self.assertEqual('desc_val1', volume.description)
self.assertEqual(2, volume.size)
self.assertEqual('type_val1', volume.type)
volume = Volume(attachments=[{'attached_at': 'foo'}],
multi_attach=True, availability_zone='avail_zone2',
vol_type='type_val2', size=3, description='desc_val2',
- volume_id='id2', name='name2')
+ volume_id='id2', name='name2', project_id='proj_id1')
self.assertEqual('name2', volume.name)
self.assertEqual('id2', volume.id)
+ self.assertEqual('proj_id1', volume.project_id)
self.assertEqual('desc_val2', volume.description)
self.assertEqual(3, volume.size)
self.assertEqual('type_val2', volume.type)
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, attachments=list()):
+ def __init__(self, name, volume_id, project_id, description, size,
+ vol_type, availability_zone, multi_attach,
+ attachments=list()):
"""
Constructor
:param name: the volume's name
:param volume_id: the volume's id
+ :param project_id: the volume's associated project id
:param description: the volume's description
:param size: the volume's size in GB
:param vol_type: the volume's type
"""
self.name = name
self.id = volume_id
+ self.project_id = project_id
self.description = description
self.size = size
self.type = vol_type
self.attachments = attachments
def __eq__(self, other):
- return (self.name == other.name and self.id == other.id
+ return (self.name == other.name
+ and self.id == other.id
+ and self.project_id == other.project_id
and self.description == other.description
and self.size == other.size
and self.type == other.type
for volume_name in self.instance_settings.volume_names:
cinder = cinder_utils.cinder_client(self._os_creds)
volume = cinder_utils.get_volume(
- cinder, volume_name=volume_name)
+ cinder, self.__keystone, volume_name=volume_name,
+ project_name=self._os_creds.project_name)
if volume and self.vm_active(block=True):
vm = nova_utils.attach_volume(
from snaps.config.volume import VolumeConfig
from snaps.openstack.openstack_creator import OpenStackVolumeObject
-from snaps.openstack.utils import cinder_utils
+from snaps.openstack.utils import cinder_utils, keystone_utils
__author__ = 'spisarski'
self.volume_settings = volume_settings
self.__volume = None
+ self.__keystone = None
def initialize(self):
"""
"""
super(self.__class__, self).initialize()
+ self.__keystone = keystone_utils.keystone_client(self._os_creds)
self.__volume = cinder_utils.get_volume(
- self._cinder, volume_settings=self.volume_settings)
+ self._cinder, self.__keystone,
+ volume_settings=self.volume_settings,
+ project_name=self._os_creds.project_name)
return self.__volume
def create(self, block=False):
if not self.__volume:
self.__volume = cinder_utils.create_volume(
- self._cinder, self.volume_settings)
+ self._cinder, self.__keystone, self.volume_settings)
logger.info(
'Created volume with name - %s', self.volume_settings.name)
from snaps.openstack.create_volume import (
VolumeSettings, OpenStackVolume)
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
-from snaps.openstack.utils import cinder_utils
+from snaps.openstack.utils import cinder_utils, keystone_utils
__author__ = 'spisarski'
name=self.__class__.__name__ + '-' + str(guid))
self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(self.os_creds)
self.volume_creator = None
def tearDown(self):
self.assertIsNotNone(created_volume)
retrieved_volume = cinder_utils.get_volume(
- self.cinder, volume_settings=self.volume_settings)
+ self.cinder, self.keystone, volume_settings=self.volume_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(retrieved_volume)
self.assertEqual(created_volume.id, retrieved_volume.id)
self.assertIsNotNone(created_volume)
retrieved_volume = cinder_utils.get_volume(
- self.cinder, volume_settings=self.volume_settings)
+ self.cinder, self.keystone, volume_settings=self.volume_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(retrieved_volume)
self.assertEqual(created_volume, retrieved_volume)
self.volume_creator.clean()
self.assertIsNone(cinder_utils.get_volume(
- self.cinder, volume_settings=self.volume_settings))
+ self.cinder, self.keystone, volume_settings=self.volume_settings,
+ project_name=self.os_creds.project_name))
# Must not throw an exception when attempting to cleanup non-existent
# volume
volume1 = self.volume_creator.create(block=True)
retrieved_volume = cinder_utils.get_volume(
- self.cinder, volume_settings=self.volume_settings)
+ self.cinder, self.keystone, volume_settings=self.volume_settings,
+ project_name=self.os_creds.project_name)
self.assertEqual(volume1, retrieved_volume)
# Should be retrieving the instance data
self.cinder, created_volume.id)
self.assertEqual(created_volume, retrieved_volume)
+
+
+class CreateVolMultipleCredsTests(OSIntegrationTestCase):
+ """
+ Test for the OpenStackVolume class and how it interacts with volumes
+ created with differenct credentials and to other projects with the same
+ name
+ """
+ def setUp(self):
+ super(self.__class__, self).__start__()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.volume_creators = list()
+
+ def tearDown(self):
+ for volume_creator in self.volume_creators:
+ volume_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ # TODO - activate after cinder API bug has been fixed
+ # see https://bugs.launchpad.net/cinder/+bug/1641982 as to why this test
+ # is not activated
+ # def test_create_by_admin_to_other_proj(self):
+ # """
+ # Creates a volume as admin to the project of os_creds then instantiates
+ # a creator object with the os_creds project to ensure it initializes
+ # without creation
+ # """
+ # self.volume_creators.append(OpenStackVolume(
+ # self.admin_os_creds, VolumeConfig(
+ # name=self.guid + '-vol',
+ # project_name=self.os_creds.project_name)))
+ # admin_vol = self.volume_creators[0].create(block=True)
+ #
+ # self.volume_creators.append(OpenStackVolume(
+ # self.os_creds, VolumeConfig(name=self.guid + '-vol')))
+ # proj_vol = self.volume_creators[1].create(block=True)
+ #
+ # self.assertEqual(admin_vol, proj_vol)
+
+ def test_create_two_vol_same_name_diff_proj(self):
+ """
+ Creates a volume as admin to the project of os_creds then instantiates
+ a creator object with the os_creds project to ensure it initializes
+ without creation
+ """
+ vol_name = self.guid + '-vol'
+ self.volume_creators.append(OpenStackVolume(
+ self.admin_os_creds, VolumeConfig(name=vol_name)))
+ admin_vol = self.volume_creators[0].create(block=True)
+ self.assertIsNotNone(admin_vol)
+
+ admin_key = keystone_utils.keystone_client(self.admin_os_creds)
+ admin_proj = keystone_utils.get_project(
+ admin_key, project_name=self.admin_os_creds.project_name)
+ self.assertEqual(admin_vol.project_id, admin_proj.id)
+
+ admin_cinder = cinder_utils.cinder_client(self.admin_os_creds)
+ admin_vol_get = cinder_utils.get_volume(
+ admin_cinder, admin_key, volume_name=vol_name,
+ project_name=self.admin_os_creds.project_name)
+ self.assertIsNotNone(admin_vol_get)
+ self.assertEqual(admin_vol, admin_vol_get)
+
+ self.volume_creators.append(OpenStackVolume(
+ self.os_creds, VolumeConfig(name=vol_name)))
+ proj_vol = self.volume_creators[1].create(block=True)
+ self.assertIsNotNone(proj_vol)
+
+ self.assertNotEqual(admin_vol, proj_vol)
+
+ proj_key = keystone_utils.keystone_client(self.os_creds)
+ proj_cinder = cinder_utils.cinder_client(self.os_creds)
+ proj_vol_get = cinder_utils.get_volume(
+ proj_cinder, proj_key, volume_name=vol_name,
+ project_name=self.os_creds.project_name)
+
+ self.assertIsNotNone(proj_vol_get)
+ self.assertEqual(proj_vol, proj_vol_get)
region_name=os_creds.region_name)
-def get_volume(cinder, volume_name=None, volume_settings=None):
+def get_volume(cinder, keystone=None, volume_name=None, volume_settings=None,
+ project_name=None):
"""
Returns an OpenStack volume object for a given name
:param cinder: the Cinder client
+ :param keystone: the Keystone client (required if project_name or
+ volume_settings.project_name is not None
:param volume_name: the volume name to lookup
:param volume_settings: the volume settings used for lookups
+ :param project_name: the name of the project associated with the volume
: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,
- attachments=volume.attachments)
+ for os_volume in volumes:
+ if os_volume.name == volume_name:
+ project_id = None
+ if hasattr(os_volume, 'os-vol-tenant-attr:tenant_id'):
+ project_id = getattr(
+ os_volume, 'os-vol-tenant-attr:tenant_id')
+
+ if volume_settings and volume_settings.project_name:
+ project_name = volume_settings.project_name
+
+ if project_name:
+ project = keystone_utils.get_project_by_id(
+ keystone, project_id)
+
+ if project and project.name == project_name:
+ return __map_os_volume_to_domain(os_volume)
+ else:
+ return __map_os_volume_to_domain(os_volume)
def __get_os_volume_by_id(cinder, volume_id):
:param volume_id: the volume ID to lookup
:return: the SNAPS-OO Domain Volume object or None
"""
- volume = __get_os_volume_by_id(cinder, volume_id)
+ os_volume = __get_os_volume_by_id(cinder, volume_id)
+ return __map_os_volume_to_domain(os_volume)
+
+
+def __map_os_volume_to_domain(os_volume):
+ """
+ Returns a SNAPS-OO domain Volume object that is created by an OpenStack
+ Volume object
+ :param os_volume: the OpenStack volume object
+ :return: Volume domain object
+ """
+ project_id = None
+ if hasattr(os_volume, 'os-vol-tenant-attr:tenant_id'):
+ project_id = getattr(
+ os_volume, 'os-vol-tenant-attr:tenant_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, attachments=volume.attachments)
+ name=os_volume.name, volume_id=os_volume.id,
+ project_id=project_id, description=os_volume.description,
+ size=os_volume.size, vol_type=os_volume.volume_type,
+ availability_zone=os_volume.availability_zone,
+ multi_attach=os_volume.multiattach,
+ attachments=os_volume.attachments)
def get_volume_status(cinder, volume):
return os_volume.status
-def create_volume(cinder, volume_settings):
+def create_volume(cinder, keystone, volume_settings):
"""
Creates and returns OpenStack volume object with an external URL
:param cinder: the cinder client
+ :param keystone: the keystone client
:param volume_settings: the volume settings object
:return: the OpenStack volume object
:raise Exception if using a file and it cannot be found
"""
- volume = cinder.volumes.create(
- name=volume_settings.name, description=volume_settings.description,
- size=volume_settings.size, imageRef=volume_settings.image_name,
+ project_id = None
+ if volume_settings.project_name:
+ project = keystone_utils.get_project(
+ keystone, project_name=volume_settings.project_name)
+ if project:
+ project_id = project.id
+ else:
+ raise KeystoneUtilsException(
+ 'Project cannot be found with name - '
+ + volume_settings.project_name)
+ os_volume = cinder.volumes.create(
+ name=volume_settings.name,
+ project_id=project_id,
+ 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=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, attachments=volume.attachments)
+ return __map_os_volume_to_domain(os_volume)
def delete_volume(cinder, volume):
"""
logger.info('Deleting QoS named - %s', qos.name)
cinder.qos_specs.delete(qos.id)
+
+
+class KeystoneUtilsException(Exception):
+ """
+ Exception when calls to the Keystone client cannot be served properly
+ """
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'
self.volume_name = self.__class__.__name__ + '-' + str(guid)
self.volume = None
self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(self.os_creds)
def tearDown(self):
"""
"""
volume_settings = VolumeConfig(name=self.volume_name)
self.volume = cinder_utils.create_volume(
- self.cinder, volume_settings)
+ 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, volume_settings=volume_settings)
+ 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)
"""
volume_settings = VolumeConfig(name=self.volume_name)
self.volume = cinder_utils.create_volume(
- self.cinder, volume_settings)
+ 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, volume_settings=volume_settings)
+ 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, volume_settings))
+ cinder_utils.get_volume(
+ self.cinder, self.keystone, volume_settings,
+ project_name=self.os_creds.project_name))
def volume_active(cinder, volume):
def test_vol_settings_from_vol(self):
volume = Volume(
- name='vol-name', volume_id='vol-id', description='desc', size=99,
- vol_type='vol-type', availability_zone='zone1', multi_attach=True)
+ name='vol-name', volume_id='vol-id', project_id='proj-id',
+ description='desc', size=99, vol_type='vol-type',
+ availability_zone='zone1', multi_attach=True)
settings = settings_utils.create_volume_config(volume)
self.assertEqual(volume.name, settings.name)
self.assertEqual(volume.description, settings.description)
from snaps.openstack.tests.create_volume_tests import (
VolumeSettingsUnitTests, CreateSimpleVolumeSuccessTests,
CreateVolumeWithTypeTests, CreateVolumeWithImageTests,
- CreateSimpleVolumeFailureTests)
+ CreateSimpleVolumeFailureTests, CreateVolMultipleCredsTests)
from snaps.openstack.tests.create_volume_type_tests import (
VolumeTypeSettingsUnitTests, CreateSimpleVolumeTypeSuccessTests,
CreateVolumeTypeComplexTests)
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(
+ CreateVolMultipleCredsTests, 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