| test_create_same_image | 1 | Ensures the OpenStackImage.create() method does not create|
| | | another image when one already exists with the same name |
+---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_same_image_new_settings | 1 | Tests the creation of an OpenStack image when the image |
+| | | already exists and the configuration only contains the |
+| | | the name. |
++---------------------------------------+---------------+-----------------------------------------------------------+
create_image_tests.py - CreateImageNegativeTests
------------------------------------------------
+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Glance API | Description |
+=======================================+===============+===========================================================+
-| test_none_image_name | 1 | Ensures OpenStackImage.create() results in an Exception |
+| test_bad_image_name | 1 | Ensures OpenStackImage.create() results in an Exception |
| | | being raised when the ImageSettings.name attribute has |
| | | not been set |
+---------------------------------------+---------------+-----------------------------------------------------------+
| test_bad_image_file | 1 | Ensures OpenStackImage.create() results in an Exception |
| | | being raised when the image file does not exist |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_none_proj_name | 1 | Ensures OpenStackImage.create() results in an Exception |
-| | | being raised when the credentials project name is None |
-+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_none_auth_url | 1 | Ensures OpenStackImage.create() results in an Exception |
-| | | being raised when the credentials URL is None |
-+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_none_password | 1 | Ensures OpenStackImage.create() results in an Exception |
-| | | being raised when the credentials password is None |
-+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_none_user | 1 | Ensures OpenStackImage.create() results in an Exception |
-| | | being raised when the credentials user is None |
-+---------------------------------------+---------------+-----------------------------------------------------------+
create_image_tests.py - CreateMultiPartImageTests
-------------------------------------------------
--- /dev/null
+---
+# Example of how to override the image settings for SNAPS testing for 3part images
+# This snippet can be placed verbatim into the snaps.images configuration for Functest
+# in config_functest.yaml as long as it is properly nested in that document
+glance_tests:
+ disk_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img
+ kernel_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel
+ ramdisk_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs
+cirros:
+ disk_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img
+ kernel_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel
+ ramdisk_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs
+centos:
+ disk_url: http://cloud.centos.org/centos/7/images/CentOS-7-x86_64-GenericCloud.qcow2
+ubuntu:
+ disk_url: http://uec-images.ubuntu.com/releases/trusty/14.04/ubuntu-14.04-server-cloudimg-amd64-disk1.img
--- /dev/null
+---
+# Example of how to override the image settings for SNAPS testing for file-based Images
+# generally for offline (no Internet available) OpenStack testing
+# This snippet can be placed verbatim into the snaps.images configuration for Functest
+# in config_functest.yaml as long as it is properly nested in that document
+glance_tests:
+ disk_file: ../images/cirros-0.3.4-x86_64-disk.img
+cirros:
+ disk_file: ../images/cirros-0.3.4-x86_64-disk.img
+centos:
+ disk_file: ../images/CentOS-7-x86_64-GenericCloud.qcow2
+ubuntu:
+ disk_file: ../images/ubuntu-14.04-server-cloudimg-amd64-disk1.img
--- /dev/null
+---
+# Example of how to override the default image URL for SNAPS testing
+# This snippet can be placed verbatim into the snaps.images configuration for Functest
+# in config_functest.yaml as long as it is properly nested in that document
+glance_tests:
+ disk_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img
+cirros:
+ disk_url: http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img
+centos:
+ disk_url: http://cloud.centos.org/centos/7/images/CentOS-7-x86_64-GenericCloud.qcow2
+ubuntu:
+ disk_url: http://uec-images.ubuntu.com/releases/trusty/14.04/ubuntu-14.04-server-cloudimg-amd64-disk1.img
--- /dev/null
+---
+# Example of how to override the image settings for SNAPS VM instance testing to leverage
+# images that already exist on your pod.
+# This snippet can be placed verbatim into the snaps.images configuration for Functest
+# in config_functest.yaml as long as it is properly nested in that document
+glance_tests:
+ disk_file: ../images/cirros-0.3.4-x86_64-disk.img
+cirros:
+ config:
+ name: static_image_test-cirros
+ exists: True
+ image_user: cirros
+centos:
+ config:
+ name: static_image_test-centos
+ exists: True
+ image_user: centos
+ubuntu:
+ config:
+ name: static_image_test-ubuntu
+ exists: True
+ image_user: ubuntu
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import argparse
+import logging
+import os
+import unittest
+
+from snaps import test_suite_builder
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.tests import openstack_tests
+from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
+from snaps.openstack.utils.tests.glance_utils_tests import GlanceUtilsTests
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('custom_image_test_runner')
+
+ARG_NOT_SET = "argument not set"
+LOG_LEVELS = {'FATAL': logging.FATAL, 'CRITICAL': logging.CRITICAL, 'ERROR': logging.ERROR, 'WARN': logging.WARN,
+ 'INFO': logging.INFO, 'DEBUG': logging.DEBUG}
+
+
+def __run_tests(source_filename, ext_net_name, proxy_settings, ssh_proxy_cmd, use_keystone, use_floating_ips,
+ log_level):
+ """
+ Compiles the tests that should run
+ :param source_filename: the OpenStack credentials file (required)
+ :param ext_net_name: the name of the external network to use for floating IPs (required)
+ :param proxy_settings: <host>:<port> of the proxy server (optional)
+ :param ssh_proxy_cmd: the command used to connect via SSH over some proxy server (optional)
+ :param use_keystone: when true, tests creating users and projects will be exercised and must be run on a host that
+ has access to the cloud's administrative network
+ :param use_floating_ips: when true, tests requiring floating IPs will be executed
+ :param log_level: the logging level
+ :return:
+ """
+ os_creds = openstack_tests.get_credentials(os_env_file=source_filename, proxy_settings_str=proxy_settings,
+ ssh_proxy_cmd=ssh_proxy_cmd)
+ # To ensure any files referenced via a relative path will begin from the diectory in which this file resides
+ os.chdir(os.path.dirname(os.path.realpath(__file__)))
+
+ image_creators = __create_images(os_creds)
+
+ meta_list = list()
+
+ # Create images from default
+ meta_list.append(None)
+
+ # Create images from specified URL
+ meta_list.append(
+ {'glance_tests': {'disk_url': openstack_tests.CIRROS_DEFAULT_IMAGE_URL},
+ 'cirros': {'disk_url': openstack_tests.CIRROS_DEFAULT_IMAGE_URL},
+ 'centos': {'disk_url': openstack_tests.CENTOS_DEFAULT_IMAGE_URL},
+ 'ubuntu': {'disk_url': openstack_tests.UBUNTU_DEFAULT_IMAGE_URL}})
+
+ # Create images from file
+ meta_list.append(
+ {'glance_tests': {'disk_file': '../images/cirros-0.3.4-x86_64-disk.img'},
+ 'cirros': {'disk_file': '../images/cirros-0.3.4-x86_64-disk.img'},
+ 'centos': {'disk_file': '../images/CentOS-7-x86_64-GenericCloud.qcow2'},
+ 'ubuntu': {'disk_file': '../images/ubuntu-14.04-server-cloudimg-amd64-disk1.img'}})
+
+ # Create images from Existing
+ meta_list.append(
+ {'glance_tests': {'disk_file': '../images/cirros-0.3.4-x86_64-disk.img'},
+ 'cirros': {'config': {'name': image_creators['cirros'].image_settings.name,
+ 'exists': True, 'image_user': 'cirros'}},
+ 'centos': {'config': {'name': image_creators['centos'].image_settings.name,
+ 'exists': True, 'image_user': 'centos'}},
+ 'ubuntu': {'config': {'name': image_creators['ubuntu'].image_settings.name,
+ 'exists': True, 'image_user': 'ubuntu'}}})
+
+ failure_count = 0
+ error_count = 0
+
+ try:
+ for metadata in meta_list:
+ logger.info('Starting tests with image metadata of - ' + str(metadata))
+ suite = unittest.TestSuite()
+
+ # Long running integration type tests
+ suite.addTest(OSComponentTestCase.parameterize(
+ GlanceUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name, image_metadata=metadata,
+ log_level=log_level))
+
+ test_suite_builder.add_openstack_integration_tests(
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone,
+ image_metadata=metadata, use_floating_ips=use_floating_ips, log_level=log_level)
+
+ result = unittest.TextTestRunner(verbosity=2).run(suite)
+ if result.errors:
+ logger.error('Number of errors in test suite - ' + str(len(result.errors)))
+ for test, message in result.errors:
+ logger.error(str(test) + " ERROR with " + message)
+ error_count += 1
+
+ if result.failures:
+ logger.error('Number of failures in test suite - ' + str(len(result.failures)))
+ for test, message in result.failures:
+ logger.error(str(test) + " FAILED with " + message)
+ failure_count += 1
+
+ if (result.errors and len(result.errors) > 0) or (result.failures and len(result.failures) > 0):
+ logger.error('See above for test failures')
+ else:
+ logger.info('All tests completed successfully in run')
+
+ logger.info('Total number of errors = ' + str(error_count))
+ logger.info('Total number of failures = ' + str(failure_count))
+
+ if error_count + failure_count > 0:
+ exit(1)
+ except Exception as e:
+ logger.warn('Unexpected error running tests - %s', e)
+ pass
+ finally:
+ for image_creator in image_creators.values():
+ try:
+ image_creator.clean()
+ except Exception as e:
+ logger.error('Exception thrown while cleaning image - %s', e)
+
+
+def __create_images(os_creds):
+ """
+ Returns a dictionary of
+ :param os_creds:
+ :return:
+ """
+ image_meta = {'cirros': {'disk_url': openstack_tests.CIRROS_DEFAULT_IMAGE_URL,
+ 'kernel_url': openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL,
+ 'ramdisk_url': openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL},
+ 'centos': {'disk_file': '../images/CentOS-7-x86_64-GenericCloud.qcow2'},
+ 'ubuntu': {'disk_file': '../images/ubuntu-14.04-server-cloudimg-amd64-disk1.img'}}
+ cirros_image_settings = openstack_tests.cirros_image_settings(name='static_image_test-cirros',
+ image_metadata=image_meta, public=True)
+ centos_image_settings = openstack_tests.centos_image_settings(name='static_image_test-centos',
+ image_metadata=image_meta, public=True)
+ ubuntu_image_settings = openstack_tests.ubuntu_image_settings(name='static_image_test-ubuntu',
+ image_metadata=image_meta, public=True)
+
+ out = dict()
+ out['cirros'] = OpenStackImage(os_creds, cirros_image_settings)
+ out['cirros'].create()
+ out['centos'] = OpenStackImage(os_creds, centos_image_settings)
+ out['centos'].create()
+ out['ubuntu'] = OpenStackImage(os_creds, ubuntu_image_settings)
+ out['ubuntu'].create()
+
+ return out
+
+
+def main(arguments):
+ """
+ Begins running tests with different image_metadata configuration.
+ argv[1] if used must be the source filename else os_env.yaml will be leveraged instead
+ argv[2] if used must be the proxy server <host>:<port>
+ """
+ log_level = LOG_LEVELS.get(arguments.log_level, logging.DEBUG)
+ logging.basicConfig(level=log_level)
+ logger.info('Starting test suite')
+
+ __run_tests(arguments.env, arguments.ext_net, arguments.proxy, arguments.ssh_proxy_cmd,
+ arguments.use_keystone != ARG_NOT_SET, arguments.floating_ips != ARG_NOT_SET, log_level)
+
+ exit(0)
+
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument('-e', '--env', dest='env', required=True, help='OpenStack credentials file')
+ parser.add_argument('-n', '--net', dest='ext_net', required=True, help='External network name')
+ parser.add_argument('-p', '--proxy', dest='proxy', nargs='?', default=None,
+ help='Optonal HTTP proxy socket (<host>:<port>)')
+ parser.add_argument('-s', '--ssh-proxy-cmd', dest='ssh_proxy_cmd', nargs='?', default=None,
+ help='Optonal SSH proxy command value')
+ parser.add_argument('-l', '--log-level', dest='log_level', default='INFO',
+ help='Logging Level (FATAL|CRITICAL|ERROR|WARN|INFO|DEBUG)')
+ parser.add_argument('-f', '--floating-ips', dest='floating_ips', default=ARG_NOT_SET, nargs='?',
+ help='When argument is set, all integration tests requiring Floating IPs will be executed')
+ parser.add_argument('-k', '--use-keystone', dest='use_keystone', default=ARG_NOT_SET, nargs='?',
+ help='When argument is set, the tests will exercise the keystone APIs and must be run on a ' +
+ 'machine that has access to the admin network' +
+ ' and is able to create users and groups')
+
+ args = parser.parse_args()
+
+ main(args)
return False
-def get_file(file_path):
- """
- Returns True if the image file has already been downloaded
- :return: the image file object
- :raise Exception when file cannot be found
- """
- if file_exists(file_path):
- return open(file_path, 'rb')
- else:
- raise Exception('File with path cannot be found - ' + file_path)
-
-
def download(url, dest_path, name=None):
"""
Download a file to a destination path given a URL
+ :param url: the endpoint to the file to download
+ :param dest_path: the directory to save the file
+ :param name: the file name (optional)
:rtype : File object
"""
if not name:
if self.__image:
logger.info('Found image with name - ' + self.image_settings.name)
return self.__image
+ elif self.image_settings.exists and not self.image_settings.url and not self.image_settings.image_file:
+ raise ImageCreationError('Image with does not exist with name - ' + self.image_settings.name)
elif not cleanup:
extra_properties = self.image_settings.extra_properties or dict()
self.__glance, self.image_settings.kernel_image_settings.name)
if not self.__kernel_image and not cleanup:
- logger.info('Creating associated kernel image')
+ logger.info('Creating associated kernel image with name - '
+ + self.image_settings.kernel_image_settings.name)
self.__kernel_image = glance_utils.create_image(
self.__glance, self.image_settings.kernel_image_settings)
extra_properties['kernel_id'] = self.__kernel_image.id
self.__glance, self.image_settings.ramdisk_image_settings.name)
if not self.__ramdisk_image and not cleanup:
- logger.info('Creating associated ramdisk image')
+ logger.info('Creating associated ramdisk image with name - '
+ + self.image_settings.ramdisk_image_settings.name)
self.__ramdisk_image = glance_utils.create_image(
self.__glance, self.image_settings.ramdisk_image_settings)
extra_properties['ramdisk_id'] = self.__ramdisk_image.id
self.image_settings.extra_properties = extra_properties
self.__image = glance_utils.create_image(self.__glance, self.image_settings)
- logger.info('Creating image')
+
+ logger.info('Created image with name - ' + self.image_settings.name)
if self.__image and self.image_active(block=True):
logger.info('Image is now active with name - ' + self.image_settings.name)
return self.__image
else:
- raise Exception('Image was not created or activated in the alloted amount of time')
+ raise ImageCreationError('Image was not created or activated in the alloted amount of time')
else:
logger.info('Did not create image due to cleanup mode')
while timeout > time.time() - start:
status = self._status(expected_status_code)
if status:
- logger.info('Image is active with name - ' + self.image_settings.name)
+ logger.debug('Image is active with name - ' + self.image_settings.name)
return True
logger.debug('Retry querying image status in ' + str(poll_interval) + ' seconds')
return False
if status == 'ERROR':
- raise Exception('Instance had an error during deployment')
+ raise ImageCreationError('Instance had an error during deployment')
logger.debug('Instance status is - ' + status)
return status == expected_status_code
class ImageSettings:
def __init__(self, config=None, name=None, image_user=None, img_format=None, url=None, image_file=None,
extra_properties=None, nic_config_pb_loc=None, kernel_image_settings=None,
- ramdisk_image_settings=None):
+ ramdisk_image_settings=None, exists=False, public=False):
"""
:param config: dict() object containing the configuration settings using the attribute names below as each
:param nic_config_pb_loc: the file location to the Ansible Playbook that can configure multiple NICs
:param kernel_image_settings: the settings for a kernel image
:param ramdisk_image_settings: the settings for a kernel image
+ :param exists: When True, an image with the given name must exist
+ :param public: When True, an image will be created with public visibility
"""
if config:
else:
self.ramdisk_image_settings = None
+ if 'exists' in config and config['exists'] is True:
+ self.exists = True
+ else:
+ self.exists = False
+
+ if 'public' in config and config['public'] is True:
+ self.public = True
+ else:
+ self.public = False
else:
self.name = name
self.image_user = image_user
self.nic_config_pb_loc = nic_config_pb_loc
self.kernel_image_settings = kernel_image_settings
self.ramdisk_image_settings = ramdisk_image_settings
+ self.exists = exists
+ self.public = public
- if not self.name or not self.image_user or not self.format:
- raise Exception("The attributes name, image_user, format, and url are required for ImageSettings")
+ if not self.name:
+ raise ImageSettingsError("The attribute name is required")
- if not self.url and not self.image_file:
- raise Exception('URL or image file must be set')
+ if not (self.url or self.image_file) and not self.exists:
+ raise ImageSettingsError('URL or image file must be set or image must already exist')
if self.url and self.image_file:
- raise Exception('Please set either URL or image file, not both')
+ raise ImageSettingsError('Please set either URL or image file, not both')
+
+ if not self.image_user:
+ raise ImageSettingsError('Image user is required')
+
+ if not self.format and not self.exists:
+ raise ImageSettingsError('Format is required when the image should not already exist')
+
+
+class ImageSettingsError(Exception):
+ """
+ Exception to be thrown when an image settings are incorrect
+ """
+ def __init__(self, message):
+ Exception.__init__(self, message)
+
+
+class ImageCreationError(Exception):
+ """
+ Exception to be thrown when an image cannot be created
+ """
+ def __init__(self, message):
+ Exception.__init__(self, message)
variables, self.__os_creds.proxy_settings)
else:
logger.warning('VM ' + self.instance_settings.name + ' cannot self configure NICs eth1++. ' +
- 'No playbook or keypairs found.')
+ 'No playbook or keypairs found.')
def get_image_user(self):
"""
:param expected_status_code: instance status evaluated with this string value
:return: T/F
"""
+ if not self.__vm:
+ return False
+
instance = self.__nova.servers.get(self.__vm.id)
if not instance:
logger.warning('Cannot find instance with id - ' + self.__vm.id)
# 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.
+
+try:
+ from urllib.request import URLError
+except ImportError:
+ from urllib2 import URLError
+
+import logging
import os
import shutil
import uuid
import unittest
from snaps import file_utils
-from snaps.openstack.create_image import ImageSettings
+from snaps.openstack.create_image import ImageSettings, ImageCreationError, ImageSettingsError
from snaps.openstack.tests import openstack_tests
from snaps.openstack.utils import glance_utils
from snaps.openstack import create_image
-from snaps.openstack import os_credentials
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
__author__ = 'spisarski'
+logger = logging.getLogger('create_image_tests')
+
class ImageSettingsUnitTests(unittest.TestCase):
"""
Tests the construction of the ImageSettings class
"""
-
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(config=dict())
def test_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(name='foo')
def test_config_with_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(config={'name': 'foo'})
def test_name_user_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(name='foo', image_user='bar')
def test_config_with_name_user_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(config={'name': 'foo', 'image_user': 'bar'})
def test_name_user_format_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(name='foo', image_user='bar', img_format='qcow2')
def test_config_with_name_user_format_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2'})
def test_name_user_format_url_file_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com',
image_file='/foo/bar.qcow')
def test_config_with_name_user_format_url_file_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ImageSettingsError):
ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
'download_url': 'http://foo.com', 'image_file': '/foo/bar.qcow'})
self.assertEqual('qcow2', settings.format)
self.assertEqual('http://foo.com', settings.url)
self.assertIsNone(settings.image_file)
+ self.assertFalse(settings.exists)
+ self.assertFalse(settings.public)
self.assertIsNone(settings.nic_config_pb_loc)
def test_name_user_format_url_only_properties(self):
self.assertEqual('http://foo.com', settings.url)
self.assertEqual(properties, settings.extra_properties)
self.assertIsNone(settings.image_file)
+ self.assertFalse(settings.exists)
+ self.assertFalse(settings.public)
self.assertIsNone(settings.nic_config_pb_loc)
def test_config_with_name_user_format_url_only(self):
self.assertEqual('qcow2', settings.format)
self.assertEqual('http://foo.com', settings.url)
self.assertIsNone(settings.image_file)
+ self.assertFalse(settings.exists)
+ self.assertFalse(settings.public)
self.assertIsNone(settings.nic_config_pb_loc)
def test_name_user_format_file_only(self):
self.assertEqual('qcow2', settings.format)
self.assertIsNone(settings.url)
self.assertEqual('/foo/bar.qcow', settings.image_file)
+ self.assertFalse(settings.exists)
+ self.assertFalse(settings.public)
self.assertIsNone(settings.nic_config_pb_loc)
def test_config_with_name_user_format_file_only(self):
self.assertEqual('qcow2', settings.format)
self.assertIsNone(settings.url)
self.assertEqual('/foo/bar.qcow', settings.image_file)
+ self.assertFalse(settings.exists)
+ self.assertFalse(settings.public)
self.assertIsNone(settings.nic_config_pb_loc)
def test_all_url(self):
ramdisk_settings = ImageSettings(name='ramdisk', url='http://ramdisk.com', image_user='bar', img_format='qcow2')
settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com',
extra_properties=properties, nic_config_pb_loc='/foo/bar',
- kernel_image_settings=kernel_settings, ramdisk_image_settings=ramdisk_settings)
+ kernel_image_settings=kernel_settings, ramdisk_image_settings=ramdisk_settings,
+ exists=True, public=True)
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.image_user)
self.assertEqual('qcow2', settings.format)
self.assertEqual('http://ramdisk.com', settings.ramdisk_image_settings.url)
self.assertEqual('bar', settings.ramdisk_image_settings.image_user)
self.assertEqual('qcow2', settings.ramdisk_image_settings.format)
+ self.assertTrue(settings.exists)
+ self.assertTrue(settings.public)
def test_config_all_url(self):
settings = ImageSettings(
'kernel_image_settings': {'name': 'kernel', 'download_url': 'http://kernel.com',
'image_user': 'bar', 'format': 'qcow2'},
'ramdisk_image_settings': {'name': 'ramdisk', 'download_url': 'http://ramdisk.com',
- 'image_user': 'bar', 'format': 'qcow2'}})
+ 'image_user': 'bar', 'format': 'qcow2'},
+ 'exists': True, 'public': True})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.image_user)
self.assertEqual('qcow2', settings.format)
self.assertEqual('http://kernel.com', settings.kernel_image_settings.url)
self.assertEqual('ramdisk', settings.ramdisk_image_settings.name)
self.assertEqual('http://ramdisk.com', settings.ramdisk_image_settings.url)
+ self.assertTrue(settings.exists)
+ self.assertTrue(settings.public)
def test_all_file(self):
properties = {'hw_video_model': 'vga'}
settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', image_file='/foo/bar.qcow',
- extra_properties=properties, nic_config_pb_loc='/foo/bar')
+ extra_properties=properties, nic_config_pb_loc='/foo/bar', exists=True, public=True)
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.image_user)
self.assertEqual('qcow2', settings.format)
self.assertEqual('/foo/bar.qcow', settings.image_file)
self.assertEqual(properties, settings.extra_properties)
self.assertEqual('/foo/bar', settings.nic_config_pb_loc)
+ self.assertTrue(settings.exists)
+ self.assertTrue(settings.public)
def test_config_all_file(self):
settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
'image_file': '/foo/bar.qcow',
'extra_properties': '{\'hw_video_model\' : \'vga\'}',
- 'nic_config_pb_loc': '/foo/bar'})
+ 'nic_config_pb_loc': '/foo/bar', 'exists': True, 'public': True})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.image_user)
self.assertEqual('qcow2', settings.format)
self.assertEqual('/foo/bar.qcow', settings.image_file)
self.assertEqual('{\'hw_video_model\' : \'vga\'}', settings.extra_properties)
self.assertEqual('/foo/bar', settings.nic_config_pb_loc)
+ self.assertTrue(settings.exists)
+ self.assertTrue(settings.public)
class CreateImageSuccessTests(OSIntegrationTestCase):
self.glance = glance_utils.glance_client(self.os_creds)
self.image_creator = None
+ if self.image_metadata and 'glance_tests' in self.image_metadata:
+ glance_test_meta = self.image_metadata['glance_tests']
+ else:
+ glance_test_meta = None
+
self.tmp_dir = 'tmp/' + str(guid)
if not os.path.exists(self.tmp_dir):
os.makedirs(self.tmp_dir)
self.image_settings = openstack_tests.cirros_image_settings(name=self.image_name,
- image_metadata=self.image_metadata)
+ image_metadata=glance_test_meta)
def tearDown(self):
"""
"""
Tests the creation of an OpenStack image from a file.
"""
+ if not self.image_settings.image_file and self.image_settings.url:
+ # Download the file of the image
+ image_file_name = file_utils.download(self.image_settings.url, self.tmp_dir).name
+ else:
+ image_file_name = self.image_settings.image_file
- # Create Image
-
- # Download the file of the image
- image_file = file_utils.download(self.image_settings.url, self.tmp_dir)
- file_image_settings = openstack_tests.file_image_test_settings(
- name=self.image_name, file_path=image_file.name)
+ if image_file_name:
+ file_image_settings = openstack_tests.file_image_test_settings(
+ name=self.image_name, file_path=image_file_name)
- self.image_creator = create_image.OpenStackImage(self.os_creds, file_image_settings)
- created_image = self.image_creator.create()
- self.assertIsNotNone(created_image)
- self.assertEqual(self.image_name, created_image.name)
+ self.image_creator = create_image.OpenStackImage(self.os_creds, file_image_settings)
+ created_image = self.image_creator.create()
+ self.assertIsNotNone(created_image)
+ self.assertEqual(self.image_name, created_image.name)
- retrieved_image = glance_utils.get_image(self.glance, file_image_settings.name)
- self.assertIsNotNone(retrieved_image)
- self.assertEqual(self.image_creator.get_image().size, retrieved_image.size)
- self.assertEqual(get_image_size(file_image_settings), retrieved_image.size)
+ retrieved_image = glance_utils.get_image(self.glance, file_image_settings.name)
+ self.assertIsNotNone(retrieved_image)
+ self.assertEqual(self.image_creator.get_image().size, retrieved_image.size)
+ self.assertEqual(get_image_size(file_image_settings), retrieved_image.size)
- self.assertEqual(created_image.name, retrieved_image.name)
- self.assertEqual(created_image.id, retrieved_image.id)
+ self.assertEqual(created_image.name, retrieved_image.name)
+ self.assertEqual(created_image.id, retrieved_image.id)
+ else:
+ logger.warn('Test not executed as the image metadata requires image files')
def test_create_delete_image(self):
"""
image2 = os_image_2.create()
self.assertEqual(image1.id, image2.id)
+ def test_create_same_image_new_settings(self):
+ """
+ Tests the creation of an OpenStack image when the image already exists and the configuration only contains
+ the name.
+ """
+ # Create Image
+ self.image_creator = create_image.OpenStackImage(self.os_creds, self.image_settings)
+ image1 = self.image_creator.create()
+
+ retrieved_image = glance_utils.get_image(self.glance, self.image_settings.name)
+ self.assertIsNotNone(retrieved_image)
+ self.assertEqual(self.image_creator.get_image().size, retrieved_image.size)
+ self.assertEqual(get_image_size(self.image_settings), retrieved_image.size)
+ self.assertEqual(image1.name, retrieved_image.name)
+ self.assertEqual(image1.id, retrieved_image.id)
+ self.assertEqual(image1.properties, retrieved_image.properties)
+
+ # Should be retrieving the instance data
+ image_2_settings = ImageSettings(name=self.image_settings.name, image_user='foo', exists=True)
+ os_image_2 = create_image.OpenStackImage(self.os_creds, image_2_settings)
+ image2 = os_image_2.create()
+ self.assertEqual(image1.id, image2.id)
+
class CreateImageNegativeTests(OSIntegrationTestCase):
"""
super(self.__class__, self).__clean__()
- def test_none_image_name(self):
+ def test_bad_image_name(self):
"""
- Expect an exception when the image name is None
+ Expect an ImageCreationError when the image name does not exist when a file or URL has not been configured
"""
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
- with self.assertRaises(Exception):
- self.image_creator = create_image.OpenStackImage(
- self.os_creds, create_image.ImageSettings(
- name=None, image_user=os_image_settings.image_user, img_format=os_image_settings.format,
- url=os_image_settings.url))
+ os_image_settings = ImageSettings(name='foo', image_user='bar', exists=True)
+ self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings)
- self.fail('Exception should have been thrown prior to this line')
+ with self.assertRaises(ImageCreationError):
+ self.image_creator.create()
+
+ self.fail('ImageCreationError should have been raised prior to this line')
def test_bad_image_url(self):
"""
- Expect an exception when the image download url is bad
+ Expect an ImageCreationError when the image download url is bad
"""
os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
self.image_creator = create_image.OpenStackImage(self.os_creds, create_image.ImageSettings(
name=os_image_settings.name, image_user=os_image_settings.image_user,
img_format=os_image_settings.format, url="http://foo.bar"))
- with self.assertRaises(Exception):
+
+ with self.assertRaises(URLError):
self.image_creator.create()
def test_bad_image_file(self):
"""
- Expect an exception when the image file does not exist
+ Expect an ImageCreationError when the image file does not exist
"""
os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
self.image_creator = create_image.OpenStackImage(
self.os_creds,
create_image.ImageSettings(name=os_image_settings.name, image_user=os_image_settings.image_user,
img_format=os_image_settings.format, image_file="/foo/bar.qcow"))
- with self.assertRaises(Exception):
+ with self.assertRaises(IOError):
self.image_creator.create()
- def test_none_proj_name(self):
- """
- Expect an exception when the project name is None
- """
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
- with self.assertRaises(Exception):
- self.image_creator = create_image.OpenStackImage(
- os_credentials.OSCreds(self.os_creds.username, self.os_creds.password, self.os_creds.auth_url, None,
- proxy_settings=self.os_creds.proxy_settings),
- os_image_settings)
- self.image_creator.create()
-
- def test_none_auth_url(self):
- """
- Expect an exception when the project name is None
- """
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
- with self.assertRaises(Exception):
- self.image_creator = create_image.OpenStackImage(
- os_credentials.OSCreds(self.os_creds.username, self.os_creds.password, None,
- self.os_creds.project_name, proxy_settings=self.os_creds.proxy_settings),
- os_image_settings)
- self.image_creator.create()
-
- def test_none_password(self):
- """
- Expect an exception when the project name is None
- """
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
- with self.assertRaises(Exception):
- self.image_creator = create_image.OpenStackImage(
- os_credentials.OSCreds(self.os_creds.username, None, self.os_creds.os_auth_url,
- self.os_creds.project_name, proxy_settings=self.os_creds.proxy_settings),
- os_image_settings)
-
- def test_none_user(self):
- """
- Expect an exception when the project name is None
- """
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
- with self.assertRaises(Exception):
- self.image_creator = create_image.OpenStackImage(
- os_credentials.OSCreds(None, self.os_creds.password, self.os_creds.os_auth_url,
- self.os_creds.project_name,
- proxy_settings=self.os_creds.proxy_settings),
- os_image_settings)
-
class CreateMultiPartImageTests(OSIntegrationTestCase):
"""
if not os.path.exists(self.tmp_dir):
os.makedirs(self.tmp_dir)
+ if self.image_metadata and 'glance_tests' in self.image_metadata:
+ self.glance_test_meta = self.image_metadata['glance_tests']
+ else:
+ self.glance_test_meta = dict()
+
def tearDown(self):
"""
Cleans the images and downloaded image file
Tests the creation of a 3-part OpenStack image from a URL.
"""
# Create the kernel image
- image_settings = openstack_tests.cirros_image_settings(
- name=self.image_name,
- image_metadata={'disk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img',
- 'kernel_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel',
- 'ramdisk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'})
-
- image_creator = create_image.OpenStackImage(self.os_creds, image_settings)
- self.image_creators.append(image_creator)
- image_creator.create()
-
- main_image = glance_utils.get_image(self.glance, image_settings.name)
- self.assertIsNotNone(main_image)
- self.assertIsNotNone(image_creator.get_image())
- self.assertEqual(image_creator.get_image().id, main_image.id)
-
- kernel_image = glance_utils.get_image(self.glance, image_settings.kernel_image_settings.name)
- self.assertIsNotNone(kernel_image)
- self.assertIsNotNone(image_creator.get_kernel_image())
- self.assertEqual(kernel_image.id, image_creator.get_kernel_image().id)
-
- ramdisk_image = glance_utils.get_image(self.glance, image_settings.ramdisk_image_settings.name)
- self.assertIsNotNone(ramdisk_image)
- self.assertIsNotNone(image_creator.get_ramdisk_image())
- self.assertEqual(ramdisk_image.id, image_creator.get_ramdisk_image().id)
+ if 'disk_file' not in self.glance_test_meta:
+ image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name,
+ image_metadata={'disk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img',
+ 'kernel_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel',
+ 'ramdisk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'})
+
+ image_creator = create_image.OpenStackImage(self.os_creds, image_settings)
+ self.image_creators.append(image_creator)
+ image_creator.create()
+
+ main_image = glance_utils.get_image(self.glance, image_settings.name)
+ self.assertIsNotNone(main_image)
+ self.assertIsNotNone(image_creator.get_image())
+ self.assertEqual(image_creator.get_image().id, main_image.id)
+
+ kernel_image = glance_utils.get_image(self.glance, image_settings.kernel_image_settings.name)
+ self.assertIsNotNone(kernel_image)
+ self.assertIsNotNone(image_creator.get_kernel_image())
+ self.assertEqual(kernel_image.id, image_creator.get_kernel_image().id)
+
+ ramdisk_image = glance_utils.get_image(self.glance, image_settings.ramdisk_image_settings.name)
+ self.assertIsNotNone(ramdisk_image)
+ self.assertIsNotNone(image_creator.get_ramdisk_image())
+ self.assertEqual(ramdisk_image.id, image_creator.get_ramdisk_image().id)
+ else:
+ logger.warn('Test not executed as the image metadata requires image files')
def test_create_three_part_image_from_file_3_creators(self):
"""
"""
# Set properties
properties = {}
- if self.image_metadata:
- if 'extra_properties' in self.image_metadata:
- properties = self.image_metadata['extra_properties']
+ if self.glance_test_meta:
+ if 'extra_properties' in self.glance_test_meta:
+ properties = self.glance_test_meta['extra_properties']
+
# Create the kernel image
- kernel_url = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel'
- if self.image_metadata:
- if 'kernel_url' in self.image_metadata:
- kernel_url = self.image_metadata['kernel_url']
- kernel_image_file = file_utils.download(kernel_url, self.tmp_dir)
+ kernel_file_name = None
+ kernel_url = openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL
+ if 'kernel_file' in self.glance_test_meta:
+ kernel_file_name = self.glance_test_meta['kernel_file']
+ elif 'kernel_url' in self.glance_test_meta:
+ kernel_url = self.glance_test_meta['kernel_url']
+ else:
+ kernel_url = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel'
+
+ if not kernel_file_name:
+ kernel_file_name = file_utils.download(kernel_url, self.tmp_dir).name
+
kernel_file_image_settings = openstack_tests.file_image_test_settings(
- name=self.image_name+'_kernel', file_path=kernel_image_file.name)
+ name=self.image_name+'_kernel', file_path=kernel_file_name)
+
self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_file_image_settings))
kernel_image = self.image_creators[-1].create()
self.assertIsNotNone(kernel_image)
self.assertEqual(get_image_size(kernel_file_image_settings), kernel_image.size)
# Create the ramdisk image
- ramdisk_url = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'
- if self.image_metadata:
- if 'ramdisk_url' in self.image_metadata:
- ramdisk_url = self.image_metadata['ramdisk_url']
- ramdisk_image_file = file_utils.download(ramdisk_url, self.tmp_dir)
+ ramdisk_file_name = None
+ ramdisk_url = openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL
+ if 'ramdisk_file' in self.glance_test_meta:
+ ramdisk_file_name = self.glance_test_meta['ramdisk_file']
+ elif 'ramdisk_url' in self.glance_test_meta:
+ ramdisk_url = self.glance_test_meta['ramdisk_url']
+
+ if not ramdisk_file_name:
+ ramdisk_file_name = file_utils.download(ramdisk_url, self.tmp_dir).name
+
ramdisk_file_image_settings = openstack_tests.file_image_test_settings(
- name=self.image_name+'_ramdisk', file_path=ramdisk_image_file.name)
+ name=self.image_name+'_ramdisk', file_path=ramdisk_file_name)
self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_file_image_settings))
ramdisk_image = self.image_creators[-1].create()
self.assertIsNotNone(ramdisk_image)
self.assertEqual(get_image_size(ramdisk_file_image_settings), ramdisk_image.size)
- # Create the main image
- image_url = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img'
- image_file = file_utils.download(image_url, self.tmp_dir)
- file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=image_file.name)
+ # Create the main disk image
+ disk_file_name = None
+ disk_url = openstack_tests.CIRROS_DEFAULT_IMAGE_URL
+ if 'disk_file' in self.glance_test_meta:
+ disk_file_name = self.glance_test_meta['disk_file']
+ elif 'disk_url' in self.glance_test_meta:
+ disk_url = self.glance_test_meta['disk_url']
+
+ if not disk_file_name:
+ disk_file_name = file_utils.download(disk_url, self.tmp_dir).name
+
+ file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=disk_file_name)
properties['kernel_id'] = kernel_image.id
properties['ramdisk_id'] = ramdisk_image.id
file_image_settings.extra_properties = properties
"""
Tests the creation of a 3-part OpenStack image from a URL.
"""
- # Set properties
- properties = {}
- if self.image_metadata and 'extra_properties' in self.image_metadata:
- properties = self.image_metadata['extra_properties']
-
- # Create the kernel image
- kernel_image_settings = openstack_tests.cirros_image_settings(
- name=self.image_name+'_kernel',
- url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel')
-
- if self.image_metadata:
- if 'kernel_url' in self.image_metadata:
- kernel_image_settings.url = self.image_metadata['kernel_url']
- self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.assertIsNotNone(kernel_image)
- self.assertEqual(get_image_size(kernel_image_settings), kernel_image.size)
-
- # Create the ramdisk image
- ramdisk_image_settings = openstack_tests.cirros_image_settings(
- name=self.image_name+'_ramdisk',
- url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs')
- if self.image_metadata:
- if 'ramdisk_url' in self.image_metadata:
- ramdisk_image_settings.url = self.image_metadata['ramdisk_url']
- self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.assertIsNotNone(ramdisk_image)
- self.assertEqual(get_image_size(ramdisk_image_settings), ramdisk_image.size)
-
- # Create the main image
- os_image_settings = openstack_tests.cirros_image_settings(
- name=self.image_name,
- url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img')
- if self.image_metadata:
- if 'disk_url' in self.image_metadata:
- os_image_settings.url = self.image_metadata['disk_url']
-
- properties['kernel_id'] = kernel_image.id
- properties['ramdisk_id'] = ramdisk_image.id
- os_image_settings.extra_properties = properties
-
- self.image_creators.append(create_image.OpenStackImage(self.os_creds, os_image_settings))
- created_image = self.image_creators[-1].create()
- self.assertIsNotNone(created_image)
- self.assertEqual(self.image_name, created_image.name)
-
- retrieved_image = glance_utils.get_image(self.glance, os_image_settings.name)
- self.assertIsNotNone(retrieved_image)
-
- self.assertEqual(self.image_creators[-1].get_image().size, retrieved_image.size)
- self.assertEqual(get_image_size(os_image_settings), retrieved_image.size)
-
- self.assertEqual(created_image.name, retrieved_image.name)
- self.assertEqual(created_image.id, retrieved_image.id)
- self.assertEqual(created_image.properties, retrieved_image.properties)
+ if 'disk_file' not in self.glance_test_meta:
+ # Set properties
+ properties = {}
+ if self.glance_test_meta and 'extra_properties' in self.glance_test_meta:
+ properties = self.glance_test_meta['extra_properties']
+
+ # Create the kernel image
+ kernel_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name+'_kernel',
+ url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel')
+
+ if self.glance_test_meta:
+ if 'kernel_url' in self.glance_test_meta:
+ kernel_image_settings.url = self.glance_test_meta['kernel_url']
+ self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_image_settings))
+ kernel_image = self.image_creators[-1].create()
+ self.assertIsNotNone(kernel_image)
+ self.assertEqual(get_image_size(kernel_image_settings), kernel_image.size)
+
+ # Create the ramdisk image
+ ramdisk_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name+'_ramdisk',
+ url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs')
+ if self.glance_test_meta:
+ if 'ramdisk_url' in self.glance_test_meta:
+ ramdisk_image_settings.url = self.glance_test_meta['ramdisk_url']
+ self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_image_settings))
+ ramdisk_image = self.image_creators[-1].create()
+ self.assertIsNotNone(ramdisk_image)
+ self.assertEqual(get_image_size(ramdisk_image_settings), ramdisk_image.size)
+
+ # Create the main image
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name,
+ url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img')
+ if self.glance_test_meta:
+ if 'disk_url' in self.glance_test_meta:
+ os_image_settings.url = self.glance_test_meta['disk_url']
+
+ properties['kernel_id'] = kernel_image.id
+ properties['ramdisk_id'] = ramdisk_image.id
+ os_image_settings.extra_properties = properties
+
+ self.image_creators.append(create_image.OpenStackImage(self.os_creds, os_image_settings))
+ created_image = self.image_creators[-1].create()
+ self.assertIsNotNone(created_image)
+ self.assertEqual(self.image_name, created_image.name)
+
+ retrieved_image = glance_utils.get_image(self.glance, os_image_settings.name)
+ self.assertIsNotNone(retrieved_image)
+
+ self.assertEqual(self.image_creators[-1].get_image().size, retrieved_image.size)
+ self.assertEqual(get_image_size(os_image_settings), retrieved_image.size)
+
+ self.assertEqual(created_image.name, retrieved_image.name)
+ self.assertEqual(created_image.id, retrieved_image.id)
+ self.assertEqual(created_image.properties, retrieved_image.properties)
+ else:
+ logger.warn('Test not executed as the image metadata requires image files')
def get_image_size(image_settings):
import unittest
import uuid
+import shutil
+from snaps import file_utils
from snaps.openstack.create_instance import VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings
from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
from snaps.openstack.create_keypairs import OpenStackKeypair, KeypairSettings
from snaps.openstack.create_network import OpenStackNetwork, PortSettings
from snaps.openstack.create_router import OpenStackRouter
-from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_image import OpenStackImage, ImageSettings
from snaps.openstack.create_security_group import SecurityGroupSettings, OpenStackSecurityGroup
from snaps.openstack.tests import openstack_tests, validation_utils
from snaps.openstack.utils import nova_utils
-from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
+from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase, OSComponentTestCase
__author__ = 'spisarski'
super(self.__class__, self).__start__()
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.keypair_priv_filepath = 'tmp/' + guid
- self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
- self.keypair_name = guid + '-kp'
self.vm_inst_name = guid + '-inst'
self.port_1_name = guid + 'port-1'
- self.port_2_name = guid + 'port-2'
- self.floating_ip_name = guid + 'fip1'
# Initialize for tearDown()
self.image_creator = None
# Create Image
# Set the default image settings, then set any custom parameters sent from the app
- self.os_image_settings = openstack_tests.cirros_image_settings(
+ os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
try:
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
# Create Network
except Exception as e:
logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
- if os.path.isfile(self.keypair_pub_filepath):
- os.remove(self.keypair_pub_filepath)
-
- if os.path.isfile(self.keypair_priv_filepath):
- os.remove(self.keypair_priv_filepath)
-
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.nova = nova_utils.nova_client(self.os_creds)
- self.os_image_settings = openstack_tests.cirros_image_settings(
+ os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
net_config = openstack_tests.get_priv_net_config(
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
# Create Flavor
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
self.pub_net_config = openstack_tests.get_pub_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name)
- self.os_image_settings = openstack_tests.cirros_image_settings(
+ os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
# Create Network
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
self.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.os_image_settings = openstack_tests.cirros_image_settings(
+ os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
# Create Network
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
self.priv_net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-priv-net', subnet_name=guid + '-priv-subnet')
- self.os_image_settings = openstack_tests.cirros_image_settings(
+ os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
try:
self.flavor_creator.create()
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
except Exception as e:
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
router_name=self.guid + '-pub-router', external_net=self.ext_net_name)
image_name = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.os_image_settings = openstack_tests.centos_image_settings(name=image_name,
- image_metadata=self.image_metadata)
+ os_image_settings = openstack_tests.centos_image_settings(name=image_name, image_metadata=self.image_metadata)
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
# First network is public
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = self.guid + '-inst'
self.nova = nova_utils.nova_client(self.os_creds)
- self.os_image_settings = openstack_tests.cirros_image_settings(
+ os_image_settings = openstack_tests.cirros_image_settings(
name=self.guid + '-image', image_metadata=self.image_metadata)
self.vm_inst_name = self.guid + '-inst'
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
# Create Network
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
self.inst_creator = None
try:
+ if self.image_metadata and 'disk_file' in self.image_metadata:
+ metadata = self.image_metadata
+ elif self.image_metadata and 'cirros' in self.image_metadata \
+ and 'disk_file' in self.image_metadata['cirros']:
+ metadata = self.image_metadata['cirros']
+ else:
+ metadata = {'disk_url': openstack_tests.CIRROS_DEFAULT_IMAGE_URL,
+ 'kernel_url': openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL,
+ 'ramdisk_url': openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL}
+
image_settings = openstack_tests.cirros_image_settings(
name=self.image_name,
- image_metadata={'disk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img',
- 'kernel_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel',
- 'ramdisk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'})
+ image_metadata=metadata)
+
+ if not image_settings.ramdisk_image_settings or not image_settings.kernel_image_settings:
+ logger.warn('3 Part image will not be tested. Image metadata has overridden this functionality')
self.image_creator = OpenStackImage(self.os_creds, image_settings)
self.image_creator.create()
except Exception as e:
logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
try:
self.image_creator.clean()
except Exception as e:
vm_inst = self.inst_creator.create()
self.assertIsNotNone(vm_inst)
self.assertTrue(self.inst_creator.vm_active(block=True))
+
+
+class CreateInstanceMockOfflineTests(OSComponentTestCase):
+ """
+ Tests the custom image_metadata that can be set by clients for handling images differently than the
+ default behavior of the existing tests primarily for offline testing
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
+ within OpenStack
+ """
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+
+ self.tmpDir = 'tmp/' + str(self.guid)
+ if not os.path.exists(self.tmpDir):
+ os.makedirs(self.tmpDir)
+
+ self.image_name = self.guid + '-image'
+ self.vm_inst_name = self.guid + '-inst'
+ self.port_1_name = self.guid + 'port-1'
+
+ # Initialize for tearDown()
+ self.image_creator = None
+ self.network_creator = None
+ self.flavor_creator = None
+ self.inst_creator = None
+
+ self.priv_net_config = openstack_tests.get_priv_net_config(
+ net_name=self.guid + '-priv-net', subnet_name=self.guid + '-priv-subnet')
+ self.port_settings = PortSettings(
+ name=self.port_1_name, network_name=self.priv_net_config.network_settings.name)
+
+ try:
+ # Download image file
+ self.image_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_IMAGE_URL, self.tmpDir)
+
+ # Create Network
+ self.network_creator = OpenStackNetwork(self.os_creds, self.priv_net_config.network_settings)
+ self.network_creator.create()
+
+ # Create Flavor
+ self.flavor_creator = OpenStackFlavor(
+ self.os_creds,
+ FlavorSettings(
+ name=self.guid + '-flavor-name', ram=128, disk=10, vcpus=1))
+ self.flavor_creator.create()
+ 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 - ' + str(e))
+
+ if self.network_creator:
+ try:
+ self.network_creator.clean()
+ except Exception as e:
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
+
+ if self.flavor_creator:
+ try:
+ self.flavor_creator.clean()
+ except Exception as e:
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
+
+ if self.image_creator:
+ try:
+ self.image_creator.clean()
+ except Exception as e:
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
+
+ if os.path.exists(self.tmpDir) and os.path.isdir(self.tmpDir):
+ shutil.rmtree(self.tmpDir)
+
+ def test_inst_from_file_image_simple_flat(self):
+ """
+ Creates a VM instance from a locally sourced file image using simply the 'disk_file' attribute vs.
+ using the 'config' option which completely overrides all image settings
+ :return:
+ """
+ metadata = {'disk_file': self.image_file.name}
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ self.assertEqual(self.image_file.name, os_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertIsNone(os_image_settings.url)
+ self.assertFalse(os_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+
+ self.assertIsNone(os_image_settings.kernel_image_settings)
+ self.assertIsNone(os_image_settings.ramdisk_image_settings)
+
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ self.image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_file_image_simple_nested(self):
+ """
+ Creates a VM instance from a locally sourced file image using simply the 'disk_file' attribute under 'cirros'
+ vs. using the 'config' option which completely overrides all image settings
+ :return:
+ """
+ metadata = {'cirros': {'disk_file': self.image_file.name}}
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ self.assertEqual(self.image_file.name, os_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertIsNone(os_image_settings.url)
+ self.assertFalse(os_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+
+ self.assertIsNone(os_image_settings.kernel_image_settings)
+ self.assertIsNone(os_image_settings.ramdisk_image_settings)
+
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ self.image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_existing(self):
+ """
+ Creates a VM instance from a image creator that has been configured to use an existing image
+ :return:
+ """
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ test_image_creator = OpenStackImage(
+ self.os_creds, ImageSettings(name=self.image_creator.image_settings.name,
+ image_user=self.image_creator.image_settings.image_user, exists=True))
+ test_image_creator.create()
+ self.assertEqual(self.image_creator.get_image().id, test_image_creator.get_image().id)
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ test_image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_file_image_complex(self):
+ """
+ Creates a VM instance from a locally sourced file image by overriding the default settings by using a dict()
+ that can be read in by ImageSettings
+ :return:
+ """
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ metadata = {'cirros': {'config':
+ {'name': os_image_settings.name, 'image_user': os_image_settings.image_user,
+ 'exists': True}}}
+ test_image_settings = openstack_tests.cirros_image_settings(image_metadata=metadata)
+ test_image = OpenStackImage(self.os_creds, test_image_settings)
+ test_image.create()
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ test_image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_file_3part_image_complex(self):
+ """
+ Creates a VM instance from a locally sourced file image by overriding the default settings by using a dict()
+ that can be read in by ImageSettings
+ :return:
+ """
+
+ kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+
+ metadata = {'cirros':
+ {'config':
+ {'name': self.image_name,
+ 'image_user': openstack_tests.CIRROS_USER,
+ 'image_file': self.image_file.name,
+ 'format': openstack_tests.DEFAULT_IMAGE_FORMAT,
+ 'kernel_image_settings':
+ {'name': self.image_name + '-kernel',
+ 'image_user': openstack_tests.CIRROS_USER,
+ 'image_file': kernel_file.name,
+ 'format': openstack_tests.DEFAULT_IMAGE_FORMAT},
+ 'ramdisk_image_settings':
+ {'name': self.image_name + '-ramdisk',
+ 'image_user': openstack_tests.CIRROS_USER,
+ 'image_file': ramdisk_file.name,
+ 'format': openstack_tests.DEFAULT_IMAGE_FORMAT}}}}
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ self.assertEqual(self.image_name, os_image_settings.name)
+ self.assertEqual(self.image_file.name, os_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertIsNone(os_image_settings.url)
+ self.assertFalse(os_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+
+ self.assertIsNotNone(os_image_settings.kernel_image_settings)
+ self.assertEqual(self.image_name + '-kernel', os_image_settings.kernel_image_settings.name)
+ self.assertEqual(kernel_file.name, os_image_settings.kernel_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.kernel_image_settings.image_user)
+ self.assertIsNone(os_image_settings.kernel_image_settings.url)
+ self.assertFalse(os_image_settings.kernel_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.kernel_image_settings.format)
+
+ self.assertIsNotNone(os_image_settings.ramdisk_image_settings)
+ self.assertEqual(self.image_name + '-ramdisk', os_image_settings.ramdisk_image_settings.name)
+ self.assertEqual(ramdisk_file.name, os_image_settings.ramdisk_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.ramdisk_image_settings.image_user)
+ self.assertIsNone(os_image_settings.ramdisk_image_settings.url)
+ self.assertFalse(os_image_settings.ramdisk_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.ramdisk_image_settings.format)
+
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings, self.image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_file_3part_image_simple_flat(self):
+ """
+ Creates a VM instance from a 3-part image locally sourced from file images using simply the 'disk_file',
+ 'kernel_file', and 'ramdisk_file' attributes vs. using the 'config' option which completely overrides all
+ image settings
+ :return:
+ """
+ kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+
+ metadata = {'disk_file': self.image_file.name, 'kernel_file': kernel_file.name,
+ 'ramdisk_file': ramdisk_file.name}
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+
+ self.assertEqual(self.image_name, os_image_settings.name)
+ self.assertEqual(self.image_file.name, os_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertIsNone(os_image_settings.url)
+ self.assertFalse(os_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+
+ self.assertIsNotNone(os_image_settings.kernel_image_settings)
+ self.assertEqual(self.image_name + '-kernel', os_image_settings.kernel_image_settings.name)
+ self.assertEqual(kernel_file.name, os_image_settings.kernel_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.kernel_image_settings.image_user)
+ self.assertIsNone(os_image_settings.kernel_image_settings.url)
+ self.assertFalse(os_image_settings.kernel_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.kernel_image_settings.format)
+
+ self.assertIsNotNone(os_image_settings.ramdisk_image_settings)
+ self.assertEqual(self.image_name + '-ramdisk', os_image_settings.ramdisk_image_settings.name)
+ self.assertEqual(ramdisk_file.name, os_image_settings.ramdisk_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.ramdisk_image_settings.image_user)
+ self.assertIsNone(os_image_settings.ramdisk_image_settings.url)
+ self.assertFalse(os_image_settings.ramdisk_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.ramdisk_image_settings.format)
+
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ self.assertIsNotNone(self.image_creator.get_kernel_image())
+ self.assertIsNotNone(self.image_creator.get_ramdisk_image())
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ self.image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_file_3part_image_simple_nested(self):
+ """
+ Creates a VM instance from a 3-part image locally sourced from file images using simply the 'disk_file',
+ 'kernel_file', and 'ramdisk_file' attributes under 'cirros' vs. using the 'config' option which completely
+ overrides all image settings
+ :return:
+ """
+ kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+
+ metadata = {'cirros': {'disk_file': self.image_file.name, 'kernel_file': kernel_file.name,
+ 'ramdisk_file': ramdisk_file.name}}
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+
+ self.assertEqual(self.image_name, os_image_settings.name)
+ self.assertEqual(self.image_file.name, os_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertIsNone(os_image_settings.url)
+ self.assertFalse(os_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+
+ self.assertIsNotNone(os_image_settings.kernel_image_settings)
+ self.assertEqual(self.image_name + '-kernel', os_image_settings.kernel_image_settings.name)
+ self.assertEqual(kernel_file.name, os_image_settings.kernel_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.kernel_image_settings.image_user)
+ self.assertIsNone(os_image_settings.kernel_image_settings.url)
+ self.assertFalse(os_image_settings.kernel_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.kernel_image_settings.format)
+
+ self.assertIsNotNone(os_image_settings.ramdisk_image_settings)
+ self.assertEqual(self.image_name + '-ramdisk', os_image_settings.ramdisk_image_settings.name)
+ self.assertEqual(ramdisk_file.name, os_image_settings.ramdisk_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.ramdisk_image_settings.image_user)
+ self.assertIsNone(os_image_settings.ramdisk_image_settings.url)
+ self.assertFalse(os_image_settings.ramdisk_image_settings.exists)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.ramdisk_image_settings.format)
+
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ self.assertIsNotNone(self.image_creator.get_kernel_image())
+ self.assertIsNotNone(self.image_creator.get_ramdisk_image())
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ self.image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
+
+ def test_inst_from_file_3part_image_existing(self):
+ """
+ Creates a VM instance from a 3-part image that is existing
+ :return:
+ """
+ kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+
+ metadata = {'cirros': {'disk_file': self.image_file.name, 'kernel_file': kernel_file.name,
+ 'ramdisk_file': ramdisk_file.name}}
+
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator.create()
+
+ test_image_creator = OpenStackImage(
+ self.os_creds, ImageSettings(name=self.image_creator.image_settings.name,
+ image_user=self.image_creator.image_settings.image_user, exists=True))
+ test_image_creator.create()
+ self.assertEqual(self.image_creator.get_image().id, test_image_creator.get_image().id)
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ test_image_creator.image_settings)
+ self.inst_creator.create()
+
+ self.assertTrue(self.inst_creator.vm_active(block=True))
logger = logging.getLogger('openstack_tests')
+CIRROS_DEFAULT_IMAGE_URL = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img'
+CIRROS_DEFAULT_KERNEL_IMAGE_URL = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel'
+CIRROS_DEFAULT_RAMDISK_IMAGE_URL = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'
+CIRROS_USER = 'cirros'
+
+CENTOS_DEFAULT_IMAGE_URL = 'http://cloud.centos.org/centos/7/images/CentOS-7-x86_64-GenericCloud.qcow2'
+CENTOS_USER = 'centos'
+
+UBUNTU_DEFAULT_IMAGE_URL =\
+ 'http://uec-images.ubuntu.com/releases/trusty/14.04/ubuntu-14.04-server-cloudimg-amd64-disk1.img'
+UBUNTU_USER = 'ubuntu'
+
+DEFAULT_IMAGE_FORMAT = 'qcow2'
+
def get_credentials(os_env_file=None, proxy_settings_str=None, ssh_proxy_cmd=None, dev_os_env_file=None):
"""
return os_creds
-def cirros_image_settings(name, url=None, image_metadata=None, kernel_settings=None, ramdisk_settings=None):
+def create_image_settings(image_name, image_user, image_format, metadata, disk_url=None, default_url=None,
+ kernel_settings=None, ramdisk_settings=None, public=False):
"""
- Returns the image settings for a Cirros QCOW2 image
- :param name: the name of the image
- :param url: the image's URL
- :param image_metadata: dict() values to override URLs for disk, kernel, and ramdisk
+ Returns the image settings object
+ :param image_name: the name of the image
+ :param image_user: the image's sudo user
+ :param image_format: the image's format string
+ :param metadata: custom metadata for overriding default behavior for test image settings
+ :param disk_url: the disk image's URL
+ :param default_url: the default URL for the disk image
:param kernel_settings: override to the kernel settings from the image_metadata
:param ramdisk_settings: override to the ramdisk settings from the image_metadata
+ :param public: True denotes image can be used by other projects where False indicates the converse (default: False)
:return:
"""
- if image_metadata and 'cirros' in image_metadata:
- metadata = image_metadata['cirros']
- else:
- metadata = image_metadata
- if metadata and 'disk_url' in metadata:
- url = metadata['disk_url']
- if not url:
- url = 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img'
+ logger.debug('Image metadata - ' + str(metadata))
- if metadata and 'kernel_url' in metadata and kernel_settings is None:
- kernel_image_settings = ImageSettings(name=name + '-kernel', image_user='cirros', img_format='qcow2',
- url=metadata['kernel_url'])
+ if metadata and 'config' in metadata:
+ return ImageSettings(config=metadata['config'])
+
+ disk_file = None
+ if metadata:
+ disk_url = metadata.get('disk_url')
+ disk_file = metadata.get('disk_file')
+ elif not disk_url:
+ disk_url = default_url
+ else:
+ disk_url = disk_url
+
+ if metadata and ('kernel_file' in metadata or 'kernel_url' in metadata) and kernel_settings is None:
+ kernel_image_settings = ImageSettings(
+ name=image_name + '-kernel', image_user=image_user, img_format=image_format,
+ image_file=metadata.get('kernel_file'), url=metadata.get('kernel_url'), public=public)
else:
kernel_image_settings = kernel_settings
- if metadata and 'ramdisk_url' in metadata and ramdisk_settings is None:
- ramdisk_image_settings = ImageSettings(name=name + '-ramdisk', image_user='cirros', img_format='qcow2',
- url=metadata['ramdisk_url'])
+ if metadata and ('ramdisk_file' in metadata or 'ramdisk_url' in metadata) and ramdisk_settings is None:
+ ramdisk_image_settings = ImageSettings(
+ name=image_name + '-ramdisk', image_user=image_user, img_format=image_format,
+ image_file=metadata.get('ramdisk_file'), url=metadata.get('ramdisk_url'), public=public)
else:
ramdisk_image_settings = ramdisk_settings
if metadata and 'extra_properties' in metadata:
extra_properties = metadata['extra_properties']
- return ImageSettings(name=name, image_user='cirros', img_format='qcow2', url=url,
- extra_properties=extra_properties,
- kernel_image_settings=kernel_image_settings,
- ramdisk_image_settings=ramdisk_image_settings)
+ return ImageSettings(name=image_name, image_user=image_user, img_format=image_format, image_file=disk_file,
+ url=disk_url, extra_properties=extra_properties, kernel_image_settings=kernel_image_settings,
+ ramdisk_image_settings=ramdisk_image_settings, public=public)
-def file_image_test_settings(name, file_path, image_user='cirros'):
- return ImageSettings(name=name, image_user=image_user, img_format='qcow2', image_file=file_path)
+def cirros_image_settings(name=None, url=None, image_metadata=None, kernel_settings=None, ramdisk_settings=None,
+ public=False):
+ """
+ Returns the image settings for a Cirros QCOW2 image
+ :param name: the name of the image
+ :param url: the image's URL
+ :param image_metadata: dict() values to override URLs for disk, kernel, and ramdisk
+ :param kernel_settings: override to the kernel settings from the image_metadata
+ :param ramdisk_settings: override to the ramdisk settings from the image_metadata
+ :param public: True denotes image can be used by other projects where False indicates the converse
+ :return:
+ """
+ if image_metadata and 'cirros' in image_metadata:
+ metadata = image_metadata['cirros']
+ else:
+ metadata = image_metadata
+
+ return create_image_settings(
+ image_name=name, image_user=CIRROS_USER, image_format=DEFAULT_IMAGE_FORMAT, metadata=metadata, disk_url=url,
+ default_url=CIRROS_DEFAULT_IMAGE_URL,
+ kernel_settings=kernel_settings, ramdisk_settings=ramdisk_settings, public=public)
+
+def file_image_test_settings(name, file_path, image_user=CIRROS_USER):
+ return ImageSettings(name=name, image_user=image_user, img_format=DEFAULT_IMAGE_FORMAT, image_file=file_path)
-def centos_image_settings(name, url=None, image_metadata=None):
+
+def centos_image_settings(name, url=None, image_metadata=None, kernel_settings=None, ramdisk_settings=None,
+ public=False):
"""
Returns the image settings for a Centos QCOW2 image
:param name: the name of the image
:param url: the image's URL
:param image_metadata: dict() values to override URLs for disk, kernel, and ramdisk
+ :param kernel_settings: override to the kernel settings from the image_metadata
+ :param ramdisk_settings: override to the ramdisk settings from the image_metadata
+ :param public: True denotes image can be used by other projects where False indicates the converse
:return:
"""
if image_metadata and 'centos' in image_metadata:
else:
metadata = image_metadata
- if metadata and 'disk_url' in metadata:
- url = metadata['disk_url']
- if not url:
- url = 'http://cloud.centos.org/centos/7/images/CentOS-7-x86_64-GenericCloud.qcow2'
+ return create_image_settings(
+ image_name=name, image_user=CENTOS_USER, image_format=DEFAULT_IMAGE_FORMAT, metadata=metadata, disk_url=url,
+ default_url=CENTOS_DEFAULT_IMAGE_URL,
+ kernel_settings=kernel_settings, ramdisk_settings=ramdisk_settings, public=public)
- kernel_image_settings = None
- if metadata and 'kernel_url' in metadata:
- kernel_image_settings = ImageSettings(name=name + '-kernel', image_user='centos', img_format='qcow2',
- url=metadata['kernel_url'])
- ramdisk_image_settings = None
- if metadata and 'ramdisk_url' in metadata:
- ramdisk_image_settings = ImageSettings(name=name + '-ramdisk', image_user='centos', img_format='qcow2',
- url=metadata['ramdisk_url'])
- extra_properties = None
- if metadata and 'extra_properties' in metadata:
- extra_properties = metadata['extra_properties']
-
- return ImageSettings(
- name=name, image_user='centos', img_format='qcow2',
- extra_properties=extra_properties, url=url,
- nic_config_pb_loc='./provisioning/ansible/centos-network-setup/playbooks/configure_host.yml',
- kernel_image_settings=kernel_image_settings,
- ramdisk_image_settings=ramdisk_image_settings)
-
-
-def ubuntu_image_settings(name, url=None, image_metadata=None):
+def ubuntu_image_settings(name, url=None, image_metadata=None, kernel_settings=None, ramdisk_settings=None,
+ public=False):
"""
Returns the image settings for a Ubuntu QCOW2 image
:param name: the name of the image
:param url: the image's URL
:param image_metadata: dict() values to override URLs for disk, kernel, and ramdisk
+ :param kernel_settings: override to the kernel settings from the image_metadata
+ :param ramdisk_settings: override to the ramdisk settings from the image_metadata
+ :param public: True denotes image can be used by other projects where False indicates the converse
:return:
"""
if image_metadata and 'ubuntu' in image_metadata:
else:
metadata = image_metadata
- if metadata and 'disk_url' in metadata:
- url = metadata['disk_url']
- if not url:
- url = 'http://uec-images.ubuntu.com/releases/trusty/14.04/ubuntu-14.04-server-cloudimg-amd64-disk1.img'
-
- kernel_image_settings = None
- if metadata and 'kernel_url' in metadata:
- kernel_image_settings = ImageSettings(name=name + '-kernel', url=metadata['kernel_url'])
- ramdisk_image_settings = None
- if metadata and 'ramdisk_url' in metadata:
- ramdisk_image_settings = ImageSettings(name=name + '-ramdisk', url=metadata['ramdisk_url'])
-
- extra_properties = None
- if metadata and 'extra_properties' in metadata:
- extra_properties = metadata['extra_properties']
-
- return ImageSettings(
- name=name, image_user='ubuntu', img_format='qcow2',
- extra_properties=extra_properties, url=url,
- nic_config_pb_loc='./provisioning/ansible/ubuntu-network-setup/playbooks/configure_host.yml',
- kernel_image_settings=kernel_image_settings,
- ramdisk_image_settings=ramdisk_image_settings)
+ return create_image_settings(
+ image_name=name, image_user=CENTOS_USER, image_format=DEFAULT_IMAGE_FORMAT, metadata=metadata, disk_url=url,
+ default_url=UBUNTU_DEFAULT_IMAGE_URL,
+ kernel_settings=kernel_settings, ramdisk_settings=ramdisk_settings, public=public)
def get_priv_net_config(net_name, subnet_name, router_name=None, cidr='10.55.0.0/24', external_net=None):
class OSComponentTestCase(unittest.TestCase):
- def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None, log_level=logging.DEBUG):
+ def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None, image_metadata=None,
+ log_level=logging.DEBUG):
"""
Super for test classes requiring a connection to OpenStack
:param method_name: default 'runTest'
:param os_creds: the OSCreds object, when null it searches for the file {cwd}/openstack/tests/conf/os_env.yaml
:param ext_net_name: the name of the external network that is used for creating routers for floating IPs
+ :param image_metadata: ability to override images being used in the tests (see examples/image-metadata)
:param log_level: the logging level of your test run (default DEBUG)
"""
super(OSComponentTestCase, self).__init__(method_name)
test_conf = file_utils.read_yaml(dev_os_env_file)
self.ext_net_name = test_conf.get('ext_net')
+ self.image_metadata = image_metadata
+
@staticmethod
- def parameterize(testcase_klass, os_creds, ext_net_name, log_level=logging.DEBUG):
+ def parameterize(testcase_klass, os_creds, ext_net_name, image_metadata=None, log_level=logging.DEBUG):
""" Create a suite containing all tests taken from the given
subclass, passing them the parameter 'param'.
"""
test_names = test_loader.getTestCaseNames(testcase_klass)
suite = unittest.TestSuite()
for name in test_names:
- suite.addTest(testcase_klass(name, os_creds, ext_net_name, log_level))
+ suite.addTest(testcase_klass(name, os_creds, ext_net_name, image_metadata, log_level))
return suite
:param log_level: the logging level of your test run (default DEBUG)
"""
super(OSIntegrationTestCase, self).__init__(method_name=method_name, os_creds=os_creds,
- ext_net_name=ext_net_name, log_level=log_level)
+ ext_net_name=ext_net_name, image_metadata=image_metadata,
+ log_level=log_level)
self.use_keystone = use_keystone
self.keystone = None
self.flavor_metadata = flavor_metadata
- self.image_metadata = image_metadata
@staticmethod
def parameterize(testcase_klass, os_creds, ext_net_name, use_keystone=False, flavor_metadata=None,
"""
created_image = None
+ # TODO/REFACTOR - replace each call with one including kwargs
if image_settings.url:
if image_settings.extra_properties:
created_image = glance.images.create(
name=image_settings.name, disk_format=image_settings.format, container_format="bare",
- location=image_settings.url, properties=image_settings.extra_properties)
+ location=image_settings.url, properties=image_settings.extra_properties,
+ is_public=image_settings.public)
else:
created_image = glance.images.create(name=image_settings.name, disk_format=image_settings.format,
- container_format="bare", location=image_settings.url)
+ container_format="bare", location=image_settings.url,
+ is_public=image_settings.public)
elif image_settings.image_file:
- image_file = file_utils.get_file(image_settings.image_file)
+ image_file = open(image_settings.image_file, 'rb')
if image_settings.extra_properties:
created_image = glance.images.create(
name=image_settings.name, disk_format=image_settings.format, container_format="bare", data=image_file,
- properties=image_settings.extra_properties)
+ properties=image_settings.extra_properties, is_public=image_settings.public)
else:
created_image = glance.images.create(
- name=image_settings.name, disk_format=image_settings.format, container_format="bare", data=image_file)
+ name=image_settings.name, disk_format=image_settings.format, container_format="bare", data=image_file,
+ is_public=image_settings.public)
return Image(name=image_settings.name, image_id=created_image.id, size=created_image.size,
properties=created_image.properties)
:return: the OpenStack image object
:raise Exception if using a file and it cannot be found
"""
- cleanup_file = False
+ cleanup_temp_file = False
if image_settings.image_file:
image_filename = image_settings.image_file
elif image_settings.url:
image_file = file_utils.download(image_settings.url, '/tmp', str(uuid.uuid4()))
image_filename = image_file.name
- cleanup_file = True
+ cleanup_temp_file = True
else:
raise Exception('Filename or URL of image not configured')
kwargs['name'] = image_settings.name
kwargs['disk_format'] = image_settings.format
kwargs['container_format'] = 'bare'
+
+ if image_settings.public:
+ kwargs['visibility'] = 'public'
+
if image_settings.extra_properties:
kwargs.update(image_settings.extra_properties)
created_image = glance.images.create(**kwargs)
- image_file = file_utils.get_file(image_filename)
+ image_file = open(image_filename, 'rb')
glance.images.upload(created_image['id'], image_file)
- except Exception as e:
+ except:
logger.error('Unexpected exception creating image. Rolling back')
if created_image:
delete_image(glance, created_image)
- raise e
+ raise
finally:
- if cleanup_file:
+ if cleanup_temp_file:
os.remove(image_filename)
updated_image = glance.images.get(created_image['id'])
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+import logging
import os
import shutil
import uuid
__author__ = 'spisarski'
+logger = logging.getLogger('glance_utils_tests')
+
+
class GlanceSmokeTests(OSComponentTestCase):
"""
Tests to ensure that the neutron client can communicate with the cloud
self.image_name = self.__class__.__name__ + '-' + str(guid)
self.image = None
self.glance = glance_utils.glance_client(self.os_creds)
+ if self.image_metadata:
+ self.glance_test_meta = self.image_metadata.get('glance_tests')
+ else:
+ self.glance_test_meta = dict()
self.tmp_dir = 'tmp/' + str(guid)
if not os.path.exists(self.tmp_dir):
def test_create_image_minimal_url(self):
"""
- Tests the glance_utils.create_image() function with a URL
+ Tests the glance_utils.create_image() function with a URL unless the self.glance_test_meta has configured a
+ file to be used.
"""
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
+ if 'disk_file' not in self.glance_test_meta:
+ os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name,
+ image_metadata=self.glance_test_meta)
- self.image = glance_utils.create_image(self.glance, os_image_settings)
- self.assertIsNotNone(self.image)
+ self.image = glance_utils.create_image(self.glance, os_image_settings)
+ self.assertIsNotNone(self.image)
- self.assertEqual(self.image_name, self.image.name)
+ self.assertEqual(self.image_name, self.image.name)
- image = glance_utils.get_image(self.glance, os_image_settings.name)
- self.assertIsNotNone(image)
+ image = glance_utils.get_image(self.glance, os_image_settings.name)
+ self.assertIsNotNone(image)
- validation_utils.objects_equivalent(self.image, image)
+ validation_utils.objects_equivalent(self.image, image)
+ else:
+ logger.warn('Test not executed as the image metadata requires image files')
def test_create_image_minimal_file(self):
"""
Tests the glance_utils.create_image() function with a file
"""
- url_image_settings = openstack_tests.cirros_image_settings('foo')
- image_file = file_utils.download(url_image_settings.url, self.tmp_dir)
- file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=image_file.name)
+ if 'disk_file' not in self.glance_test_meta:
+ url_image_settings = openstack_tests.cirros_image_settings(name='foo', image_metadata=self.glance_test_meta)
+ image_file_name = file_utils.download(url_image_settings.url, self.tmp_dir).name
+ else:
+ image_file_name = self.glance_test_meta['disk_file']
+
+ file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=image_file_name)
self.image = glance_utils.create_image(self.glance, file_image_settings)
self.assertIsNotNone(self.image)
if self.network_creator:
self.network_creator.clean()
- if self.image_creator:
+ if self.image_creator and not self.image_creator.image_settings.exists:
self.image_creator.clean()
if os.path.isfile(self.test_file_local_path):
-# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs")
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
# and others. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
import os
import unittest
-from snaps import test_suite_builder
+from snaps import test_suite_builder, file_utils
from snaps.openstack.tests import openstack_tests
__author__ = 'spisarski'
def __create_test_suite(source_filename, ext_net_name, proxy_settings, ssh_proxy_cmd, run_unit_tests,
- run_connection_tests, run_api_tests, run_integration_tests, flavor_metadata, use_keystone,
- use_floating_ips, log_level):
+ run_connection_tests, run_api_tests, run_integration_tests, run_staging_tests, flavor_metadata,
+ image_metadata, use_keystone, use_floating_ips, log_level):
"""
Compiles the tests that should run
:param source_filename: the OpenStack credentials file (required)
:param run_connection_tests: when true, the tests that perform simple connections to OpenStack are executed
:param run_api_tests: when true, the tests that perform simple API calls to OpenStack are executed
:param run_integration_tests: when true, the integration tests are executed
+ :param run_staging_tests: when true, the staging tests are executed
:param proxy_settings: <host>:<port> of the proxy server (optional)
:param ssh_proxy_cmd: the command used to connect via SSH over some proxy server (optional)
:param flavor_metadata: dict() object containing the metadata for flavors created for test VM instance
+ :param image_metadata: dict() object containing the metadata for overriding default images within the tests
:param use_keystone: when true, tests creating users and projects will be exercised and must be run on a host that
has access to the cloud's administrative network
:param use_floating_ips: when true, tests requiring floating IPs will be executed
# Tests the OpenStack API calls
if run_api_tests:
test_suite_builder.add_openstack_api_tests(
- suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone, log_level=log_level)
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone,
+ image_metadata=image_metadata, log_level=log_level)
# Long running integration type tests
if run_integration_tests:
test_suite_builder.add_openstack_integration_tests(
suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone,
- flavor_metadata=flavor_metadata, use_floating_ips=use_floating_ips, log_level=log_level)
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata, use_floating_ips=use_floating_ips,
+ log_level=log_level)
+
+ if run_staging_tests:
+ test_suite_builder.add_openstack_staging_tests(
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level)
return suite
log_level = LOG_LEVELS.get(arguments.log_level, logging.DEBUG)
flavor_metadata = None
-
if arguments.flavor_metadata:
flavor_metadata = json.loads(arguments.flavor_metadata)
+ image_metadata = None
+ if arguments.image_metadata_file:
+ image_metadata = file_utils.read_yaml(arguments.image_metadata_file)
+
suite = None
if arguments.env and arguments.ext_net:
unit = arguments.include_unit != ARG_NOT_SET
connection = arguments.include_connection != ARG_NOT_SET
api = arguments.include_api != ARG_NOT_SET
integration = arguments.include_integration != ARG_NOT_SET
- if not unit and not connection and not api and not integration:
+ staging = arguments.include_staging != ARG_NOT_SET
+ if not unit and not connection and not api and not integration and not staging:
unit = True
connection = True
api = True
integration = True
suite = __create_test_suite(arguments.env, arguments.ext_net, arguments.proxy, arguments.ssh_proxy_cmd,
- unit, connection, api, integration,
- flavor_metadata,
+ unit, connection, api, integration, staging, flavor_metadata, image_metadata,
arguments.use_keystone != ARG_NOT_SET,
arguments.floating_ips != ARG_NOT_SET, log_level)
else:
help='Optonal SSH proxy command value')
parser.add_argument('-l', '--log-level', dest='log_level', default='INFO',
help='Logging Level (FATAL|CRITICAL|ERROR|WARN|INFO|DEBUG)')
-
parser.add_argument('-u', '--unit-tests', dest='include_unit', default=ARG_NOT_SET, nargs='?',
help='When argument is set, all tests not requiring OpenStack will be executed')
parser.add_argument('-c', '--connection-tests', dest='include_connection', default=ARG_NOT_SET, nargs='?',
help='When argument is set, OpenStack API tests will be executed')
parser.add_argument('-i', '--integration-tests', dest='include_integration', default=ARG_NOT_SET, nargs='?',
help='When argument is set, OpenStack integrations tests will be executed')
+ parser.add_argument('-st', '--staging-tests', dest='include_staging', default=ARG_NOT_SET, nargs='?',
+ help='When argument is set, OpenStack staging tests will be executed')
parser.add_argument('-f', '--floating-ips', dest='floating_ips', default=ARG_NOT_SET, nargs='?',
help='When argument is set, all integration tests requiring Floating IPs will be executed')
-
parser.add_argument('-k', '--use-keystone', dest='use_keystone', default=ARG_NOT_SET, nargs='?',
help='When argument is set, the tests will exercise the keystone APIs and must be run on a ' +
'machine that has access to the admin network' +
' and is able to create users and groups')
-
parser.add_argument('-fm', '--flavor-meta', dest='flavor_metadata',
default='{\"hw:mem_page_size\": \"any\"}',
help='JSON string to be used as flavor metadata for all test instances created')
-
+ parser.add_argument('-im', '--image-meta', dest='image_metadata_file',
+ default=None,
+ help='Location of YAML file containing the image metadata')
parser.add_argument('-r', '--num-runs', dest='num_runs', default=1,
help='Number of test runs to execute (default 1)')
from snaps.openstack.tests.create_instance_tests import CreateInstanceSingleNetworkTests, \
CreateInstancePubPrivNetTests, CreateInstanceOnComputeHost, CreateInstanceSimpleTests, \
FloatingIpSettingsUnitTests, InstanceSecurityGroupTests, VmInstanceSettingsUnitTests, \
- CreateInstancePortManipulationTests, SimpleHealthCheck, CreateInstanceFromThreePartImage
+ CreateInstancePortManipulationTests, SimpleHealthCheck, CreateInstanceFromThreePartImage, \
+ CreateInstanceMockOfflineTests
from snaps.provisioning.tests.ansible_utils_tests import AnsibleProvisioningTests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase, OSIntegrationTestCase
from snaps.openstack.utils.tests.nova_utils_tests import NovaSmokeTests, NovaUtilsKeypairTests, NovaUtilsFlavorTests
log_level=log_level))
-def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True, log_level=logging.INFO):
+def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True, image_metadata=None,
+ log_level=logging.INFO):
"""
Adds tests written to exercise all existing OpenStack APIs
:param suite: the unittest.TestSuite object to which to add the tests
:param ext_net_name: the name of an external network on the cloud under test
:param use_keystone: when True, tests requiring direct access to Keystone are added as these need to be running on
a host that has access to the cloud's private network
+ :param image_metadata: dict() object containing metadata for creating an image with custom config
+ (see YAML files in examples/image-metadata)
:param log_level: the logging level
:return: None as the tests will be adding to the 'suite' parameter object
"""
CreateProjectUserTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- GlanceUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ GlanceUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name, image_metadata=image_metadata,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
NeutronUtilsNetworkTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
:param ext_net_name: the name of an external network on the cloud under test
:param use_keystone: when True, tests requiring direct access to Keystone are added as these need to be running on
a host that has access to the cloud's private network
- :param image_metadata: dict() object containing metadata for creating an image with custom config:
- (i.e. {'hw_video_model' : 'vga'}). It can be used to override the default url and
- create 3-part images by passing kerner_url and ramdisk_url info
+ :param image_metadata: dict() object containing metadata for creating an image with custom config
+ (see YAML files in examples/image-metadata)
:param flavor_metadata: dict() object containing the metadata required by your flavor based on your configuration:
(i.e. {'hw:mem_page_size': 'large'})
:param use_floating_ips: when true, all tests requiring Floating IPs will be added to the suite
"""
suite.addTest(OSComponentTestCase.parameterize(
CreateNetworkTypeTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ suite.addTest(OSComponentTestCase.parameterize(
+ CreateInstanceMockOfflineTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
Code Review / snaps.git / commitdiff