diff options
-rw-r--r-- | snaps/openstack/create_flavor.py | 118 | ||||
-rw-r--r-- | snaps/openstack/create_image.py | 210 | ||||
-rw-r--r-- | snaps/openstack/tests/create_flavor_tests.py | 154 | ||||
-rw-r--r-- | snaps/openstack/tests/create_image_tests.py | 401 |
4 files changed, 545 insertions, 338 deletions
diff --git a/snaps/openstack/create_flavor.py b/snaps/openstack/create_flavor.py index 2d78be4..f1c7ee3 100644 --- a/snaps/openstack/create_flavor.py +++ b/snaps/openstack/create_flavor.py @@ -46,18 +46,23 @@ class OpenStackFlavor: def create(self, cleanup=False): """ Creates the image in OpenStack if it does not already exist - :param cleanup: Denotes whether or not this is being called for cleanup or not + :param cleanup: Denotes whether or not this is being called for cleanup + or not :return: The OpenStack flavor object """ self.__nova = nova_utils.nova_client(self.__os_creds) - self.__flavor = nova_utils.get_flavor_by_name(self.__nova, self.flavor_settings.name) + self.__flavor = nova_utils.get_flavor_by_name( + self.__nova, self.flavor_settings.name) if self.__flavor: - logger.info('Found flavor with name - ' + self.flavor_settings.name) + logger.info( + 'Found flavor with name - ' + self.flavor_settings.name) elif not cleanup: - self.__flavor = nova_utils.create_flavor(self.__nova, self.flavor_settings) + self.__flavor = nova_utils.create_flavor( + self.__nova, self.flavor_settings) if self.flavor_settings.metadata: self.__flavor.set_keys(self.flavor_settings.metadata) - self.__flavor = nova_utils.get_flavor_by_name(self.__nova, self.flavor_settings.name) + self.__flavor = nova_utils.get_flavor_by_name( + self.__nova, self.flavor_settings.name) else: logger.info('Did not create flavor due to cleanup mode') @@ -89,12 +94,12 @@ class FlavorSettings: Configuration settings for OpenStack flavor creation """ - def __init__(self, config=None, name=None, flavor_id='auto', ram=None, disk=None, vcpus=None, ephemeral=0, swap=0, - rxtx_factor=1.0, is_public=True, metadata=None): + def __init__(self, **kwargs): """ Constructor - :param config: dict() object containing the configuration settings using the attribute names below as each - member's the key and overrides any of the other parameters. + :param config: dict() object containing the configuration settings + using the attribute names below as each member's the + key and overrides any of the other parameters. :param name: the flavor's name (required) :param flavor_id: the string ID (default 'auto') :param ram: the required RAM in MB (required) @@ -102,62 +107,52 @@ class FlavorSettings: :param vcpus: the number of virtual CPUs (required) :param ephemeral: the size of the ephemeral disk in GB (default 0) :param swap: the size of the dedicated swap disk in GB (default 0) - :param rxtx_factor: the receive/transmit factor to be set on ports if backend supports - QoS extension (default 1.0) - :param is_public: denotes whether or not the flavor is public (default True) - :param metadata: freeform dict() for special metadata (default hw:mem_page_size=any) + :param rxtx_factor: the receive/transmit factor to be set on ports if + backend supports QoS extension (default 1.0) + :param is_public: denotes whether or not the flavor is public + (default True) + :param metadata: freeform dict() for special metadata """ + self.name = kwargs.get('name') - if config: - self.name = config.get('name') - - if config.get('flavor_id'): - self.flavor_id = config['flavor_id'] - else: - self.flavor_id = flavor_id - - self.ram = config.get('ram') - self.disk = config.get('disk') - self.vcpus = config.get('vcpus') - - if config.get('ephemeral'): - self.ephemeral = config['ephemeral'] - else: - self.ephemeral = ephemeral - - if config.get('swap'): - self.swap = config['swap'] - else: - self.swap = swap - - if config.get('rxtx_factor'): - self.rxtx_factor = config['rxtx_factor'] - else: - self.rxtx_factor = rxtx_factor - - if config.get('is_public') is not None: - self.is_public = config['is_public'] - else: - self.is_public = is_public - - if config.get('metadata'): - self.metadata = config['metadata'] - else: - self.metadata = metadata + if kwargs.get('flavor_id'): + self.flavor_id = kwargs['flavor_id'] else: - self.name = name - self.flavor_id = flavor_id - self.ram = ram - self.disk = disk - self.vcpus = vcpus - self.ephemeral = ephemeral - self.swap = swap - self.rxtx_factor = rxtx_factor - self.is_public = is_public - self.metadata = metadata + self.flavor_id = 'auto' + + self.ram = kwargs.get('ram') + self.disk = kwargs.get('disk') + self.vcpus = kwargs.get('vcpus') + + if kwargs.get('ephemeral'): + self.ephemeral = kwargs['ephemeral'] + else: + self.ephemeral = 0 + + if kwargs.get('swap'): + self.swap = kwargs['swap'] + else: + self.swap = 0 + + if kwargs.get('rxtx_factor'): + self.rxtx_factor = kwargs['rxtx_factor'] + else: + self.rxtx_factor = 1.0 + + if kwargs.get('is_public') is not None: + self.is_public = kwargs['is_public'] + else: + self.is_public = True + + if kwargs.get('metadata'): + self.metadata = kwargs['metadata'] + else: + self.metadata = None if not self.name or not self.ram or not self.disk or not self.vcpus: - raise Exception('The attributes name, ram, disk, and vcpus are required for FlavorSettings') + raise Exception( + 'The attributes name, ram, disk, and vcpus are required for' + 'FlavorSettings') if not isinstance(self.ram, int): raise Exception('The ram attribute must be a integer') @@ -175,7 +170,8 @@ class FlavorSettings: raise Exception('The swap attribute must be an integer') if self.rxtx_factor and not isinstance(self.rxtx_factor, (int, float)): - raise Exception('The is_public attribute must be an integer or float') + raise Exception( + 'The is_public attribute must be an integer or float') if self.is_public and not isinstance(self.is_public, bool): raise Exception('The is_public attribute must be a boolean') diff --git a/snaps/openstack/create_image.py b/snaps/openstack/create_image.py index 401e845..9ed813c 100644 --- a/snaps/openstack/create_image.py +++ b/snaps/openstack/create_image.py @@ -49,50 +49,68 @@ class OpenStackImage: def create(self, cleanup=False): """ - Creates the image in OpenStack if it does not already exist and returns the domain Image object - :param cleanup: Denotes whether or not this is being called for cleanup or not + Creates the image in OpenStack if it does not already exist and returns + the domain Image object + :param cleanup: Denotes whether or not this is being called for cleanup + or not :return: The OpenStack Image object """ self.__glance = glance_utils.glance_client(self.__os_creds) - self.__image = glance_utils.get_image(self.__glance, self.image_settings.name) + self.__image = glance_utils.get_image(self.__glance, + self.image_settings.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 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() if self.image_settings.kernel_image_settings: self.__kernel_image = glance_utils.get_image( - self.__glance, self.image_settings.kernel_image_settings.name) + self.__glance, + self.image_settings.kernel_image_settings.name) if not self.__kernel_image and not cleanup: - logger.info('Creating associated kernel image with name - ' - + self.image_settings.kernel_image_settings.name) + logger.info( + 'Creating associated kernel image with name - %s', + self.image_settings.kernel_image_settings.name) self.__kernel_image = glance_utils.create_image( - self.__glance, self.image_settings.kernel_image_settings) + self.__glance, + self.image_settings.kernel_image_settings) extra_properties['kernel_id'] = self.__kernel_image.id if self.image_settings.ramdisk_image_settings: self.__ramdisk_image = glance_utils.get_image( - self.__glance, self.image_settings.ramdisk_image_settings.name) + self.__glance, + self.image_settings.ramdisk_image_settings.name) if not self.__ramdisk_image and not cleanup: - logger.info('Creating associated ramdisk image with name - ' - + self.image_settings.ramdisk_image_settings.name) + logger.info( + 'Creating associated ramdisk image with name - %s', + self.image_settings.ramdisk_image_settings.name) self.__ramdisk_image = glance_utils.create_image( - self.__glance, self.image_settings.ramdisk_image_settings) + 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) + self.__image = glance_utils.create_image(self.__glance, + self.image_settings) - logger.info('Created image with name - ' + self.image_settings.name) + logger.info( + 'Created image with name - %s', 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) + logger.info( + 'Image is now active with name - %s', + self.image_settings.name) return self.__image else: - raise ImageCreationError('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') @@ -116,40 +134,51 @@ class OpenStackImage: def get_image(self): """ - Returns the domain Image object as it was populated when create() was called + Returns the domain Image object as it was populated when create() was + called :return: the object """ return self.__image def get_kernel_image(self): """ - Returns the OpenStack kernel image object as it was populated when create() was called + Returns the OpenStack kernel image object as it was populated when + create() was called :return: the object """ return self.__kernel_image def get_ramdisk_image(self): """ - Returns the OpenStack ramdisk image object as it was populated when create() was called + Returns the OpenStack ramdisk image object as it was populated when + create() was called :return: the object """ return self.__ramdisk_image - def image_active(self, block=False, timeout=IMAGE_ACTIVE_TIMEOUT, poll_interval=POLL_INTERVAL): + def image_active(self, block=False, timeout=IMAGE_ACTIVE_TIMEOUT, + poll_interval=POLL_INTERVAL): """ - Returns true when the image status returns the value of expected_status_code - :param block: When true, thread will block until active or timeout value in seconds has been exceeded (False) + Returns true when the image status returns the value of + expected_status_code + :param block: When true, thread will block until active or timeout + value in seconds has been exceeded (False) :param timeout: The timeout value :param poll_interval: The polling interval in seconds :return: T/F """ - return self._image_status_check(STATUS_ACTIVE, block, timeout, poll_interval) + return self._image_status_check(STATUS_ACTIVE, block, timeout, + poll_interval) - def _image_status_check(self, expected_status_code, block, timeout, poll_interval): + def _image_status_check(self, expected_status_code, block, timeout, + poll_interval): """ - Returns true when the image status returns the value of expected_status_code - :param expected_status_code: instance status evaluated with this string value - :param block: When true, thread will block until active or timeout value in seconds has been exceeded (False) + Returns true when the image status returns the value of + expected_status_code + :param expected_status_code: instance status evaluated with this string + value + :param block: When true, thread will block until active or timeout + value in seconds has been exceeded (False) :param timeout: The timeout value :param poll_interval: The polling interval in seconds :return: T/F @@ -163,26 +192,31 @@ class OpenStackImage: while timeout > time.time() - start: status = self._status(expected_status_code) if status: - logger.debug('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') + logger.debug('Retry querying image status in ' + str( + poll_interval) + ' seconds') time.sleep(poll_interval) - logger.debug('Image status query timeout in ' + str(timeout - (time.time() - start))) + logger.debug('Image status query timeout in ' + str( + timeout - (time.time() - start))) - logger.error('Timeout checking for image status for ' + expected_status_code) + logger.error( + 'Timeout checking for image status for ' + expected_status_code) return False def _status(self, expected_status_code): """ Returns True when active else False - :param expected_status_code: instance status evaluated with this string value + :param expected_status_code: instance status evaluated with this string + value :return: T/F """ - # TODO - Place this API call into glance_utils. status = glance_utils.get_image_status(self.__glance, self.__image) if not status: - logger.warning('Cannot image status for image with ID - ' + self.__image.id) + logger.warning( + 'Cannot image status for image with ID - ' + self.__image.id) return False if status == 'ERROR': @@ -192,87 +226,96 @@ class OpenStackImage: 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, exists=False, public=False): + def __init__(self, **kwargs): """ - - :param config: dict() object containing the configuration settings using the attribute names below as each - member's the key and overrides any of the other parameters. + Constructor :param name: the image's name (required) :param image_user: the image's default sudo user (required) - :param img_format: the image type (required) - :param url: the image download location (requires url or img_file) + :param format or img_format: the image type (required) + :param url or download_url: the image download location (requires url + or img_file) :param image_file: the image file location (requires url or img_file) - :param extra_properties: dict() object containing extra parameters to pass when loading the image; - can be ids of kernel and initramfs images for a 3-part image - :param nic_config_pb_loc: the file location to the Ansible Playbook that can configure multiple NICs + :param extra_properties: dict() object containing extra parameters to + pass when loading the image; + can be ids of kernel and initramfs images for + a 3-part image + :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 + :param public: When True, an image will be created with public + visibility """ - if config: - self.name = config.get('name') - self.image_user = config.get('image_user') - self.format = config.get('format') - self.url = config.get('download_url') - self.image_file = config.get('image_file') - self.extra_properties = config.get('extra_properties') - self.nic_config_pb_loc = config.get('nic_config_pb_loc') - if config.get('kernel_image_settings'): - self.kernel_image_settings = ImageSettings(config=config['kernel_image_settings']) + self.name = kwargs.get('name') + self.image_user = kwargs.get('image_user') + self.format = kwargs.get('format') + if not self.format: + self.format = kwargs.get('img_format') + + self.url = kwargs.get('url') + if not self.url: + self.url = kwargs.get('download_url') + + self.image_file = kwargs.get('image_file') + self.extra_properties = kwargs.get('extra_properties') + self.nic_config_pb_loc = kwargs.get('nic_config_pb_loc') + + kernel_image_settings = kwargs.get('kernel_image_settings') + if kernel_image_settings: + if isinstance(kernel_image_settings, dict): + self.kernel_image_settings = ImageSettings( + **kernel_image_settings) else: - self.kernel_image_settings = None + self.kernel_image_settings = kernel_image_settings + else: + self.kernel_image_settings = None - if config.get('ramdisk_image_settings'): - self.ramdisk_image_settings = ImageSettings(config=config['ramdisk_image_settings']) + ramdisk_image_settings = kwargs.get('ramdisk_image_settings') + if ramdisk_image_settings: + if isinstance(ramdisk_image_settings, dict): + self.ramdisk_image_settings = ImageSettings( + **ramdisk_image_settings) else: - self.ramdisk_image_settings = None + self.ramdisk_image_settings = ramdisk_image_settings + else: + self.ramdisk_image_settings = None - if 'exists' in config and config['exists'] is True: - self.exists = True - else: - self.exists = False + if 'exists' in kwargs and kwargs['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 + if 'public' in kwargs and kwargs['public'] is True: + self.public = True else: - self.name = name - self.image_user = image_user - self.format = img_format - self.url = url - self.image_file = image_file - self.extra_properties = extra_properties - 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 + self.public = False if not self.name: raise ImageSettingsError("The attribute name is required") 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') + raise ImageSettingsError( + 'URL or image file must be set or image must already exist') if self.url and self.image_file: - raise ImageSettingsError('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') + 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) @@ -281,5 +324,6 @@ class ImageCreationError(Exception): """ Exception to be thrown when an image cannot be created """ + def __init__(self, message): Exception.__init__(self, message) diff --git a/snaps/openstack/tests/create_flavor_tests.py b/snaps/openstack/tests/create_flavor_tests.py index be7ac64..5d7e4c4 100644 --- a/snaps/openstack/tests/create_flavor_tests.py +++ b/snaps/openstack/tests/create_flavor_tests.py @@ -12,8 +12,8 @@ # 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 uuid import unittest +import uuid from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor from snaps.openstack.tests.os_source_file_test import OSComponentTestCase @@ -61,107 +61,140 @@ class FlavorSettingsUnitTests(unittest.TestCase): def test_ram_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram='bar', disk=2, vcpus=3, ephemeral=4, swap=5, rxtx_factor=6.0, + FlavorSettings(name='foo', ram='bar', disk=2, vcpus=3, ephemeral=4, + swap=5, rxtx_factor=6.0, is_public=False) def test_config_ram_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 'bar', 'disk': 2, 'vcpus': 3, 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 'bar', 'disk': 2, 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_ram_float(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1.5, disk=2, vcpus=3, ephemeral=4, swap=5, rxtx_factor=6.0, is_public=False) + FlavorSettings(name='foo', ram=1.5, disk=2, vcpus=3, ephemeral=4, + swap=5, rxtx_factor=6.0, is_public=False) def test_config_ram_float(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1.5, 'disk': 2, 'vcpus': 3, 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1.5, 'disk': 2, 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_disk_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk='bar', vcpus=3, ephemeral=4, swap=5, rxtx_factor=6.0, + FlavorSettings(name='foo', ram=1, disk='bar', vcpus=3, ephemeral=4, + swap=5, rxtx_factor=6.0, is_public=False) def test_config_disk_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 'bar', 'vcpus': 3, 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 'bar', 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_disk_float(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2.5, vcpus=3, ephemeral=4, swap=5, rxtx_factor=6.0, is_public=False) + FlavorSettings(name='foo', ram=1, disk=2.5, vcpus=3, ephemeral=4, + swap=5, rxtx_factor=6.0, is_public=False) def test_config_disk_float(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2.5, 'vcpus': 3, 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2.5, 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_vcpus_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus='bar', ephemeral=4, swap=5, rxtx_factor=6.0, + FlavorSettings(name='foo', ram=1, disk=2, vcpus='bar', ephemeral=4, + swap=5, rxtx_factor=6.0, is_public=False) def test_config_vcpus_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 'bar', 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 'bar', + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_ephemeral_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral='bar', swap=5, rxtx_factor=6.0, + FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral='bar', + swap=5, rxtx_factor=6.0, is_public=False) def test_config_ephemeral_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, 'ephemeral': 'bar', 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, + 'ephemeral': 'bar', 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_ephemeral_float(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4.5, swap=5, rxtx_factor=6.0, is_public=False) + FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4.5, + swap=5, rxtx_factor=6.0, is_public=False) def test_config_ephemeral_float(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, 'ephemeral': 4.5, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, + 'ephemeral': 4.5, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_swap_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, swap='bar', rxtx_factor=6.0, + FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, + swap='bar', rxtx_factor=6.0, is_public=False) def test_config_swap_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, 'ephemeral': 4, 'swap': 'bar', - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, + 'ephemeral': 4, 'swap': 'bar', + 'rxtx_factor': 6.0, 'is_public': False}) def test_swap_float(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, swap=5.5, rxtx_factor=6.0, is_public=False) + FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, + swap=5.5, rxtx_factor=6.0, is_public=False) def test_config_swap_float(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, 'ephemeral': 4, 'swap': 5.5, - 'rxtx_factor': 6.0, 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, + 'ephemeral': 4, 'swap': 5.5, + 'rxtx_factor': 6.0, 'is_public': False}) def test_rxtx_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, swap=5, rxtx_factor='bar', is_public=False) + FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, + swap=5, rxtx_factor='bar', is_public=False) def test_config_rxtx_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 'bar', 'is_public': False}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 'bar', 'is_public': False}) def test_is_pub_string(self): with self.assertRaises(Exception): - FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, swap=5, rxtx_factor=6.0, is_public='bar') + FlavorSettings(name='foo', ram=1, disk=2, vcpus=3, ephemeral=4, + swap=5, rxtx_factor=6.0, is_public='bar') def test_config_is_pub_string(self): with self.assertRaises(Exception): - FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, 'ephemeral': 4, 'swap': 5, - 'rxtx_factor': 6.0, 'is_public': 'bar'}) + FlavorSettings( + config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, + 'rxtx_factor': 6.0, 'is_public': 'bar'}) def test_name_ram_disk_vcpus_only(self): settings = FlavorSettings(name='foo', ram=1, disk=2, vcpus=3) @@ -177,7 +210,8 @@ class FlavorSettingsUnitTests(unittest.TestCase): self.assertEqual(None, settings.metadata) def test_config_with_name_ram_disk_vcpus_only(self): - settings = FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3}) + settings = FlavorSettings( + **{'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3}) self.assertEqual('foo', settings.name) self.assertEqual('auto', settings.flavor_id) self.assertEqual(1, settings.ram) @@ -191,8 +225,9 @@ class FlavorSettingsUnitTests(unittest.TestCase): def test_all(self): metadata = {'foo': 'bar'} - settings = FlavorSettings(name='foo', flavor_id='bar', ram=1, disk=2, vcpus=3, ephemeral=4, swap=5, - rxtx_factor=6.0, is_public=False, metadata=metadata) + settings = FlavorSettings( + name='foo', flavor_id='bar', ram=1, disk=2, vcpus=3, ephemeral=4, + swap=5, rxtx_factor=6.0, is_public=False, metadata=metadata) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.flavor_id) self.assertEqual(1, settings.ram) @@ -206,9 +241,12 @@ class FlavorSettingsUnitTests(unittest.TestCase): def test_config_all(self): metadata = {'foo': 'bar'} - settings = FlavorSettings(config={'name': 'foo', 'flavor_id': 'bar', 'ram': 1, 'disk': 2, 'vcpus': 3, - 'ephemeral': 4, 'swap': 5, 'rxtx_factor': 6.0, 'is_public': False, - 'metadata': metadata}) + settings = FlavorSettings( + **{'name': 'foo', 'flavor_id': 'bar', 'ram': 1, 'disk': 2, + 'vcpus': 3, + 'ephemeral': 4, 'swap': 5, 'rxtx_factor': 6.0, + 'is_public': False, + 'metadata': metadata}) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.flavor_id) self.assertEqual(1, settings.ram) @@ -228,8 +266,8 @@ class CreateFlavorTests(OSComponentTestCase): def setUp(self): """ - Instantiates the CreateSecurityGroup object that is responsible for downloading and creating an OS image file - within OpenStack + Instantiates the CreateSecurityGroup object that is responsible for + downloading and creating an OS image file within OpenStack """ guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) self.flavor_name = guid + 'name' @@ -251,17 +289,20 @@ class CreateFlavorTests(OSComponentTestCase): Tests the creation of an OpenStack flavor. """ # Create Flavor - flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, vcpus=1) + flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, + vcpus=1) self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings) flavor = self.flavor_creator.create() self.assertTrue(validate_flavor(flavor_settings, flavor)) def test_create_flavor_existing(self): """ - Tests the creation of an OpenStack flavor then starts another creator to ensure it has not been done twice. + Tests the creation of an OpenStack flavor then starts another creator + to ensure it has not been done twice. """ # Create Flavor - flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, vcpus=1) + flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, + vcpus=1) self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings) flavor = self.flavor_creator.create() self.assertTrue(validate_flavor(flavor_settings, flavor)) @@ -276,7 +317,8 @@ class CreateFlavorTests(OSComponentTestCase): Tests the creation and cleanup of an OpenStack flavor. """ # Create Flavor - flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, vcpus=1) + flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, + vcpus=1) self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings) flavor = self.flavor_creator.create() self.assertTrue(validate_flavor(flavor_settings, flavor)) @@ -285,22 +327,26 @@ class CreateFlavorTests(OSComponentTestCase): self.flavor_creator.clean() self.assertIsNone(self.flavor_creator.get_flavor()) - self.assertIsNone(nova_utils.get_flavor_by_name(self.nova, flavor_settings.name)) + self.assertIsNone( + nova_utils.get_flavor_by_name(self.nova, flavor_settings.name)) def test_create_delete_flavor(self): """ - Tests the creation of an OpenStack Security Group, the deletion, then cleanup to ensure clean() does not + Tests the creation of an OpenStack Security Group, the deletion, then + cleanup to ensure clean() does not raise any exceptions. """ # Create Flavor - flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, vcpus=1) + flavor_settings = FlavorSettings(name=self.flavor_name, ram=1, disk=1, + vcpus=1) self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings) flavor = self.flavor_creator.create() self.assertTrue(validate_flavor(flavor_settings, flavor)) # Delete Flavor nova_utils.delete_flavor(self.nova, flavor) - self.assertIsNone(nova_utils.get_flavor_by_name(self.nova, flavor_settings.name)) + self.assertIsNone( + nova_utils.get_flavor_by_name(self.nova, flavor_settings.name)) # Attempt to cleanup self.flavor_creator.clean() @@ -317,7 +363,7 @@ def validate_flavor(flavor_settings, flavor): :param flavor: the OpenStack flavor object """ return flavor is not None \ - and flavor_settings.name == flavor.name \ - and flavor_settings.ram == flavor.ram \ - and flavor_settings.disk == flavor.disk \ - and flavor_settings.vcpus == flavor.vcpus + and flavor_settings.name == flavor.name \ + and flavor_settings.ram == flavor.ram \ + and flavor_settings.disk == flavor.disk \ + and flavor_settings.vcpus == flavor.vcpus diff --git a/snaps/openstack/tests/create_image_tests.py b/snaps/openstack/tests/create_image_tests.py index 7cb46e4..cadbe5c 100644 --- a/snaps/openstack/tests/create_image_tests.py +++ b/snaps/openstack/tests/create_image_tests.py @@ -19,18 +19,19 @@ except ImportError: from urllib2 import URLError import logging -import os import shutil -import uuid import unittest +import uuid -from snaps import file_utils -from snaps.openstack.create_image import ImageSettings, ImageCreationError, ImageSettingsError +import os -from snaps.openstack.tests import openstack_tests -from snaps.openstack.utils import glance_utils +from snaps import file_utils from snaps.openstack import create_image +from snaps.openstack.create_image import (ImageSettings, ImageCreationError, + ImageSettingsError) +from snaps.openstack.tests import openstack_tests from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase +from snaps.openstack.utils import glance_utils __author__ = 'spisarski' @@ -41,13 +42,14 @@ class ImageSettingsUnitTests(unittest.TestCase): """ Tests the construction of the ImageSettings class """ + def test_no_params(self): with self.assertRaises(ImageSettingsError): ImageSettings() def test_empty_config(self): with self.assertRaises(ImageSettingsError): - ImageSettings(config=dict()) + ImageSettings(**dict()) def test_name_only(self): with self.assertRaises(ImageSettingsError): @@ -55,7 +57,7 @@ class ImageSettingsUnitTests(unittest.TestCase): def test_config_with_name_only(self): with self.assertRaises(ImageSettingsError): - ImageSettings(config={'name': 'foo'}) + ImageSettings(**{'name': 'foo'}) def test_name_user_only(self): with self.assertRaises(ImageSettingsError): @@ -63,7 +65,7 @@ class ImageSettingsUnitTests(unittest.TestCase): def test_config_with_name_user_only(self): with self.assertRaises(ImageSettingsError): - ImageSettings(config={'name': 'foo', 'image_user': 'bar'}) + ImageSettings(**{'name': 'foo', 'image_user': 'bar'}) def test_name_user_format_only(self): with self.assertRaises(ImageSettingsError): @@ -71,20 +73,25 @@ class ImageSettingsUnitTests(unittest.TestCase): def test_config_with_name_user_format_only(self): with self.assertRaises(ImageSettingsError): - ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2'}) + ImageSettings( + **{'name': 'foo', 'image_user': 'bar', 'format': 'qcow2'}) def test_name_user_format_url_file_only(self): with self.assertRaises(ImageSettingsError): - ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com', + 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(ImageSettingsError): - ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', - 'download_url': 'http://foo.com', 'image_file': '/foo/bar.qcow'}) + ImageSettings( + **{'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', + 'download_url': 'http://foo.com', + 'image_file': '/foo/bar.qcow'}) def test_name_user_format_url_only(self): - settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com') + settings = ImageSettings(name='foo', image_user='bar', + img_format='qcow2', url='http://foo.com') self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) self.assertEqual('qcow2', settings.format) @@ -96,7 +103,8 @@ class ImageSettingsUnitTests(unittest.TestCase): def test_name_user_format_url_only_properties(self): properties = {'hw_video_model': 'vga'} - settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com', + settings = ImageSettings(name='foo', image_user='bar', + img_format='qcow2', url='http://foo.com', extra_properties=properties) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) @@ -109,8 +117,9 @@ class ImageSettingsUnitTests(unittest.TestCase): self.assertIsNone(settings.nic_config_pb_loc) def test_config_with_name_user_format_url_only(self): - settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', - 'download_url': 'http://foo.com'}) + settings = ImageSettings( + **{'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', + 'download_url': 'http://foo.com'}) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) self.assertEqual('qcow2', settings.format) @@ -121,7 +130,9 @@ class ImageSettingsUnitTests(unittest.TestCase): self.assertIsNone(settings.nic_config_pb_loc) def test_name_user_format_file_only(self): - settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', image_file='/foo/bar.qcow') + settings = ImageSettings(name='foo', image_user='bar', + img_format='qcow2', + image_file='/foo/bar.qcow') self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) self.assertEqual('qcow2', settings.format) @@ -132,8 +143,9 @@ class ImageSettingsUnitTests(unittest.TestCase): self.assertIsNone(settings.nic_config_pb_loc) def test_config_with_name_user_format_file_only(self): - settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', - 'image_file': '/foo/bar.qcow'}) + settings = ImageSettings( + **{'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', + 'image_file': '/foo/bar.qcow'}) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) self.assertEqual('qcow2', settings.format) @@ -145,11 +157,17 @@ class ImageSettingsUnitTests(unittest.TestCase): def test_all_url(self): properties = {'hw_video_model': 'vga'} - kernel_settings = ImageSettings(name='kernel', url='http://kernel.com', image_user='bar', img_format='qcow2') - 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_settings = ImageSettings(name='kernel', url='http://kernel.com', + image_user='bar', img_format='qcow2') + 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, exists=True, public=True) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) @@ -159,11 +177,13 @@ class ImageSettingsUnitTests(unittest.TestCase): self.assertIsNone(settings.image_file) self.assertEqual('/foo/bar', settings.nic_config_pb_loc) self.assertEqual('kernel', settings.kernel_image_settings.name) - self.assertEqual('http://kernel.com', settings.kernel_image_settings.url) + self.assertEqual('http://kernel.com', + settings.kernel_image_settings.url) self.assertEqual('bar', settings.kernel_image_settings.image_user) self.assertEqual('qcow2', settings.kernel_image_settings.format) self.assertEqual('ramdisk', settings.ramdisk_image_settings.name) - self.assertEqual('http://ramdisk.com', settings.ramdisk_image_settings.url) + 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) @@ -171,33 +191,46 @@ class ImageSettingsUnitTests(unittest.TestCase): def test_config_all_url(self): settings = ImageSettings( - config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', - 'download_url': 'http://foo.com', - 'extra_properties': '{\'hw_video_model\': \'vga\'}', - 'nic_config_pb_loc': '/foo/bar', - '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'}, - 'exists': True, 'public': True}) + **{'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', + 'download_url': 'http://foo.com', + 'extra_properties': '{\'hw_video_model\': \'vga\'}', + 'nic_config_pb_loc': '/foo/bar', + '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'}, + 'exists': True, 'public': True}) self.assertEqual('foo', settings.name) self.assertEqual('bar', settings.image_user) self.assertEqual('qcow2', settings.format) self.assertEqual('http://foo.com', settings.url) - self.assertEqual('{\'hw_video_model\': \'vga\'}', settings.extra_properties) + self.assertEqual('{\'hw_video_model\': \'vga\'}', + settings.extra_properties) self.assertIsNone(settings.image_file) self.assertEqual('/foo/bar', settings.nic_config_pb_loc) self.assertEqual('kernel', settings.kernel_image_settings.name) - self.assertEqual('http://kernel.com', settings.kernel_image_settings.url) + 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.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', exists=True, public=True) + settings = ImageSettings(name='foo', image_user='bar', + img_format='qcow2', + image_file='/foo/bar.qcow', + 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) @@ -209,16 +242,19 @@ class ImageSettingsUnitTests(unittest.TestCase): 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', 'exists': True, 'public': True}) + settings = ImageSettings( + **{'name': 'foo', 'image_user': 'bar', 'format': 'qcow2', + 'image_file': '/foo/bar.qcow', + 'extra_properties': '{\'hw_video_model\' : \'vga\'}', + '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.assertIsNone(settings.url) self.assertEqual('/foo/bar.qcow', settings.image_file) - self.assertEqual('{\'hw_video_model\' : \'vga\'}', settings.extra_properties) + 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) @@ -231,8 +267,8 @@ class CreateImageSuccessTests(OSIntegrationTestCase): def setUp(self): """ - Instantiates the CreateImage object that is responsible for downloading and creating an OS image file - within OpenStack + Instantiates the CreateImage object that is responsible for downloading + and creating an OS image file within OpenStack """ super(self.__class__, self).__start__() @@ -250,8 +286,9 @@ class CreateImageSuccessTests(OSIntegrationTestCase): 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=glance_test_meta) + self.image_settings = openstack_tests.cirros_image_settings( + name=self.image_name, + image_metadata=glance_test_meta) def tearDown(self): """ @@ -270,16 +307,20 @@ class CreateImageSuccessTests(OSIntegrationTestCase): Tests the creation of an OpenStack image from a URL. """ # Create Image - # Set the default image settings, then set any custom parameters sent from the app + # Set the default image settings, then set any custom parameters sent + # from the app - self.image_creator = create_image.OpenStackImage(self.os_creds, self.image_settings) + self.image_creator = create_image.OpenStackImage(self.os_creds, + self.image_settings) created_image = self.image_creator.create() self.assertIsNotNone(created_image) - retrieved_image = glance_utils.get_image(self.glance, self.image_settings.name) + retrieved_image = glance_utils.get_image(self.glance, + self.image_settings.name) self.assertIsNotNone(retrieved_image) self.assertEqual(created_image.size, retrieved_image.size) - self.assertEqual(get_image_size(self.image_settings), retrieved_image.size) + self.assertEqual(get_image_size(self.image_settings), + retrieved_image.size) self.assertEqual(created_image.name, retrieved_image.name) self.assertEqual(created_image.id, retrieved_image.id) @@ -288,15 +329,20 @@ class CreateImageSuccessTests(OSIntegrationTestCase): Tests the creation of an OpenStack image from a URL and set properties. """ # Create Image - # Set the default image settings, then set any custom parameters sent from the app - self.image_creator = create_image.OpenStackImage(self.os_creds, self.image_settings) + # Set the default image settings, then set any custom parameters sent + # from the app + self.image_creator = create_image.OpenStackImage(self.os_creds, + self.image_settings) created_image = self.image_creator.create() self.assertIsNotNone(created_image) - retrieved_image = glance_utils.get_image(self.glance, self.image_settings.name) + 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(self.image_creator.get_image().size, + retrieved_image.size) + self.assertEqual(get_image_size(self.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) @@ -307,7 +353,8 @@ class CreateImageSuccessTests(OSIntegrationTestCase): """ 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 + image_file_name = file_utils.download(self.image_settings.url, + self.tmp_dir).name else: image_file_name = self.image_settings.image_file @@ -315,41 +362,53 @@ class CreateImageSuccessTests(OSIntegrationTestCase): 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) + 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) + 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(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) else: - logger.warn('Test not executed as the image metadata requires image files') + logger.warn( + 'Test not executed as the image metadata requires image files') def test_create_delete_image(self): """ - Tests the creation then deletion of an OpenStack image to ensure clean() does not raise an Exception. + Tests the creation then deletion of an OpenStack image to ensure + clean() does not raise an Exception. """ # Create Image - self.image_creator = create_image.OpenStackImage(self.os_creds, self.image_settings) + self.image_creator = create_image.OpenStackImage(self.os_creds, + self.image_settings) created_image = self.image_creator.create() self.assertIsNotNone(created_image) - retrieved_image = glance_utils.get_image(self.glance, self.image_settings.name) + 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(self.image_creator.get_image().size, + retrieved_image.size) + self.assertEqual(get_image_size(self.image_settings), + retrieved_image.size) # Delete Image manually glance_utils.delete_image(self.glance, created_image) - self.assertIsNone(glance_utils.get_image(self.glance, self.image_creator.image_settings.name)) + self.assertIsNone(glance_utils.get_image( + self.glance, self.image_creator.image_settings.name)) - # Must not throw an exception when attempting to cleanup non-existent image + # Must not throw an exception when attempting to cleanup non-existent + # image self.image_creator.clean() self.assertIsNone(self.image_creator.get_image()) @@ -358,42 +417,53 @@ class CreateImageSuccessTests(OSIntegrationTestCase): Tests the creation of an OpenStack image when the image already exists. """ # Create Image - self.image_creator = create_image.OpenStackImage(self.os_creds, self.image_settings) + 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) + 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(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 - os_image_2 = create_image.OpenStackImage(self.os_creds, self.image_settings) + os_image_2 = create_image.OpenStackImage(self.os_creds, + self.image_settings) 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. + 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) + 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) + 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(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) + 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) @@ -417,24 +487,32 @@ class CreateImageNegativeTests(OSIntegrationTestCase): def test_bad_image_name(self): """ - Expect an ImageCreationError when the image name does not exist when a file or URL has not been configured + Expect an ImageCreationError when the image name does not exist when a + file or URL has not been configured """ - os_image_settings = ImageSettings(name='foo', image_user='bar', exists=True) - self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings) + os_image_settings = ImageSettings(name='foo', image_user='bar', + exists=True) + self.image_creator = create_image.OpenStackImage(self.os_creds, + os_image_settings) with self.assertRaises(ImageCreationError): self.image_creator.create() - self.fail('ImageCreationError should have been raised prior to this line') + self.fail('ImageCreationError should have been raised prior to' + 'this line') def test_bad_image_url(self): """ 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")) + 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(URLError): self.image_creator.create() @@ -443,11 +521,14 @@ class CreateImageNegativeTests(OSIntegrationTestCase): """ Expect an ImageCreationError when the image file does not exist """ - os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name) + 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")) + 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(IOError): self.image_creator.create() @@ -456,6 +537,7 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): """ Test different means for creating a 3-part images """ + def setUp(self): """ Instantiates the CreateImage object that is responsible for @@ -497,30 +579,41 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): 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) + image_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_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) + 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) + 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) + 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) + 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) + self.assertEqual(ramdisk_image.id, + image_creator.get_ramdisk_image().id) else: - logger.warn('Test not executed as the image metadata requires image files') + logger.warn( + 'Test not executed as the image metadata requires image files') def test_create_three_part_image_from_file_3_creators(self): """ @@ -540,18 +633,21 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): 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' + kernel_url = openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL if not kernel_file_name: - kernel_file_name = file_utils.download(kernel_url, self.tmp_dir).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_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)) + 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) + self.assertEqual(get_image_size(kernel_file_image_settings), + kernel_image.size) # Create the ramdisk image ramdisk_file_name = None @@ -562,14 +658,17 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): 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_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_file_name) - self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_file_image_settings)) + 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) + self.assertEqual(get_image_size(ramdisk_file_image_settings), + ramdisk_image.size) # Create the main disk image disk_file_name = None @@ -582,20 +681,25 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): 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) + 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 - self.image_creators.append(create_image.OpenStackImage(self.os_creds, file_image_settings)) + self.image_creators.append( + create_image.OpenStackImage(self.os_creds, file_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, file_image_settings.name) + retrieved_image = glance_utils.get_image(self.glance, + file_image_settings.name) self.assertIsNotNone(retrieved_image) - self.assertEqual(self.image_creators[-1].get_image().size, retrieved_image.size) - self.assertEqual(get_image_size(file_image_settings), retrieved_image.size) + self.assertEqual(self.image_creators[-1].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.properties, retrieved_image.properties) @@ -607,38 +711,47 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): 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: + 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') + name=self.image_name + '_kernel', + url=openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL) 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_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) + 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') + name=self.image_name + '_ramdisk', + url=openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL) 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_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) + 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') + url=openstack_tests.CIRROS_DEFAULT_IMAGE_URL) if self.glance_test_meta: if 'disk_url' in self.glance_test_meta: os_image_settings.url = self.glance_test_meta['disk_url'] @@ -647,22 +760,28 @@ class CreateMultiPartImageTests(OSIntegrationTestCase): 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)) + 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) + 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(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) + self.assertEqual(created_image.properties, + retrieved_image.properties) else: - logger.warn('Test not executed as the image metadata requires image files') + logger.warn( + 'Test not executed as the image metadata requires image files') def get_image_size(image_settings): @@ -675,4 +794,6 @@ def get_image_size(image_settings): elif image_settings.url: return int(file_utils.get_content_length(image_settings.url)) else: - raise Exception('Cannot retrieve expected image size. Image filename or URL has not been configured') + raise Exception( + 'Cannot retrieve expected image size. Image filename or URL has ' + 'not been configured') |