| 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 |
+| | | heat_utils.create_stack() to return the correct |
| | | Volume domain objects deployed with Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
| test_create_vol_types_with_stack | 1 | Tests ability of the function |
| | | VolumeType domain objects deployed with Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
+heat_utils_tests.py - HeatUtilsKeypairTests
+-------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_create_keypair_with_stack | 1 | Tests ability of the function |
+| | | heat_utils.create_stack() to return the correct |
+| | | Keypair domain objects deployed with Heat |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
settings_utils_tests.py - SettingsUtilsNetworkingTests
------------------------------------------------------
| | | deploying |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_stack_tests.py - CreateStackKeypairTests
+-----------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_retrieve_keypair_creator | 1 | Ensures that an OpenStackHeatStack instance can return a |
+| | | OpenStackKeypair instance that it was responsible for |
+| | | deploying |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_stack_tests.py - CreateComplexStackTests
-----------------------------------------------
from heatclient.exc import HTTPNotFound
from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_keypairs import OpenStackKeypair
from snaps.openstack.create_volume import OpenStackVolume
from snaps.openstack.create_volume_type import OpenStackVolumeType
from snaps.openstack.openstack_creator import OpenStackCloudObject
return out
+ def get_keypair_creators(self, outputs_pk_key=None):
+ """
+ Returns a list of keypair creator objects as configured by the heat
+ template
+ :return: list() of OpenStackKeypair objects
+ """
+
+ out = list()
+ nova = nova_utils.nova_client(self._os_creds)
+
+ keypairs = heat_utils.get_stack_keypairs(
+ self.__heat_cli, nova, self.__stack)
+
+ for keypair in keypairs:
+ settings = settings_utils.create_keypair_settings(
+ self.__heat_cli, self.__stack, keypair, outputs_pk_key)
+ creator = OpenStackKeypair(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):
"""
class CreateStackVolumeTests(OSIntegrationTestCase):
"""
- Tests for the CreateStack class defined in create_stack.py
+ Tests for the CreateStack class as they pertain to volumes
"""
def setUp(self):
self.assertEqual(volume_type.id, encryption.volume_type_id)
+class CreateStackKeypairTests(OSIntegrationTestCase):
+ """
+ Tests for the CreateStack class as they pertain to keypairs
+ """
+
+ 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.nova = nova_utils.nova_client(self.heat_creds)
+ self.stack_creator = None
+
+ self.keypair_name = self.guid + '-kp'
+
+ self.env_values = {
+ 'keypair_name': self.keypair_name}
+
+ self.heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'keypair_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_keypair_creator(self):
+ """
+ Tests the creation of an OpenStack stack from Heat template file and
+ the retrieval of an OpenStackKeypair creator/state machine instance
+ """
+ kp_creators = self.stack_creator.get_keypair_creators('private_key')
+ self.assertEqual(1, len(kp_creators))
+
+ creator = kp_creators[0]
+
+ self.assertEqual(self.keypair_name, creator.get_keypair().name)
+ self.assertIsNotNone(creator.keypair_settings.private_filepath)
+
+ private_file_contents = file_utils.read_file(
+ creator.keypair_settings.private_filepath)
+ self.assertTrue(private_file_contents.startswith(
+ '-----BEGIN RSA PRIVATE KEY-----'))
+
+ keypair = nova_utils.get_keypair_by_id(
+ self.nova, creator.get_keypair().id)
+ self.assertIsNotNone(keypair)
+ self.assertEqual(creator.get_keypair(), keypair)
+
+
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: >
+ Test template that simply deploys a keypair with a generated key
+
+parameters:
+ keypair_name:
+ type: string
+ label: Keypair name
+ description: The name of the stack's keypair
+ default: keypair_name
+
+resources:
+ keypair:
+ type: OS::Nova::KeyPair
+ properties:
+ name: { get_param: keypair_name }
+ save_private_key: True
+
+outputs:
+ private_key:
+ description: "SSH Private Key"
+ value: { get_attr: [ keypair, private_key ]}
return out
+def get_stack_keypairs(heat_cli, nova, stack):
+ """
+ Returns a list of Keypair 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 VMInst domain objects
+ """
+
+ out = list()
+ resources = get_resources(heat_cli, stack, 'OS::Nova::KeyPair')
+ for resource in resources:
+ try:
+ keypair = nova_utils.get_keypair_by_id(nova, resource.id)
+ if keypair:
+ out.append(keypair)
+ except NotFound:
+ logger.warn('Keypair 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
+ Returns an instance of Volume domain objects created 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
def get_stack_volume_types(heat_cli, cinder, stack):
"""
- Returns an instance of NetworkSettings for each network owned by this stack
+ Returns an instance of VolumeType domain objects created 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 None
+def get_keypair_by_id(nova, kp_id):
+ """
+ Returns a list of all available keypairs
+ :param nova: the Nova client
+ :param kp_id: the ID of the keypair to return
+ :return: the keypair object
+ """
+ keypair = nova.keypairs.get(kp_id)
+ return Keypair(name=keypair.name, kp_id=keypair.id,
+ public_key=keypair.public_key)
+
+
def delete_keypair(nova, key):
"""
Deletes a keypair object from OpenStack
qos_spec_name=qos_spec_name, public=volume_type.public)
+def create_keypair_settings(heat_cli, stack, keypair, pk_output_key):
+ """
+ Instantiates a KeypairSettings object from a Keypair domain objects
+ :param heat_cli: the heat client
+ :param stack: the Stack domain object
+ :param keypair: the Keypair SNAPS domain object
+ :param pk_output_key: the key to the heat template's outputs for retrieval
+ of the private key file
+ :return: a KeypairSettings object
+ """
+ if pk_output_key:
+ outputs = heat_utils.get_outputs(heat_cli, stack)
+ for output in outputs:
+ if output.key == pk_output_key:
+ # Save to file
+ guid = uuid.uuid4()
+ key_file = file_utils.save_string_to_file(
+ output.value, str(guid), 0o400)
+
+ # Use outputs, file and resources for the KeypairSettings
+ return KeypairSettings(
+ name=keypair.name, private_filepath=key_file.name)
+
+ return KeypairSettings(name=keypair.name)
+
+
def create_vm_inst_settings(nova, neutron, server):
"""
Returns a NetworkSettings object
self.assertIsNotNone(outputs)
self.assertEqual(0, len(outputs))
- # 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.stack1.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)
+ self.assertTrue(stack_active(self.heat_client, self.stack1))
neutron = neutron_utils.neutron_client(self.os_creds)
networks = heat_utils.get_stack_networks(
self.stack1.id)
self.assertEqual(self.stack1, stack1_query_3)
- 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.stack1.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)
+ self.assertTrue(stack_active(self.heat_client, self.stack1))
self.stack2 = heat_utils.create_stack(self.heat_client,
self.stack_settings2)
self.heat_client = heat_utils.heat_client(self.os_creds)
self.stack = heat_utils.create_stack(self.heat_client, 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)
+ self.assertTrue(stack_active(self.heat_client, self.stack))
def tearDown(self):
"""
"""
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)
+ self.assertTrue(stack_active(self.heat_client, self.stack))
volumes = heat_utils.get_stack_volumes(
self.heat_client, self.cinder, self.stack)
"""
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)
+ self.assertTrue(stack_active(self.heat_client, self.stack))
volume_types = heat_utils.get_stack_volume_types(
self.heat_client, self.cinder, self.stack)
self.assertEqual(u'nova.volume.encryptors.luks.LuksEncryptor',
encryption.provider)
self.assertEqual(volume_type.id, encryption.volume_type_id)
+
+
+class HeatUtilsKeypairTests(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.keypair_name = guid + '-kp'
+
+ env_values = {
+ 'keypair_name': self.keypair_name}
+
+ heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'keypair_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.nova = nova_utils.nova_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_keypair_with_stack(self):
+ """
+ Tests the creation of an OpenStack keypair with Heat.
+ """
+ self.stack = heat_utils.create_stack(
+ self.heat_client, self.stack_settings)
+ self.assertTrue(stack_active(self.heat_client, self.stack))
+
+ keypairs = heat_utils.get_stack_keypairs(
+ self.heat_client, self.nova, self.stack)
+
+ self.assertEqual(1, len(keypairs))
+ keypair = keypairs[0]
+
+ self.assertEqual(self.keypair_name, keypair.name)
+
+ outputs = heat_utils.get_outputs(self.heat_client, self.stack)
+
+ for output in outputs:
+ if output.key == 'private_key':
+ self.assertTrue(output.value.startswith(
+ '-----BEGIN RSA PRIVATE KEY-----'))
+
+ keypair = nova_utils.get_keypair_by_id(self.nova, keypair.id)
+ self.assertIsNotNone(keypair)
+
+ self.assertEqual(self.keypair_name, keypair.name)
+
+
+def stack_active(heat_cli, stack):
+ """
+ Blocks until stack application has successfully completed or failed
+ :param heat_cli: the Heat client
+ :param stack: the Stack domain object
+ :return: T/F
+ """
+ # 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(heat_cli, 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)
+
+ return is_active
SecurityGroupSettingsUnitTests)
from snaps.openstack.tests.create_stack_tests import (
StackSettingsUnitTests, CreateStackSuccessTests, CreateStackNegativeTests,
- CreateStackFloatingIpTests, CreateStackVolumeTests)
+ CreateStackFloatingIpTests, CreateStackKeypairTests,
+ 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, HeatUtilsVolumeTests)
+ HeatUtilsCreateComplexStackTests, HeatUtilsVolumeTests,
+ HeatUtilsKeypairTests)
from snaps.openstack.utils.tests.keystone_utils_tests import (
KeystoneSmokeTests, KeystoneUtilsTests)
from snaps.openstack.utils.tests.neutron_utils_tests import (
HeatUtilsVolumeTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level,
image_metadata=image_metadata))
+ suite.addTest(OSComponentTestCase.parameterize(
+ HeatUtilsKeypairTests, 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(
+ CreateStackKeypairTests, 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,
Code Review / snaps.git / commitdiff