"""
def test_construction_positional(self):
- vm_inst = VmInst('name', 'id', '456', '123', dict(), 'kp-name', list())
+ vm_inst = VmInst('name', 'id', '456', '123', dict(), 'kp-name',
+ ['foo', 'bar'], ['123', '456'])
self.assertEqual('name', vm_inst.name)
self.assertEqual('id', vm_inst.id)
self.assertEqual('456', vm_inst.image_id)
self.assertEqual('123', vm_inst.flavor_id)
self.assertEqual(dict(), vm_inst.networks)
self.assertEqual('kp-name', vm_inst.keypair_name)
- self.assertEqual(list(), vm_inst.sec_grp_names)
+ self.assertEqual(['foo', 'bar'], vm_inst.sec_grp_names)
+ self.assertEqual(['123', '456'], vm_inst.volume_ids)
def test_construction_named(self):
- vm_inst = VmInst(sec_grp_names=list(), networks=dict(), inst_id='id',
- name='name', flavor_id='123', image_id='456',
- keypair_name='kp-name')
+ vm_inst = VmInst(
+ volume_ids=['123', '456'], sec_grp_names=['foo', 'bar'],
+ networks=dict(), inst_id='id', name='name', flavor_id='123',
+ image_id='456', keypair_name='kp-name')
self.assertEqual('name', vm_inst.name)
self.assertEqual('id', vm_inst.id)
self.assertEqual('456', vm_inst.image_id)
self.assertEqual('123', vm_inst.flavor_id)
self.assertEqual(dict(), vm_inst.networks)
self.assertEqual('kp-name', vm_inst.keypair_name)
- self.assertEqual(list(), vm_inst.sec_grp_names)
+ self.assertEqual(['foo', 'bar'], vm_inst.sec_grp_names)
+ self.assertEqual(['123', '456'], vm_inst.volume_ids)
class FloatingIpDomainObjectTests(unittest.TestCase):
def test_construction_positional(self):
volume = Volume('name1', 'id1', 'desc_val1', 2, 'type_val1',
- 'avail_zone1', False)
+ 'avail_zone1', False, [{'attached_at': 'foo'}])
self.assertEqual('name1', volume.name)
self.assertEqual('id1', volume.id)
self.assertEqual('desc_val1', volume.description)
self.assertEqual('type_val1', volume.type)
self.assertEqual('avail_zone1', volume.availability_zone)
self.assertFalse(volume.multi_attach)
+ self.assertIsNotNone(volume.attachments)
+ self.assertTrue(isinstance(volume.attachments[0], dict))
+ self.assertEqual(1, len(volume.attachments))
def test_construction_named(self):
- volume = Volume(multi_attach=True, availability_zone='avail_zone2',
+ 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')
self.assertEqual('name2', volume.name)
self.assertEqual('type_val2', volume.type)
self.assertEqual('avail_zone2', volume.availability_zone)
self.assertTrue(volume.multi_attach)
+ self.assertIsNotNone(volume.attachments)
+ self.assertTrue(isinstance(volume.attachments[0], dict))
+ self.assertEqual(1, len(volume.attachments))
class VolumeTypeDomainObjectTests(unittest.TestCase):
are shared amongst cloud providers
"""
def __init__(self, name, inst_id, image_id, flavor_id, networks,
- keypair_name, sec_grp_names):
+ keypair_name, sec_grp_names, volume_ids):
"""
Constructor
:param name: the image's name
value is a list of associated IPs
:param keypair_name: the name of the associated keypair
:param sec_grp_names: list of security group names
+ :param volume_ids: list of attached volume IDs
"""
self.name = name
self.id = inst_id
self.networks = networks
self.keypair_name = keypair_name
self.sec_grp_names = sec_grp_names
+ self.volume_ids = volume_ids
def __eq__(self, other):
return (self.name == other.name and
self.flavor_id == other.flavor_id and
self.networks == other.networks and
self.keypair_name == other.keypair_name and
- self.sec_grp_names == other.sec_grp_names)
+ self.sec_grp_names == other.sec_grp_names and
+ self.volume_ids == other.volume_ids)
class FloatingIp:
are shared amongst cloud providers
"""
def __init__(self, name, volume_id, description, size, vol_type,
- availability_zone, multi_attach):
+ availability_zone, multi_attach, attachments=list()):
"""
Constructor
:param name: the volume's name
: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
+ :param attachments: List of dict objects containing the info on where
+ this volume is attached
"""
self.name = name
self.id = volume_id
self.type = vol_type
self.availability_zone = availability_zone
self.multi_attach = multi_attach
+ self.attachments = attachments
def __eq__(self, other):
return (self.name == other.name and self.id == other.id
from snaps.openstack.create_network import PortSettings
from snaps.openstack.openstack_creator import OpenStackComputeObject
-from snaps.openstack.utils import glance_utils
+from snaps.openstack.utils import glance_utils, cinder_utils
from snaps.openstack.utils import neutron_utils
from snaps.openstack.utils import nova_utils
from snaps.provisioning import ansible_utils
self.initialize()
if len(self.__ports) == 0:
- self.__ports = self.__create_ports(self.instance_settings.port_settings)
+ self.__ports = self.__create_ports(
+ self.instance_settings.port_settings)
if not self.__vm:
self.__create_vm(block)
' to VM that did not activate with name - ' +
self.instance_settings.name)
+ if self.instance_settings.volume_names:
+ 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)
+
+ if volume and self.vm_active(block=True):
+ timeout = 30
+ vm = nova_utils.attach_volume(
+ self._nova, self.__vm, volume, timeout)
+
+ if vm:
+ self.__vm = vm
+ else:
+ logger.warn('Volume [%s] not attached within timeout '
+ 'of [%s]', volume.name, timeout)
+ else:
+ logger.warn('Unable to attach volume named [%s]',
+ volume_name)
+
self.__apply_floating_ips()
def __apply_floating_ips(self):
logger.error('Error deleting Floating IP - ' + str(e))
self.__floating_ip_dict = dict()
+ # Detach Volume
+ for volume_rec in self.__vm.volume_ids:
+ cinder = cinder_utils.cinder_client(self._os_creds)
+ volume = cinder_utils.get_volume_by_id(cinder, volume_rec['id'])
+ if volume:
+ try:
+ vm = nova_utils.detach_volume(
+ self._nova, self.__vm, volume, 30)
+ if vm:
+ self.__vm = vm
+ else:
+ logger.warn(
+ 'Timeout waiting to detach volume %s', volume.name)
+ except Exception as e:
+ logger.error('Unexpected error detaching volume %s '
+ 'with error %s', volume.name, e)
+ else:
+ logger.warn('Unable to detach volume with ID - [%s]',
+ volume_rec['id'])
+
# Cleanup ports
for name, port in self.__ports:
- logger.info('Deleting Port with ID - %S ' + port.id)
+ logger.info('Deleting Port with ID - %s ', port.id)
try:
neutron_utils.delete_port(self.__neutron, port)
except PortNotFoundClient as e:
port = neutron_utils.get_port(
self.__neutron, port_settings=port_setting)
if not port:
- port = neutron_utils.create_port(self.__neutron, self._os_creds, port_setting)
+ port = neutron_utils.create_port(
+ self.__neutron, self._os_creds, port_setting)
if port:
ports.append((port_setting.name, port))
waiting obtaining an SSH connection to a VM
:param availability_zone: the name of the compute server on which to
deploy the VM (optional)
+ :param volume_names: a list of the names of the volume to attach
+ (optional)
:param userdata: the string contents of any optional cloud-init script
to execute after the VM has been activated.
This value may also contain a dict who's key value
self.floating_ip_settings.append(FloatingIpSettings(
**floating_ip_config['floating_ip']))
- if kwargs.get('vm_boot_timeout'):
- self.vm_boot_timeout = kwargs['vm_boot_timeout']
- else:
- self.vm_boot_timeout = 900
-
- if kwargs.get('vm_delete_timeout'):
- self.vm_delete_timeout = kwargs['vm_delete_timeout']
- else:
- self.vm_delete_timeout = 300
+ self.vm_boot_timeout = kwargs.get('vm_boot_timeout', 900)
+ self.vm_delete_timeout = kwargs.get('vm_delete_timeout', 300)
+ self.ssh_connect_timeout = kwargs.get('ssh_connect_timeout', 180)
+ self.availability_zone = kwargs.get('availability_zone')
+ self.volume_names = kwargs.get('volume_names')
- if kwargs.get('ssh_connect_timeout'):
- self.ssh_connect_timeout = kwargs['ssh_connect_timeout']
- else:
- self.ssh_connect_timeout = 180
-
- if kwargs.get('availability_zone'):
- self.availability_zone = kwargs['availability_zone']
- else:
- self.availability_zone = None
+ if self.volume_names and not isinstance(self.volume_names, list):
+ raise VmInstanceSettingsError('volume_names must be a list')
if not self.name or not self.flavor:
raise VmInstanceSettingsError(
VOLUME_DELETE_TIMEOUT = 60
POLL_INTERVAL = 3
STATUS_ACTIVE = 'available'
-STATUS_FAILED = 'failed'
+STATUS_IN_USE = 'in-use'
+STATUS_FAILED = 'error'
STATUS_DELETED = 'deleted'
"""
if self.__volume:
try:
- if self.volume_active(block=True):
+ if self.volume_active():
cinder_utils.delete_volume(self._cinder, self.__volume)
else:
logger.warn('Timeout waiting to delete volume %s',
return self._volume_status_check(STATUS_ACTIVE, block, timeout,
poll_interval)
+ def volume_in_use(self):
+ """
+ Returns true when the volume status returns the value of
+ expected_status_code
+ :return: T/F
+ """
+ return self._volume_status_check(STATUS_IN_USE, False, 0, 0)
+
def volume_deleted(self, block=False, poll_interval=POLL_INTERVAL):
"""
Returns true when the VM status returns the value of
if block:
start = time.time()
else:
- start = time.time() - timeout + 10
+ start = time.time() - timeout + 1
while timeout > time.time() - start:
status = self._status(expected_status_code)
from snaps.openstack.create_security_group import (
SecurityGroupSettings, OpenStackSecurityGroup, SecurityGroupRuleSettings,
Direction, Protocol)
+from snaps.openstack.create_volume import OpenStackVolume, VolumeSettings
from snaps.openstack.tests import openstack_tests, validation_utils
from snaps.openstack.tests.os_source_file_test import (
OSIntegrationTestCase, OSComponentTestCase)
self.assertEqual(300, settings.vm_delete_timeout)
self.assertEqual(180, settings.ssh_connect_timeout)
self.assertIsNone(settings.availability_zone)
+ self.assertIsNone(settings.volume_names)
def test_config_with_name_flavor_port_only(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
self.assertEqual(300, settings.vm_delete_timeout)
self.assertEqual(180, settings.ssh_connect_timeout)
self.assertIsNone(settings.availability_zone)
+ self.assertIsNone(settings.volume_names)
def test_all(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port',
router_name='foo-bar-router')
- settings = VmInstanceSettings(name='foo', flavor='bar',
- port_settings=[port_settings],
- security_group_names=['sec_grp_1'],
- floating_ip_settings=[fip_settings],
- sudo_user='joe', vm_boot_timeout=999,
- vm_delete_timeout=333,
- ssh_connect_timeout=111,
- availability_zone='server name')
+ settings = VmInstanceSettings(
+ name='foo', flavor='bar', port_settings=[port_settings],
+ security_group_names=['sec_grp_1'],
+ floating_ip_settings=[fip_settings], sudo_user='joe',
+ vm_boot_timeout=999, vm_delete_timeout=333,
+ ssh_connect_timeout=111, availability_zone='server name',
+ volume_names=['vol1'])
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.flavor)
self.assertEqual(1, len(settings.port_settings))
self.assertEqual(333, settings.vm_delete_timeout)
self.assertEqual(111, settings.ssh_connect_timeout)
self.assertEqual('server name', settings.availability_zone)
+ self.assertEqual('vol1', settings.volume_names[0])
def test_config_all(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
'security_group_names': ['sec_grp_1'],
'floating_ips': [fip_settings], 'sudo_user': 'joe',
'vm_boot_timeout': 999, 'vm_delete_timeout': 333,
- 'ssh_connect_timeout': 111, 'availability_zone': 'server name'})
+ 'ssh_connect_timeout': 111, 'availability_zone': 'server name',
+ 'volume_names': ['vol2']})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.flavor)
self.assertEqual(1, len(settings.port_settings))
self.assertEqual(333, settings.vm_delete_timeout)
self.assertEqual(111, settings.ssh_connect_timeout)
self.assertEqual('server name', settings.availability_zone)
+ self.assertEqual('vol2', settings.volume_names[0])
class FloatingIpSettingsUnitTests(unittest.TestCase):
self.assertTrue(check_ping(self.inst_creators[1]))
+class CreateInstanceVolumeTests(OSIntegrationTestCase):
+ """
+ Simple instance creation with an attached volume
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file
+ within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.vm_inst_name = guid + '-inst'
+ self.nova = nova_utils.nova_client(self.os_creds)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
+
+ net_config = openstack_tests.get_priv_net_config(
+ net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
+ router_name=guid + '-pub-router', external_net=self.ext_net_name)
+
+ self.volume_settings1 = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(guid) + '-1')
+ self.volume_settings2 = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(guid) + '-2')
+
+ # Initialize for tearDown()
+ self.image_creator = None
+ self.flavor_creator = None
+
+ self.network_creator = None
+ self.inst_creator = None
+ self.volume_creator1 = None
+ self.volume_creator2 = None
+
+ try:
+ # Create Image
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
+ self.image_creator.create()
+
+ # Create Flavor
+ self.flavor_creator = OpenStackFlavor(
+ self.admin_os_creds,
+ FlavorSettings(name=guid + '-flavor-name', ram=256, disk=1,
+ vcpus=2, metadata=self.flavor_metadata))
+ self.flavor_creator.create()
+
+ # Create Network
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, net_config.network_settings)
+ self.network_creator.create()
+
+ self.port_settings = PortSettings(
+ name=guid + '-port',
+ network_name=net_config.network_settings.name)
+
+ self.volume_creator1 = OpenStackVolume(
+ self.os_creds, self.volume_settings1)
+ self.volume_creator1.create(block=True)
+
+ self.volume_creator2 = OpenStackVolume(
+ self.os_creds, self.volume_settings2)
+ self.volume_creator2.create(block=True)
+
+ except Exception as e:
+ self.tearDown()
+ raise e
+
+ def tearDown(self):
+ """
+ Cleans the created object
+ """
+ if self.inst_creator:
+ try:
+ self.inst_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message '
+ '- %s', e)
+
+ if self.flavor_creator:
+ try:
+ self.flavor_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
+
+ if self.network_creator:
+ try:
+ self.network_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning network with message - %s',
+ e)
+
+ if self.volume_creator2:
+ try:
+ self.volume_creator2.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning volume with message - %s',
+ e)
+
+ if self.volume_creator1:
+ try:
+ self.volume_creator1.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning volume with message - %s',
+ e)
+
+ if self.image_creator and not self.image_creator.image_settings.exists:
+ try:
+ self.image_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
+
+ super(self.__class__, self).__clean__()
+
+ def test_create_instance_with_one_volume(self):
+ """
+ Tests the creation of an OpenStack instance with a single volume.
+ """
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings],
+ volume_names=[self.volume_settings1.name])
+
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
+
+ vm_inst = self.inst_creator.create(block=True)
+ self.assertIsNotNone(nova_utils.get_server(
+ self.nova, vm_inst_settings=instance_settings))
+
+ self.assertIsNotNone(vm_inst)
+ self.assertEqual(1, len(vm_inst.volume_ids))
+ self.assertEqual(self.volume_creator1.get_volume().id,
+ vm_inst.volume_ids[0]['id'])
+
+ def test_create_instance_with_two_volumes(self):
+ """
+ Tests the creation of an OpenStack instance with a single volume.
+ """
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings],
+ volume_names=[self.volume_settings1.name,
+ self.volume_settings2.name])
+
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
+
+ vm_inst = self.inst_creator.create(block=True)
+ self.assertIsNotNone(nova_utils.get_server(
+ self.nova, vm_inst_settings=instance_settings))
+
+ self.assertIsNotNone(vm_inst)
+ self.assertEqual(2, len(vm_inst.volume_ids))
+ self.assertEqual(self.volume_creator1.get_volume().id,
+ vm_inst.volume_ids[0]['id'])
+ self.assertEqual(self.volume_creator2.get_volume().id,
+ vm_inst.volume_ids[1]['id'])
+
+
def check_dhcp_lease(inst_creator, ip, timeout=160):
"""
Returns true if the expected DHCP lease has been acquired
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
def test_create_volume_simple(self):
"""
- Tests the creation of an OpenStack volume from a URL.
+ Tests the creation of a simple OpenStack volume.
"""
# Create Volume
- self.volume_creator = create_volume.OpenStackVolume(
+ self.volume_creator = OpenStackVolume(
self.os_creds, self.volume_settings)
created_volume = self.volume_creator.create(block=True)
self.assertIsNotNone(created_volume)
clean() does not raise an Exception.
"""
# Create Volume
- self.volume_creator = create_volume.OpenStackVolume(
+ self.volume_creator = OpenStackVolume(
self.os_creds, self.volume_settings)
created_volume = self.volume_creator.create(block=True)
self.assertIsNotNone(created_volume)
self.assertEqual(volume1, volume2)
+class CreateSimpleVolumeFailureTests(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__()
+
+ self.guid = uuid.uuid4()
+ 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_bad_size(self):
+ """
+ Tests the creation of an OpenStack volume with a negative size to
+ ensure it raises a BadRequest exception.
+ """
+ volume_settings = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid), size=-1)
+
+ # Create Volume
+ self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
+
+ with self.assertRaises(BadRequest):
+ self.volume_creator.create(block=True)
+
+ def test_create_volume_bad_type(self):
+ """
+ Tests the creation of an OpenStack volume with a type that does not
+ exist to ensure it raises a NotFound exception.
+ """
+ volume_settings = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid),
+ type_name='foo')
+
+ # Create Volume
+ self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
+
+ with self.assertRaises(NotFound):
+ self.volume_creator.create(block=True)
+
+ def test_create_volume_bad_image(self):
+ """
+ Tests the creation of an OpenStack volume with an image that does not
+ exist to ensure it raises a BadRequest exception.
+ """
+ volume_settings = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid),
+ image_name='foo')
+
+ # Create Volume
+ self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
+
+ with self.assertRaises(BadRequest):
+ self.volume_creator.create(block=True)
+
+ def test_create_volume_bad_zone(self):
+ """
+ Tests the creation of an OpenStack volume with an availability zone
+ that does not exist to ensure it raises a BadRequest exception.
+ """
+ volume_settings = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid),
+ availability_zone='foo')
+
+ # Create Volume
+ self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
+
+ with self.assertRaises(BadRequest):
+ self.volume_creator.create(block=True)
+
+
class CreateVolumeWithTypeTests(OSIntegrationTestCase):
"""
Test cases for the CreateVolume when attempting to associate it to a
VolumeSettings(name=self.volume_name,
type_name=self.volume_type_name))
- created_volume = self.volume_creator.create()
+ created_volume = self.volume_creator.create(block=True)
self.assertIsNotNone(created_volume)
self.assertEqual(self.volume_type_name, created_volume.type)
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_name = guid + '-vol'
self.image_name = guid + '-image'
def test_bad_image_name(self):
"""
- Expect a NotFound to be raised when the volume type does not exist
+ Tests OpenStackVolume#create() method to ensure a volume is NOT created
+ when associating it to an invalid image name
"""
self.volume_creator = OpenStackVolume(
self.os_creds,
def test_valid_volume_image(self):
"""
- Expect a NotFound to be raised when the volume type does not exist
+ Tests OpenStackVolume#create() method to ensure a volume is NOT created
+ when associating it to an invalid image name
"""
self.volume_creator = OpenStackVolume(
self.os_creds,
created_volume = self.volume_creator.create(block=True)
self.assertIsNotNone(created_volume)
- self.assertIsNone(created_volume.type)
+ self.assertEqual(
+ self.volume_creator.volume_settings.name, created_volume.name)
self.assertTrue(self.volume_creator.volume_active())
+
+ retrieved_volume = cinder_utils.get_volume_by_id(
+ self.cinder, created_volume.id)
+
+ self.assertEqual(created_volume, retrieved_volume)
description=volume.description, size=volume.size,
vol_type=volume.volume_type,
availability_zone=volume.availability_zone,
- multi_attach=volume.multiattach)
+ multi_attach=volume.multiattach,
+ attachments=volume.attachments)
+
+
+def __get_os_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
+ """
+ return cinder.volumes.get(volume_id)
def get_volume_by_id(cinder, volume_id):
:param volume_id: the volume ID to lookup
:return: the SNAPS-OO Domain Volume object or None
"""
- volume = cinder.volumes.get(volume_id)
+ volume = __get_os_volume_by_id(cinder, 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)
+ multi_attach=volume.multiattach, attachments=volume.attachments)
def get_volume_status(cinder, volume):
:return: the OpenStack volume object
:raise Exception if using a file and it cannot be found
"""
- created_volume = cinder.volumes.create(
+ 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,
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)
+ 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)
def delete_volume(cinder, volume):
:param volume: the volume to delete
"""
logger.info('Deleting volume named - %s', volume.name)
- cinder.volumes.delete(volume.id)
+ return cinder.volumes.delete(volume.id)
def get_volume_type(cinder, volume_type_name=None, volume_type_settings=None):
import logging
import os
+import time
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
return __map_os_server_obj_to_vm_inst(server)
+def get_server_connection(nova, vm_inst_settings=None, server_name=None):
+ """
+ Returns a VmInst object for the first server instance found.
+ :param nova: the Nova client
+ :param vm_inst_settings: the VmInstanceSettings object from which to build
+ the query if not None
+ :param server_name: the server with this name to return if vm_inst_settings
+ is not None
+ :return: a snaps.domain.VmInst object or None if not found
+ """
+ search_opts = dict()
+ if vm_inst_settings:
+ search_opts['name'] = vm_inst_settings.name
+ elif server_name:
+ search_opts['name'] = server_name
+
+ servers = nova.servers.list(search_opts=search_opts)
+ for server in servers:
+ return server.links[0]
+
+
def __map_os_server_obj_to_vm_inst(os_server):
"""
Returns a VmInst object for an OpenStack Server object
if sec_group.get('name'):
sec_grp_names.append(sec_group.get('name'))
+ volumes = None
+ if hasattr(os_server, 'os-extended-volumes:volumes_attached'):
+ volumes = getattr(os_server, 'os-extended-volumes:volumes_attached')
+
return VmInst(
name=os_server.name, inst_id=os_server.id,
image_id=os_server.image['id'], flavor_id=os_server.flavor['id'],
networks=os_server.networks, keypair_name=os_server.key_name,
- sec_grp_names=sec_grp_names)
+ sec_grp_names=sec_grp_names, volume_ids=volumes)
def __get_latest_server_os_object(nova, server):
update_values['cores'] = compute_quotas.cores
update_values['instances'] = compute_quotas.instances
update_values['injected_files'] = compute_quotas.injected_files
- update_values['injected_file_content_bytes'] = compute_quotas.injected_file_content_bytes
+ update_values['injected_file_content_bytes'] = (
+ compute_quotas.injected_file_content_bytes)
update_values['ram'] = compute_quotas.ram
update_values['fixed_ips'] = compute_quotas.fixed_ips
update_values['key_pairs'] = compute_quotas.key_pairs
return nova.quotas.update(project_id, **update_values)
+def attach_volume(nova, server, volume, timeout=None):
+ """
+ Attaches a volume to a server
+ :param nova: the nova client
+ :param server: the VMInst domain object
+ :param volume: the Volume domain object
+ :param timeout: denotes the amount of time to block to determine if the
+ has been properly attached. When None, do not wait.
+ :return: the value from the nova call
+ """
+ nova.volumes.create_server_volume(server.id, volume.id)
+
+ if timeout:
+ start_time = time.time()
+ while time.time() < start_time + timeout:
+ vm = get_server_object_by_id(nova, server.id)
+ for vol_dict in vm.volume_ids:
+ if volume.id == vol_dict['id']:
+ return vm
+
+ return None
+ else:
+ return get_server_object_by_id(nova, server.id)
+
+
+def detach_volume(nova, server, volume, timeout=None):
+ """
+ Attaches a volume to a server
+ :param nova: the nova client
+ :param server: the VMInst domain object
+ :param volume: the Volume domain object
+ :param timeout: denotes the amount of time to block to determine if the
+ has been properly detached. When None, do not wait.
+ :return: the value from the nova call
+ """
+ nova.volumes.delete_server_volume(server.id, volume.id)
+
+ if timeout:
+ start_time = time.time()
+ while time.time() < start_time + timeout:
+ vm = get_server_object_by_id(nova, server.id)
+ found = False
+ for vol_dict in vm.volume_ids:
+ if volume.id == vol_dict['id']:
+ found = True
+
+ if not found:
+ return vm
+
+ return None
+ else:
+ return get_server_object_by_id(nova, server.id)
+
+
class NovaException(Exception):
"""
Exception when calls to the Keystone client cannot be served properly
from snaps.openstack import create_instance
from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
from snaps.openstack.create_image import OpenStackImage
-from snaps.openstack.create_instance import VmInstanceSettings
+from snaps.openstack.create_instance import (
+ VmInstanceSettings, OpenStackVmInstance)
from snaps.openstack.create_network import OpenStackNetwork, PortSettings
+from snaps.openstack.create_volume import OpenStackVolume, VolumeSettings
from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
-from snaps.openstack.utils import nova_utils, neutron_utils, glance_utils
+from snaps.openstack.utils import (
+ nova_utils, neutron_utils, glance_utils, cinder_utils)
__author__ = 'spisarski'
self.assertEqual(self.vm_inst.name, vm_inst.name)
self.assertEqual(self.vm_inst.id, vm_inst.id)
+
+
+class NovaUtilsInstanceVolumeTests(OSComponentTestCase):
+ """
+ Tests the creation of VM instances via nova_utils.py
+ """
+
+ def setUp(self):
+ """
+ Setup objects required by VM instances
+ :return:
+ """
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+
+ self.nova = nova_utils.nova_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(self.os_creds)
+
+ self.image_creator = None
+ self.network_creator = None
+ self.flavor_creator = None
+ self.volume_creator = None
+ self.instance_creator = None
+
+ try:
+ image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
+ self.image_creator = OpenStackImage(
+ self.os_creds, image_settings=image_settings)
+ self.image_creator.create()
+
+ network_settings = openstack_tests.get_priv_net_config(
+ guid + '-net', guid + '-subnet').network_settings
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, network_settings)
+ self.network_creator.create()
+
+ self.flavor_creator = OpenStackFlavor(
+ self.os_creds,
+ FlavorSettings(
+ name=guid + '-flavor-name', ram=256, disk=10, vcpus=1))
+ self.flavor_creator.create()
+
+ # Create Volume
+ volume_settings = VolumeSettings(
+ name=self.__class__.__name__ + '-' + str(guid))
+ self.volume_creator = OpenStackVolume(
+ self.os_creds, volume_settings)
+ self.volume_creator.create(block=True)
+
+ port_settings = PortSettings(
+ name=guid + '-port', network_name=network_settings.name)
+ instance_settings = VmInstanceSettings(
+ name=guid + '-vm_inst',
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
+ self.instance_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings, image_settings)
+ self.instance_creator.create(block=True)
+ except:
+ self.tearDown()
+ raise
+
+ def tearDown(self):
+ """
+ Cleanup deployed resources
+ :return:
+ """
+ if self.instance_creator:
+ try:
+ self.instance_creator.clean()
+ except:
+ pass
+ if self.volume_creator:
+ try:
+ self.volume_creator.clean()
+ except:
+ pass
+ if self.flavor_creator:
+ try:
+ self.flavor_creator.clean()
+ except:
+ pass
+ if self.network_creator:
+ try:
+ self.network_creator.clean()
+ except:
+ pass
+ if self.image_creator:
+ try:
+ self.image_creator.clean()
+ except:
+ pass
+
+ def test_add_remove_volume(self):
+ """
+ Tests the nova_utils.create_server() method
+ :return:
+ """
+
+ self.assertIsNotNone(self.volume_creator.get_volume())
+ self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
+
+ # Attach volume to VM
+ nova_utils.attach_volume(
+ self.nova, self.instance_creator.get_vm_inst(),
+ self.volume_creator.get_volume())
+
+ time.sleep(10)
+
+ vol_attach = cinder_utils.get_volume_by_id(
+ self.cinder, self.volume_creator.get_volume().id)
+ vm_attach = nova_utils.get_server_object_by_id(
+ self.nova, self.instance_creator.get_vm_inst().id)
+
+ # Detach volume to VM
+ nova_utils.detach_volume(
+ self.nova, self.instance_creator.get_vm_inst(),
+ self.volume_creator.get_volume())
+
+ time.sleep(10)
+
+ vol_detach = cinder_utils.get_volume_by_id(
+ self.cinder, self.volume_creator.get_volume().id)
+ vm_detach = nova_utils.get_server_object_by_id(
+ self.nova, self.instance_creator.get_vm_inst().id)
+
+ # Validate Attachment
+ self.assertIsNotNone(vol_attach)
+ self.assertEqual(self.volume_creator.get_volume().id, vol_attach.id)
+ self.assertEqual(1, len(vol_attach.attachments))
+ self.assertEqual(vm_attach.volume_ids[0]['id'],
+ vol_attach.attachments[0]['volume_id'])
+
+ # Validate Detachment
+ self.assertIsNotNone(vol_detach)
+ self.assertEqual(self.volume_creator.get_volume().id, vol_detach.id)
+ self.assertEqual(0, len(vol_detach.attachments))
+ self.assertEqual(0, len(vm_detach.volume_ids))
FloatingIpSettingsUnitTests, InstanceSecurityGroupTests,
VmInstanceSettingsUnitTests, CreateInstancePortManipulationTests,
SimpleHealthCheck, CreateInstanceFromThreePartImage,
- CreateInstanceMockOfflineTests, CreateInstanceTwoNetTests)
+ CreateInstanceMockOfflineTests, CreateInstanceTwoNetTests,
+ CreateInstanceVolumeTests)
from snaps.openstack.tests.create_keypairs_tests import (
CreateKeypairsTests, KeypairSettingsUnitTests, CreateKeypairsCleanupTests)
from snaps.openstack.tests.create_network_tests import (
UserSettingsUnitTests, CreateUserSuccessTests)
from snaps.openstack.tests.create_volume_tests import (
VolumeSettingsUnitTests, CreateSimpleVolumeSuccessTests,
- CreateVolumeWithTypeTests, CreateVolumeWithImageTests)
+ CreateVolumeWithTypeTests, CreateVolumeWithImageTests,
+ CreateSimpleVolumeFailureTests)
from snaps.openstack.tests.create_volume_type_tests import (
VolumeTypeSettingsUnitTests, CreateSimpleVolumeTypeSuccessTests,
CreateVolumeTypeComplexTests)
NeutronUtilsFloatingIpTests)
from snaps.openstack.utils.tests.nova_utils_tests import (
NovaSmokeTests, NovaUtilsKeypairTests, NovaUtilsFlavorTests,
- NovaUtilsInstanceTests)
+ NovaUtilsInstanceTests, NovaUtilsInstanceVolumeTests)
from snaps.provisioning.tests.ansible_utils_tests import (
AnsibleProvisioningTests)
from snaps.tests.file_utils_tests import FileUtilsTests
suite.addTest(OSComponentTestCase.parameterize(
NovaUtilsInstanceTests, os_creds=os_creds, ext_net_name=ext_net_name,
log_level=log_level, image_metadata=image_metadata))
+ suite.addTest(OSComponentTestCase.parameterize(
+ NovaUtilsInstanceVolumeTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level,
+ image_metadata=image_metadata))
suite.addTest(OSComponentTestCase.parameterize(
CreateFlavorTests, 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(
+ CreateSimpleVolumeFailureTests, 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,
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(
+ CreateInstanceVolumeTests, 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(
CreateStackSuccessTests, os_creds=os_creds, ext_net_name=ext_net_name,
use_keystone=use_keystone,
Code Review / snaps.git / commitdiff