| | Neutron 2 | port and it's assigned IP address |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_instance_tests.py - CreateInstanceTwoNetTests
+----------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | API Versions | Description |
++=======================================+===============+===========================================================+
+| test_ping_via_router | Nova 2 | Tests the ability of two VMs on different private overlay |
+| | Neutron 2 | networks tied together with a router to ping each other |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_instance_tests.py - CreateInstanceSingleNetworkTests
-----------------------------------------------------------
Ensures that all required members are included when constructing a
Project domain object
+DomainDomainObjectTests
+-----------------------
+
+Ensures that all required members are included when constructing a
+Domain domain object
+
RoleDomainObjectTests
---------------------
# Flavors
from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-flavor_settings = FlavorSettings(name='test-flavor', ram=128, disk=10, vcpus=2)
+flavor_settings = FlavorSettings(name='test-flavor', ram=256, disk=10, vcpus=2)
flavor = OpenStackFlavor(os_creds, flavor_settings)
flavor.create()
def __eq__(self, other):
return self.name == other.name and self.id == other.id
+
+
+class Domain:
+ """
+ SNAPS domain object for OpenStack Keystone v3+ domains.
+ """
+ def __init__(self, name, domain_id=None):
+ """
+ Constructor
+ :param name: the project's name
+ :param domain_id: the project's domain id
+ """
+ self.name = name
+ self.id = domain_id
+
+ def __eq__(self, other):
+ return self.name == other.name and self.id == other.id
"""
self.name = name
self.id = stack_id
+
+ def __eq__(self, other):
+ return (self.name == other.name and
+ self.id == other.id)
# limitations under the License.
import unittest
-from snaps.domain.project import Project
+from snaps.domain.project import Project, Domain
class ProjectDomainObjectTests(unittest.TestCase):
"""
- Tests the construction of the snaps.domain.test.Project class
+ Tests the construction of the snaps.domain.project.Project class
"""
def test_construction_positional_minimal(self):
self.assertEqual('foo', project.name)
self.assertEqual('123-456', project.id)
self.assertEqual('hello', project.domain_id)
+
+
+class DomainDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.project.Domain class
+ """
+
+ def test_construction_positional(self):
+ domain = Domain('foo', '123-456')
+ self.assertEqual('foo', domain.name)
+ self.assertEqual('123-456', domain.id)
+
+ def test_construction_named_minimal(self):
+ domain = Domain(domain_id='123-456', name='foo')
+ self.assertEqual('foo', domain.name)
+ self.assertEqual('123-456', domain.id)
: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, image_settings=self.image_settings)
if self.__image:
logger.info('Found image with name - ' + self.image_settings.name)
return self.__image
if self.image_settings.kernel_image_settings:
self.__kernel_image = glance_utils.get_image(
self.__glance,
- self.image_settings.kernel_image_settings.name)
+ image_settings=self.image_settings.kernel_image_settings)
if not self.__kernel_image and not cleanup:
logger.info(
if self.image_settings.ramdisk_image_settings:
self.__ramdisk_image = glance_utils.get_image(
self.__glance,
- self.image_settings.ramdisk_image_settings.name)
+ image_settings=self.image_settings.ramdisk_image_settings)
if not self.__ramdisk_image and not cleanup:
logger.info(
VM with the same name already exists
within the project
"""
- servers = nova_utils.get_servers_by_name(self.__nova,
- self.instance_settings.name)
- for server in servers:
+ server = nova_utils.get_server(
+ self.__nova, vm_inst_settings=self.instance_settings)
+ if server:
if server.name == self.instance_settings.name:
self.__vm = server
logger.info(
ext_gateway = self.__ext_gateway_by_router(
floating_ip_setting.router_name)
if ext_gateway:
- subnet = neutron_utils.get_subnet_by_name(
- self.__neutron, floating_ip_setting.subnet_name)
+ subnet = neutron_utils.get_subnet(
+ self.__neutron,
+ subnet_name=floating_ip_setting.subnet_name)
floating_ip = neutron_utils.create_floating_ip(
self.__neutron, ext_gateway)
self.__floating_ip_dict[floating_ip_setting.name] = floating_ip
:param router_name: The name of the router to lookup
:return: the external network name or None
"""
- router = neutron_utils.get_router_by_name(self.__neutron, router_name)
+ router = neutron_utils.get_router(
+ self.__neutron, router_name=router_name)
if router and router.external_gateway_info:
network = neutron_utils.get_network_by_id(
self.__neutron,
ports = list()
for port_setting in port_settings:
- port = neutron_utils.get_port_by_name(self.__neutron,
- port_setting.name)
+ port = neutron_utils.get_port(
+ self.__neutron, port_settings=port_setting)
if port:
ports.append((port_setting.name, port))
elif not cleanup:
port = self.get_port_by_name(port_name)
if port:
if subnet_name:
- subnet = neutron_utils.get_subnet_by_name(self.__neutron,
- subnet_name)
+ subnet = neutron_utils.get_subnet(
+ self.__neutron, subnet_name=subnet_name)
if not subnet:
logger.warning('Cannot retrieve port IP as subnet could '
'not be located with name - %s',
logger.info(
'Creating neutron network %s...' % self.network_settings.name)
net_inst = neutron_utils.get_network(
- self.__neutron, self.network_settings.name,
- self.network_settings.get_project_id(self.__os_creds))
+ self.__neutron, network_settings=self.network_settings,
+ project_id=self.network_settings.get_project_id(self.__os_creds))
if net_inst:
self.__network = net_inst
else:
logger.debug('Creating Subnets....')
for subnet_setting in self.network_settings.subnet_settings:
- sub_inst = neutron_utils.get_subnet_by_name(
- self.__neutron, subnet_setting.name)
+ sub_inst = neutron_utils.get_subnet(
+ self.__neutron, subnet_settings=subnet_setting)
if sub_inst:
self.__subnets.append(sub_inst)
logger.debug(
self.fixed_ips = list()
for ip_addr_dict in self.ip_addrs:
- subnet = neutron_utils.get_subnet_by_name(neutron,
- ip_addr_dict[
- 'subnet_name'])
- if subnet:
+ subnet = neutron_utils.get_subnet(
+ neutron, subnet_name=ip_addr_dict['subnet_name'])
+ if subnet and 'ip' in ip_addr_dict:
self.fixed_ips.append({'ip_address': ip_addr_dict['ip'],
'subnet_id': subnet.id})
else:
project_id = project.id
if not self.network:
- self.network = neutron_utils.get_network(neutron,
- self.network_name,
- project_id)
+ self.network = neutron_utils.get_network(
+ neutron, network_name=self.network_name, project_id=project_id)
if not self.network:
raise PortSettingsError(
'Cannot locate network with name - ' + self.network_name)
import logging
from keystoneclient.exceptions import NotFound
-from snaps.openstack.utils import keystone_utils
+from snaps.openstack.utils import keystone_utils, neutron_utils
__author__ = 'spisarski'
"""
self.__keystone = keystone_utils.keystone_client(self.__os_creds)
self.__project = keystone_utils.get_project(
- keystone=self.__keystone, project_name=self.project_settings.name)
+ keystone=self.__keystone, project_settings=self.project_settings)
if self.__project:
logger.info(
'Found project with name - ' + self.project_settings.name)
:return: void
"""
if self.__project:
+ # Delete security group 'default' if exists
+ neutron = neutron_utils.neutron_client(self.__os_creds)
+ default_sec_grp = neutron_utils.get_security_group(
+ neutron, sec_grp_name='default',
+ project_id=self.__project.id)
+ if default_sec_grp:
+ try:
+ neutron_utils.delete_security_group(
+ neutron, default_sec_grp)
+ except:
+ pass
+
+ # Delete Project
try:
keystone_utils.delete_project(self.__keystone, self.__project)
except NotFound:
"""
Constructor
:param name: the project's name (required)
- :param domain: the project's domain name (default 'default'). Field is
- used for v3 clients
+ :param domain or domain_name: the project's domain name
+ (default = 'Default').
+ Field is used for v3 clients
:param description: the description (optional)
:param enabled: denotes whether or not the user is enabled
(default True)
"""
self.name = kwargs.get('name')
- if kwargs.get('domain'):
- self.domain = kwargs['domain']
- else:
- self.domain = 'default'
+ self.domain_name = kwargs.get(
+ 'domain', kwargs.get('domain', 'Default'))
self.description = kwargs.get('description')
if kwargs.get('enabled') is not None:
logger.debug(
'Creating Router with name - ' + self.router_settings.name)
existing = False
- router_inst = neutron_utils.get_router_by_name(
- self.__neutron, self.router_settings.name)
+ router_inst = neutron_utils.get_router(
+ self.__neutron, router_settings=self.router_settings)
if router_inst:
self.__router = router_inst
existing = True
self.__neutron, self.__os_creds, self.router_settings)
for internal_subnet_name in self.router_settings.internal_subnets:
- internal_subnet = neutron_utils.get_subnet_by_name(
- self.__neutron, internal_subnet_name)
+ internal_subnet = neutron_utils.get_subnet(
+ self.__neutron, subnet_name=internal_subnet_name)
if internal_subnet:
self.__internal_subnets.append(internal_subnet)
if internal_subnet and not cleanup and not existing:
'Subnet not found with name ' + internal_subnet_name)
for port_setting in self.router_settings.port_settings:
- port = neutron_utils.get_port_by_name(self.__neutron,
- port_setting.name)
+ port = neutron_utils.get_port(
+ self.__neutron, port_settings=port_setting)
logger.info(
'Retrieved port %s for router - %s', port_setting.name,
self.router_settings.name)
out = dict()
ext_gw = dict()
- project_id = None
-
if self.name:
out['name'] = self.name
if self.project_name:
if self.admin_state_up is not None:
out['admin_state_up'] = self.admin_state_up
if self.external_gateway:
- ext_net = neutron_utils.get_network(neutron, self.external_gateway,
- project_id)
+ ext_net = neutron_utils.get_network(
+ neutron, network_name=self.external_gateway)
if ext_net:
ext_gw['network_id'] = ext_net.id
out['external_gateway_info'] = ext_gw
'Creating security group %s...' % self.sec_grp_settings.name)
self.__security_group = neutron_utils.get_security_group(
- self.__neutron, self.sec_grp_settings.name)
+ self.__neutron, sec_grp_settings=self.sec_grp_settings)
if not self.__security_group and not cleanup:
# Create the security group
self.__security_group = neutron_utils.create_security_group(
# Refresh security group object to reflect the new rules added
self.__security_group = neutron_utils.get_security_group(
- self.__neutron, self.sec_grp_settings.name)
+ self.__neutron, sec_grp_settings=self.sec_grp_settings)
else:
# Populate rules
existing_rules = neutron_utils.get_rules_by_security_group(
out['protocol'] = self.protocol.name
if self.sec_grp_name:
sec_grp = neutron_utils.get_security_group(
- neutron, self.sec_grp_name)
+ neutron, sec_grp_name=self.sec_grp_name)
if sec_grp:
out['security_group_id'] = sec_grp.id
else:
:return: The OpenStack Stack object
"""
self.__heat_cli = heat_utils.heat_client(self.__os_creds)
- self.__stack = heat_utils.get_stack_by_name(self.__heat_cli,
- self.stack_settings.name)
+ self.__stack = heat_utils.get_stack(
+ self.__heat_cli, stack_settings=self.stack_settings)
if self.__stack:
logger.info('Found stack with name - ' + self.stack_settings.name)
return self.__stack
if not self.template and not self.template_path:
raise StackSettingsError('A Heat template is required')
+ def __eq__(self, other):
+ return (self.name == other.name and
+ self.template == other.template and
+ self.template_path == other.template_path and
+ self.env_values == other.env_values and
+ self.stack_create_timeout == other.stack_create_timeout)
+
class StackSettingsError(Exception):
"""
-# 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");
auth_url=self.__os_creds.auth_url,
project_name=project_name,
identity_api_version=self.__os_creds.identity_api_version,
+ user_domain_name=self.__os_creds.user_domain_name,
user_domain_id=self.__os_creds.user_domain_id,
+ project_domain_name=self.__os_creds.project_domain_name,
project_domain_id=self.__os_creds.project_domain_id,
interface=self.__os_creds.interface,
proxy_settings=self.__os_creds.proxy_settings,
self.password = kwargs.get('password')
self.project_name = kwargs.get('project_name')
self.email = kwargs.get('email')
- self.domain_name = kwargs.get('domain_name', 'default')
+ self.domain_name = kwargs.get('domain_name', 'Default')
self.enabled = kwargs.get('enabled', True)
self.roles = kwargs.get('roles', dict())
clients
:param heat_api_version: The OpenStack's API version to use for Heat
clients
- :param user_domain_id: Used for v3 APIs (default=None)
- :param user_domain_name: Used for v3 APIs (default='default')
- :param project_domain_id: Used for v3 APIs (default=None)
- :param project_domain_name: Used for v3 APIs (default='default')
+ :param user_domain_id: Used for v3 APIs (default='default')
+ :param user_domain_name: Used for v3 APIs (default='Default')
+ :param project_domain_id: Used for v3 APIs (default='default')
+ :param project_domain_name: Used for v3 APIs (default='Default')
:param interface: Used to specify the endpoint type for keystone as
public, admin, internal
:param proxy_settings: instance of os_credentials.ProxySettings class
else:
self.heat_api_version = float(kwargs['heat_api_version'])
- self.user_domain_id = kwargs.get('user_domain_id')
+ self.user_domain_id = kwargs.get('user_domain_id', 'default')
if kwargs.get('user_domain_name') is None:
- self.user_domain_name = 'default'
+ self.user_domain_name = 'Default'
else:
self.user_domain_name = kwargs['user_domain_name']
- self.project_domain_id = kwargs.get('project_domain_id')
+ self.project_domain_id = kwargs.get('project_domain_id', 'default')
if kwargs.get('project_domain_name') is None:
- self.project_domain_name = 'default'
+ self.project_domain_name = 'Default'
else:
self.project_domain_name = kwargs['project_domain_name']
self.assertEqual(2, os_creds.image_api_version)
self.assertEqual(2, os_creds.compute_api_version)
self.assertEqual(1, os_creds.heat_api_version)
- self.assertIsNone(os_creds.user_domain_id)
- self.assertEqual('default', os_creds.user_domain_name)
- self.assertIsNone(os_creds.project_domain_id)
- self.assertEqual('default', os_creds.project_domain_name)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('Default', os_creds.user_domain_name)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('Default', os_creds.project_domain_name)
self.assertEqual('admin', os_creds.interface)
self.assertFalse(os_creds.cacert)
self.assertIsNone(os_creds.proxy_settings)
self.assertEqual(2, os_creds.image_api_version)
self.assertEqual(2, os_creds.compute_api_version)
self.assertEqual(1, os_creds.heat_api_version)
- self.assertIsNone(os_creds.user_domain_id)
- self.assertEqual('default', os_creds.user_domain_name)
- self.assertIsNone(os_creds.project_domain_id)
- self.assertEqual('default', os_creds.project_domain_name)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('Default', os_creds.user_domain_name)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('Default', os_creds.project_domain_name)
self.assertEqual('admin', os_creds.interface)
self.assertFalse(os_creds.cacert)
self.assertIsNone(os_creds.proxy_settings)
self.assertEqual(6, os_creds.image_api_version)
self.assertEqual(7, os_creds.compute_api_version)
self.assertEqual(8.0, os_creds.heat_api_version)
- self.assertIsNone(os_creds.user_domain_id)
- self.assertEqual('default', os_creds.user_domain_name)
- self.assertIsNone(os_creds.project_domain_id)
- self.assertEqual('default', os_creds.project_domain_name)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('Default', os_creds.user_domain_name)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('Default', os_creds.project_domain_name)
self.assertEqual('admin', os_creds.interface)
self.assertTrue(os_creds.cacert)
self.assertIsNone(os_creds.proxy_settings)
self.assertEqual(6, os_creds.image_api_version)
self.assertEqual(7, os_creds.compute_api_version)
self.assertEqual(8.0, os_creds.heat_api_version)
- self.assertIsNone(os_creds.user_domain_id)
- self.assertEqual('default', os_creds.user_domain_name)
- self.assertIsNone(os_creds.project_domain_id)
- self.assertEqual('default', os_creds.project_domain_name)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('Default', os_creds.user_domain_name)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('Default', os_creds.project_domain_name)
self.assertEqual('admin', os_creds.interface)
self.assertTrue(os_creds.cacert)
self.assertIsNone(os_creds.proxy_settings)
self.assertEqual(2, os_creds.image_api_version)
self.assertEqual(2, os_creds.compute_api_version)
self.assertEqual(1, os_creds.heat_api_version)
- self.assertIsNone(os_creds.user_domain_id)
- self.assertEqual('default', os_creds.user_domain_name)
- self.assertIsNone(os_creds.project_domain_id)
- self.assertEqual('default', os_creds.project_domain_name)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('Default', os_creds.user_domain_name)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('Default', os_creds.project_domain_name)
self.assertEqual('admin', os_creds.interface)
self.assertFalse(os_creds.cacert)
self.assertEqual('foo', os_creds.proxy_settings.host)
self.assertEqual(2, os_creds.image_api_version)
self.assertEqual(2, os_creds.compute_api_version)
self.assertEqual(1, os_creds.heat_api_version)
- self.assertIsNone(os_creds.user_domain_id)
- self.assertEqual('default', os_creds.user_domain_name)
- self.assertIsNone(os_creds.project_domain_id)
- self.assertEqual('default', os_creds.project_domain_name)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('Default', os_creds.user_domain_name)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('Default', os_creds.project_domain_name)
self.assertEqual('admin', os_creds.interface)
self.assertFalse(os_creds.cacert)
self.assertEqual('foo', os_creds.proxy_settings.host)
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, image_settings=self.image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(created_image.size, retrieved_image.size)
self.assertEqual(get_image_size(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, image_settings=self.image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creator.get_image().size,
retrieved_image.size)
self.assertEqual(self.image_name, created_image.name)
retrieved_image = glance_utils.get_image(
- self.glance, file_image_settings.name)
+ self.glance, image_settings=file_image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creator.get_image().size,
retrieved_image.size)
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, image_settings=self.image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creator.get_image().size,
retrieved_image.size)
glance_utils.delete_image(self.glance, created_image)
self.assertIsNone(glance_utils.get_image(
- self.glance, self.image_creator.image_settings.name))
+ self.glance, image_settings=self.image_creator.image_settings))
# Must not throw an exception when attempting to cleanup non-existent
# image
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, image_settings=self.image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creator.get_image().size,
retrieved_image.size)
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, image_settings=self.image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creator.get_image().size,
retrieved_image.size)
image_creator.create()
main_image = glance_utils.get_image(self.glance,
- image_settings.name)
+ image_settings=image_settings)
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.glance,
+ image_settings=image_settings.kernel_image_settings)
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.glance,
+ image_settings=image_settings.ramdisk_image_settings)
self.assertIsNotNone(ramdisk_image)
self.assertIsNotNone(image_creator.get_ramdisk_image())
self.assertEqual(ramdisk_image.id,
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, image_settings=file_image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creators[-1].get_image().size,
retrieved_image.size)
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, image_settings=os_image_settings)
self.assertIsNotNone(retrieved_image)
self.assertEqual(self.image_creators[-1].get_image().size,
from neutronclient.common.exceptions import InvalidIpForSubnetClient
from snaps import file_utils
+from snaps.openstack import create_network, create_router
from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
from snaps.openstack.create_image import OpenStackImage, ImageSettings
from snaps.openstack.create_instance import (
VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings,
VmInstanceSettingsError, FloatingIpSettingsError)
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_network import (
+ OpenStackNetwork, PortSettings, NetworkSettings)
+from snaps.openstack.create_router import OpenStackRouter, RouterSettings
from snaps.openstack.create_security_group import (
SecurityGroupSettings, OpenStackSecurityGroup, SecurityGroupRuleSettings,
Direction, Protocol)
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ FlavorSettings(name=guid + '-flavor-name', ram=256, disk=10,
vcpus=1, metadata=self.flavor_metadata))
self.flavor_creator.create()
except Exception as e:
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ FlavorSettings(name=guid + '-flavor-name', ram=256, disk=10,
vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
self.image_creator.image_settings)
vm_inst = self.inst_creator.create()
- self.assertEqual(1, len(
- nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
+ self.assertIsNotNone(nova_utils.get_server(
+ self.nova, vm_inst_settings=instance_settings))
# Delete instance
nova_utils.delete_vm_instance(self.nova, vm_inst)
self.assertTrue(self.inst_creator.vm_deleted(block=True))
- self.assertEqual(0, len(
- nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
+ self.assertIsNone(nova_utils.get_server(
+ self.nova, vm_inst_settings=instance_settings))
# Exception should not be thrown
self.inst_creator.clean()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ FlavorSettings(name=guid + '-flavor-name', ram=256, disk=10,
vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ FlavorSettings(name=guid + '-flavor-name', ram=256, disk=10,
vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
except Exception as e:
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=self.guid + '-flavor-name', ram=128,
+ FlavorSettings(name=self.guid + '-flavor-name', ram=256,
disk=10, vcpus=2,
metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ FlavorSettings(name=guid + '-flavor-name', ram=256, disk=10,
vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
self.flavor_creator = OpenStackFlavor(
self.os_creds,
FlavorSettings(
- name=self.guid + '-flavor-name', ram=128, disk=10,
+ name=self.guid + '-flavor-name', ram=256, disk=10,
vcpus=1))
self.flavor_creator.create()
except Exception as e:
self.assertTrue(self.inst_creator.vm_active(block=True))
+class CreateInstanceTwoNetTests(OSIntegrationTestCase):
+ """
+ Tests the ability of two VMs to communicate when attached to separate
+ private networks that are tied together with a router.
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ cidr1 = '10.200.201.0/24'
+ cidr2 = '10.200.202.0/24'
+ static_gateway_ip1 = '10.200.201.1'
+ static_gateway_ip2 = '10.200.202.1'
+ self.ip1 = '10.200.201.5'
+ self.ip2 = '10.200.202.5'
+
+ self.nova = nova_utils.nova_client(self.os_creds)
+
+ # Initialize for tearDown()
+ self.image_creator = None
+ self.network_creators = list()
+ self.router_creator = None
+ self.flavor_creator = None
+ self.sec_grp_creator = None
+ self.inst_creators = list()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.vm_inst1_name = self.guid + '-inst1'
+ self.vm_inst2_name = self.guid + '-inst2'
+ self.port_1_name = self.guid + '-vm1-port'
+ self.port_2_name = self.guid + '-vm2-port'
+ self.net_config_1 = NetworkSettings(
+ name=self.guid + '-net1',
+ subnet_settings=[
+ create_network.SubnetSettings(
+ cidr=cidr1, name=self.guid + '-subnet1',
+ gateway_ip=static_gateway_ip1)])
+ self.net_config_2 = NetworkSettings(
+ name=self.guid + '-net2',
+ subnet_settings=[
+ create_network.SubnetSettings(
+ cidr=cidr2, name=self.guid + '-subnet2',
+ gateway_ip=static_gateway_ip2)])
+
+ image_name = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=image_name, image_metadata=self.image_metadata)
+
+ try:
+ # Create Image
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
+ self.image_creator.create()
+
+ # First network is public
+ self.network_creators.append(OpenStackNetwork(
+ self.os_creds, self.net_config_1))
+ # Second network is private
+ self.network_creators.append(OpenStackNetwork(
+ self.os_creds, self.net_config_2))
+ for network_creator in self.network_creators:
+ network_creator.create()
+
+ port_settings = [
+ create_network.PortSettings(
+ name=self.guid + '-router-port1',
+ ip_addrs=[{
+ 'subnet_name':
+ self.net_config_1.subnet_settings[0].name,
+ 'ip': static_gateway_ip1
+ }],
+ network_name=self.net_config_1.name,
+ project_name=self.os_creds.project_name),
+ create_network.PortSettings(
+ name=self.guid + '-router-port2',
+ ip_addrs=[{
+ 'subnet_name':
+ self.net_config_2.subnet_settings[0].name,
+ 'ip': static_gateway_ip2
+ }],
+ network_name=self.net_config_2.name,
+ project_name=self.os_creds.project_name)]
+
+ router_settings = RouterSettings(name=self.guid + '-pub-router',
+ port_settings=port_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
+ self.router_creator.create()
+
+ # Create Flavor
+ self.flavor_creator = OpenStackFlavor(
+ self.admin_os_creds,
+ FlavorSettings(name=self.guid + '-flavor-name', ram=512,
+ disk=10, vcpus=2,
+ metadata=self.flavor_metadata))
+ self.flavor_creator.create()
+
+ sec_grp_name = self.guid + '-sec-grp'
+ rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
+ protocol=Protocol.icmp)
+ self.sec_grp_creator = OpenStackSecurityGroup(
+ self.os_creds,
+ SecurityGroupSettings(name=sec_grp_name,
+ rule_settings=[rule1]))
+ self.sec_grp_creator.create()
+ except:
+ self.tearDown()
+ raise
+
+ def tearDown(self):
+ """
+ Cleans the created objects
+ """
+ for inst_creator in self.inst_creators:
+ try:
+ inst_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message '
+ '- %s', e)
+
+ if self.flavor_creator:
+ try:
+ self.flavor_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
+
+ if self.router_creator:
+ try:
+ self.router_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning router with message - %s',
+ e)
+
+ for network_creator in self.network_creators:
+ try:
+ network_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning network with message - %s',
+ e)
+
+ if self.sec_grp_creator:
+ try:
+ self.sec_grp_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning security group with message'
+ ' - %s', e)
+
+ if self.image_creator and not self.image_creator.image_settings.exists:
+ try:
+ self.image_creator.clean()
+ except Exception as e:
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
+
+ super(self.__class__, self).__clean__()
+
+ def test_ping_via_router(self):
+ """
+ Tests the creation of two OpenStack instances with one port on
+ different private networks wit a router in between to ensure that they
+ can ping
+ through
+ """
+ # Create ports/NICs for instance
+ ports_settings = []
+ ctr = 1
+ for network_creator in self.network_creators:
+ ports_settings.append(PortSettings(
+ name=self.guid + '-port-' + str(ctr),
+ network_name=network_creator.network_settings.name))
+ ctr += 1
+
+ # Configure instances
+ instance1_settings = VmInstanceSettings(
+ name=self.vm_inst1_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ userdata=_get_ping_userdata(self.ip2),
+ port_settings=[PortSettings(
+ name=self.port_1_name,
+ ip_addrs=[{
+ 'subnet_name':
+ self.net_config_1.subnet_settings[0].name,
+ 'ip': self.ip1
+ }],
+ network_name=self.network_creators[0].network_settings.name)])
+ instance2_settings = VmInstanceSettings(
+ name=self.vm_inst2_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ userdata=_get_ping_userdata(self.ip1),
+ port_settings=[PortSettings(
+ name=self.port_2_name,
+ ip_addrs=[{
+ 'subnet_name':
+ self.net_config_2.subnet_settings[0].name,
+ 'ip': self.ip2
+ }],
+ network_name=self.network_creators[1].network_settings.name)])
+
+ # Create instances
+ self.inst_creators.append(OpenStackVmInstance(
+ self.os_creds, instance1_settings,
+ self.image_creator.image_settings))
+ self.inst_creators.append(OpenStackVmInstance(
+ self.os_creds, instance2_settings,
+ self.image_creator.image_settings))
+
+ for inst_creator in self.inst_creators:
+ inst_creator.create(block=True)
+
+ # Check for DHCP lease
+ self.assertTrue(check_dhcp_lease(self.inst_creators[0], self.ip1))
+ self.assertTrue(check_dhcp_lease(self.inst_creators[1], self.ip2))
+
+ # Effectively blocks until VM has been properly activated
+ self.assertTrue(check_ping(self.inst_creators[0]))
+ self.assertTrue(check_ping(self.inst_creators[1]))
+
+
def check_dhcp_lease(inst_creator, ip, timeout=160):
"""
Returns true if the expected DHCP lease has been acquired
logger.debug('Full console output -\n' + full_log)
return found
+
+
+def _get_ping_userdata(test_ip):
+ """
+ Returns the post VM creation script to be added into the VM's userdata
+ :param test_ip: the IP value to substitute into the script
+ :return: the bash script contents
+ """
+ if test_ip:
+ return ("#!/bin/sh\n\n"
+ "while true; do\n"
+ " ping -c 1 %s 2>&1 >/dev/null\n"
+ " RES=$?\n"
+ " if [ \"Z$RES\" = \"Z0\" ] ; then\n"
+ " echo 'vPing OK'\n"
+ " break\n"
+ " else\n"
+ " echo 'vPing KO'\n"
+ " fi\n"
+ " sleep 1\n"
+ "done\n" % test_ip)
+ return None
+
+
+def check_ping(vm_creator, timeout=160):
+ """
+ Check for VM for ping result
+ """
+ tries = 0
+
+ while tries < timeout:
+ time.sleep(1)
+ p_console = vm_creator.get_console_output()
+ if "vPing OK" in p_console:
+ return True
+ elif "failed to read iid from metadata" in p_console or tries > 5:
+ return False
+ tries += 1
+
+ return False
self.router_creator.clean()
if self.net_creator:
- if len(self.net_creator.get_subnets()) > 0:
- # Validate subnet has been deleted
- neutron_utils_tests.validate_subnet(
- self.neutron,
- self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr,
- False)
-
- if self.net_creator.get_network():
- # Validate network has been deleted
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name,
- False)
self.net_creator.clean()
super(self.__class__, self).__clean__()
self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True))
# Validate subnets
- neutron_utils_tests.validate_subnet(
+ self.assertTrue(neutron_utils_tests.validate_subnet(
self.neutron,
self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr, True)
+ self.net_creator.network_settings.subnet_settings[0].cidr, True))
def test_create_delete_network(self):
"""
self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True))
neutron_utils.delete_network(self.neutron,
self.net_creator.get_network())
self.assertIsNone(neutron_utils.get_network(
- self.neutron, self.net_creator.network_settings.name))
+ self.neutron, network_settings=self.net_creator.network_settings))
# This shall not throw an exception here
self.net_creator.clean()
self.router_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True))
# Validate subnets
- neutron_utils_tests.validate_subnet(
+ self.assertTrue(neutron_utils_tests.validate_subnet(
self.neutron,
self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr, True)
+ self.net_creator.network_settings.subnet_settings[0].cidr, True))
# Validate routers
neutron_utils_tests.validate_router(
self.net_creator.create()
retrieved_net = neutron_utils.get_network(
- self.neutron, self.net_config.network_settings.name)
+ self.neutron, network_settings=self.net_config.network_settings)
self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
self.os_creds, self.net_config.router_settings)
self.router_creator.create()
- retrieved_router = neutron_utils.get_router_by_name(
- self.neutron, self.router_creator.get_router().name)
+ retrieved_router = neutron_utils.get_router(
+ self.neutron, router_settings=self.router_creator.router_settings)
self.assertEqual(
self.router_creator.get_router().id, retrieved_router.id)
self.net_creator.create()
retrieved_net = neutron_utils.get_network(
- self.neutron, self.net_config.network_settings.name)
+ self.neutron, network_settings=self.net_config.network_settings)
self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
self.admin_os_creds, self.net_config.router_settings)
self.router_creator.create()
- retrieved_router = neutron_utils.get_router_by_name(
- self.neutron, self.router_creator.get_router().name)
+ retrieved_router = neutron_utils.get_router(
+ self.neutron, router_settings=self.router_creator.router_settings)
self.assertEqual(
self.router_creator.get_router().id, retrieved_router.id)
Cleans the network
"""
if self.net_creator:
- if len(self.net_creator.get_subnets()) > 0:
- # Validate subnet has been deleted
- neutron_utils_tests.validate_subnet(
- self.neutron,
- self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr,
- False)
-
- if self.net_creator.get_network():
- # Validate network has been deleted
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name,
- False)
self.net_creator.clean()
def test_create_network_type_vlan(self):
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, net_settings.name, True))
self.assertEquals(network_type, network.type)
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(self.neutron, net_settings.name,
- True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, net_settings.name, True))
self.assertEqual(network_type, network.type)
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, net_settings.name, True))
self.assertEquals(network_type, network.type)
def test_name_only(self):
settings = ProjectSettings(name='foo')
self.assertEqual('foo', settings.name)
- self.assertEqual('default', settings.domain)
+ self.assertEqual('Default', settings.domain_name)
self.assertIsNone(settings.description)
self.assertTrue(settings.enabled)
def test_config_with_name_only(self):
settings = ProjectSettings(**{'name': 'foo'})
self.assertEqual('foo', settings.name)
- self.assertEqual('default', settings.domain)
+ self.assertEqual('Default', settings.domain_name)
self.assertIsNone(settings.description)
self.assertTrue(settings.enabled)
settings = ProjectSettings(name='foo', domain='bar',
description='foobar', enabled=False)
self.assertEqual('foo', settings.name)
- self.assertEqual('bar', settings.domain)
+ self.assertEqual('bar', settings.domain_name)
self.assertEqual('foobar', settings.description)
self.assertFalse(settings.enabled)
**{'name': 'foo', 'domain': 'bar', 'description': 'foobar',
'enabled': False})
self.assertEqual('foo', settings.name)
- self.assertEqual('bar', settings.domain)
+ self.assertEqual('bar', settings.domain_name)
self.assertEqual('foobar', settings.description)
self.assertFalse(settings.enabled)
"""
guid = str(uuid.uuid4())[:-19]
guid = self.__class__.__name__ + '-' + guid
- self.project_settings = ProjectSettings(name=guid + '-name')
+ self.project_settings = ProjectSettings(
+ name=guid + '-name',
+ domain=self.os_creds.project_domain_name)
self.keystone = keystone_utils.keystone_client(self.os_creds)
value. This test will not do anything with a keystone v2.0 client.
"""
if self.keystone.version != keystone_utils.V2_VERSION_STR:
- self.project_settings.domain = 'foo'
+ self.project_settings.domain_name = 'foo'
self.project_creator = OpenStackProject(self.os_creds,
self.project_settings)
self.assertIsNotNone(created_project)
retrieved_project = keystone_utils.get_project(
- keystone=self.keystone, project_name=self.project_settings.name)
+ keystone=self.keystone, project_settings=self.project_settings)
self.assertIsNotNone(retrieved_project)
self.assertEqual(created_project, retrieved_project)
self.assertTrue(validate_project(self.keystone, self.project_settings,
self.assertIsNotNone(created_project)
retrieved_project = keystone_utils.get_project(
- keystone=self.keystone, project_name=self.project_settings.name)
+ keystone=self.keystone, project_settings=self.project_settings)
self.assertIsNotNone(retrieved_project)
self.assertEqual(created_project, retrieved_project)
"""
guid = str(uuid.uuid4())[:-19]
self.guid = self.__class__.__name__ + '-' + guid
- self.project_settings = ProjectSettings(name=self.guid + '-name')
+ self.project_settings = ProjectSettings(
+ name=self.guid + '-name',
+ domain=self.os_creds.project_domain_name)
self.keystone = keystone_utils.keystone_client(self.os_creds)
user_creator = OpenStackUser(
self.os_creds, UserSettings(
name=self.guid + '-user',
- password=self.guid, roles={'admin': 'admin'}))
+ password=self.guid, roles={'admin': 'admin'},
+ domain_name=self.os_creds.user_domain_name))
self.project_creator.assoc_user(user_creator.create())
self.user_creators.append(user_creator)
user_creator_1 = OpenStackUser(
self.os_creds, UserSettings(
name=self.guid + '-user1', password=self.guid,
- roles={'admin': 'admin'}))
+ roles={'admin': 'admin'},
+ domain_name=self.os_creds.user_domain_name))
self.project_creator.assoc_user(user_creator_1.create())
self.user_creators.append(user_creator_1)
user_creator_2 = OpenStackUser(
self.os_creds, UserSettings(
name=self.guid + '-user2', password=self.guid,
- roles={'admin': 'admin'}))
+ roles={'admin': 'admin'},
+ domain_name=self.os_creds.user_domain_name))
self.project_creator.assoc_user(user_creator_2.create())
self.user_creators.append(user_creator_2)
if keystone.version == keystone_utils.V2_VERSION_STR:
return project_settings.name == project.name
else:
+ domain = keystone_utils.get_domain_by_id(keystone, project.domain_id)
return (project_settings.name == project.name and
- project_settings.domain == project.domain_id)
+ project_settings.domain_name == domain.name)
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
self.assertTrue(verify_router_attributes(
self.admin_os_creds, router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
self.assertTrue(verify_router_attributes(
self.os_creds, router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
self.assertTrue(verify_router_attributes(
self.os_creds, self.router_settings)
created_router = self.router_creator.create()
self.assertIsNotNone(created_router)
- retrieved_router = neutron_utils.get_router_by_name(
- self.neutron, self.router_settings.name)
+ retrieved_router = neutron_utils.get_router(
+ self.neutron, router_settings=self.router_settings)
self.assertIsNotNone(retrieved_router)
neutron_utils.delete_router(self.neutron, created_router)
- retrieved_router = neutron_utils.get_router_by_name(
- self.neutron, self.router_settings.name)
+ retrieved_router = neutron_utils.get_router(
+ self.neutron, router_settings=self.router_settings)
self.assertIsNone(retrieved_router)
# Should not raise an exception
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
self.assertTrue(verify_router_attributes(router, self.router_creator,
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
self.assertTrue(verify_router_attributes(router, self.router_creator,
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertTrue(verify_router_attributes(router, self.router_creator))
+ # Instantiate second identical creator to ensure a second router
+ # has not been created
+ router_creator2 = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
+ router2 = router_creator2.create()
+ self.assertIsNotNone(self.router_creator.get_router(), router2)
+
def test_create_router_external_network(self):
"""
Test creation of a router connected to an external network and a
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertTrue(verify_router_attributes(
router, self.router_creator, ext_gateway=self.ext_net_name))
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNotNone(sec_grp)
validation_utils.objects_equivalent(
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNotNone(sec_grp)
validation_utils.objects_equivalent(
self.admin_os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNotNone(sec_grp)
validation_utils.objects_equivalent(
neutron_utils.delete_security_group(self.neutron, created_sec_grp)
self.assertIsNone(neutron_utils.get_security_group(
- self.neutron, self.sec_grp_creator.sec_grp_settings.name))
+ self.neutron,
+ sec_grp_settings=self.sec_grp_creator.sec_grp_settings))
self.sec_grp_creator.clean()
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
validation_utils.objects_equivalent(
self.sec_grp_creator.get_security_group(), sec_grp)
rules = neutron_utils.get_rules_by_security_group(
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
validation_utils.objects_equivalent(
self.sec_grp_creator.get_security_group(), sec_grp)
rules = neutron_utils.get_rules_by_security_group(
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
validation_utils.objects_equivalent(
self.sec_grp_creator.get_security_group(), sec_grp)
rules = neutron_utils.get_rules_by_security_group(
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
validation_utils.objects_equivalent(
self.sec_grp_creator.get_security_group(), sec_grp)
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
validation_utils.objects_equivalent(
self.sec_grp_creator.get_security_group(), sec_grp)
rules = neutron_utils.get_rules_by_security_group(
self.os_creds, sec_grp_settings)
self.sec_grp_creator.create()
- sec_grp = neutron_utils.get_security_group(self.neutron,
- self.sec_grp_name)
+ sec_grp = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
validation_utils.objects_equivalent(
self.sec_grp_creator.get_security_group(), sec_grp)
setting_proto = rule_setting.protocol.name
sec_grp = neutron_utils.get_security_group(
- neutron, rule_setting.sec_grp_name)
+ neutron, sec_grp_name=rule_setting.sec_grp_name)
setting_eth_type = create_security_group.Ethertype.IPv4
if rule_setting.ethertype:
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=self.guid + '-flavor-name', ram=128, disk=10,
+ FlavorSettings(name=self.guid + '-flavor-name', ram=256, disk=10,
vcpus=1))
self.flavor_creator.create()
"""
guid = str(uuid.uuid4())[:-19]
guid = self.__class__.__name__ + '-' + guid
- self.user_settings = UserSettings(name=guid + '-name',
- password=guid + '-password',
- roles={'admin': 'admin'})
+ self.user_settings = UserSettings(
+ name=guid + '-name',
+ password=guid + '-password',
+ roles={'admin': 'admin'},
+ domain_name=self.os_creds.user_domain_name)
self.keystone = keystone_utils.keystone_client(self.os_creds)
'compute_api_version': config.get('compute_api_version'),
'heat_api_version': config.get('heat_api_version'),
'user_domain_id': config.get('user_domain_id'),
+ 'user_domain_name': config.get('user_domain_name'),
'project_domain_id': config.get('project_domain_id'),
+ 'project_domain_name': config.get('project_domain_name'),
'interface': config.get('interface'),
'proxy_settings': proxy_settings,
'cacert': config.get('cacert'),
creds_dict.update(overrides)
os_creds = OSCreds(**creds_dict)
- logger.info('OS Credentials = %s', os_creds)
+ logger.info('OS Credentials = %s', os_creds.__str__)
return os_creds
self.role = None
if self.use_keystone:
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(self.admin_os_creds)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())[:-19]
project_name = guid + '-proj'
self.project_creator = deploy_utils.create_project(
- self.admin_os_creds, ProjectSettings(name=project_name))
+ self.admin_os_creds, ProjectSettings(
+ name=project_name,
+ domain=self.admin_os_creds.project_domain_name))
self.user_creator = deploy_utils.create_user(
self.admin_os_creds, UserSettings(
name=guid + '-user', password=guid,
project_name=project_name, roles={
- 'admin': self.project_creator.project_settings.name}))
+ 'admin': self.project_creator.project_settings.name},
+ domain_name=self.admin_os_creds.user_domain_name))
self.os_creds = self.user_creator.get_os_creds(
self.project_creator.project_settings.name)
region_name=os_creds.region_name)
-def get_image(glance, image_name=None):
+def get_image(glance, image_name=None, image_settings=None):
"""
Returns an OpenStack image object for a given name
:param glance: the Glance client
:param image_name: the image name to lookup
+ :param image_settings: the image settings used for lookups
:return: the image object or None
"""
- images = glance.images.list()
+ img_filter = dict()
+ if image_settings:
+ if image_settings.exists:
+ img_filter = {'name': image_settings.name}
+ else:
+ img_filter = {'name': image_settings.name,
+ 'disk_format': image_settings.format}
+ elif image_name:
+ img_filter = {'name': image_name}
+
+ images = glance.images.list(**{'filters': img_filter})
for image in images:
if glance.version == VERSION_1:
- if image.name == image_name:
- image = glance.images.get(image.id)
- return Image(name=image.name, image_id=image.id,
- size=image.size, properties=image.properties)
+ image = glance.images.get(image.id)
+ return Image(name=image.name, image_id=image.id,
+ size=image.size, properties=image.properties)
elif glance.version == VERSION_2:
- if image['name'] == image_name:
- return Image(
- name=image['name'], image_id=image['id'],
- size=image['size'], properties=image.get('properties'))
- return None
+ return Image(
+ name=image['name'], image_id=image['id'],
+ size=image['size'], properties=image.get('properties'))
def get_image_by_id(glance, image_id):
region_name=os_creds.region_name)
-def get_stack_by_name(heat_cli, stack_name):
+def get_stack(heat_cli, stack_settings=None, stack_name=None):
"""
- Returns a domain Stack object
+ Returns the first domain Stack object found. When stack_setting
+ is not None, the filter created will take the name attribute. When
+ stack_settings is None and stack_name is not, stack_name will be used
+ instead. When both are None, the first stack object received will be
+ returned, else None
:param heat_cli: the OpenStack heat client
- :param stack_name: the name of the heat stack
+ :param stack_settings: a StackSettings object
+ :param stack_name: the name of the heat stack to return
:return: the Stack domain object else None
"""
- stacks = heat_cli.stacks.list(**{'name': stack_name})
+
+ stack_filter = dict()
+ if stack_settings:
+ stack_filter['stack_name'] = stack_settings.name
+ elif stack_name:
+ stack_filter['stack_name'] = stack_name
+
+ stacks = heat_cli.stacks.list(**stack_filter)
for stack in stacks:
return Stack(name=stack.identifier, stack_id=stack.id)
- return None
-
def get_stack_by_id(heat_cli, stack_id):
"""
-# 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 keystoneauth1 import session
import requests
-from snaps.domain.project import Project
+from snaps.domain.project import Project, Domain
from snaps.domain.role import Role
from snaps.domain.user import User
auth=auth, service_type=service_type, interface=interface)
-def get_project(keystone=None, os_creds=None, project_name=None):
+def get_project(keystone=None, os_creds=None, project_settings=None,
+ project_name=None):
"""
Returns the first project object or None if not found
:param keystone: the Keystone client
:param os_creds: the OpenStack credentials used to obtain the Keystone
client if the keystone parameter is None
+ :param project_settings: a ProjectSettings object
:param project_name: the name to query
:return: the SNAPS-OO Project domain object or None
"""
- if not project_name:
- return None
-
if not keystone:
if os_creds:
keystone = keystone_client(os_creds)
raise KeystoneException(
'Cannot lookup project without the proper credentials')
+ proj_filter = dict()
+
+ if project_name:
+ proj_filter['name'] = project_name
+ elif project_settings:
+ proj_filter['name'] = project_settings.name
+ proj_filter['description'] = project_settings.description
+ proj_filter['domain_name'] = project_settings.domain_name
+ proj_filter['enabled'] = project_settings.enabled
+
if keystone.version == V2_VERSION_STR:
projects = keystone.tenants.list()
else:
- projects = keystone.projects.list(**{'name': project_name})
+ projects = keystone.projects.list(**proj_filter)
for project in projects:
- domain_id = None
- if keystone.version != V2_VERSION_STR:
- domain_id = project.domain_id
- if project.name == project_name:
+ if project.name == proj_filter['name']:
+ domain_id = None
+ if keystone.version != V2_VERSION_STR:
+ domain_id = project.domain_id
+
return Project(name=project.name, project_id=project.id,
domain_id=domain_id)
- return None
-
def create_project(keystone, project_settings):
"""
project_settings.name, project_settings.description,
project_settings.enabled)
else:
+ os_domain = __get_os_domain_by_name(
+ keystone, project_settings.domain_name)
+ if not os_domain:
+ os_domain = project_settings.domain_name
os_project = keystone.projects.create(
- project_settings.name, project_settings.domain,
+ project_settings.name, os_domain,
description=project_settings.description,
enabled=project_settings.enabled)
domain_id = os_project.domain_id
:param project_name: the associated project (optional)
:return: a SNAPS-OO User domain object or None
"""
- project = get_project(keystone=keystone, project_name=project_name)
+ project = None
+ if project_name:
+ project = get_project(keystone=keystone, project_name=project_name)
if project:
users = keystone.users.list(tenant_id=project.id)
email=user_settings.email, tenant_id=project_id,
enabled=user_settings.enabled)
else:
+ os_domain = __get_os_domain_by_name(
+ keystone, user_settings.domain_name)
+ if not os_domain:
+ os_domain = user_settings.domain_name
os_user = keystone.users.create(
name=user_settings.name, password=user_settings.password,
email=user_settings.email, project=project,
- domain=user_settings.domain_name, enabled=user_settings.enabled)
+ domain=os_domain, enabled=user_settings.enabled)
for role_name, role_project in user_settings.roles.items():
os_role = get_role_by_name(keystone, role_name)
os_project = get_project(keystone=keystone, project_name=role_project)
if os_role and os_project:
- existing_roles = get_roles_by_user(keystone, os_user,
- os_project)
+ existing_roles = get_roles_by_user(keystone, os_user, os_project)
found = False
for role in existing_roles:
if role.id == os_role.id:
keystone.roles.grant(os_role, user=user, project=project)
+def get_domain_by_id(keystone, domain_id):
+ """
+ Returns the first OpenStack domain with the given name else None
+ :param keystone: the Keystone client
+ :param domain_id: the domain ID to retrieve
+ :return: the SNAPS-OO Domain domain object
+ """
+ domain = keystone.domains.get(domain_id)
+ if domain:
+ return Domain(name=domain.name, domain_id=domain.id)
+
+
+def __get_os_domain_by_name(keystone, domain_name):
+ """
+ Returns the first OpenStack domain with the given name else None
+ :param keystone: the Keystone client
+ :param domain_name: the domain name to lookup
+ :return: the OpenStack domain object
+ """
+ domains = keystone.domains.list(name=domain_name)
+ for domain in domains:
+ if domain.name == domain_name:
+ return domain
+
+
class KeystoneException(Exception):
"""
Exception when calls to the Keystone client cannot be served properly
neutron.delete_network(network.id)
-def get_network(neutron, network_name, project_id=None):
+def get_network(neutron, network_settings=None, network_name=None,
+ project_id=None):
"""
- Returns an object (dictionary) of the first network found with a given name
- and project_id (if included)
+ Returns Network SNAPS-OO domain object the first network found with
+ either the given attributes from the network_settings object if not None,
+ else the query will use just the name from the network_name parameter.
+ When the project_id is included, that will be added to the query filter.
:param neutron: the client
+ :param network_settings: the NetworkSettings object used to create filter
:param network_name: the name of the network to retrieve
:param project_id: the id of the network's project
:return: a SNAPS-OO Network domain object
"""
net_filter = dict()
- if network_name:
+ if network_settings:
+ net_filter['name'] = network_settings.name
+ elif network_name:
net_filter['name'] = network_name
+
if project_id:
net_filter['project_id'] = project_id
networks = neutron.list_networks(**net_filter)
for network, netInsts in networks.items():
for inst in netInsts:
- if inst.get('name') == network_name:
- return Network(**inst)
- return None
+ return Network(**inst)
def get_network_by_id(neutron, network_id):
"""
- Returns the network object (dictionary) with the given ID
+ Returns the network object (dictionary) with the given ID else None
:param neutron: the client
:param network_id: the id of the network to retrieve
:return: a SNAPS-OO Network domain object
"""
networks = neutron.list_networks(**{'id': network_id})
- for network, netInsts in networks.items():
- for inst in netInsts:
- if inst.get('id') == network_id:
- return Network(**inst)
- return None
+ for network in networks['networks']:
+ if network['id'] == network_id:
+ return Network(**network)
def create_subnet(neutron, subnet_settings, os_creds, network=None):
neutron.delete_subnet(subnet.id)
-def get_subnet_by_name(neutron, subnet_name):
+def get_subnet(neutron, subnet_settings=None, subnet_name=None):
"""
- Returns the first subnet object (dictionary) found with a given name
+ Returns the first subnet object that fits the query else None including
+ if subnet_settings or subnet_name parameters are None.
:param neutron: the client
- :param subnet_name: the name of the network to retrieve
- :return: a SNAPS-OO Subnet domain object
+ :param subnet_settings: the subnet settings of the object to retrieve
+ :param subnet_name: the name of the subnet to retrieve
+ :return: a SNAPS-OO Subnet domain object or None
"""
- subnets = neutron.list_subnets(**{'name': subnet_name})
- for subnet, subnetInst in subnets.items():
- for inst in subnetInst:
- if inst['name'] == subnet_name:
- return Subnet(**inst)
- return None
+ sub_filter = dict()
+ if subnet_settings:
+ sub_filter['name'] = subnet_settings.name
+ sub_filter['cidr'] = subnet_settings.cidr
+ if subnet_settings.gateway_ip:
+ sub_filter['gateway_ip'] = subnet_settings.gateway_ip
+
+ if subnet_settings.enable_dhcp is not None:
+ sub_filter['enable_dhcp'] = subnet_settings.enable_dhcp
+
+ if subnet_settings.destination:
+ sub_filter['destination'] = subnet_settings.destination
+
+ if subnet_settings.nexthop:
+ sub_filter['nexthop'] = subnet_settings.nexthop
+
+ if subnet_settings.ipv6_ra_mode:
+ sub_filter['ipv6_ra_mode'] = subnet_settings.ipv6_ra_mode
+
+ if subnet_settings.ipv6_address_mode:
+ sub_filter['ipv6_address_mode'] = subnet_settings.ipv6_address_mode
+ elif subnet_name:
+ sub_filter['name'] = subnet_name
+ else:
+ return None
+
+ subnets = neutron.list_subnets(**sub_filter)
+ for subnet in subnets['subnets']:
+ return Subnet(**subnet)
def create_router(neutron, os_creds, router_settings):
neutron.delete_router(router=router.id)
-def get_router_by_name(neutron, router_name):
+def get_router(neutron, router_settings=None, router_name=None):
"""
- Returns the first router object (dictionary) found with a given name
+ Returns the first router object (dictionary) found the given the settings
+ values if not None, else finds the first with the value of the router_name
+ parameter, else None
:param neutron: the client
+ :param router_settings: the RouterSettings object
:param router_name: the name of the network to retrieve
:return: a SNAPS-OO Router domain object
"""
- routers = neutron.list_routers(**{'name': router_name})
- for router, routerInst in routers.items():
- for inst in routerInst:
- if inst.get('name') == router_name:
- return Router(**inst)
+ router_filter = dict()
+ if router_settings:
+ router_filter['name'] = router_settings.name
+ if router_settings.admin_state_up is not None:
+ router_filter['admin_state_up'] = router_settings.admin_state_up
+ elif router_name:
+ router_filter['name'] = router_name
+ else:
+ return None
+
+ routers = neutron.list_routers(**router_filter)
+ for routerInst in routers['routers']:
+ return Router(**routerInst)
return None
neutron.delete_port(port.id)
-def get_port_by_name(neutron, port_name):
+def get_port(neutron, port_settings=None, port_name=None):
"""
- Returns the first port object (dictionary) found with a given name
+ Returns the first port object (dictionary) found for the given query
:param neutron: the client
- :param port_name: the name of the port to retrieve
+ :param port_settings: the PortSettings object used for generating the query
+ :param port_name: if port_settings is None, this name is the value to place
+ into the query
:return: a SNAPS-OO Port domain object
"""
- ports = neutron.list_ports(**{'name': port_name})
+ port_filter = dict()
+
+ if port_settings:
+ port_filter['name'] = port_settings.name
+ if port_settings.admin_state_up:
+ port_filter['admin_state_up'] = port_settings.admin_state_up
+ if port_settings.device_id:
+ port_filter['device_id'] = port_settings.device_id
+ if port_settings.mac_address:
+ port_filter['mac_address'] = port_settings.mac_address
+ elif port_name:
+ port_filter['name'] = port_name
+
+ ports = neutron.list_ports(**port_filter)
for port in ports['ports']:
- if port['name'] == port_name:
- return Port(name=port['name'], id=port['id'],
- ips=port['fixed_ips'], mac_address=port['mac_address'])
+ return Port(name=port['name'], id=port['id'],
+ ips=port['fixed_ips'], mac_address=port['mac_address'])
return None
neutron.delete_security_group(sec_grp.id)
-def get_security_group(neutron, name):
+def get_security_group(neutron, sec_grp_settings=None, sec_grp_name=None,
+ project_id=None):
"""
- Returns the first security group object of the given name else None
+ Returns the first security group for a given query. The query gets built
+ from the sec_grp_settings parameter if not None, else only the name of
+ the security group will be used, else if the query parameters are None then
+ None will be returned
:param neutron: the client
- :param name: the name of security group object to retrieve
+ :param sec_grp_settings: an instance of SecurityGroupSettings config object
+ :param sec_grp_name: the name of security group object to retrieve
+ :param project_id: the ID of the project/tentant object that owns the
+ secuity group to retrieve
:return: a SNAPS-OO SecurityGroup domain object or None if not found
"""
- logger.info('Retrieving security group with name - ' + name)
- groups = neutron.list_security_groups(**{'name': name})
+ sec_grp_filter = dict()
+ if project_id:
+ sec_grp_filter['tenant_id'] = project_id
+
+ if sec_grp_settings:
+ sec_grp_filter['name'] = sec_grp_settings.name
+
+ if sec_grp_settings.description:
+ sec_grp_filter['description'] = sec_grp_settings.description
+ elif sec_grp_name:
+ sec_grp_filter['name'] = sec_grp_name
+ else:
+ return None
+
+ groups = neutron.list_security_groups(**sec_grp_filter)
for group in groups['security_groups']:
- if group['name'] == name:
- return SecurityGroup(**group)
- return None
+ return SecurityGroup(**group)
def get_security_group_by_id(neutron, sec_grp_id):
:return: the SNAPS FloatingIp object
"""
logger.info('Creating floating ip to external network - ' + ext_net_name)
- ext_net = get_network(neutron, ext_net_name)
+ ext_net = get_network(neutron, network_name=ext_net_name)
if ext_net:
fip = neutron.create_floatingip(
body={'floatingip':
ports = list()
for port_setting in instance_settings.port_settings:
- ports.append(neutron_utils.get_port_by_name(
- neutron, port_setting.name))
+ ports.append(neutron_utils.get_port(
+ neutron, port_settings=port_setting))
nics = []
for port in ports:
kv = dict()
raise NovaException(
'Flavor not found with name - %s', instance_settings.flavor)
- image = glance_utils.get_image(glance, image_settings.name)
+ image = glance_utils.get_image(glance, image_settings=image_settings)
if image:
args = {'name': instance_settings.name,
'flavor': flavor,
image_settings.name)
-def get_servers_by_name(nova, name):
+def get_server(nova, vm_inst_settings=None, server_name=None):
"""
- Returns a list of servers with a given name
+ Returns a VmInst object for the first server instance found.
:param nova: the Nova client
- :param name: the server name
- :return: the list of snaps.domain.VmInst objects
- """
- out = list()
- servers = nova.servers.list(search_opts={'name': name})
+ :param vm_inst_settings: the VmInstanceSettings object from which to build
+ the query if not None
+ :param server_name: the server with this name to return if vm_inst_settings
+ is not None
+ :return: a snaps.domain.VmInst object or None if not found
+ """
+ search_opts = dict()
+ if vm_inst_settings:
+ search_opts['name'] = vm_inst_settings.name
+ elif server_name:
+ search_opts['name'] = server_name
+
+ servers = nova.servers.list(search_opts=search_opts)
for server in servers:
- out.append(VmInst(name=server.name, inst_id=server.id,
- networks=server.networks))
- return out
+ return VmInst(name=server.name, inst_id=server.id,
+ networks=server.networks)
def __get_latest_server_os_object(nova, server):
Tests to ensure that the proper credentials can connect.
"""
glance = glance_utils.glance_client(self.os_creds)
- image = glance_utils.get_image(glance, 'foo')
+ image = glance_utils.get_image(glance, image_name='foo')
self.assertIsNone(image)
def test_glance_connect_fail(self):
glance = glance_utils.glance_client(OSCreds(
username='user', password='pass', auth_url='url',
project_name='project'))
- glance_utils.get_image(glance, 'foo')
+ glance_utils.get_image(glance, image_name='foo')
class GlanceUtilsTests(OSComponentTestCase):
self.assertEqual(self.image_name, self.image.name)
- image = glance_utils.get_image(self.glance, os_image_settings.name)
+ image = glance_utils.get_image(self.glance,
+ image_settings=os_image_settings)
self.assertIsNotNone(image)
validation_utils.objects_equivalent(self.image, image)
self.assertIsNotNone(self.image)
self.assertEqual(self.image_name, self.image.name)
- image = glance_utils.get_image(self.glance, file_image_settings.name)
+ image = glance_utils.get_image(
+ self.glance, image_settings=file_image_settings)
self.assertIsNotNone(image)
validation_utils.objects_equivalent(self.image, image)
class HeatSmokeTests(OSComponentTestCase):
"""
- Tests to ensure that the nova client can communicate with the cloud
+ Tests to ensure that the heat client can communicate with the cloud
"""
def test_nova_connect_success(self):
class HeatUtilsCreateStackTests(OSComponentTestCase):
"""
- Test basic nova keypair functionality
+ Test basic Heat functionality
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading
- and creating an OS image file within OpenStack
+ Instantiates OpenStack instances that cannot be spawned by Heat
"""
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- stack_name = self.__class__.__name__ + '-' + str(guid) + '-stack'
+ stack_name1 = self.__class__.__name__ + '-' + str(guid) + '-stack1'
+ stack_name2 = self.__class__.__name__ + '-' + str(guid) + '-stack2'
self.image_creator = OpenStackImage(
self.os_creds, openstack_tests.cirros_image_settings(
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.os_creds,
- FlavorSettings(name=guid + '-flavor', ram=128, disk=10, vcpus=1))
+ FlavorSettings(name=guid + '-flavor', ram=256, disk=10, vcpus=1))
self.flavor_creator.create()
env_values = {'image_name': self.image_creator.image_settings.name,
'flavor_name': self.flavor_creator.flavor_settings.name}
heat_tmplt_path = pkg_resources.resource_filename(
'snaps.openstack.tests.heat', 'test_heat_template.yaml')
- self.stack_settings = StackSettings(
- name=stack_name, template_path=heat_tmplt_path,
+ self.stack_settings1 = StackSettings(
+ name=stack_name1, template_path=heat_tmplt_path,
env_values=env_values)
- self.stack = None
+ self.stack_settings2 = StackSettings(
+ name=stack_name2, template_path=heat_tmplt_path,
+ env_values=env_values)
+ self.stack1 = None
+ self.stack2 = None
self.heat_client = heat_utils.heat_client(self.os_creds)
def tearDown(self):
"""
Cleans the image and downloaded image file
"""
- if self.stack:
+ if self.stack1:
try:
- heat_utils.delete_stack(self.heat_client, self.stack)
+ heat_utils.delete_stack(self.heat_client, self.stack1)
+ except:
+ pass
+
+ if self.stack2:
+ try:
+ heat_utils.delete_stack(self.heat_client, self.stack2)
except:
pass
pass
def test_create_stack(self):
+ """
+ Tests the creation of an OpenStack Heat stack1 that does not exist.
+ """
+ self.stack1 = heat_utils.create_stack(self.heat_client,
+ self.stack_settings1)
+
+ stack_query_1 = heat_utils.get_stack(
+ self.heat_client, stack_settings=self.stack_settings1)
+ self.assertEqual(self.stack1, stack_query_1)
+
+ stack_query_2 = heat_utils.get_stack(
+ self.heat_client, stack_name=self.stack_settings1.name)
+ self.assertEqual(self.stack1, stack_query_2)
+
+ stack_query_3 = heat_utils.get_stack_by_id(self.heat_client,
+ self.stack1.id)
+ self.assertEqual(self.stack1, stack_query_3)
+
+ outputs = heat_utils.get_stack_outputs(
+ self.heat_client, self.stack1.id)
+ self.assertIsNotNone(outputs)
+ self.assertEqual(0, len(outputs))
+
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+
+ is_active = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack1.id)
+ if status == create_stack.STATUS_CREATE_COMPLETE:
+ is_active = True
+ break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
+
+ time.sleep(3)
+
+ self.assertTrue(is_active)
+
+ def test_create_stack_x2(self):
"""
Tests the creation of an OpenStack keypair that does not exist.
"""
- self.stack = heat_utils.create_stack(self.heat_client,
- self.stack_settings)
+ self.stack1 = heat_utils.create_stack(self.heat_client,
+ self.stack_settings1)
+
+ stack1_query_1 = heat_utils.get_stack(
+ self.heat_client, stack_settings=self.stack_settings1)
+ self.assertEqual(self.stack1, stack1_query_1)
+
+ stack1_query_2 = heat_utils.get_stack(
+ self.heat_client, stack_name=self.stack_settings1.name)
+ self.assertEqual(self.stack1, stack1_query_2)
+
+ stack1_query_3 = heat_utils.get_stack_by_id(self.heat_client,
+ self.stack1.id)
+ self.assertEqual(self.stack1, stack1_query_3)
+
+ outputs = heat_utils.get_stack_outputs(self.heat_client,
+ self.stack1.id)
+ self.assertIsNotNone(outputs)
+ self.assertEqual(0, len(outputs))
+
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+
+ is_active = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack1.id)
+ if status == create_stack.STATUS_CREATE_COMPLETE:
+ is_active = True
+ break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
+
+ time.sleep(3)
+
+ self.assertTrue(is_active)
+
+ self.stack2 = heat_utils.create_stack(self.heat_client,
+ self.stack_settings2)
+
+ stack2_query_1 = heat_utils.get_stack(
+ self.heat_client, stack_settings=self.stack_settings2)
+ self.assertEqual(self.stack2, stack2_query_1)
- stack_query_1 = heat_utils.get_stack_by_name(self.heat_client,
- self.stack_settings.name)
- self.assertEqual(self.stack.id, stack_query_1.id)
+ stack2_query_2 = heat_utils.get_stack(
+ self.heat_client, stack_name=self.stack_settings2.name)
+ self.assertEqual(self.stack2, stack2_query_2)
- stack_query_2 = heat_utils.get_stack_by_id(self.heat_client,
- self.stack.id)
- self.assertEqual(self.stack.id, stack_query_2.id)
+ stack2_query_3 = heat_utils.get_stack_by_id(self.heat_client,
+ self.stack2.id)
+ self.assertEqual(self.stack2, stack2_query_3)
- outputs = heat_utils.get_stack_outputs(self.heat_client, self.stack.id)
+ outputs = heat_utils.get_stack_outputs(self.heat_client,
+ self.stack2.id)
self.assertIsNotNone(outputs)
self.assertEqual(0, len(outputs))
is_active = False
while time.time() < end_time:
status = heat_utils.get_stack_status(self.heat_client,
- self.stack.id)
+ self.stack2.id)
if status == create_stack.STATUS_CREATE_COMPLETE:
is_active = True
break
from snaps.openstack.create_project import ProjectSettings
from snaps.openstack.create_user import UserSettings
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
-from snaps.openstack.utils import keystone_utils
+from snaps.openstack.utils import keystone_utils, neutron_utils
__author__ = 'spisarski'
Cleans the remote OpenStack objects
"""
if self.project:
- keystone_utils.delete_project(self.keystone, self.project)
+ neutron = neutron_utils.neutron_client(self.os_creds)
+ default_sec_grp = neutron_utils.get_security_group(
+ neutron, sec_grp_name='default',
+ project_id=self.project.id)
+ if default_sec_grp:
+ try:
+ neutron_utils.delete_security_group(
+ neutron, default_sec_grp)
+ except:
+ pass
+
+ keystone_utils.delete_project(self.keystone, self.project)
if self.user:
keystone_utils.delete_user(self.keystone, self.user)
"""
Tests the keystone_utils.create_user() function
"""
- user_settings = UserSettings(name=self.username,
- password=str(uuid.uuid4()))
+ user_settings = UserSettings(
+ name=self.username,
+ password=str(uuid.uuid4()),
+ domain_name=self.os_creds.user_domain_name)
self.user = keystone_utils.create_user(self.keystone, user_settings)
self.assertEqual(self.username, self.user.name)
"""
Tests the keyston_utils.create_project() funtion
"""
- project_settings = ProjectSettings(name=self.project_name)
+ project_settings = ProjectSettings(
+ name=self.project_name, domain=self.os_creds.project_domain_name)
self.project = keystone_utils.create_project(self.keystone,
project_settings)
self.assertEqual(self.project_name, self.project.name)
project = keystone_utils.get_project(
- keystone=self.keystone, project_name=project_settings.name)
+ keystone=self.keystone, project_settings=project_settings)
self.assertIsNotNone(project)
self.assertEqual(self.project_name, self.project.name)
+ domain = keystone_utils.get_domain_by_id(
+ self.keystone, project.domain_id)
+ self.assertIsNotNone(domain)
+ self.assertEqual(domain.id, project.domain_id)
+
def test_get_endpoint_success(self):
"""
Tests to ensure that proper credentials and proper service type can
Tests the keystone_utils function grant_user_role_to_project()
:return:
"""
- user_settings = UserSettings(name=self.username,
- password=str(uuid.uuid4()))
+ user_settings = UserSettings(
+ name=self.username, password=str(uuid.uuid4()),
+ domain_name=self.os_creds.user_domain_name)
self.user = keystone_utils.create_user(self.keystone, user_settings)
self.assertEqual(self.username, self.user.name)
- project_settings = ProjectSettings(name=self.project_name)
+ project_settings = ProjectSettings(
+ name=self.project_name, domain=self.os_creds.project_domain_name)
self.project = keystone_utils.create_project(self.keystone,
project_settings)
self.assertEqual(self.project_name, self.project.name)
"""
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network.name, False)
def test_create_network(self):
"""
"""
if self.subnet:
neutron_utils.delete_subnet(self.neutron, self.subnet)
- validate_subnet(self.neutron, self.subnet.name,
- self.net_config.network_settings.subnet_settings[
- 0].cidr, False)
-
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network.name, False)
def test_create_subnet(self):
"""
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting, self.os_creds, network=self.network)
- validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
def test_create_subnet_null_name(self):
"""
subnet_setting = self.net_config.network_settings.subnet_settings[0]
neutron_utils.create_subnet(
self.neutron, subnet_setting, self.os_creds, network=self.network)
- validate_subnet(self.neutron, '', subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.assertFalse(validate_subnet(
+ self.neutron, '', subnet_setting.cidr, True))
def test_create_subnet_null_cidr(self):
"""
if self.subnet:
neutron_utils.delete_subnet(self.neutron, self.subnet)
- validate_subnet(
- self.neutron, self.subnet.name,
- self.net_config.network_settings.subnet_settings[0].cidr,
- False)
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network.name, False)
def test_create_router_simple(self):
"""
validate_router(self.neutron, self.net_config.router_settings.name,
True)
- ext_net = neutron_utils.get_network(self.neutron, self.ext_net_name)
+ ext_net = neutron_utils.get_network(
+ self.neutron, network_name=self.ext_net_name)
self.assertEqual(
self.router.external_gateway_info['network_id'], ext_net.id)
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting,
self.os_creds, self.network)
- validate_subnet(
- self.neutron,
- subnet_setting.name,
- subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
self.router = neutron_utils.create_router(
self.neutron, self.os_creds, self.net_config.router_settings)
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting,
self.os_creds, self.network)
- validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
with self.assertRaises(NeutronException):
self.interface_router = neutron_utils.add_interface_router(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting, self.os_creds, self.network)
- validate_subnet(self.neutron, subnet_setting.name,
- subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, PortSettings(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting, self.os_creds, self.network)
- validate_subnet(self.neutron, subnet_setting.name, subnet_setting.cidr,
- True)
+ self.assertTrue(validate_subnet(self.neutron, subnet_setting.name,
+ subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, PortSettings(
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting,
self.os_creds, self.network)
- validate_subnet(
- self.neutron,
- subnet_setting.name,
- subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting,
self.os_creds, self.network)
- validate_subnet(
- self.neutron,
- subnet_setting.name,
- subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting, self.os_creds, self.network)
- validate_subnet(self.neutron,
- subnet_setting.name,
- subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.subnet = neutron_utils.create_subnet(
self.neutron, subnet_setting, self.os_creds, self.network)
- validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True)
+ self.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
self.assertTrue(sec_grp_settings.name, security_group.name)
- sec_grp_get = neutron_utils.get_security_group(self.neutron,
- sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNotNone(sec_grp_get)
self.assertTrue(validation_utils.objects_equivalent(
security_group, sec_grp_get))
neutron_utils.delete_security_group(self.neutron, security_group)
- sec_grp_get = neutron_utils.get_security_group(self.neutron,
- sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNone(sec_grp_get)
def test_create_sec_grp_no_name(self):
self.assertTrue(sec_grp_settings.name, self.security_groups[0].name)
self.assertEqual(sec_grp_settings.name, self.security_groups[0].name)
- sec_grp_get = neutron_utils.get_security_group(self.neutron,
- sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNotNone(sec_grp_get)
self.assertEqual(self.security_groups[0], sec_grp_get)
# Refresh object so it is populated with the newly added rule
security_group = neutron_utils.get_security_group(
- self.neutron, sec_grp_settings.name)
+ self.neutron, sec_grp_settings=sec_grp_settings)
rules = neutron_utils.get_rules_by_security_group(self.neutron,
security_group)
self.assertTrue(sec_grp_settings.name, security_group.name)
- sec_grp_get = neutron_utils.get_security_group(self.neutron,
- sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings=sec_grp_settings)
self.assertIsNotNone(sec_grp_get)
self.assertEqual(security_group, sec_grp_get)
:param exists: Whether or not the network name should exist or not
:return: True/False
"""
- network = neutron_utils.get_network(neutron, name)
+ network = neutron_utils.get_network(neutron, network_name=name)
if exists and network:
return True
if not exists and not network:
:param exists: Whether or not the network name should exist or not
:return: True/False
"""
- subnet = neutron_utils.get_subnet_by_name(neutron, name)
- if exists and subnet:
+ subnet = neutron_utils.get_subnet(neutron, subnet_name=name)
+ if exists and subnet and subnet.name == name:
return subnet.cidr == cidr
if not exists and not subnet:
return True
:param exists: Whether or not the network name should exist or not
:return: True/False
"""
- router = neutron_utils.get_router_by_name(neutron, name)
+ router = neutron_utils.get_router(neutron, router_name=name)
if exists and router:
return True
return False
self.flavor_creator = OpenStackFlavor(
self.os_creds,
FlavorSettings(
- name=guid + '-flavor-name', ram=128, disk=10, vcpus=1))
+ name=guid + '-flavor-name', ram=256, disk=10, vcpus=1))
self.flavor_creator.create()
port_settings = PortSettings(name=guid + '-port',
SecurityGroupDomainObjectTests, SecurityGroupRuleDomainObjectTests,
PortDomainObjectTests, RouterDomainObjectTests,
InterfaceRouterDomainObjectTests, NetworkObjectTests, SubnetObjectTests)
-from snaps.domain.test.project_tests import ProjectDomainObjectTests
+from snaps.domain.test.project_tests import (
+ ProjectDomainObjectTests, DomainDomainObjectTests)
from snaps.domain.test.role_tests import RoleDomainObjectTests
from snaps.domain.test.stack_tests import StackDomainObjectTests
from snaps.domain.test.user_tests import UserDomainObjectTests
FloatingIpSettingsUnitTests, InstanceSecurityGroupTests,
VmInstanceSettingsUnitTests, CreateInstancePortManipulationTests,
SimpleHealthCheck, CreateInstanceFromThreePartImage,
- CreateInstanceMockOfflineTests)
+ CreateInstanceMockOfflineTests, CreateInstanceTwoNetTests)
from snaps.openstack.tests.create_keypairs_tests import (
CreateKeypairsTests, KeypairSettingsUnitTests, CreateKeypairsCleanupTests)
from snaps.openstack.tests.create_network_tests import (
ProjectSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
ProjectDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ DomainDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
RoleDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
- CreateKeypairsCleanupTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ CreateKeypairsCleanupTests, os_creds=os_creds,
+ ext_net_name=ext_net_name,
use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateInstanceTwoNetTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
CreateInstanceSimpleTests, os_creds=os_creds,
ext_net_name=ext_net_name, use_keystone=use_keystone,
Code Review / snaps.git / commitdiff