| | | by Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
+heat_utils_tests.py - HeatUtilsVolumeTests
+------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_create_vol_with_stack | 1 | Tests ability of the function |
+| | | heat_utils.get_stack_volumes() to return the correct |
+| | | Volume domain objects deployed with Heat |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_vol_types_with_stack | 1 | Tests ability of the function |
+| | | heat_utils.get_stack_volumes_types() to return the correct|
+| | | VolumeType domain objects deployed with Heat |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
settings_utils_tests.py - SettingsUtilsNetworkingTests
------------------------------------------------------
-----------------------------------------------
+---------------------------------------+---------------+-----------------------------------------------------------+
-| Test Name | Neutron API | Description |
+| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_create_stack_template_file | 2 | Ensures that a Heat stack can be created with a file-based|
+| test_create_stack_template_file | 1 | Ensures that a Heat stack can be created with a file-based|
| | | Heat template file |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_stack_template_dict | 2 | Ensures that a Heat stack can be created with a dictionary|
+| test_create_stack_template_dict | 1 | Ensures that a Heat stack can be created with a dictionary|
| | | Heat template |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_delete_stack | 2 | Ensures that a Heat stack can be created and deleted |
+| test_create_delete_stack | 1 | Ensures that a Heat stack can be created and deleted |
| | | while having clean() called 2x without an exception |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_same_stack | 2 | Ensures that a Heat stack with the same name cannot be |
+| test_create_same_stack | 1 | Ensures that a Heat stack with the same name cannot be |
| | | created 2x |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_retrieve_network_creators | 2 | Ensures that an OpenStackHeatStack instance can return an |
+| test_retrieve_network_creators | 1 | Ensures that an OpenStackHeatStack instance can return an |
| | | OpenStackNetwork instance configured as deployed |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_retrieve_vm_inst_creators | 2 | Ensures that an OpenStackHeatStack instance can return an |
+| test_retrieve_vm_inst_creators | 1 | Ensures that an OpenStackHeatStack instance can return an |
| | | OpenStackVmInstance instance configured as deployed |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_stack_tests.py - CreateStackVolumeTests
+----------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_retrieve_volume_creator | 1 | Ensures that an OpenStackHeatStack instance can return a |
+| | | OpenStackVolume instance that it was responsible for |
+| | | deploying |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_retrieve_volume_type_creator | 1 | Ensures that an OpenStackHeatStack instance can return a |
+| | | OpenStackVolumeType instance that it was responsible for |
+| | | deploying |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_stack_tests.py - CreateComplexStackTests
-----------------------------------------------
+---------------------------------------+---------------+-----------------------------------------------------------+
-| Test Name | Neutron API | Description |
+| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_connect_via_ssh_heat_vm | 2 | Ensures that two OpenStackHeatStack instances can return |
+| test_connect_via_ssh_heat_vm | 1 | Ensures that two OpenStackHeatStack instances can return |
| | | OpenStackVmInstance instances one configured with a |
| | | floating IP and keypair and can be access via SSH |
+---------------------------------------+---------------+-----------------------------------------------------------+
------------------------------------------------
+----------------------------------------+---------------+-----------------------------------------------------------+
-| Test Name | Neutron API | Description |
+| Test Name | Heat API | Description |
+========================================+===============+===========================================================+
-| test_missing_dependencies | 2 | Ensures that a Heat template fails to deploy when expected|
+| test_missing_dependencies | 1 | Ensures that a Heat template fails to deploy when expected|
| | | dependencies are missing |
+----------------------------------------+---------------+-----------------------------------------------------------+
-| test_bad_stack_file | 2 | Ensures that a Heat template fails to deploy when the Heat|
+| test_bad_stack_file | 1 | Ensures that a Heat template fails to deploy when the Heat|
| | | template file does not exist |
+----------------------------------------+---------------+-----------------------------------------------------------+
Ensures that all required members are included when constructing a
VmInst domain object
+
+SettingsUtilsVolumeTests
+------------------------
+
+Ensures that the settings_utils.py#create_volume_settings() function properly
+maps a snaps.domain.Volume object correctly to a
+snaps.openstack.create_volume.VolumeSettings object as well as a
+snaps.domain.VolumeType object to a
+snaps.openstack.create_volume.VolumeSettings object
from heatclient.exc import HTTPNotFound
from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_volume import OpenStackVolume
+from snaps.openstack.create_volume_type import OpenStackVolumeType
from snaps.openstack.openstack_creator import OpenStackCloudObject
-from snaps.openstack.utils import nova_utils, settings_utils, glance_utils
+from snaps.openstack.utils import (
+ nova_utils, settings_utils, glance_utils, cinder_utils)
from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.utils import heat_utils, neutron_utils
self.initialize()
if self.__stack:
- logger.info('Found stack with name - ' + self.stack_settings.name)
+ logger.info('Found stack with name - %s', self.stack_settings.name)
return self.__stack
else:
self.__stack = heat_utils.create_stack(self.__heat_cli,
self.stack_settings)
logger.info(
- 'Created stack with name - ' + self.stack_settings.name)
+ 'Created stack with name - %s', self.stack_settings.name)
if self.__stack and self.stack_complete(block=True):
- logger.info(
- 'Stack is now active with name - ' +
- self.stack_settings.name)
+ logger.info('Stack is now active with name - %s',
+ self.stack_settings.name)
return self.__stack
else:
status = heat_utils.get_stack_status_reason(self.__heat_cli,
self.__stack.id)
- logger.error(
- 'ERROR: STACK CREATION FAILED: ' + status)
- raise StackCreationError(
- 'Failure while creating stack')
+ logger.error('ERROR: STACK CREATION FAILED: %s', status)
+ raise StackCreationError('Failure while creating stack')
def clean(self):
"""
"""
if self.__stack:
try:
- logger.info('Deleting stack - %s' + self.__stack.name)
+ logger.info('Deleting stack - %s', self.__stack.name)
heat_utils.delete_stack(self.__heat_cli, self.__stack)
try:
return out
+ def get_volume_creators(self):
+ """
+ Returns a list of Volume creator objects as configured by the heat
+ template
+ :return: list() of OpenStackVolume objects
+ """
+
+ out = list()
+ cinder = cinder_utils.cinder_client(self._os_creds)
+
+ volumes = heat_utils.get_stack_volumes(
+ self.__heat_cli, cinder, self.__stack)
+
+ for volume in volumes:
+ settings = settings_utils.create_volume_settings(volume)
+ creator = OpenStackVolume(self._os_creds, settings)
+ out.append(creator)
+
+ try:
+ creator.initialize()
+ except Exception as e:
+ logger.error(
+ 'Unexpected error initializing volume creator - %s', e)
+
+ return out
+
+ def get_volume_type_creators(self):
+ """
+ Returns a list of VolumeType creator objects as configured by the heat
+ template
+ :return: list() of OpenStackVolumeType objects
+ """
+
+ out = list()
+ cinder = cinder_utils.cinder_client(self._os_creds)
+
+ vol_types = heat_utils.get_stack_volume_types(
+ self.__heat_cli, cinder, self.__stack)
+
+ for volume in vol_types:
+ settings = settings_utils.create_volume_type_settings(volume)
+ creator = OpenStackVolumeType(self._os_creds, settings)
+ out.append(creator)
+
+ try:
+ creator.initialize()
+ except Exception as e:
+ logger.error(
+ 'Unexpected error initializing volume type creator - %s',
+ e)
+
+ return out
+
def _stack_status_check(self, expected_status_code, block, timeout,
poll_interval, fail_status):
"""
nova, vm_inst_creators[0].get_vm_inst().id))
-class CreateComplexStackTests(OSIntegrationTestCase):
+class CreateStackFloatingIpTests(OSIntegrationTestCase):
"""
Tests for the CreateStack class defined in create_stack.py
"""
self.assertEqual(0, len(vm_settings.floating_ip_settings))
+class CreateStackVolumeTests(OSIntegrationTestCase):
+ """
+ Tests for the CreateStack class defined in create_stack.py
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateStack object that is responsible for downloading
+ and creating an OS stack file within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+
+ self.heat_creds = self.admin_os_creds
+ self.heat_creds.project_name = self.admin_os_creds.project_name
+
+ self.heat_cli = heat_utils.heat_client(self.heat_creds)
+ self.stack_creator = None
+
+ self.volume_name = self.guid + '-volume'
+ self.volume_type_name = self.guid + '-volume-type'
+
+ self.env_values = {
+ 'volume_name': self.volume_name,
+ 'volume_type_name': self.volume_type_name}
+
+ self.heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'volume_heat_template.yaml')
+
+ stack_settings = StackSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid) + '-stack',
+ template_path=self.heat_tmplt_path,
+ env_values=self.env_values)
+ self.stack_creator = create_stack.OpenStackHeatStack(
+ self.heat_creds, stack_settings)
+ self.created_stack = self.stack_creator.create()
+ self.assertIsNotNone(self.created_stack)
+
+ def tearDown(self):
+ """
+ Cleans the stack and downloaded stack file
+ """
+ if self.stack_creator:
+ try:
+ self.stack_creator.clean()
+ except:
+ pass
+
+ super(self.__class__, self).__clean__()
+
+ def test_retrieve_volume_creator(self):
+ """
+ Tests the creation of an OpenStack stack from Heat template file and
+ the retrieval of an OpenStackVolume creator/state machine instance
+ """
+ volume_creators = self.stack_creator.get_volume_creators()
+ self.assertEqual(1, len(volume_creators))
+
+ creator = volume_creators[0]
+ self.assertEqual(self.volume_name, creator.volume_settings.name)
+ self.assertEqual(self.volume_name, creator.get_volume().name)
+ self.assertEqual(self.volume_type_name,
+ creator.volume_settings.type_name)
+ self.assertEqual(self.volume_type_name, creator.get_volume().type)
+ self.assertEqual(1, creator.volume_settings.size)
+ self.assertEqual(1, creator.get_volume().size)
+
+ def test_retrieve_volume_type_creator(self):
+ """
+ Tests the creation of an OpenStack stack from Heat template file and
+ the retrieval of an OpenStackVolume creator/state machine instance
+ """
+ volume_type_creators = self.stack_creator.get_volume_type_creators()
+ self.assertEqual(1, len(volume_type_creators))
+
+ creator = volume_type_creators[0]
+ self.assertIsNotNone(creator)
+
+ volume_type = creator.get_volume_type()
+ self.assertIsNotNone(volume_type)
+
+ self.assertEqual(self.volume_type_name, volume_type.name)
+ self.assertTrue(volume_type.public)
+ self.assertIsNone(volume_type.qos_spec)
+
+ encryption = volume_type.encryption
+ self.assertIsNotNone(encryption)
+ self.assertIsNone(encryption.cipher)
+ self.assertEqual('front-end', encryption.control_location)
+ self.assertIsNone(encryption.key_size)
+ self.assertEqual(u'nova.volume.encryptors.luks.LuksEncryptor',
+ encryption.provider)
+ self.assertEqual(volume_type.id, encryption.volume_type_id)
+
+
class CreateStackNegativeTests(OSIntegrationTestCase):
"""
Negative test cases for the CreateStack class
--- /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.
+##############################################################################
+heat_template_version: 2015-04-30
+
+description: Simple template to deploy a single volume with encryption
+
+parameters:
+ volume_name:
+ type: string
+ label: Volume name
+ description: The name of the volume
+ default: test-vol-name
+ volume_type_name:
+ type: string
+ label: Volume Type name
+ description: The name of the volume type
+ default: test-vol-type-name
+
+resources:
+ volume_type:
+ type: OS::Cinder::VolumeType
+ properties:
+ name: { get_param: volume_type_name }
+
+ encryption_vol_type:
+ type: OS::Cinder::EncryptedVolumeType
+ properties:
+ provider: nova.volume.encryptors.luks.LuksEncryptor
+ control_location: front-end
+ volume_type: { get_resource: volume_type }
+
+ volume:
+ type: OS::Cinder::Volume
+ properties:
+ name: { get_param: volume_name }
+ size: 1
+ volume_type: { get_resource: encryption_vol_type }
from snaps import file_utils
from snaps.domain.stack import Stack, Resource, Output
-from snaps.openstack.utils import keystone_utils, neutron_utils, nova_utils
+from snaps.openstack.utils import keystone_utils, neutron_utils, nova_utils, \
+ cinder_utils
__author__ = 'spisarski'
return heat_cli.resources.list(stack.id)
-def get_resources(heat_cli, stack):
+def get_resources(heat_cli, stack, res_type=None):
"""
Returns all of the OpenStack resource objects for a given stack
:param heat_cli: the OpenStack heat client
:param stack: the SNAPS-OO Stack domain object
- :return: a list
+ :param res_type: the type name to filter
+ :return: a list of Resource domain objects
"""
os_resources = __get_os_resources(heat_cli, stack)
if os_resources:
out = list()
for os_resource in os_resources:
- out.append(Resource(resource_type=os_resource.resource_type,
- resource_id=os_resource.physical_resource_id))
+ if ((res_type and os_resource.resource_type == res_type)
+ or not res_type):
+ out.append(Resource(
+ resource_type=os_resource.resource_type,
+ resource_id=os_resource.physical_resource_id))
return out
for given stack
:param heat_cli: the OpenStack heat client
:param stack: the SNAPS-OO Stack domain object
- :return: a list
+ :return: a list of Output domain objects
"""
out = list()
def get_stack_networks(heat_cli, neutron, stack):
"""
- Returns an instance of NetworkSettings for each network owned by this stack
+ Returns a list of Network domain objects deployed by this stack
:param heat_cli: the OpenStack heat client object
:param neutron: the OpenStack neutron client object
:param stack: the SNAPS-OO Stack domain object
- :return: a list of NetworkSettings
+ :return: a list of Network objects
"""
out = list()
- resources = get_resources(heat_cli, stack)
+ resources = get_resources(heat_cli, stack, 'OS::Neutron::Net')
for resource in resources:
- if resource.type == 'OS::Neutron::Net':
- network = neutron_utils.get_network_by_id(
- neutron, resource.id)
- if network:
- out.append(network)
+ network = neutron_utils.get_network_by_id(
+ neutron, resource.id)
+ if network:
+ out.append(network)
return out
def get_stack_servers(heat_cli, nova, stack):
"""
- Returns an instance of NetworkSettings for each network owned by this stack
+ Returns a list of VMInst domain objects associated with a Stack
:param heat_cli: the OpenStack heat client object
:param nova: the OpenStack nova client object
:param stack: the SNAPS-OO Stack domain object
- :return: a list of NetworkSettings
+ :return: a list of VMInst domain objects
"""
out = list()
- resources = get_resources(heat_cli, stack)
+ resources = get_resources(heat_cli, stack, 'OS::Nova::Server')
for resource in resources:
- if resource.type == 'OS::Nova::Server':
- try:
- server = nova_utils.get_server_object_by_id(
- nova, resource.id)
- if server:
- out.append(server)
- except NotFound:
- logger.warn(
- 'VmInst cannot be located with ID %s', resource.id)
+ try:
+ server = nova_utils.get_server_object_by_id(nova, resource.id)
+ if server:
+ out.append(server)
+ except NotFound:
+ logger.warn('VmInst cannot be located with ID %s', resource.id)
+
+ return out
+
+
+def get_stack_volumes(heat_cli, cinder, stack):
+ """
+ Returns an instance of NetworkSettings for each network owned by this stack
+ :param heat_cli: the OpenStack heat client object
+ :param cinder: the OpenStack cinder client object
+ :param stack: the SNAPS-OO Stack domain object
+ :return: a list of Volume domain objects
+ """
+
+ out = list()
+ resources = get_resources(heat_cli, stack, 'OS::Cinder::Volume')
+ for resource in resources:
+ try:
+ server = cinder_utils.get_volume_by_id(cinder, resource.id)
+ if server:
+ out.append(server)
+ except NotFound:
+ logger.warn('Volume cannot be located with ID %s', resource.id)
+
+ return out
+
+
+def get_stack_volume_types(heat_cli, cinder, stack):
+ """
+ Returns an instance of NetworkSettings for each network owned by this stack
+ :param heat_cli: the OpenStack heat client object
+ :param cinder: the OpenStack cinder client object
+ :param stack: the SNAPS-OO Stack domain object
+ :return: a list of VolumeType domain objects
+ """
+
+ out = list()
+ resources = get_resources(heat_cli, stack, 'OS::Cinder::VolumeType')
+ for resource in resources:
+ try:
+ vol_type = cinder_utils.get_volume_type_by_id(cinder, resource.id)
+ if vol_type:
+ out.append(vol_type)
+ except NotFound:
+ logger.warn('VolumeType cannot be located with ID %s', resource.id)
return out
from snaps.openstack.create_keypairs import KeypairSettings
from snaps.openstack.create_network import (
PortSettings, SubnetSettings, NetworkSettings)
+from snaps.openstack.create_volume import VolumeSettings
+from snaps.openstack.create_volume_type import (
+ VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation)
from snaps.openstack.utils import (
neutron_utils, nova_utils, heat_utils, glance_utils)
return out
+def create_volume_settings(volume):
+ """
+ Returns a VolumeSettings object
+ :param volume: a SNAPS-OO Volume object
+ """
+
+ return VolumeSettings(
+ name=volume.name, description=volume.description,
+ size=volume.size, type_name=volume.type,
+ availability_zone=volume.availability_zone,
+ multi_attach=volume.multi_attach)
+
+
+def create_volume_type_settings(volume_type):
+ """
+ Returns a VolumeTypeSettings object
+ :param volume_type: a SNAPS-OO VolumeType object
+ """
+
+ control = None
+ if volume_type.encryption:
+ if (volume_type.encryption.control_location
+ == ControlLocation.front_end.value):
+ control = ControlLocation.front_end
+ else:
+ control = ControlLocation.back_end
+
+ encrypt_settings = VolumeTypeEncryptionSettings(
+ name=volume_type.encryption.__class__,
+ provider_class=volume_type.encryption.provider,
+ control_location=control,
+ cipher=volume_type.encryption.cipher,
+ key_size=volume_type.encryption.key_size)
+
+ qos_spec_name = None
+ if volume_type.qos_spec:
+ qos_spec_name = volume_type.qos_spec.name
+
+ return VolumeTypeSettings(
+ name=volume_type.name, encryption=encrypt_settings,
+ qos_spec_name=qos_spec_name, public=volume_type.public)
+
+
def create_vm_inst_settings(nova, neutron, server):
"""
Returns a NetworkSettings object
from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
from snaps.openstack.utils import (
- heat_utils, neutron_utils, nova_utils, settings_utils, glance_utils)
+ heat_utils, neutron_utils, nova_utils, settings_utils, glance_utils,
+ cinder_utils)
__author__ = 'spisarski'
priv_key_key='private_key')
self.assertIsNotNone(keypair2_settings)
self.assertEqual(self.keypair_name, keypair2_settings.name)
+
+
+class HeatUtilsVolumeTests(OSComponentTestCase):
+ """
+ Test Heat volume functionality
+ """
+
+ def setUp(self):
+ """
+ Instantiates OpenStack instances that cannot be spawned by Heat
+ """
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ stack_name = guid + '-stack'
+ self.volume_name = guid + '-vol'
+ self.volume_type_name = guid + '-vol-type'
+
+ env_values = {
+ 'volume_name': self.volume_name,
+ 'volume_type_name': self.volume_type_name}
+
+ heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'volume_heat_template.yaml')
+ self.stack_settings = StackSettings(
+ name=stack_name, template_path=heat_tmplt_path,
+ env_values=env_values)
+ self.stack = None
+ self.heat_client = heat_utils.heat_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(self.os_creds)
+
+ def tearDown(self):
+ """
+ Cleans the image and downloaded image file
+ """
+ if self.stack:
+ try:
+ heat_utils.delete_stack(self.heat_client, self.stack)
+ except:
+ pass
+
+ def test_create_vol_with_stack(self):
+ """
+ Tests the creation of an OpenStack volume with Heat.
+ """
+ self.stack = heat_utils.create_stack(
+ self.heat_client, self.stack_settings)
+
+ # Wait until stack deployment has completed
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+ is_active = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack.id)
+ if status == create_stack.STATUS_CREATE_COMPLETE:
+ is_active = True
+ break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
+
+ time.sleep(3)
+
+ self.assertTrue(is_active)
+
+ volumes = heat_utils.get_stack_volumes(
+ self.heat_client, self.cinder, self.stack)
+
+ self.assertEqual(1, len(volumes))
+
+ volume = volumes[0]
+ self.assertEqual(self.volume_name, volume.name)
+ self.assertEqual(self.volume_type_name, volume.type)
+ self.assertEqual(1, volume.size)
+ self.assertEqual(False, volume.multi_attach)
+
+ def test_create_vol_types_with_stack(self):
+ """
+ Tests the creation of an OpenStack volume with Heat.
+ """
+ self.stack = heat_utils.create_stack(
+ self.heat_client, self.stack_settings)
+
+ # Wait until stack deployment has completed
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+ is_active = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack.id)
+ if status == create_stack.STATUS_CREATE_COMPLETE:
+ is_active = True
+ break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
+
+ time.sleep(3)
+
+ self.assertTrue(is_active)
+
+ volume_types = heat_utils.get_stack_volume_types(
+ self.heat_client, self.cinder, self.stack)
+
+ self.assertEqual(1, len(volume_types))
+
+ volume_type = volume_types[0]
+
+ self.assertEqual(self.volume_type_name, volume_type.name)
+ self.assertTrue(volume_type.public)
+ self.assertIsNone(volume_type.qos_spec)
+
+ encryption = volume_type.encryption
+ self.assertIsNotNone(encryption)
+ self.assertIsNone(encryption.cipher)
+ self.assertEqual('front-end', encryption.control_location)
+ self.assertIsNone(encryption.key_size)
+ self.assertEqual(u'nova.volume.encryptors.luks.LuksEncryptor',
+ encryption.provider)
+ self.assertEqual(volume_type.id, encryption.volume_type_id)
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
+import unittest
+
import os
import uuid
+from snaps.domain.volume import (
+ Volume, VolumeType, VolumeTypeEncryption, QoSSpec)
from snaps.openstack import (
create_image, create_network, create_router, create_flavor,
create_keypairs, create_instance)
from snaps.openstack.create_network import (
NetworkSettings, OpenStackNetwork, SubnetSettings)
+from snaps.openstack.create_qos import Consumer
from snaps.openstack.create_security_group import (
SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
SecurityGroupSettings)
self.assertIsNotNone(derived_image_settings)
self.assertEqual(self.image_creator.image_settings.name,
derived_image_settings.name)
+
+
+class SettingsUtilsVolumeTests(unittest.TestCase):
+ """
+ Exercises the settings_utils.py functions around volumes
+ """
+
+ 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)
+ settings = settings_utils.create_volume_settings(volume)
+ self.assertEqual(volume.name, settings.name)
+ self.assertEqual(volume.description, settings.description)
+ self.assertEqual(volume.size, settings.size)
+ self.assertEqual(volume.type, settings.type_name)
+ self.assertEqual(volume.availability_zone, settings.availability_zone)
+ self.assertEqual(volume.multi_attach, settings.multi_attach)
+
+ def test_vol_type_settings_from_vol(self):
+ encryption = VolumeTypeEncryption(
+ volume_encryption_id='vol-encrypt-id', volume_type_id='vol-typ-id',
+ control_location='front-end', provider='FooClass', cipher='1',
+ key_size=1)
+ qos_spec = QoSSpec(name='qos-spec-name', spec_id='qos-spec-id',
+ consumer=Consumer.back_end)
+ volume_type = VolumeType(
+ name='vol-type-name', volume_type_id='vol-type-id', public=True,
+ encryption=encryption, qos_spec=qos_spec)
+
+ settings = settings_utils.create_volume_type_settings(volume_type)
+ self.assertEqual(volume_type.name, settings.name)
+ self.assertEqual(volume_type.public, settings.public)
+
+ encrypt_settings = settings.encryption
+ self.assertIsNotNone(encrypt_settings)
+ self.assertEqual(encryption.control_location,
+ encrypt_settings.control_location.value)
+ self.assertEqual(encryption.cipher, encrypt_settings.cipher)
+ self.assertEqual(encryption.key_size, encrypt_settings.key_size)
+
+ self.assertEqual(qos_spec.name, settings.qos_spec_name)
SecurityGroupSettingsUnitTests)
from snaps.openstack.tests.create_stack_tests import (
StackSettingsUnitTests, CreateStackSuccessTests, CreateStackNegativeTests,
- CreateComplexStackTests)
+ CreateStackFloatingIpTests, CreateStackVolumeTests)
from snaps.openstack.tests.create_user_tests import (
UserSettingsUnitTests, CreateUserSuccessTests)
from snaps.openstack.tests.create_volume_tests import (
GlanceSmokeTests, GlanceUtilsTests)
from snaps.openstack.utils.tests.heat_utils_tests import (
HeatSmokeTests, HeatUtilsCreateSimpleStackTests,
- HeatUtilsCreateComplexStackTests)
+ HeatUtilsCreateComplexStackTests, HeatUtilsVolumeTests)
from snaps.openstack.utils.tests.keystone_utils_tests import (
KeystoneSmokeTests, KeystoneUtilsTests)
from snaps.openstack.utils.tests.neutron_utils_tests import (
from snaps.openstack.utils.tests.nova_utils_tests import (
NovaSmokeTests, NovaUtilsKeypairTests, NovaUtilsFlavorTests,
NovaUtilsInstanceTests, NovaUtilsInstanceVolumeTests)
+from snaps.openstack.utils.tests.settings_utils_tests import (
+ SettingsUtilsVolumeTests)
from snaps.provisioning.tests.ansible_utils_tests import (
AnsibleProvisioningTests)
from snaps.tests.file_utils_tests import FileUtilsTests
VolumeTypeSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
VolumeSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ SettingsUtilsVolumeTests))
def add_openstack_client_tests(suite, os_creds, ext_net_name,
HeatUtilsCreateComplexStackTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level,
image_metadata=image_metadata))
+ suite.addTest(OSComponentTestCase.parameterize(
+ HeatUtilsVolumeTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level,
+ image_metadata=image_metadata))
suite.addTest(OSComponentTestCase.parameterize(
CinderUtilsQoSTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level,
use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateStackVolumeTests, 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(
CreateStackNegativeTests, 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(
- CreateComplexStackTests, os_creds=os_creds,
+ CreateStackFloatingIpTests, 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))
Code Review / snaps.git / commitdiff