return self.project_id
else:
if self.project_name:
- keystone = keystone_utils.keystone_client(os_creds)
- project = keystone_utils.get_project(
- keystone=keystone, project_name=self.project_name)
- if project:
- return project.id
+ session = keystone_utils.keystone_session(os_creds)
+ keystone = keystone_utils.keystone_client(os_creds, session)
+ try:
+ project = keystone_utils.get_project(
+ keystone=keystone, project_name=self.project_name)
+ if project:
+ return project.id
+ finally:
+ keystone_utils.close_session(session)
return None
if self.name:
out['name'] = self.name
if self.project_name:
- keystone = keystone_utils.keystone_client(os_creds)
- project = keystone_utils.get_project(
- keystone=keystone, project_name=self.project_name)
+ session = keystone_utils.keystone_session(os_creds)
+ keystone = keystone_utils.keystone_client(os_creds, session)
+ try:
+ project = keystone_utils.get_project(
+ keystone=keystone, project_name=self.project_name)
+ finally:
+ keystone_utils.close_session(session)
project_id = None
if project:
project_id = project.id
out = dict()
- keystone = keystone_utils.keystone_client(os_creds)
- network = neutron_utils.get_network(
- neutron, keystone, network_name=self.network_name,
- project_name=self.project_name)
+ session = keystone_utils.keystone_session(os_creds)
+ keystone = keystone_utils.keystone_client(os_creds, session)
+ try:
+ network = neutron_utils.get_network(
+ neutron, keystone, network_name=self.network_name,
+ project_name=self.project_name)
+ finally:
+ keystone_utils.close_session(session)
+
if not network:
raise PortConfigError(
'Cannot locate network with name - ' + self.network_name
out = dict()
ext_gw = dict()
- keystone = keystone_utils.keystone_client(os_creds)
-
- if self.name:
- out['name'] = self.name
- if self.project_name:
- project = keystone_utils.get_project(
- keystone=keystone, project_name=self.project_name)
- if project:
- out['tenant_id'] = project.id
- else:
- raise RouterConfigError(
- 'Could not find project ID for project named - ' +
- 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, keystone, network_name=self.external_gateway)
- if ext_net:
- ext_gw['network_id'] = ext_net.id
- out['external_gateway_info'] = ext_gw
- else:
- raise RouterConfigError(
- 'Could not find the external network named - ' +
- self.external_gateway)
+ session = keystone_utils.keystone_session(os_creds)
+ keystone = keystone_utils.keystone_client(os_creds, session)
+ try:
+ if self.name:
+ out['name'] = self.name
+ if self.project_name:
+ project = keystone_utils.get_project(
+ keystone=keystone, project_name=self.project_name)
+ if project:
+ out['tenant_id'] = project.id
+ else:
+ raise RouterConfigError(
+ 'Could not find project ID for project named - ' +
+ 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, keystone, network_name=self.external_gateway)
+ if ext_net:
+ ext_gw['network_id'] = ext_net.id
+ out['external_gateway_info'] = ext_gw
+ else:
+ raise RouterConfigError(
+ 'Could not find the external network named - ' +
+ self.external_gateway)
+ finally:
+ keystone_utils.close_session(session)
return {'router': out}
self.admin_state_up == other.admin_state_up and
self.shared == other.shared and
self.external == other.external and
- self.type == other.type and
self.subnets == other.subnets)
self.__cluster_template = None
+ super(self.__class__, self).clean()
+
def get_cluster_template(self):
"""
Returns the domain Volume object as it was populated when create() was
self.__flavor = None
+ super(self.__class__, self).clean()
+
def get_flavor(self):
"""
Returns the OpenStack flavor object
Loads the existing Image
:return: The Image domain object or None
"""
- self.__glance = glance_utils.glance_client(self._os_creds)
+ super(self.__class__, self).initialize()
+
+ self.__glance = glance_utils.glance_client(
+ self._os_creds, self._os_session)
self.__image = glance_utils.get_image(
self.__glance, image_settings=self.image_settings)
self.__kernel_image = None
self.__ramdisk_image = None
+ if self.__glance:
+ self.__glance.http_client.session.session.close()
+
+ super(self.__class__, self).clean()
+
def get_image(self):
"""
Returns the domain Image object as it was populated when create() was
"""
super(self.__class__, self).initialize()
- self.__neutron = neutron_utils.neutron_client(self._os_creds)
- self.__keystone = keystone_utils.keystone_client(self._os_creds)
+ self.__neutron = neutron_utils.neutron_client(
+ self._os_creds, self._os_session)
+ self.__keystone = keystone_utils.keystone_client(
+ self._os_creds, self._os_session)
+ self.__cinder = cinder_utils.cinder_client(
+ self._os_creds, self._os_session)
+ self.__glance = glance_utils.glance_client(
+ self._os_creds, self._os_session)
self.__ports = self.__query_ports(self.instance_settings.port_settings)
self.__lookup_existing_vm_by_name()
active, error, or timeout waiting. Floating IPs will be
assigned after active when block=True
"""
- glance = glance_utils.glance_client(self._os_creds)
self.__vm = nova_utils.create_server(
- self._nova, self.__keystone, self.__neutron, glance,
+ self._nova, self.__keystone, self.__neutron, self.__glance,
self.instance_settings, self.image_settings,
self._os_creds.project_name, self.keypair_settings)
logger.info('Created instance with name - %s',
if self.instance_settings.volume_names:
for volume_name in self.instance_settings.volume_names:
- cinder = cinder_utils.cinder_client(self._os_creds)
volume = cinder_utils.get_volume(
- cinder, self.__keystone, volume_name=volume_name,
+ self.__cinder, self.__keystone, volume_name=volume_name,
project_name=self._os_creds.project_name)
if volume and self.vm_active(block=True):
if self.__vm:
# Detach Volume
for volume_rec in self.__vm.volume_ids:
- cinder = cinder_utils.cinder_client(self._os_creds)
volume = cinder_utils.get_volume_by_id(
- cinder, volume_rec['id'])
+ self.__cinder, volume_rec['id'])
if volume:
vm = nova_utils.detach_volume(
self._nova, self.__neutron, self.__keystone, self.__vm,
'VM not deleted within the timeout period of %s '
'seconds', self.instance_settings.vm_delete_timeout)
+ super(self.__class__, self).clean()
+
def __query_ports(self, port_settings):
"""
Returns the previously configured ports or an empty list if none
:param keypair_config: the associated KeypairConfig object (optional)
:return: an initialized OpenStackVmInstance object
"""
- nova = nova_utils.nova_client(os_creds)
- keystone = keystone_utils.keystone_client(os_creds)
- neutron = neutron_utils.neutron_client(os_creds)
- derived_inst_config = settings_utils.create_vm_inst_config(
- nova, keystone, neutron, vm_inst, project_name)
-
- derived_inst_creator = OpenStackVmInstance(
- os_creds, derived_inst_config, image_config, keypair_config)
- derived_inst_creator.initialize()
- return derived_inst_creator
+ session = keystone_utils.keystone_session(os_creds)
+ nova = nova_utils.nova_client(os_creds, session)
+ keystone = keystone_utils.keystone_client(os_creds, session)
+ neutron = neutron_utils.neutron_client(os_creds, session)
+
+ try:
+ derived_inst_config = settings_utils.create_vm_inst_config(
+ nova, keystone, neutron, vm_inst, project_name)
+
+ derived_inst_creator = OpenStackVmInstance(
+ os_creds, derived_inst_config, image_config, keypair_config)
+ derived_inst_creator.initialize()
+ return derived_inst_creator
+ finally:
+ keystone_utils.close_session(session)
class VmInstanceSettings(VmInstanceConfig):
os.remove(expanded_path)
logger.info('Deleted private key file [%s]', expanded_path)
+ super(self.__class__, self).clean()
+
def get_keypair(self):
"""
Returns the OpenStack keypair object
from snaps.config.network import NetworkConfig, SubnetConfig, PortConfig
from snaps.openstack.openstack_creator import OpenStackNetworkObject
-from snaps.openstack.utils import neutron_utils, keystone_utils
+from snaps.openstack.utils import neutron_utils
__author__ = 'spisarski'
super(self.__class__, self).initialize()
try:
- keystone = keystone_utils.keystone_client(self._os_creds)
self.__network = neutron_utils.get_network(
- self._neutron, keystone,
+ self._neutron, self._keystone,
network_settings=self.network_settings,
project_name=self._os_creds.project_name)
except Unauthorized as e:
"""
Removes and deletes all items created in reverse order.
"""
- if self.__network:
- try:
- neutron_utils.delete_network(self._neutron, self.__network)
- except NetworkNotFoundClient:
- pass
- self.__network = None
+ try:
+ neutron_utils.delete_network(self._neutron, self.__network)
+ except NetworkNotFoundClient:
+ pass
+ self.__network = None
+
+ super(self.__class__, self).clean()
def get_network(self):
"""
"""
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, self._keystone, sec_grp_name='default',
- project_name=self.__project.name)
- if default_sec_grp:
- try:
- neutron_utils.delete_security_group(
- neutron, default_sec_grp)
- except:
- pass
+ neutron = neutron_utils.neutron_client(
+ self._os_creds, self._os_session)
+ try:
+ default_sec_grp = neutron_utils.get_security_group(
+ neutron, self._keystone, sec_grp_name='default',
+ project_name=self.__project.name)
+ if default_sec_grp:
+ try:
+ neutron_utils.delete_security_group(
+ neutron, default_sec_grp)
+ except:
+ pass
+ finally:
+ neutron.httpclient.session.session.close()
# Delete Project
try:
if role:
keystone_utils.delete_role(self._keystone, role)
+ super(self.__class__, self).clean()
+
def get_project(self):
"""
Returns the OpenStack project object populated on create()
Returns the compute quotas as an instance of the ComputeQuotas class
:return:
"""
- nova = nova_utils.nova_client(self._os_creds)
- return nova_utils.get_compute_quotas(nova, self.__project.id)
+ nova = nova_utils.nova_client(self._os_creds, self._os_session)
+
+ try:
+ return nova_utils.get_compute_quotas(nova, self.__project.id)
+ finally:
+ nova.client.session.session.close()
def get_network_quotas(self):
"""
Returns the network quotas as an instance of the NetworkQuotas class
:return:
"""
- neutron = neutron_utils.neutron_client(self._os_creds)
- return neutron_utils.get_network_quotas(neutron, self.__project.id)
+ neutron = neutron_utils.neutron_client(
+ self._os_creds, self._os_session)
+ try:
+ return neutron_utils.get_network_quotas(neutron, self.__project.id)
+ finally:
+ neutron.httpclient.session.session.close()
def update_compute_quotas(self, compute_quotas):
"""
Updates the compute quotas for this project
:param compute_quotas: a ComputeQuotas object.
"""
- nova = nova_utils.nova_client(self._os_creds)
- nova_utils.update_quotas(nova, self.__project.id, compute_quotas)
+ nova = nova_utils.nova_client(self._os_creds, self._os_session)
+ try:
+ nova_utils.update_quotas(nova, self.__project.id, compute_quotas)
+ finally:
+ nova.client.session.session.close()
def update_network_quotas(self, network_quotas):
"""
Updates the network quotas for this project
:param network_quotas: a NetworkQuotas object.
"""
- neutron = neutron_utils.neutron_client(self._os_creds)
- neutron_utils.update_quotas(neutron, self.__project.id, network_quotas)
+ neutron = neutron_utils.neutron_client(
+ self._os_creds, self._os_session)
+ try:
+ neutron_utils.update_quotas(
+ neutron, self.__project.id, network_quotas)
+ finally:
+ neutron.httpclient.session.session.close()
class ProjectSettings(ProjectConfig):
from warnings import warn
warn('Use snaps.config.project.ProjectConfig instead',
DeprecationWarning)
- super(self.__class__, self).__init__(**kwargs)
\ No newline at end of file
+ super(self.__class__, self).__init__(**kwargs)
self.__qos = None
+ super(self.__class__, self).clean()
+
def get_qos(self):
"""
Returns the domain QoS object as it was populated when create() was
if not router_settings:
raise RouterCreationError('router_settings is required')
- self.__keystone = None
-
self.router_settings = router_settings
# Attributes instantiated on create()
"""
super(self.__class__, self).initialize()
- self.__keystone = keystone_utils.keystone_client(self._os_creds)
-
try:
self.__router = neutron_utils.get_router(
- self._neutron, self.__keystone,
+ self._neutron, self._keystone,
router_settings=self.router_settings,
project_name=self._os_creds.project_name)
except Unauthorized as e:
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
- self._neutron, self.__keystone, port_settings=port_setting,
+ self._neutron, self._keystone, port_settings=port_setting,
project_name=self._os_creds.project_name)
if port:
self.__ports.append(port)
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
- self._neutron, self.__keystone, port_settings=port_setting,
+ self._neutron, self._keystone, port_settings=port_setting,
project_name=self._os_creds.project_name)
logger.info(
'Retrieved port %s for router - %s', port_setting.name,
pass
self.__router = None
+ super(self.__class__, self).clean()
+
def get_router(self):
"""
Returns the OpenStack router object
from snaps.config.security_group import (
SecurityGroupConfig, SecurityGroupRuleConfig)
from snaps.openstack.openstack_creator import OpenStackNetworkObject
-from snaps.openstack.utils import keystone_utils
from snaps.openstack.utils import neutron_utils
__author__ = 'spisarski'
"""
super(self.__class__, self).initialize()
- keystone = keystone_utils.keystone_client(self._os_creds)
self.__security_group = neutron_utils.get_security_group(
- self._neutron, keystone, sec_grp_settings=self.sec_grp_settings,
+ self._neutron, self._keystone,
+ sec_grp_settings=self.sec_grp_settings,
project_name=self._os_creds.project_name)
if self.__security_group:
# Populate rules
logger.info(
'Creating security group %s...' % self.sec_grp_settings.name)
- keystone = keystone_utils.keystone_client(self._os_creds)
self.__security_group = neutron_utils.create_security_group(
- self._neutron, keystone, self.sec_grp_settings)
+ self._neutron, self._keystone, self.sec_grp_settings)
# Get the rules added for free
auto_rules = neutron_utils.get_rules_by_security_group(
for sec_grp_rule_setting in self.sec_grp_settings.rule_settings:
try:
custom_rule = neutron_utils.create_security_group_rule(
- self._neutron, keystone, sec_grp_rule_setting,
+ self._neutron, self._keystone, sec_grp_rule_setting,
self._os_creds.project_name)
self.__rules[sec_grp_rule_setting] = custom_rule
except Conflict as e:
self.__security_group = None
+ super(self.__class__, self).clean()
+
def get_security_group(self):
"""
Returns the OpenStack security group object
:param rule_setting: the rule configuration
"""
rule_setting.sec_grp_name = self.sec_grp_settings.name
- keystone = keystone_utils.keystone_client(self._os_creds)
new_rule = neutron_utils.create_security_group_rule(
- self._neutron, keystone, rule_setting, self._os_creds.project_name)
+ self._neutron, self._keystone, rule_setting,
+ self._os_creds.project_name)
self.__rules[rule_setting] = new_rule
self.sec_grp_settings.rule_settings.append(rule_setting)
from snaps.openstack.create_volume_type import OpenStackVolumeType
from snaps.openstack.openstack_creator import OpenStackCloudObject
from snaps.openstack.utils import (
- nova_utils, settings_utils, glance_utils, cinder_utils, keystone_utils)
+ nova_utils, settings_utils, glance_utils, cinder_utils)
from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.utils import heat_utils, neutron_utils
self.__stack = None
self.__heat_cli = None
+ self.__neutron = None
+ self.__nova = None
+ self.__glance = None
+ self.__cinder = None
+
def initialize(self):
"""
Loads the existing heat stack
:return: The Stack domain object or None
"""
- self.__heat_cli = heat_utils.heat_client(self._os_creds)
+ super(self.__class__, self).initialize()
+
+ self.__neutron = neutron_utils.neutron_client(
+ self._os_creds, self._os_session)
+ self.__nova = nova_utils.nova_client(self._os_creds, self._os_session)
+ self.__glance = glance_utils.glance_client(
+ self._os_creds, self._os_session)
+ self.__cinder = cinder_utils.cinder_client(
+ self._os_creds, self._os_session)
+
+ self.__heat_cli = heat_utils.heat_client(
+ self._os_creds, self._os_session)
self.__stack = heat_utils.get_stack(
self.__heat_cli, stack_settings=self.stack_settings)
if self.__stack:
self.__stack = None
+ self.__neutron.httpclient.session.session.close()
+ self.__nova.client.session.session.close()
+ self.__glance.http_client.session.session.close()
+ self.__cinder.client.session.session.close()
+
+ super(self.__class__, self).clean()
+
def get_stack(self):
"""
Returns the domain Stack object as it was populated when create() was
:return: list() of OpenStackNetwork objects
"""
- neutron = neutron_utils.neutron_client(self._os_creds)
-
out = list()
stack_networks = heat_utils.get_stack_networks(
- self.__heat_cli, neutron, self.__stack)
+ self.__heat_cli, self.__neutron, self.__stack)
for stack_network in stack_networks:
net_settings = settings_utils.create_network_config(
- neutron, stack_network)
+ self.__neutron, stack_network)
net_creator = OpenStackNetwork(self._os_creds, net_settings)
out.append(net_creator)
net_creator.initialize()
:return: list() of OpenStackNetwork objects
"""
- neutron = neutron_utils.neutron_client(self._os_creds)
-
out = list()
stack_security_groups = heat_utils.get_stack_security_groups(
- self.__heat_cli, neutron, self.__stack)
+ self.__heat_cli, self.__neutron, self.__stack)
for stack_security_group in stack_security_groups:
settings = settings_utils.create_security_group_config(
- neutron, stack_security_group)
+ self.__neutron, stack_security_group)
creator = OpenStackSecurityGroup(self._os_creds, settings)
out.append(creator)
creator.initialize()
:return: list() of OpenStackRouter objects
"""
- neutron = neutron_utils.neutron_client(self._os_creds)
-
out = list()
stack_routers = heat_utils.get_stack_routers(
- self.__heat_cli, neutron, self.__stack)
+ self.__heat_cli, self.__neutron, self.__stack)
for routers in stack_routers:
settings = settings_utils.create_router_config(
- neutron, routers)
+ self.__neutron, routers)
creator = OpenStackRouter(self._os_creds, settings)
out.append(creator)
creator.initialize()
out = list()
- nova = nova_utils.nova_client(self._os_creds)
- keystone = keystone_utils.keystone_client(self._os_creds)
- neutron = neutron_utils.neutron_client(self._os_creds)
stack_servers = heat_utils.get_stack_servers(
- self.__heat_cli, nova, neutron, keystone, self.__stack,
- self._os_creds.project_name)
-
- glance = glance_utils.glance_client(self._os_creds)
+ self.__heat_cli, self.__nova, self.__neutron, self._keystone,
+ self.__stack, self._os_creds.project_name)
for stack_server in stack_servers:
vm_inst_settings = settings_utils.create_vm_inst_config(
- nova, keystone, neutron, stack_server,
+ self.__nova, self._keystone, self.__neutron, stack_server,
self._os_creds.project_name)
image_settings = settings_utils.determine_image_config(
- glance, stack_server, self.image_settings)
+ self.__glance, stack_server, self.image_settings)
keypair_settings = settings_utils.determine_keypair_config(
self.__heat_cli, self.__stack, stack_server,
keypair_settings=self.keypair_settings,
"""
out = list()
- cinder = cinder_utils.cinder_client(self._os_creds)
-
volumes = heat_utils.get_stack_volumes(
- self.__heat_cli, cinder, self.__stack)
+ self.__heat_cli, self.__cinder, self.__stack)
for volume in volumes:
settings = settings_utils.create_volume_config(volume)
"""
out = list()
- cinder = cinder_utils.cinder_client(self._os_creds)
-
vol_types = heat_utils.get_stack_volume_types(
- self.__heat_cli, cinder, self.__stack)
+ self.__heat_cli, self.__cinder, self.__stack)
for volume in vol_types:
settings = settings_utils.create_volume_type_config(volume)
"""
out = list()
- nova = nova_utils.nova_client(self._os_creds)
keypairs = heat_utils.get_stack_keypairs(
- self.__heat_cli, nova, self.__stack)
+ self.__heat_cli, self.__nova, self.__stack)
for keypair in keypairs:
settings = settings_utils.create_keypair_config(
"""
out = list()
- nova = nova_utils.nova_client(self._os_creds)
flavors = heat_utils.get_stack_flavors(
- self.__heat_cli, nova, self.__stack)
+ self.__heat_cli, self.__nova, self.__stack)
for flavor in flavors:
settings = settings_utils.create_flavor_config(flavor)
pass
self.__user = None
+ super(self.__class__, self).clean()
+
def get_user(self):
"""
Returns the OpenStack user object populated in create()
from snaps.config.volume import VolumeConfig
from snaps.openstack.openstack_creator import OpenStackVolumeObject
-from snaps.openstack.utils import cinder_utils, keystone_utils
+from snaps.openstack.utils import cinder_utils
__author__ = 'spisarski'
self.volume_settings = volume_settings
self.__volume = None
- self.__keystone = None
def initialize(self):
"""
"""
super(self.__class__, self).initialize()
- self.__keystone = keystone_utils.keystone_client(self._os_creds)
self.__volume = cinder_utils.get_volume(
- self._cinder, self.__keystone,
+ self._cinder, self._keystone,
volume_settings=self.volume_settings,
project_name=self._os_creds.project_name)
return self.__volume
if not self.__volume:
self.__volume = cinder_utils.create_volume(
- self._cinder, self.__keystone, self.volume_settings)
+ self._cinder, self._keystone, self.volume_settings)
logger.info(
'Created volume with name - %s', self.volume_settings.name)
self.__volume = None
+ super(self.__class__, self).clean()
+
def get_volume(self):
"""
Returns the domain Volume object as it was populated when create() was
self.__volume_type = None
+ super(self.__class__, self).clean()
+
def get_volume_type(self):
"""
Returns the domain Volume object as it was populated when create() was
:param os_creds: the OpenStack credentials object
"""
self._os_creds = os_creds
+ self._os_session = None
+ self._keystone = None
def initialize(self):
- raise NotImplementedError('Do not override abstract method')
+ self._os_session = keystone_utils.keystone_session(self._os_creds)
+ self._keystone = keystone_utils.keystone_client(
+ self._os_creds, session=self._os_session)
def create(self):
raise NotImplementedError('Do not override abstract method')
def clean(self):
- raise NotImplementedError('Do not override abstract method')
+ if self._os_session:
+ keystone_utils.close_session(self._os_session)
class OpenStackComputeObject(OpenStackCloudObject):
self._nova = None
def initialize(self):
- self._nova = nova_utils.nova_client(self._os_creds)
+ super(OpenStackComputeObject, self).initialize()
+ self._nova = nova_utils.nova_client(self._os_creds, self._os_session)
def create(self):
raise NotImplementedError('Do not override abstract method')
- def clean(self):
- raise NotImplementedError('Do not override abstract method')
-
class OpenStackNetworkObject(OpenStackCloudObject):
"""
self._neutron = None
def initialize(self):
- self._neutron = neutron_utils.neutron_client(self._os_creds)
+ super(OpenStackNetworkObject, self).initialize()
+ self._neutron = neutron_utils.neutron_client(
+ self._os_creds, self._os_session)
def create(self):
raise NotImplementedError('Do not override abstract method')
- def clean(self):
- raise NotImplementedError('Do not override abstract method')
-
class OpenStackIdentityObject(OpenStackCloudObject):
"""
:param os_creds: the OpenStack credentials object
"""
super(OpenStackIdentityObject, self).__init__(os_creds)
- self._keystone = None
def initialize(self):
- self._keystone = keystone_utils.keystone_client(self._os_creds)
+ super(OpenStackIdentityObject, self).initialize()
def create(self):
raise NotImplementedError('Do not override abstract method')
- def clean(self):
- raise NotImplementedError('Do not override abstract method')
-
class OpenStackVolumeObject(OpenStackCloudObject):
"""
self._cinder = None
def initialize(self):
- self._cinder = cinder_utils.cinder_client(self._os_creds)
+ super(OpenStackVolumeObject, self).initialize()
+ self._cinder = cinder_utils.cinder_client(
+ self._os_creds, self._os_session)
def create(self):
raise NotImplementedError('Do not override abstract method')
- def clean(self):
- raise NotImplementedError('Do not override abstract method')
-
class OpenStackMagnumObject(OpenStackCloudObject):
"""
self._magnum = None
def initialize(self):
- self._magnum = magnum_utils.magnum_client(self._os_creds)
+ super(OpenStackMagnumObject, self).initialize()
+ self._magnum = magnum_utils.magnum_client(
+ self._os_creds, self._os_session)
def create(self):
raise NotImplementedError('Do not override abstract method')
-
- def clean(self):
- raise NotImplementedError('Do not override abstract method')
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.cluster_type_name = self.guid + '-cluster-type'
- self.magnum = magnum_utils.magnum_client(self.os_creds)
+ self.magnum = magnum_utils.magnum_client(
+ self.os_creds, self.os_session)
metadata = self.image_metadata
if not metadata:
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.flavor_name = guid + 'name'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
# Initialize for cleanup
self.flavor_creator = None
if self.flavor_creator:
self.flavor_creator.clean()
+ super(self.__class__, self).__clean__()
+
def test_create_flavor(self):
"""
Tests the creation of an OpenStack flavor.
guid = uuid.uuid4()
self.image_name = self.__class__.__name__ + '-' + str(guid)
- self.glance = glance_utils.glance_client(self.os_creds)
+ self.glance = glance_utils.glance_client(
+ self.os_creds, self.os_session)
self.image_creator = None
if self.image_metadata and 'glance_tests' in self.image_metadata:
clean() does not raise an Exception.
"""
# Create Image
- self.image_creator = create_image.OpenStackImage(self.os_creds,
- self.image_settings)
+ self.image_creator = create_image.OpenStackImage(
+ self.os_creds, self.image_settings)
created_image = self.image_creator.create()
self.assertIsNotNone(created_image)
guid = uuid.uuid4()
self.image_creators = list()
self.image_name = self.__class__.__name__ + '-' + str(guid)
- self.glance = glance_utils.glance_client(self.os_creds)
+ self.glance = glance_utils.glance_client(
+ self.os_creds, self.os_session)
self.tmp_dir = 'tmp/' + str(guid)
if not os.path.exists(self.tmp_dir):
OSIntegrationTestCase, OSComponentTestCase)
from snaps.openstack.utils import nova_utils, keystone_utils, neutron_utils
from snaps.openstack.utils.nova_utils import RebootType
+from snaps.openstack.utils import nova_utils, settings_utils, neutron_utils
__author__ = 'spisarski'
"""
super(self.__class__, self).__start__()
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.port_1_name = guid + 'port-1'
self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
- self.assertIsNotNone(nova_utils.get_server(
+ vm_inst_get = nova_utils.get_server(
self.nova, self.neutron, self.keystone,
- vm_inst_settings=instance_settings))
+ vm_inst_settings=instance_settings)
+ self.assertEqual(vm_inst, vm_inst_get)
self.assertIsNotNone(self.inst_creator.get_vm_inst().availability_zone)
self.assertIsNone(self.inst_creator.get_vm_inst().compute_host)
"""
super(self.__class__, self).__start__()
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
"""
super(self.__class__, self).__start__()
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + self.guid
self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
Tests the creation of OpenStack VM instances to each compute node.
"""
from snaps.openstack.utils import nova_utils
- nova = nova_utils.nova_client(self.admin_os_creds)
+ nova = nova_utils.nova_client(
+ self.admin_os_creds, self.admin_os_session)
zone_hosts = nova_utils.get_availability_zone_hosts(nova)
# Create Instance on each server/zone
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = self.guid + '-inst'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
os_image_settings = openstack_tests.cirros_image_settings(
name=self.guid + '-image', image_metadata=self.image_metadata)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.image_name = guid
self.vm_inst_name = guid + '-inst'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
if os.path.exists(self.tmpDir) and os.path.isdir(self.tmpDir):
shutil.rmtree(self.tmpDir)
+ super(self.__class__, self).__clean__()
+
def test_inst_from_file_image_simple_flat(self):
"""
Creates a VM instance from a locally sourced file image using simply
self.ip1 = '10.200.201.5'
self.ip2 = '10.200.202.5'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
# Initialize for tearDown()
self.image_creator = None
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
- self.nova = nova_utils.nova_client(self.os_creds)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.nova = nova_utils.nova_client(
+ self.os_creds, self.os_session)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.priv_file_path = 'tmp/' + guid
self.pub_file_path = self.priv_file_path + '.pub'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.keypair_name = guid
self.keypair_creator = None
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.priv_file_path = 'tmp/' + guid
self.pub_file_path = self.priv_file_path + '.pub'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.keypair_name = guid
self.keypair_creator = None
router_name=guid + '-pub-router', external_net=self.ext_net_name,
netconf_override=self.netconf_override)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.net_creator = None
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
self.ip_prfx = '10.1.0.'
super(self.__class__, self).__start__()
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.net_creator = None
self.net_config = openstack_tests.get_pub_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.net_creator = None
if self.net_creator:
self.net_creator.clean()
+ super(self.__class__, self).__clean__()
+
def test_create_network_type_vlan(self):
"""
Tests the creation of an OpenStack network of type vlan.
self.net_config = openstack_tests.get_pub_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.admin_net_creator = None
name=guid + '-name',
domain=self.os_creds.project_domain_name)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.project_creator = None
if self.project_creator:
self.project_creator.clean()
+ super(self.__class__, self).__clean__()
+
def test_create_project_bad_domain(self):
"""
Tests the creation of an OpenStack project with an invalid domain
self.assertEqual(update_network_quotas,
self.project_creator.get_network_quotas())
- nova = nova_utils.nova_client(self.os_creds)
+ nova = nova_utils.nova_client(self.os_creds, self.os_session)
new_compute_quotas = nova_utils.get_compute_quotas(
nova, self.project_creator.get_project().id)
self.assertEqual(update_compute_quotas, new_compute_quotas)
- neutron = neutron_utils.neutron_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
new_network_quotas = neutron_utils.get_network_quotas(
neutron, self.project_creator.get_project().id)
self.assertEqual(update_network_quotas, new_network_quotas)
name=self.guid + '-name',
domain=self.os_creds.project_domain_name)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.project_creator = None
if self.project_creator:
self.project_creator.clean()
+ super(self.__class__, self).__clean__()
+
def test_create_project_sec_grp_one_user(self):
"""
Tests the creation of an OpenStack object to a project with a new users
name=self.__class__.__name__ + '-' + str(guid),
consumer=Consumer.both)
- self.cinder = cinder_utils.cinder_client(self.admin_os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.admin_os_creds, self.admin_os_session)
self.qos_creator = create_qos.OpenStackQoS(
self.admin_os_creds, qos_settings)
self.router_creator = None
self.network_creator1 = None
self.network_creator2 = None
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.admin_router_creator = None
self.proj_router_creator = None
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
network_settings = NetworkConfig(
name=self.guid + '-pub-net', shared=True,
self.os_creds, SecurityGroupConfig(name=self.guid + '-sec_grp'))
self.sec_grp_creator.create()
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.sec_grp_name = guid + 'name'
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.sec_grp_creator = None
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.sec_grp_name = guid + 'name'
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
# Initialize for cleanup
self.admin_sec_grp_config = SecurityGroupConfig(
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
self.stack_creator = None
self.image_creator = OpenStackImage(
# Need to use 'admin' creds as heat creates objects under it's own
# project/tenant
- neutron = neutron_utils.neutron_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
net_by_name = neutron_utils.get_network(
neutron, keystone, network_name=net_creators[0].get_network().name)
self.assertEqual(net_creators[0].get_network(), net_by_name)
self.assertEqual(self.vm_inst_name,
vm_inst_creators[0].get_vm_inst().name)
- nova = nova_utils.nova_client(self.os_creds)
- neutron = neutron_utils.neutron_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ nova = nova_utils.nova_client(self.os_creds, self.os_session)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(self.os_creds, self.os_session)
vm_inst_by_name = nova_utils.get_server(
nova, neutron, keystone,
server_name=vm_inst_creators[0].get_vm_inst().name)
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
self.stack_creator = None
self.image_creator = OpenStackImage(
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
self.stack_creator = None
self.image_creator = OpenStackImage(
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
self.stack_creator = None
self.net_name = self.guid + '-net'
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
self.stack_creator = None
self.volume_name = self.guid + '-volume'
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
self.stack_creator = None
self.heat_tmplt_path = pkg_resources.resource_filename(
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.stack_creator = None
self.keypair_name = self.guid + '-kp'
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.stack_creator = None
self.security_group_name = self.guid + '-sec-grp'
def tearDown(self):
if self.stack_creator:
self.stack_creator.clean()
+
super(self.__class__, self).__clean__()
def test_missing_dependencies(self):
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.heat_cli = heat_utils.heat_client(self.os_creds)
+ self.heat_cli = heat_utils.heat_client(self.os_creds, self.os_session)
self.stack_creator = None
self.tmp_file = file_utils.save_string_to_file(
roles={'admin': self.os_creds.project_name},
domain_name=self.os_creds.user_domain_name)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(self.os_creds, self.os_session)
# Initialize for cleanup
self.user_creator = None
if self.user_creator:
self.user_creator.clean()
+ super(self.__class__, self).__clean__()
+
def test_create_user(self):
"""
Tests the creation of an OpenStack user.
self.volume_settings = VolumeConfig(
name=self.__class__.__name__ + '-' + str(guid))
- self.cinder = cinder_utils.cinder_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.volume_creator = None
def tearDown(self):
super(self.__class__, self).__start__()
self.guid = uuid.uuid4()
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
self.volume_creator = None
def tearDown(self):
def setUp(self):
super(self.__class__, self).__start__()
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.volume_name = guid + '-vol'
admin_vol = self.volume_creators[0].create(block=True)
self.assertIsNotNone(admin_vol)
- admin_key = keystone_utils.keystone_client(self.admin_os_creds)
+ admin_key = keystone_utils.keystone_client(
+ self.admin_os_creds, self.admin_os_session)
admin_proj = keystone_utils.get_project(
admin_key, project_name=self.admin_os_creds.project_name)
self.assertEqual(admin_vol.project_id, admin_proj.id)
- admin_cinder = cinder_utils.cinder_client(self.admin_os_creds)
+ admin_cinder = cinder_utils.cinder_client(
+ self.admin_os_creds, self.admin_os_session)
admin_vol_get = cinder_utils.get_volume(
admin_cinder, admin_key, volume_name=vol_name,
project_name=self.admin_os_creds.project_name)
self.assertNotEqual(admin_vol, proj_vol)
- proj_key = keystone_utils.keystone_client(self.os_creds)
- proj_cinder = cinder_utils.cinder_client(self.os_creds)
+ proj_key = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
+ proj_cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
proj_vol_get = cinder_utils.get_volume(
proj_cinder, proj_key, volume_name=vol_name,
project_name=self.os_creds.project_name)
self.volume_type_settings = VolumeTypeConfig(
name=self.__class__.__name__ + '-' + str(guid))
- self.cinder = cinder_utils.cinder_client(self.admin_os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.admin_os_creds, self.admin_os_session)
self.volume_type_creator = OpenStackVolumeType(
self.admin_os_creds, self.volume_type_settings)
def setUp(self):
super(self.__class__, self).__start__()
- self.cinder = cinder_utils.cinder_client(self.admin_os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.admin_os_creds, self.admin_os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.os_creds = openstack_tests.get_credentials(
dev_os_env_file=dev_os_env_file)
+ self.os_session = keystone_utils.keystone_session(self.os_creds)
self.ext_net_name = ext_net_name
if not self.ext_net_name and file_utils.file_exists(dev_os_env_file):
return suite
+ def __clean__(self):
+ """
+ Cleans up keystone session.
+ """
+ if self.os_session:
+ keystone_utils.close_session(self.os_session)
+
+
class OSIntegrationTestCase(OSComponentTestCase):
def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None,
self.project_creator = None
self.user_creator = None
self.admin_os_creds = self.os_creds
+ self.admin_os_session = self.os_session
+ keystone_utils.keystone_session(
+ self.admin_os_creds)
if self.use_keystone:
- self.keystone = keystone_utils.keystone_client(self.admin_os_creds)
+ self.keystone = keystone_utils.keystone_client(
+ self.admin_os_creds, self.admin_os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())[:-19]
project_name = guid + '-proj'
self.project_creator = deploy_utils.create_project(
self.os_creds = self.user_creator.get_os_creds(
self.project_creator.project_settings.name)
+ self.os_session = keystone_utils.keystone_session(self.os_creds)
# add user to project
self.project_creator.assoc_user(self.user_creator.get_user())
if self.user_creator:
self.user_creator.clean()
+
+ if self.admin_os_session:
+ keystone_utils.close_session(self.admin_os_session)
+
+ super(OSIntegrationTestCase, self).__clean__()
+
"""
-def cinder_client(os_creds):
+def cinder_client(os_creds, session=None):
"""
Creates and returns a cinder client object
+ :param os_creds: the credentials for connecting to the OpenStack remote API
+ :param session: the keystone session object (optional)
:return: the cinder client
"""
+ if not session:
+ session = keystone_utils.keystone_session(os_creds)
+
return Client(version=os_creds.volume_api_version,
- session=keystone_utils.keystone_session(os_creds),
+ session=session,
region_name=os_creds.region_name)
"""
-def glance_client(os_creds):
+def glance_client(os_creds, session=None):
"""
Creates and returns a glance client object
+ :param os_creds: the credentials for connecting to the OpenStack remote API
+ :param session: the keystone session object (optional)
:return: the glance client
"""
+ if not session:
+ session = keystone_utils.keystone_session(os_creds)
+
return Client(version=os_creds.image_api_version,
- session=keystone_utils.keystone_session(os_creds),
+ session=session,
region_name=os_creds.region_name)
logger = logging.getLogger('heat_utils')
-def heat_client(os_creds):
+def heat_client(os_creds, session=None):
"""
Retrieves the Heat client
:param os_creds: the OpenStack credentials
:return: the client
"""
logger.debug('Retrieving Heat Client')
+ if not session:
+ session = keystone_utils.keystone_session(os_creds)
return Client(os_creds.heat_api_version,
- session=keystone_utils.keystone_session(os_creds),
+ session=session,
region_name=os_creds.region_name)
# limitations under the License.
import logging
+import keystoneauth1
from keystoneclient.client import Client
from keystoneauth1.identity import v3, v2
from keystoneauth1 import session
verify=os_creds.cacert)
-def keystone_client(os_creds):
+def close_session(session):
+ """
+ Closes a keystone session
+ :param session: a session.Session object
+ """
+ if isinstance(session, keystoneauth1.session.Session):
+ session.session.close()
+
+
+def keystone_client(os_creds, session=None):
"""
Returns the keystone client
:param os_creds: the OpenStack credentials (OSCreds) object
+ :param session: the keystone session object (optional)
:return: the client
"""
+
+ if not session:
+ session = keystone_session(os_creds)
+
return Client(
version=os_creds.identity_api_version,
- session=keystone_session(os_creds),
+ session=session,
interface=os_creds.interface,
region_name=os_creds.region_name)
from snaps.openstack.create_volume import OpenStackVolume
from snaps.openstack.create_volume_type import OpenStackVolumeType
from snaps.openstack.os_credentials import OSCreds, ProxySettings
-from snaps.openstack.utils import deploy_utils, neutron_utils
+from snaps.openstack.utils import deploy_utils, neutron_utils, keystone_utils
from snaps.provisioning import ansible_utils
logger = logging.getLogger('lanuch_utils')
if router_creator:
if 'external_fixed_ip' == var_config_values.get('attr'):
- neutron = neutron_utils.neutron_client(os_creds)
- ext_nets = neutron_utils.get_external_networks(neutron)
-
- subnet_name = var_config_values.get('subnet_name')
-
- for ext_net in ext_nets:
- for subnet in ext_net.subnets:
- if subnet_name == subnet.name:
- router = router_creator.get_router()
- for fixed_ips in router.external_fixed_ips:
- if subnet.id == fixed_ips['subnet_id']:
- return fixed_ips['ip_address']
+ session = keystone_utils.keystone_session(os_creds)
+ neutron = neutron_utils.neutron_client(os_creds, session)
+ try:
+ ext_nets = neutron_utils.get_external_networks(neutron)
+
+ subnet_name = var_config_values.get('subnet_name')
+
+ for ext_net in ext_nets:
+ for subnet in ext_net.subnets:
+ if subnet_name == subnet.name:
+ router = router_creator.get_router()
+ for fixed_ips in router.external_fixed_ips:
+ if subnet.id == fixed_ips['subnet_id']:
+ return fixed_ips['ip_address']
+ finally:
+ keystone_utils.close_session(session)
def __get_vm_port_variable_value(var_config_values, vm_dict):
logger = logging.getLogger('magnum_utils')
-def magnum_client(os_creds):
+def magnum_client(os_creds, session=None):
"""
Retrieves the Magnum client
:param os_creds: the OpenStack credentialsf
+ :param session: the keystone session object (optional)
:return: the client
"""
logger.debug('Retrieving Magnum Client')
- return Client(str(os_creds.magnum_api_version),
- session=keystone_utils.keystone_session(os_creds))
+ if not session:
+ session = keystone_utils.keystone_session(os_creds)
+
+ return Client(str(os_creds.magnum_api_version), session=session)
def get_cluster_template(magnum, template_config=None, template_name=None):
"""
-def neutron_client(os_creds):
+def neutron_client(os_creds, session=None):
"""
Instantiates and returns a client for communications with OpenStack's
Neutron server
:param os_creds: the credentials for connecting to the OpenStack remote API
+ :param session: the keystone session object (optional)
:return: the client object
"""
+ if not session:
+ session = keystone_utils.keystone_session(os_creds)
return Client(api_version=os_creds.network_api_version,
- session=keystone_utils.keystone_session(os_creds),
+ session=session,
region_name=os_creds.region_name)
"""
-def nova_client(os_creds):
+def nova_client(os_creds, session=None):
"""
Instantiates and returns a client for communications with OpenStack's Nova
server
:param os_creds: The connection credentials to the OpenStack API
+ :param session: the keystone session object (optional)
:return: the client object
"""
logger.debug('Retrieving Nova Client')
+ if not session:
+ session = keystone_utils.keystone_session(os_creds)
+
return Client(os_creds.compute_api_version,
- session=keystone_utils.keystone_session(os_creds),
+ session=session,
region_name=os_creds.region_name)
"""
Tests to ensure that the proper credentials can connect.
"""
- cinder = cinder_utils.cinder_client(self.os_creds)
+ cinder = cinder_utils.cinder_client(self.os_creds, self.os_session)
volumes = cinder.volumes.list()
self.assertIsNotNone(volumes)
self.assertTrue(isinstance(volumes, list))
guid = uuid.uuid4()
self.volume_name = self.__class__.__name__ + '-' + str(guid)
self.volume = None
- self.cinder = cinder_utils.cinder_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
self.qos_name = self.__class__.__name__ + '-' + str(guid)
self.specs = {'foo': 'bar '}
self.qos = None
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except NotFound:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_qos_both(self):
"""
Tests the cinder_utils.create_qos()
volume_type_name = self.__class__.__name__ + '-' + str(guid)
self.volume_type_settings = VolumeTypeConfig(name=volume_type_name)
self.volume_type = None
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except NotFound:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_simple_volume_type(self):
"""
Tests the cinder_utils.create_volume_type(), get_volume_type(), and
self.encryption_name = self.__class__.__name__ + '-' + str(guid)
self.encryption = None
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type'
self.volume_type = cinder_utils.create_volume_type(
except NotFound:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_simple_encryption(self):
"""
Tests the cinder_utils.create_volume_encryption(),
self.qos_name = self.__class__.__name__ + '-' + str(guid) + '-qos'
self.vol_type_name = self.__class__.__name__ + '-' + str(guid)
self.specs = {'foo': 'bar'}
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
qos_settings = QoSConfig(
name=self.qos_name, specs=self.specs, consumer=Consumer.both)
self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
except NotFound:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_with_encryption(self):
"""
Tests the cinder_utils.create_volume_type() where encryption has been
"""
Tests to ensure that the proper credentials can connect.
"""
- glance = glance_utils.glance_client(self.os_creds)
+ glance = glance_utils.glance_client(self.os_creds, self.os_session)
image = glance_utils.get_image(glance, image_name='foo')
self.assertIsNone(image)
guid = uuid.uuid4()
self.image_name = self.__class__.__name__ + '-' + str(guid)
self.image = None
- self.glance = glance_utils.glance_client(self.os_creds)
+ self.glance = glance_utils.glance_client(
+ self.os_creds, self.os_session)
if self.image_metadata:
self.glance_test_meta = self.image_metadata.get('glance_tests')
else:
if os.path.exists(self.tmp_dir) and os.path.isdir(self.tmp_dir):
shutil.rmtree(self.tmp_dir)
+ super(self.__class__, self).__clean__()
+
def test_create_image_minimal_url(self):
"""
Tests the glance_utils.create_image() function with a URL unless the
"""
Tests to ensure that the proper credentials can connect.
"""
- heat = heat_utils.heat_client(self.os_creds)
+ heat = heat_utils.heat_client(self.os_creds, self.os_session)
# This should not throw an exception
stacks = heat.stacks.list()
env_values=env_values)
self.stack1 = None
self.stack2 = None
- self.heat_client = heat_utils.heat_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_stack(self):
"""
Tests the creation of an OpenStack Heat stack1 that does not exist.
self.assertTrue(stack_active(self.heat_client, self.stack1))
- neutron = neutron_utils.neutron_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
networks = heat_utils.get_stack_networks(
self.heat_client, neutron, self.stack1)
self.assertIsNotNone(networks)
self.assertEqual(1, len(subnets))
self.assertEqual(self.subnet_name, subnets[0].name)
- nova = nova_utils.nova_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ nova = nova_utils.nova_client(self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
servers = heat_utils.get_stack_servers(
self.heat_client, nova, neutron, keystone, self.stack1,
self.os_creds.project_name)
stack_settings = StackConfig(
name=stack_name, template_path=heat_tmplt_path,
env_values=env_values)
- self.heat_client = heat_utils.heat_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
self.stack = heat_utils.create_stack(self.heat_client, stack_settings)
self.assertTrue(stack_active(self.heat_client, self.stack))
time.sleep(3)
if not is_deleted:
- nova = nova_utils.nova_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
- neutron = neutron_utils.neutron_client(self.os_creds)
- glance = glance_utils.glance_client(self.os_creds)
+ nova = nova_utils.nova_client(
+ self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
+ neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ glance = glance_utils.glance_client(
+ self.os_creds, self.os_session)
+
servers = heat_utils.get_stack_servers(
self.heat_client, nova, neutron, keystone, self.stack,
self.os_creds.project_name)
os.chmod(expanded_path, 0o755)
os.remove(expanded_path)
+ super(self.__class__, self).__clean__()
+
def test_get_settings_from_stack(self):
"""
Tests that a heat template with floating IPs and can have the proper
self.assertIsNotNone(options)
self.assertEqual(1, len(options))
- neutron = neutron_utils.neutron_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
networks = heat_utils.get_stack_networks(
self.heat_client, neutron, self.stack)
self.assertIsNotNone(networks)
self.assertIsNotNone(network_settings)
self.assertEqual(self.network_name, network_settings.name)
- nova = nova_utils.nova_client(self.os_creds)
- glance = glance_utils.glance_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ nova = nova_utils.nova_client(self.os_creds, self.os_session)
+ glance = glance_utils.glance_client(self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
servers = heat_utils.get_stack_servers(
self.heat_client, nova, neutron, keystone, self.stack,
self.os_creds.project_name)
name=stack_name, template_path=heat_tmplt_path,
env_values=env_values)
self.stack = None
- self.heat_client = heat_utils.heat_client(self.os_creds)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_router_with_stack(self):
"""
Tests the creation of an OpenStack router with Heat and the retrieval
router = routers[0]
self.assertEqual(self.router_name, router.name)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
ext_net = neutron_utils.get_network(
self.neutron, keystone, network_name=self.ext_net_name)
self.assertEqual(ext_net.id, router.external_network_id)
name=stack_name, template_path=heat_tmplt_path,
env_values=env_values)
self.stack = None
- self.heat_client = heat_utils.heat_client(self.os_creds)
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_vol_with_stack(self):
"""
Tests the creation of an OpenStack volume with Heat.
self.stack_settings = StackConfig(
name=stack_name, template_path=heat_tmplt_path)
self.stack = None
- self.heat_client = heat_utils.heat_client(self.os_creds)
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
+ self.nova = nova_utils.nova_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_flavor_with_stack(self):
"""
Tests the creation of an OpenStack volume with Heat.
name=stack_name, template_path=heat_tmplt_path,
env_values=env_values)
self.stack = None
- self.heat_client = heat_utils.heat_client(self.os_creds)
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
+ self.nova = nova_utils.nova_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_keypair_with_stack(self):
"""
Tests the creation of an OpenStack keypair with Heat.
name=stack_name, template_path=heat_tmplt_path,
env_values=env_values)
self.stack = None
- self.heat_client = heat_utils.heat_client(self.os_creds)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.heat_client = heat_utils.heat_client(
+ self.os_creds, self.os_session)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_security_group_with_stack(self):
"""
Tests the creation of an OpenStack SecurityGroup with Heat.
"""
Tests to ensure that the proper credentials can connect.
"""
- keystone = keystone_utils.keystone_client(self.os_creds)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
users = keystone.users.list()
self.assertIsNotNone(users)
self.project_name = self.guid + '-projName'
self.project = None
self.role = None
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
Cleans the remote OpenStack objects
"""
if self.project:
- neutron = neutron_utils.neutron_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
default_sec_grp = neutron_utils.get_security_group(
neutron, self.keystone, sec_grp_name='default',
project_name=self.os_creds.project_name)
if self.role:
keystone_utils.delete_role(self.keystone, self.role)
+ super(self.__class__, self).__clean__()
+
def test_create_user_minimal(self):
"""
Tests the keystone_utils.create_user() function
"""
Tests to ensure that the proper credentials can connect.
"""
- magnum = magnum_utils.magnum_client(self.os_creds)
+ magnum = magnum_utils.magnum_client(
+ self.os_creds, self.os_session)
# This should not throw an exception
self.assertIsNotNone(magnum.clusters.list())
def setUp(self):
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.cluster_type_name = self.guid + '-cluster-type'
- self.magnum = magnum_utils.magnum_client(self.os_creds)
+ self.magnum = magnum_utils.magnum_client(
+ self.os_creds, self.os_session)
metadata = self.image_metadata
if not metadata:
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_cluster_template_simple(self):
config = ClusterTemplateConfig(
name=self.cluster_type_name,
"""
Tests to ensure that the proper credentials can connect.
"""
- neutron = neutron_utils.neutron_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
networks = neutron.list_networks()
configured self.ext_net_name is contained within the returned list
:return:
"""
- neutron = neutron_utils.neutron_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
ext_networks = neutron_utils.get_external_networks(neutron)
found = False
for network in ext_networks:
def setUp(self):
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.port_name = str(guid) + '-port'
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.network = None
self.net_config = openstack_tests.get_pub_net_config(
net_name=guid + '-pub-net')
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
+ super(self.__class__, self).__clean__()
+
def test_create_network(self):
"""
Tests the neutron_utils.create_network() function
def setUp(self):
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.port_name = str(guid) + '-port'
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.network = None
self.net_config = openstack_tests.get_pub_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_subnet(self):
"""
Tests the neutron_utils.create_network() function
def setUp(self):
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
self.network = None
def tearDown(self):
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_network_slaac(self):
"""
Tests the neutron_utils.create_network() with an IPv6 subnet where DHCP
def setUp(self):
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.port_name = str(guid) + '-port'
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.network = None
self.port = None
self.router = None
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
+ super(self.__class__, self).__clean__()
+
def test_create_router_simple(self):
"""
Tests the neutron_utils.create_router()
self.security_groups = list()
self.security_group_rules = list()
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_delete_simple_sec_grp(self):
"""
Tests the neutron_utils.create_security_group() function
for free_rule in free_rules:
self.security_group_rules.append(free_rule)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.security_group_rules.append(
neutron_utils.create_security_group_rule(
self.neutron, keystone, sec_grp_settings.rule_settings[0],
Instantiates the CreateImage object that is responsible for downloading
and creating an OS image file within OpenStack
"""
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.floating_ip = None
def tearDown(self):
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_floating_ips(self):
"""
Tests the creation of a floating IP
"""
Tests to ensure that the proper credentials can connect.
"""
- nova = nova_utils.nova_client(self.os_creds)
+ nova = nova_utils.nova_client(self.os_creds, self.os_session)
# This should not throw an exception
nova.flavors.list()
"""
Tests to ensure that get_hypervisors() function works.
"""
- nova = nova_utils.nova_client(self.os_creds)
+ nova = nova_utils.nova_client(self.os_creds, self.os_session)
hosts = nova_utils.get_hypervisor_hosts(nova)
# This should not throw an exception
self.priv_key_file_path = 'tmp/' + guid
self.pub_key_file_path = self.priv_key_file_path + '.pub'
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.keys = nova_utils.create_keys()
self.public_key = nova_utils.public_key_openssh(self.keys)
self.keypair_name = guid
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_keypair(self):
"""
Tests the creation of an OpenStack keypair that does not exist.
self.flavor_settings = FlavorConfig(
name=guid + '-name', flavor_id=guid + '-id', ram=1, disk=1,
vcpus=1, ephemeral=1, swap=2, rxtx_factor=3.0, is_public=False)
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
self.flavor = None
def tearDown(self):
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_flavor(self):
"""
Tests the creation of an OpenStack keypair that does not exist.
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.nova = nova_utils.nova_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
- self.glance = glance_utils.glance_client(self.os_creds)
+ self.nova = nova_utils.nova_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ self.glance = glance_utils.glance_client(
+ self.os_creds, self.os_session)
self.image_creator = None
self.network_creator = None
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_create_instance(self):
"""
Tests the nova_utils.create_server() method
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.nova = nova_utils.nova_client(self.os_creds)
- self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
+ self.cinder = cinder_utils.cinder_client(
+ self.os_creds, self.os_session)
self.image_creator = None
self.network_creator = None
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_add_remove_volume(self):
"""
Tests the nova_utils.attach_volume() and detach_volume functions with
self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
# Attach volume to VM
- neutron = neutron_utils.neutron_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
self.assertIsNotNone(nova_utils.attach_volume(
self.nova, neutron, keystone, self.instance_creator.get_vm_inst(),
self.volume_creator.get_volume(), self.os_creds.project_name))
self.assertTrue(attached)
self.assertIsNotNone(vol_attach)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
vm_attach = nova_utils.get_server_object_by_id(
self.nova, neutron, keystone,
self.instance_creator.get_vm_inst().id, self.os_creds.project_name)
self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
# Attach volume to VM
- neutron = neutron_utils.neutron_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
with self.assertRaises(NovaException):
nova_utils.attach_volume(
self.nova, neutron, keystone,
self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
# Attach volume to VM
- neutron = neutron_utils.neutron_client(self.os_creds)
- keystone = keystone_utils.keystone_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds, self.os_session)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
nova_utils.attach_volume(
self.nova, neutron, keystone, self.instance_creator.get_vm_inst(),
self.volume_creator.get_volume(), self.os_creds.project_name)
# Check VmInst for attachment
- keystone = keystone_utils.keystone_client(self.os_creds)
+ keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
latest_vm = nova_utils.get_server_object_by_id(
self.nova, neutron, keystone,
self.instance_creator.get_vm_inst().id, self.os_creds.project_name)
self.network_name = guid + '-net'
self.subnet_name = guid + '-subnet'
self.net_creator = None
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
def tearDown(self):
"""
except:
pass
+ super(self.__class__, self).__clean__()
+
def test_derive_net_settings_no_subnet(self):
"""
Validates the utility function settings_utils#create_network_config
Instantiates the CreateImage object that is responsible for downloading
and creating an OS image file within OpenStack
"""
- self.nova = nova_utils.nova_client(self.os_creds)
- self.keystone = keystone_utils.keystone_client(self.os_creds)
- self.glance = glance_utils.glance_client(self.os_creds)
- self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.nova = nova_utils.nova_client(
+ self.os_creds, self.os_session)
+ self.keystone = keystone_utils.keystone_client(
+ self.os_creds, self.os_session)
+ self.glance = glance_utils.glance_client(
+ self.os_creds, self.os_session)
+ self.neutron = neutron_utils.neutron_client(
+ self.os_creds, self.os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
if os.path.isfile(self.test_file_local_path):
os.remove(self.test_file_local_path)
- # super(self.__class__, self).__clean__()
+ super(self.__class__, self).__clean__()
def test_derive_vm_inst_config(self):
"""
"""
super(self.__class__, self).__start__()
- self.nova = nova_utils.nova_client(self.os_creds)
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
Code Review / snaps.git / commitdiff