self.enabled = True
if not self.name:
- raise Exception(
+ raise ProjectSettingsError(
"The attribute name is required for ProjectSettings")
+
+
+class ProjectSettingsError(Exception):
+ """
+ Exception to be thrown when project settings attributes are incorrect
+ """
self.__os_creds = os_creds
if not router_settings:
- raise Exception('router_settings is required')
+ raise RouterCreationError('router_settings is required')
self.router_settings = router_settings
self.__neutron = None
self.__internal_router_interface = neutron_utils.add_interface_router(
self.__neutron, self.__router, subnet=internal_subnet)
else:
- raise Exception(
+ raise RouterCreationError(
'Subnet not found with name ' + internal_subnet_name)
for port_setting in self.router_settings.port_settings:
self.__router,
port=port)
else:
- raise Exception(
+ raise RouterCreationError(
'Error creating port with name - ' + port_setting.name)
return self.__router
return self.__internal_router_interface
+class RouterCreationError(Exception):
+ """
+ Exception to be thrown when an router instance cannot be created
+ """
+
+
class RouterSettings:
"""
Class representing a router configuration
PortSettings(**interface['port']))
if not self.name:
- raise Exception('Name is required')
+ raise RouterSettingsError('Name is required')
def dict_for_neutron(self, neutron, os_creds):
"""
if project_id:
out['project_id'] = project_id
else:
- raise Exception(
+ raise RouterSettingsError(
'Could not find project ID for project named - ' +
self.project_name)
if self.admin_state_up is not None:
ext_gw['network_id'] = ext_net.id
out['external_gateway_info'] = ext_gw
else:
- raise Exception(
+ raise RouterSettingsError(
'Could not find the external network named - ' +
self.external_gateway)
# TODO: Add external_fixed_ips Tests
return {'router': out}
+
+
+class RouterSettingsError(Exception):
+ """
+ Exception to be thrown when router settings attributes are incorrect
+ """
**rule_setting))
if not self.name:
- raise Exception('The attribute name is required')
+ raise SecurityGroupSettingsError('The attribute name is required')
for rule_setting in self.rule_settings:
if rule_setting.sec_grp_name is not self.name:
- raise Exception(
+ raise SecurityGroupSettingsError(
'Rule settings must correspond with the name of this '
'security group')
if project_id:
out['project_id'] = project_id
else:
- raise Exception(
+ raise SecurityGroupSettingsError(
'Could not find project ID for project named - ' +
self.project_name)
IPv6 = 6
+class SecurityGroupSettingsError(Exception):
+ """
+ Exception to be thrown when security group settings attributes are
+ invalid
+ """
+
+
class SecurityGroupRuleSettings:
"""
Class representing a keypair configuration
self.remote_ip_prefix = kwargs.get('remote_ip_prefix')
if not self.direction or not self.sec_grp_name:
- raise Exception('direction and sec_grp_name are required')
+ raise SecurityGroupRuleSettingsError(
+ 'direction and sec_grp_name are required')
def dict_for_neutron(self, neutron):
"""
if sec_grp:
out['security_group_id'] = sec_grp.id
else:
- raise Exception(
+ raise SecurityGroupRuleSettingsError(
'Cannot locate security group with name - ' +
self.sec_grp_name)
if self.remote_group_id:
elif dir_str == 'ingress':
return Direction.ingress
else:
- raise Exception('Invalid Direction - ' + dir_str)
+ raise SecurityGroupRuleSettingsError(
+ 'Invalid Direction - ' + dir_str)
def map_protocol(protocol):
elif proto_str == 'null':
return Protocol.null
else:
- raise Exception('Invalid Protocol - ' + proto_str)
+ raise SecurityGroupRuleSettingsError(
+ 'Invalid Protocol - ' + proto_str)
def map_ethertype(ethertype):
elif eth_str == 'IPv4':
return Ethertype.IPv4
else:
- raise Exception('Invalid Ethertype - ' + eth_str)
+ raise SecurityGroupRuleSettingsError(
+ 'Invalid Ethertype - ' + eth_str)
+
+
+class SecurityGroupRuleSettingsError(Exception):
+ """
+ Exception to be thrown when security group rule settings attributes are
+ invalid
+ """
import unittest
import uuid
+from snaps.openstack import create_flavor
from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor, \
FlavorSettingsError
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
vcpus=1)
self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings)
flavor = self.flavor_creator.create()
- self.assertTrue(validate_flavor(flavor_settings, flavor))
+ self.assertTrue(validate_flavor(self.nova, flavor_settings, flavor))
def test_create_flavor_existing(self):
"""
vcpus=1)
self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings)
flavor = self.flavor_creator.create()
- self.assertTrue(validate_flavor(flavor_settings, flavor))
+ self.assertTrue(validate_flavor(self.nova, flavor_settings, flavor))
flavor_creator_2 = OpenStackFlavor(self.os_creds, flavor_settings)
flavor2 = flavor_creator_2.create()
vcpus=1)
self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings)
flavor = self.flavor_creator.create()
- self.assertTrue(validate_flavor(flavor_settings, flavor))
+ self.assertTrue(validate_flavor(self.nova, flavor_settings, flavor))
# Clean Flavor
self.flavor_creator.clean()
def test_create_delete_flavor(self):
"""
- Tests the creation of an OpenStack Security Group, the deletion, then
+ Tests the creation of an OpenStack Flavor, the deletion, then
cleanup to ensure clean() does not
raise any exceptions.
"""
vcpus=1)
self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings)
flavor = self.flavor_creator.create()
- self.assertTrue(validate_flavor(flavor_settings, flavor))
+ self.assertTrue(validate_flavor(self.nova, flavor_settings, flavor))
# Delete Flavor
nova_utils.delete_flavor(self.nova, flavor)
self.assertIsNone(self.flavor_creator.get_flavor())
- # TODO - Add more tests to exercise all configuration options
+ def test_create_flavor_all_settings(self):
+ """
+ Tests the creation of an OpenStack Flavor, 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, ephemeral=2, swap=3,
+ rxtx_factor=2.2, is_public=False,
+ metadata=create_flavor.MEM_PAGE_SIZE_ANY)
+ self.flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings)
+ flavor = self.flavor_creator.create()
+ self.assertTrue(validate_flavor(self.nova, 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))
+
+ # Attempt to cleanup
+ self.flavor_creator.clean()
+
+ self.assertIsNone(self.flavor_creator.get_flavor())
-def validate_flavor(flavor_settings, flavor):
+def validate_flavor(nova, flavor_settings, flavor):
"""
Validates the flavor_settings against the OpenStack flavor object
+ :param nova: the nova client
:param flavor_settings: the settings used to create the 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
+ setting_meta = dict()
+ if flavor_settings.metadata:
+ setting_meta = flavor_settings.metadata
+ metadata = nova_utils.get_flavor_keys(nova, flavor)
+
+ equals = True
+ for key, value in setting_meta.items():
+ if metadata[key] != value:
+ equals = False
+ break
+
+ swap = str()
+ if flavor_settings.swap != 0:
+ swap = flavor_settings.swap
+
+ 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.ephemeral == flavor.ephemeral and
+ swap == flavor.swap and
+ flavor_settings.rxtx_factor == flavor.rxtx_factor and
+ flavor_settings.is_public == flavor.is_public and
+ equals)
# 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.
+from glanceclient.exc import HTTPBadRequest
try:
from urllib.request import URLError
img_format=os_image_settings.format,
url="http://foo.bar"))
- with self.assertRaises(URLError):
+ try:
+ self.image_creator.create()
+ except HTTPBadRequest:
+ pass
+ except URLError:
+ pass
+ except Exception as e:
+ self.fail('Invalid Exception ' + str(e))
+
+ def test_bad_image_image_type(self):
+ """
+ Expect an ImageCreationError when the image type 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='foo',
+ url=os_image_settings.url))
+
+ with self.assertRaises(Exception):
self.image_creator.create()
def test_bad_image_file(self):
import unittest
import uuid
-from snaps.openstack.create_project import OpenStackProject, ProjectSettings
+from snaps.openstack.create_project import (
+ OpenStackProject, ProjectSettings, ProjectSettingsError)
from snaps.openstack.create_security_group import OpenStackSecurityGroup
from snaps.openstack.create_security_group import SecurityGroupSettings
from snaps.openstack.create_user import OpenStackUser
"""
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ProjectSettingsError):
ProjectSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(ProjectSettingsError):
ProjectSettings(**dict())
def test_name_only(self):
-# 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");
from snaps.openstack.create_network import (
NetworkSettings, PortSettings)
from snaps.openstack.create_network import OpenStackNetwork
-from snaps.openstack.create_router import RouterSettings
+from snaps.openstack.create_router import (
+ RouterSettings, RouterSettingsError, RouterCreationError)
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import neutron_utils
"""
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(RouterSettingsError):
RouterSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(RouterSettingsError):
RouterSettings(**dict())
def test_name_only(self):
"""
Test creating a router without a name.
"""
- with self.assertRaises(Exception):
+ with self.assertRaises(RouterSettingsError):
router_settings = RouterSettings(
name=None, external_gateway=self.ext_net_name)
self.router_creator = create_router.OpenStackRouter(
"""
Test creating a router without a valid network gateway name.
"""
- with self.assertRaises(Exception):
+ with self.assertRaises(RouterSettingsError):
router_settings = RouterSettings(name=self.guid + '-pub-router',
external_gateway="Invalid_name")
self.router_creator = create_router.OpenStackRouter(
from snaps.openstack.create_security_group import (SecurityGroupSettings,
SecurityGroupRuleSettings,
Direction, Ethertype,
- Protocol)
+ Protocol,
+ SecurityGroupRuleSettingsError,
+ SecurityGroupSettingsError)
from snaps.openstack.tests import validation_utils
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import neutron_utils
"""
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupRuleSettingsError):
SecurityGroupRuleSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupRuleSettingsError):
SecurityGroupRuleSettings(**dict())
def test_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupRuleSettingsError):
SecurityGroupRuleSettings(sec_grp_name='foo')
def test_config_with_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupRuleSettingsError):
SecurityGroupRuleSettings(**{'sec_grp_name': 'foo'})
def test_name_and_direction(self):
"""
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupSettingsError):
SecurityGroupSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupSettingsError):
SecurityGroupSettings(**dict())
def test_name_only(self):
def test_invalid_rule(self):
rule_setting = SecurityGroupRuleSettings(sec_grp_name='bar',
direction=Direction.ingress)
- with self.assertRaises(Exception):
+ with self.assertRaises(SecurityGroupSettingsError):
SecurityGroupSettings(name='foo', rule_settings=[rule_setting])
def test_all(self):
return None
+def get_image_by_id(glance, image_id):
+ """
+ Returns an OpenStack image object for a given name
+ :param glance: the Glance client
+ :param image_id: the image ID to lookup
+ :return: the SNAPS-OO Domain Image object or None
+ """
+ image = glance.images.get(image_id)
+ return Image(
+ name=image['name'], image_id=image['id'],
+ size=image['size'], properties=image.get('properties'))
+
+
def get_image_status(glance, image):
"""
Returns a new OpenStack Image object for a given OpenStack image object
:return: the OpenStack image object
:raise exceptions from the Glance client or IOError when opening a file
"""
- created_image = None
+ kwargs = {
+ 'name': image_settings.name, 'disk_format': image_settings.format,
+ 'container_format': 'bare', 'is_public': image_settings.public}
+
+ if image_settings.extra_properties:
+ kwargs['properties'] = image_settings.extra_properties
- # 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,
- 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,
- is_public=image_settings.public)
+ kwargs['location'] = image_settings.url
elif 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,
- 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,
- is_public=image_settings.public)
+ kwargs['data'] = image_file
+ else:
+ logger.warn('Unable to create image with name - %s. No file or URL',
+ image_settings.name)
+ return None
+ created_image = glance.images.create(**kwargs)
return Image(name=image_settings.name, image_id=created_image.id,
size=created_image.size, properties=created_image.properties)
image_settings.url, './tmp', file_name)
image_filename = image_file.name
except:
- os.remove('./tmp/' + file_name)
+ if image_file:
+ os.remove(image_file.name)
raise
cleanup_temp_file = True
else:
raise GlanceException('Filename or URL of image not configured')
- created_image = None
+ os_image = None
try:
kwargs = dict()
kwargs['name'] = image_settings.name
if image_settings.extra_properties:
kwargs.update(image_settings.extra_properties)
- created_image = glance.images.create(**kwargs)
+ os_image = glance.images.create(**kwargs)
image_file = open(image_filename, 'rb')
- glance.images.upload(created_image['id'], image_file)
+ glance.images.upload(os_image['id'], image_file)
except:
logger.error('Unexpected exception creating image. Rolling back')
- if created_image:
- delete_image(glance, created_image)
+ if os_image:
+ delete_image(glance, Image(
+ name=os_image['name'], image_id=os_image['id'],
+ size=os_image['size'], properties=os_image.get('properties')))
raise
finally:
if image_file:
+ logger.debug('Closing file %s', image_file.name)
image_file.close()
if cleanup_temp_file:
+ logger.info('Removing file %s', image_file.name)
os.remove(image_filename)
- updated_image = glance.images.get(created_image['id'])
- return Image(
- name=updated_image['name'], image_id=updated_image['id'],
- size=updated_image['size'], properties=updated_image.get('properties'))
+ return get_image_by_id(glance, os_image['id'])
def delete_image(glance, image):
:param glance: the glance client
:param image: the image to delete
"""
- if glance.version == VERSION_1:
- glance.images.delete(image)
- elif glance.version == VERSION_2:
- glance.images.delete(image.id)
- else:
- raise GlanceException('Unsupported glance client version')
+ logger.info('Deleting image named - %s', image.name)
+ glance.images.delete(image.id)
class GlanceException(Exception):
os_flavor.set_keys(metadata)
+def get_flavor_keys(nova, flavor):
+ """
+ Sets metadata on the flavor
+ :param nova: the Nova client
+ :param flavor: the SNAPS flavor domain object
+ """
+ os_flavor = __get_os_flavor(nova, flavor)
+ if os_flavor:
+ return os_flavor.get_keys()
+
+
def add_security_group(nova, vm, security_group_name):
"""
Adds a security group to an existing VM
import uuid
from snaps import file_utils
+from snaps.openstack.tests import openstack_tests
__author__ = 'spisarski'
def setUp(self):
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.tmpDir = 'tmp/' + str(guid)
- if not os.path.exists(self.tmpDir):
- os.makedirs(self.tmpDir)
+ self.tmp_dir = '.tmp/'
+ self.test_dir = self.tmp_dir + str(guid)
+ if not os.path.exists(self.test_dir):
+ os.makedirs(self.test_dir)
- self.tmpFile = self.tmpDir + '/bar.txt'
+ self.tmpFile = self.test_dir + '/bar.txt'
if not os.path.exists(self.tmpFile):
open(self.tmpFile, 'wb')
def tearDown(self):
- if os.path.exists(self.tmpDir) and os.path.isdir(self.tmpDir):
- shutil.rmtree(self.tmpDir)
+ if os.path.exists(self.test_dir) and os.path.isdir(self.test_dir):
+ shutil.rmtree(self.tmp_dir)
def testFileIsDirectory(self):
"""
- Ensure the file_utils.fileExists() method returns false with a directory
+ Ensure the file_utils.fileExists() method returns false with a
+ directory
"""
- result = file_utils.file_exists(self.tmpDir)
+ result = file_utils.file_exists(self.test_dir)
self.assertFalse(result)
- # TODO - Cleanup directory
def testFileNotExist(self):
"""
- Ensure the file_utils.fileExists() method returns false with a bogus file
+ Ensure the file_utils.fileExists() method returns false with a bogus
+ file
"""
result = file_utils.file_exists('/foo/bar.txt')
self.assertFalse(result)
def testFileExists(self):
"""
- Ensure the file_utils.fileExists() method returns false with a directory
+ Ensure the file_utils.fileExists() method returns false with a
+ directory
"""
if not os.path.exists(self.tmpFile):
os.makedirs(self.tmpFile)
Tests the file_utils.download() method when given a bad URL
"""
with self.assertRaises(Exception):
- file_utils.download('http://bunkUrl.com/foo/bar.iso', self.tmpDir)
+ file_utils.download('http://bunkUrl.com/foo/bar.iso',
+ self.test_dir)
def testDownloadBadDir(self):
"""
Tests the file_utils.download() method when given a bad URL
"""
with self.assertRaises(Exception):
- file_utils.download('http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img', '/foo/bar')
+ file_utils.download(openstack_tests.CIRROS_DEFAULT_IMAGE_URL,
+ '/foo/bar')
def testCirrosImageDownload(self):
"""
- Tests the file_utils.download() method when given a good Cirros QCOW2 URL
+ Tests the file_utils.download() method when given a good Cirros QCOW2
+ URL
"""
- image_file = file_utils.download('http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img',
- self.tmpDir)
+ image_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_IMAGE_URL, self.test_dir)
self.assertIsNotNone(image_file)
- self.assertTrue(image_file.name.endswith("cirros-0.3.4-x86_64-disk.img"))
- self.assertTrue(image_file.name.startswith(self.tmpDir))
+ self.assertTrue(
+ image_file.name.endswith("cirros-0.3.4-x86_64-disk.img"))
+ self.assertTrue(image_file.name.startswith(self.test_dir))
def testReadOSEnvFile(self):
"""
Tests that the OS Environment file is correctly parsed
:return:
"""
- rc_file_path = pkg_resources.resource_filename('snaps.openstack.tests.conf', 'overcloudrc_test')
+ rc_file_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.conf', 'overcloudrc_test')
os_env_dict = file_utils.read_os_env_file(rc_file_path)
self.assertEqual('test_pw', os_env_dict['OS_PASSWORD'])
self.assertEqual('http://foo:5000/v2.0/', os_env_dict['OS_AUTH_URL'])
Code Review / snaps.git / commitdiff