summaryrefslogtreecommitdiffstats
path: root/snaps
diff options
context:
space:
mode:
authorSteven Pisarski <s.pisarski@cablelabs.com>2017-06-21 13:35:11 +0000
committerGerrit Code Review <gerrit@opnfv.org>2017-06-21 13:35:11 +0000
commit7055a581be2496365952c88473889f4389f60fb3 (patch)
tree9b1d0d6ee80a265a3d487c3cde8e75bce3595ca9 /snaps
parent104f7e15a579b3187a0cdf819bac77406e536938 (diff)
parent003bdd3bff285e893dcaf7bab211d8b82ce488fd (diff)
Merge "Changes to VMInstanceSettings and FloatingIPSettings constructors."
Diffstat (limited to 'snaps')
-rw-r--r--snaps/openstack/create_instance.py525
-rw-r--r--snaps/openstack/tests/create_instance_tests.py1240
2 files changed, 1134 insertions, 631 deletions
diff --git a/snaps/openstack/create_instance.py b/snaps/openstack/create_instance.py
index 85e96a8..991aca5 100644
--- a/snaps/openstack/create_instance.py
+++ b/snaps/openstack/create_instance.py
@@ -18,11 +18,11 @@ import time
from neutronclient.common.exceptions import PortNotFoundClient
from novaclient.exceptions import NotFound
+from snaps.openstack.create_network import PortSettings
from snaps.openstack.utils import glance_utils
from snaps.openstack.utils import neutron_utils
-from snaps.openstack.create_network import PortSettings
-from snaps.provisioning import ansible_utils
from snaps.openstack.utils import nova_utils
+from snaps.provisioning import ansible_utils
__author__ = 'spisarski'
@@ -38,7 +38,8 @@ class OpenStackVmInstance:
Class responsible for creating a VM instance in OpenStack
"""
- def __init__(self, os_creds, instance_settings, image_settings, keypair_settings=None):
+ def __init__(self, os_creds, instance_settings, image_settings,
+ keypair_settings=None):
"""
Constructor
:param os_creds: The connection credentials to the OpenStack API
@@ -56,7 +57,8 @@ class OpenStackVmInstance:
self.image_settings = image_settings
self.keypair_settings = keypair_settings
- # TODO - get rid of FIP list and only use the dict(). Need to fix populating this object when already exists
+ # TODO - get rid of FIP list and only use the dict(). Need to fix
+ # populating this object when already exists
self.__floating_ips = list()
self.__floating_ip_dict = dict()
@@ -70,14 +72,17 @@ class OpenStackVmInstance:
"""
Creates a VM instance
:param cleanup: When true, only perform lookups for OpenStack objects.
- :param block: Thread will block until instance has either become active, error, or timeout waiting.
- Additionally, when True, floating IPs will not be applied until VM is active.
+ :param block: Thread will block until instance has either become
+ active, error, or timeout waiting.
+ Additionally, when True, floating IPs will not be applied
+ until VM is active.
:return: The VM reference object
"""
self.__nova = nova_utils.nova_client(self.__os_creds)
self.__neutron = neutron_utils.neutron_client(self.__os_creds)
- self.__ports = self.__setup_ports(self.instance_settings.port_settings, cleanup)
+ self.__ports = self.__setup_ports(self.instance_settings.port_settings,
+ cleanup)
self.__lookup_existing_vm_by_name()
if not self.__vm and not cleanup:
self.__create_vm(block)
@@ -85,25 +90,31 @@ class OpenStackVmInstance:
def __lookup_existing_vm_by_name(self):
"""
- Populates the member variables 'self.vm' and 'self.floating_ips' if a VM with the same name already exists
+ Populates the member variables 'self.vm' and 'self.floating_ips' if a
+ VM with the same name already exists
within the project
"""
- servers = nova_utils.get_servers_by_name(self.__nova, self.instance_settings.name)
+ servers = nova_utils.get_servers_by_name(self.__nova,
+ self.instance_settings.name)
for server in servers:
if server.name == self.instance_settings.name:
self.__vm = server
- logger.info('Found existing machine with name - ' + self.instance_settings.name)
+ logger.info(
+ 'Found existing machine with name - %s',
+ self.instance_settings.name)
fips = self.__nova.floating_ips.list()
for fip in fips:
if fip.instance_id == server.id:
self.__floating_ips.append(fip)
- # TODO - Determine a means to associate to the FIP configuration and add to FIP map
+ # TODO - Determine a means to associate to the FIP
+ # configuration and add to FIP map
def __create_vm(self, block=False):
"""
Responsible for creating the VM instance
- :param block: Thread will block until instance has either become active, error, or timeout waiting.
- Floating IPs will be assigned after active when block=True
+ :param block: Thread will block until instance has either become
+ active, error, or timeout waiting. Floating IPs will be
+ assigned after active when block=True
"""
nics = []
for key, port in self.__ports:
@@ -116,12 +127,16 @@ class OpenStackVmInstance:
if self.keypair_settings:
keypair_name = self.keypair_settings.name
- flavor = nova_utils.get_flavor_by_name(self.__nova, self.instance_settings.flavor)
+ flavor = nova_utils.get_flavor_by_name(self.__nova,
+ self.instance_settings.flavor)
if not flavor:
- raise Exception('Flavor not found with name - ' + self.instance_settings.flavor)
+ raise Exception(
+ 'Flavor not found with name - %s',
+ self.instance_settings.flavor)
- image = glance_utils.get_image(glance_utils.glance_client(self.__os_creds),
- self.image_settings.name)
+ image = glance_utils.get_image(
+ glance_utils.glance_client(self.__os_creds),
+ self.image_settings.name)
if image:
self.__vm = self.__nova.servers.create(
name=self.instance_settings.name,
@@ -134,22 +149,32 @@ class OpenStackVmInstance:
availability_zone=self.instance_settings.availability_zone)
else:
- raise Exception('Cannot create instance, image cannot be located with name ' + self.image_settings.name)
+ raise Exception(
+ 'Cannot create instance, image cannot be located with name %s',
+ self.image_settings.name)
- logger.info('Created instance with name - ' + self.instance_settings.name)
+ logger.info(
+ 'Created instance with name - %s', self.instance_settings.name)
if block:
if not self.vm_active(block=True):
- raise Exception('Fatal error, VM did not become ACTIVE within the alloted time')
+ raise Exception(
+ 'Fatal error, VM did not become ACTIVE within the alloted '
+ 'time')
- # TODO - the call above should add security groups. The return object shows they exist but the association
- # had never been made by OpenStack. This call is here to ensure they have been added
+ # TODO - the call above should add security groups. The return object
+ # shows they exist but the association had never been made by
+ # OpenStack. This call is here to ensure they have been added
for sec_grp_name in self.instance_settings.security_group_names:
if self.vm_active(block=True):
- nova_utils.add_security_group(self.__nova, self.__vm, sec_grp_name)
+ nova_utils.add_security_group(self.__nova, self.__vm,
+ sec_grp_name)
else:
- raise Exception('Cannot applying security group with name ' + sec_grp_name +
- ' to VM that did not activate with name - ' + self.instance_settings.name)
+ raise Exception(
+ 'Cannot applying security group with name ' +
+ sec_grp_name +
+ ' to VM that did not activate with name - ' +
+ self.instance_settings.name)
self.__apply_floating_ips()
@@ -166,33 +191,43 @@ class OpenStackVmInstance:
port = port_dict.get(floating_ip_setting.port_name)
if not port:
- raise Exception('Cannot find port object with name - ' + floating_ip_setting.port_name)
-
- # Setup Floating IP only if there is a router with an external gateway
- ext_gateway = self.__ext_gateway_by_router(floating_ip_setting.router_name)
+ raise Exception(
+ 'Cannot find port object with name - ' +
+ floating_ip_setting.port_name)
+
+ # Setup Floating IP only if there is a router with an external
+ # gateway
+ 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)
- floating_ip = nova_utils.create_floating_ip(self.__nova, ext_gateway)
+ subnet = neutron_utils.get_subnet_by_name(
+ self.__neutron, floating_ip_setting.subnet_name)
+ floating_ip = nova_utils.create_floating_ip(self.__nova,
+ ext_gateway)
self.__floating_ips.append(floating_ip)
self.__floating_ip_dict[floating_ip_setting.name] = floating_ip
- logger.info('Created floating IP ' + floating_ip.ip + ' via router - ' +
- floating_ip_setting.router_name)
+ logger.info(
+ 'Created floating IP %s via router - %s', floating_ip.ip,
+ floating_ip_setting.router_name)
self.__add_floating_ip(floating_ip, port, subnet)
else:
- raise Exception('Unable to add floating IP to port,' +
- ' cannot locate router with an external gateway ')
+ raise Exception('Unable to add floating IP to port,'
+ ' cannot locate router with an external '
+ 'gateway ')
def __ext_gateway_by_router(self, router_name):
"""
- Returns network name for the external network attached to a router or None if not found
+ Returns network name for the external network attached to a router or
+ None if not found
: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)
if router and router['router'].get('external_gateway_info'):
- network = neutron_utils.get_network_by_id(self.__neutron,
- router['router']['external_gateway_info']['network_id'])
+ network = neutron_utils.get_network_by_id(
+ self.__neutron,
+ router['router']['external_gateway_info']['network_id'])
if network:
return network['network']['name']
return None
@@ -218,48 +253,58 @@ class OpenStackVmInstance:
try:
neutron_utils.delete_port(self.__neutron, port)
except PortNotFoundClient as e:
- logger.warning('Unexpected error deleting port - ' + str(e))
+ logger.warning('Unexpected error deleting port - %s', e)
pass
self.__ports = list()
# Cleanup VM
if self.__vm:
try:
- logger.info('Deleting VM instance - ' + self.instance_settings.name)
+ logger.info(
+ 'Deleting VM instance - ' + self.instance_settings.name)
nova_utils.delete_vm_instance(self.__nova, self.__vm)
except Exception as e:
- logger.error('Error deleting VM - ' + str(e))
+ logger.error('Error deleting VM - %s', e)
- # Block until instance cannot be found or returns the status of DELETED
+ # Block until instance cannot be found or returns the status of
+ # DELETED
logger.info('Checking deletion status')
try:
if self.vm_deleted(block=True):
- logger.info('VM has been properly deleted VM with name - ' + self.instance_settings.name)
+ logger.info(
+ 'VM has been properly deleted VM with name - %s',
+ self.instance_settings.name)
self.__vm = None
else:
- logger.error('VM not deleted within the timeout period of ' +
- str(self.instance_settings.vm_delete_timeout) + ' seconds')
+ logger.error(
+ 'VM not deleted within the timeout period of %s '
+ 'seconds', self.instance_settings.vm_delete_timeout)
except Exception as e:
- logger.error('Unexpected error while checking VM instance status - ' + str(e))
+ logger.error(
+ 'Unexpected error while checking VM instance status - %s',
+ e)
def __setup_ports(self, port_settings, cleanup):
"""
- Returns the previously configured ports or creates them if they do not exist
+ Returns the previously configured ports or creates them if they do not
+ exist
:param port_settings: A list of PortSetting objects
:param cleanup: When true, only perform lookups for OpenStack objects.
- :return: a list of OpenStack port tuples where the first member is the port name and the second is the port
- object
+ :return: a list of OpenStack port tuples where the first member is the
+ port name and the second is the port object
"""
ports = list()
for port_setting in port_settings:
# First check to see if network already has this port
- # TODO/FIXME - this could potentially cause problems if another port with the same name exists
+ # TODO/FIXME - this could potentially cause problems if another
+ # port with the same name exists
# VM has the same network/port name pair
found = False
- # TODO/FIXME - should we not be iterating on ports for the specific network in question as unique port names
+ # TODO/FIXME - should we not be iterating on ports for the specific
+ # network in question as unique port names
# seem to only be important by network
existing_ports = self.__neutron.list_ports()['ports']
for existing_port in existing_ports:
@@ -269,12 +314,15 @@ class OpenStackVmInstance:
break
if not found and not cleanup:
- ports.append((port_setting.name, neutron_utils.create_port(self.__neutron, self.__os_creds,
- port_setting)))
+ ports.append((port_setting.name,
+ neutron_utils.create_port(self.__neutron,
+ self.__os_creds,
+ port_setting)))
return ports
- def __add_floating_ip(self, floating_ip, port, subnet, timeout=30, poll_interval=POLL_INTERVAL):
+ def __add_floating_ip(self, floating_ip, port, subnet, timeout=30,
+ poll_interval=POLL_INTERVAL):
"""
Returns True when active else False
TODO - Make timeout and poll_interval configurable...
@@ -282,7 +330,8 @@ class OpenStackVmInstance:
ip = None
if subnet:
- # Take IP of subnet if there is one configured on which to place the floating IP
+ # Take IP of subnet if there is one configured on which to place
+ # the floating IP
for fixed_ip in port['port']['fixed_ips']:
if fixed_ip['subnet_id'] == subnet['subnet']['id']:
ip = fixed_ip['ip_address']
@@ -297,16 +346,20 @@ class OpenStackVmInstance:
logger.debug('Attempting to add floating IP to instance')
try:
self.__vm.add_floating_ip(floating_ip, ip)
- logger.info('Added floating IP ' + floating_ip.ip + ' to port IP - ' + ip +
- ' on instance - ' + self.instance_settings.name)
+ logger.info(
+ 'Added floating IP %s to port IP %s on instance %s',
+ floating_ip.ip, ip, self.instance_settings.name)
return
except Exception as e:
- logger.debug('Retry adding floating IP to instance. Last attempt failed with - ' + str(e))
+ logger.debug(
+ 'Retry adding floating IP to instance. Last attempt '
+ 'failed with - %s', e)
time.sleep(poll_interval)
count -= 1
pass
else:
- raise Exception('Unable find IP address on which to place the floating IP')
+ raise Exception(
+ 'Unable find IP address on which to place the floating IP')
logger.error('Timeout attempting to add the floating IP to instance.')
raise Exception('Timeout while attempting add floating IP to instance')
@@ -327,8 +380,9 @@ class OpenStackVmInstance:
def get_port_ip(self, port_name, subnet_name=None):
"""
- Returns the first IP for the port corresponding with the port_name parameter when subnet_name is None
- else returns the IP address that corresponds to the subnet_name parameter
+ Returns the first IP for the port corresponding with the port_name
+ parameter when subnet_name is None else returns the IP address that
+ corresponds to the subnet_name parameter
:param port_name: the name of the port from which to return the IP
:param subnet_name: the name of the subnet attached to this IP
:return: the IP or None if not found
@@ -337,9 +391,12 @@ class OpenStackVmInstance:
if port:
port_dict = port['port']
if subnet_name:
- subnet = neutron_utils.get_subnet_by_name(self.__neutron, subnet_name)
+ subnet = neutron_utils.get_subnet_by_name(self.__neutron,
+ subnet_name)
if not subnet:
- logger.warning('Cannot retrieve port IP as subnet could not be located with name - ' + subnet_name)
+ logger.warning('Cannot retrieve port IP as subnet could '
+ 'not be located with name - %s',
+ subnet_name)
return None
for fixed_ip in port_dict['fixed_ips']:
if fixed_ip['subnet_id'] == subnet['subnet']['id']:
@@ -352,8 +409,10 @@ class OpenStackVmInstance:
def get_port_mac(self, port_name):
"""
- Returns the first IP for the port corresponding with the port_name parameter
- TODO - Add in the subnet as an additional parameter as a port may have multiple fixed_ips
+ Returns the first IP for the port corresponding with the port_name
+ parameter
+ TODO - Add in the subnet as an additional parameter as a port may have
+ multiple fixed_ips
:param port_name: the name of the port from which to return the IP
:return: the IP or None if not found
"""
@@ -377,7 +436,8 @@ class OpenStackVmInstance:
def config_nics(self):
"""
- Responsible for configuring NICs on RPM systems where the instance has more than one configured port
+ Responsible for configuring NICs on RPM systems where the instance has
+ more than one configured port
:return: None
"""
if len(self.__ports) > 1 and len(self.__floating_ips) > 0:
@@ -386,12 +446,16 @@ class OpenStackVmInstance:
port_index = self.__ports.index((key, port))
if port_index > 0:
nic_name = 'eth' + repr(port_index)
- self.__config_nic(nic_name, port, self.__get_first_provisioning_floating_ip().ip)
- logger.info('Configured NIC - ' + nic_name + ' on VM - ' + self.instance_settings.name)
+ self.__config_nic(
+ nic_name, port,
+ self.__get_first_provisioning_floating_ip().ip)
+ logger.info('Configured NIC - %s on VM - %s',
+ nic_name, self.instance_settings.name)
def __get_first_provisioning_floating_ip(self):
"""
- Returns the first floating IP tagged with the Floating IP name if exists else the first one found
+ Returns the first floating IP tagged with the Floating IP name if
+ exists else the first one found
:return:
"""
for floating_ip_setting in self.instance_settings.floating_ip_settings:
@@ -404,10 +468,8 @@ class OpenStackVmInstance:
def __config_nic(self, nic_name, port, floating_ip):
"""
- Although ports/NICs can contain multiple IPs, this code currently only supports the first.
-
- Your CWD at this point must be the <repo dir>/python directory.
- TODO - fix this restriction.
+ Although ports/NICs can contain multiple IPs, this code currently only
+ supports the first.
:param nic_name: Name of the interface
:param port: The port information containing the expected IP values.
@@ -422,27 +484,33 @@ class OpenStackVmInstance:
}
if self.image_settings.nic_config_pb_loc and self.keypair_settings:
- return self.apply_ansible_playbook(self.image_settings.nic_config_pb_loc, variables)
+ return self.apply_ansible_playbook(
+ self.image_settings.nic_config_pb_loc, variables)
else:
- logger.warning('VM ' + self.instance_settings.name + ' cannot self configure NICs eth1++. ' +
- 'No playbook or keypairs found.')
+ logger.warning(
+ 'VM %s cannot self configure NICs eth1++. No playbook or '
+ 'keypairs found.', self.instance_settings.name)
- def apply_ansible_playbook(self, pb_file_loc, variables=None, fip_name=None):
+ def apply_ansible_playbook(self, pb_file_loc, variables=None,
+ fip_name=None):
"""
Applies a playbook to a VM
:param pb_file_loc: the file location of the playbook to be applied
- :param variables: a dict() of substitution values required by the playbook
- :param fip_name: the name of the floating IP to use for applying the playbook (default - will take the first)
+ :param variables: a dict() of substitution values required by the
+ playbook
+ :param fip_name: the name of the floating IP to use for applying the
+ playbook (default - will take the first)
:return: the return value from ansible
"""
- return ansible_utils.apply_playbook(pb_file_loc, [self.get_floating_ip(fip_name=fip_name).ip],
- self.get_image_user(), self.keypair_settings.private_filepath,
- variables, self.__os_creds.proxy_settings)
+ return ansible_utils.apply_playbook(
+ pb_file_loc, [self.get_floating_ip(fip_name=fip_name).ip],
+ self.get_image_user(), self.keypair_settings.private_filepath,
+ variables, self.__os_creds.proxy_settings)
def get_image_user(self):
"""
- Returns the instance sudo_user if it has been configured in the instance_settings else it returns the
- image_settings.image_user value
+ Returns the instance sudo_user if it has been configured in the
+ instance_settings else it returns the image_settings.image_user value
"""
if self.instance_settings.sudo_user:
return self.instance_settings.sudo_user
@@ -451,33 +519,45 @@ class OpenStackVmInstance:
def vm_deleted(self, block=False, poll_interval=POLL_INTERVAL):
"""
- Returns true when the VM status returns the value of expected_status_code or instance retrieval throws
- a NotFound exception.
- :param block: When true, thread will block until active or timeout value in seconds has been exceeded (False)
+ Returns true when the VM status returns the value of
+ expected_status_code or instance retrieval throws a NotFound exception.
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
:param poll_interval: The polling interval in seconds
:return: T/F
"""
try:
- return self.__vm_status_check(STATUS_DELETED, block, self.instance_settings.vm_delete_timeout,
- poll_interval)
+ return self.__vm_status_check(
+ STATUS_DELETED, block,
+ self.instance_settings.vm_delete_timeout, poll_interval)
except NotFound as e:
- logger.debug("Instance not found when querying status for " + STATUS_DELETED + ' with message ' + str(e))
+ logger.debug(
+ "Instance not found when querying status for %s with message "
+ "%s", STATUS_DELETED, e)
return True
def vm_active(self, block=False, poll_interval=POLL_INTERVAL):
"""
- Returns true when the VM status returns the value of expected_status_code
- :param block: When true, thread will block until active or timeout value in seconds has been exceeded (False)
+ Returns true when the VM status returns the value of
+ expected_status_code
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
:param poll_interval: The polling interval in seconds
:return: T/F
"""
- return self.__vm_status_check(STATUS_ACTIVE, block, self.instance_settings.vm_boot_timeout, poll_interval)
+ return self.__vm_status_check(STATUS_ACTIVE, block,
+ self.instance_settings.vm_boot_timeout,
+ poll_interval)
- def __vm_status_check(self, expected_status_code, block, timeout, poll_interval):
+ def __vm_status_check(self, expected_status_code, block, timeout,
+ poll_interval):
"""
- Returns true when the VM status returns the value of expected_status_code
- :param expected_status_code: instance status evaluated with this string value
- :param block: When true, thread will block until active or timeout value in seconds has been exceeded (False)
+ Returns true when the VM status returns the value of
+ expected_status_code
+ :param expected_status_code: instance status evaluated with this
+ string value
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
:param timeout: The timeout value
:param poll_interval: The polling interval in seconds
:return: T/F
@@ -494,17 +574,21 @@ class OpenStackVmInstance:
logger.info('VM is - ' + expected_status_code)
return True
- logger.debug('Retry querying VM status in ' + str(poll_interval) + ' seconds')
+ logger.debug('Retry querying VM status in ' + str(
+ poll_interval) + ' seconds')
time.sleep(poll_interval)
- logger.debug('VM status query timeout in ' + str(timeout - (time.time() - start)))
+ logger.debug('VM status query timeout in ' + str(
+ timeout - (time.time() - start)))
- logger.error('Timeout checking for VM status for ' + expected_status_code)
+ logger.error(
+ 'Timeout checking for VM status for ' + expected_status_code)
return False
def __status(self, expected_status_code):
"""
Returns True when active else False
- :param expected_status_code: instance status evaluated with this string value
+ :param expected_status_code: instance status evaluated with this string
+ value
:return: T/F
"""
if not self.__vm:
@@ -517,13 +601,16 @@ class OpenStackVmInstance:
if instance.status == 'ERROR':
raise Exception('Instance had an error during deployment')
- logger.debug('Instance status [' + self.instance_settings.name + '] is - ' + instance.status)
+ logger.debug(
+ 'Instance status [%s] is - %s', self.instance_settings.name,
+ instance.status)
return instance.status == expected_status_code
def vm_ssh_active(self, block=False, poll_interval=POLL_INTERVAL):
"""
Returns true when the VM can be accessed via SSH
- :param block: When true, thread will block until active or timeout value in seconds has been exceeded (False)
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
:param poll_interval: The polling interval
:return: T/F
"""
@@ -544,9 +631,11 @@ class OpenStackVmInstance:
logger.info('SSH is active for VM instance')
return True
- logger.debug('Retry SSH connection in ' + str(poll_interval) + ' seconds')
+ logger.debug('Retry SSH connection in ' + str(
+ poll_interval) + ' seconds')
time.sleep(poll_interval)
- logger.debug('SSH connection timeout in ' + str(timeout - (time.time() - start)))
+ logger.debug('SSH connection timeout in ' + str(
+ timeout - (time.time() - start)))
logger.error('Timeout attempting to connect with VM via SSH')
return False
@@ -564,7 +653,8 @@ class OpenStackVmInstance:
def get_floating_ip(self, fip_name=None):
"""
- Returns the floating IP object byt name if found, else the first known, else None
+ Returns the floating IP object byt name if found, else the first known,
+ else None
:param fip_name: the name of the floating IP to return
:return: the SSH client or None
"""
@@ -577,17 +667,20 @@ class OpenStackVmInstance:
def ssh_client(self, fip_name=None):
"""
- Returns an SSH client using the name or the first known floating IP if exists, else None
+ Returns an SSH client using the name or the first known floating IP if
+ exists, else None
:param fip_name: the name of the floating IP to return
:return: the SSH client or None
"""
fip = self.get_floating_ip(fip_name)
if fip:
- return ansible_utils.ssh_client(self.__floating_ips[0].ip, self.get_image_user(),
- self.keypair_settings.private_filepath,
- proxy_settings=self.__os_creds.proxy_settings)
+ return ansible_utils.ssh_client(
+ self.__floating_ips[0].ip, self.get_image_user(),
+ self.keypair_settings.private_filepath,
+ proxy_settings=self.__os_creds.proxy_settings)
else:
- logger.warning('Cannot return an SSH client. No Floating IP configured')
+ logger.warning(
+ 'Cannot return an SSH client. No Floating IP configured')
def add_security_group(self, security_group):
"""
@@ -602,7 +695,9 @@ class OpenStackVmInstance:
return False
try:
- nova_utils.add_security_group(self.__nova, self.get_vm_inst(), security_group['security_group']['name'])
+ nova_utils.add_security_group(self.__nova, self.get_vm_inst(),
+ security_group['security_group'][
+ 'name'])
return True
except NotFound as e:
logger.warning('Security group not added - ' + str(e))
@@ -610,7 +705,7 @@ class OpenStackVmInstance:
def remove_security_group(self, security_group):
"""
- Removes a security group to this VM. Call will block until VM is active.
+ Removes a security group to this VM. Call will block until VM is active
:param security_group: the OpenStack security group object
:return True if successful else False
"""
@@ -621,7 +716,8 @@ class OpenStackVmInstance:
return False
try:
- nova_utils.remove_security_group(self.__nova, self.get_vm_inst(), security_group)
+ nova_utils.remove_security_group(self.__nova, self.get_vm_inst(),
+ security_group)
return True
except NotFound as e:
logger.warning('Security group not removed - ' + str(e))
@@ -632,134 +728,133 @@ class VmInstanceSettings:
"""
Class responsible for holding configuration setting for a VM Instance
"""
- def __init__(self, config=None, name=None, flavor=None, port_settings=list(), security_group_names=set(),
- floating_ip_settings=list(), sudo_user=None, vm_boot_timeout=900,
- vm_delete_timeout=300, ssh_connect_timeout=180, availability_zone=None, userdata=None):
+
+ def __init__(self, **kwargs):
"""
Constructor
- :param config: dict() object containing the configuration settings using the attribute names below as each
- member's the key and overrides any of the other parameters.
:param name: the name of the VM
:param flavor: the VM's flavor
:param port_settings: the port configuration settings (required)
- :param security_group_names: a set of names of the security groups to add to the VM
+ :param security_group_names: a set of names of the security groups to
+ add to the VM
:param floating_ip_settings: the floating IP configuration settings
- :param sudo_user: the sudo user of the VM that will override the instance_settings.image_user when trying to
+ :param sudo_user: the sudo user of the VM that will override the
+ instance_settings.image_user when trying to
connect to the VM
- :param vm_boot_timeout: the amount of time a thread will sleep waiting for an instance to boot
- :param vm_delete_timeout: the amount of time a thread will sleep waiting for an instance to be deleted
- :param ssh_connect_timeout: the amount of time a thread will sleep waiting obtaining an SSH connection to a VM
- :param availability_zone: the name of the compute server on which to deploy the VM (optional)
- :param userdata: the cloud-init script to run after the VM has been started
- """
- if config:
- self.name = config.get('name')
- self.flavor = config.get('flavor')
- self.sudo_user = config.get('sudo_user')
- self.userdata = config.get('userdata')
-
- self.port_settings = list()
- if config.get('ports'):
- for port_config in config['ports']:
- if isinstance(port_config, PortSettings):
- self.port_settings.append(port_config)
- else:
- self.port_settings.append(PortSettings(config=port_config['port']))
-
- if config.get('security_group_names'):
- if isinstance(config['security_group_names'], list):
- self.security_group_names = set(config['security_group_names'])
- elif isinstance(config['security_group_names'], set):
- self.security_group_names = config['security_group_names']
- elif isinstance(config['security_group_names'], str):
- self.security_group_names = [config['security_group_names']]
- else:
- raise Exception('Invalid data type for security_group_names attribute')
+ :param vm_boot_timeout: the amount of time a thread will sleep waiting
+ for an instance to boot
+ :param vm_delete_timeout: the amount of time a thread will sleep
+ waiting for an instance to be deleted
+ :param ssh_connect_timeout: the amount of time a thread will sleep
+ waiting obtaining an SSH connection to a VM
+ :param availability_zone: the name of the compute server on which to
+ deploy the VM (optional)
+ :param userdata: the cloud-init script to run after the VM has been
+ started
+ """
+ self.name = kwargs.get('name')
+ self.flavor = kwargs.get('flavor')
+ self.sudo_user = kwargs.get('sudo_user')
+ self.userdata = kwargs.get('userdata')
+
+ self.port_settings = list()
+ port_settings = kwargs.get('ports')
+ if not port_settings:
+ port_settings = kwargs.get('port_settings')
+ if port_settings:
+ for port_setting in port_settings:
+ if isinstance(port_setting, dict):
+ self.port_settings.append(PortSettings(**port_setting))
+ elif isinstance(port_setting, PortSettings):
+ self.port_settings.append(port_setting)
+
+ if kwargs.get('security_group_names'):
+ if isinstance(kwargs['security_group_names'], list):
+ self.security_group_names = kwargs['security_group_names']
+ elif isinstance(kwargs['security_group_names'], set):
+ self.security_group_names = kwargs['security_group_names']
+ elif isinstance(kwargs['security_group_names'], str):
+ self.security_group_names = [kwargs['security_group_names']]
else:
- self.security_group_names = set()
-
- self.floating_ip_settings = list()
- if config.get('floating_ips'):
- for floating_ip_config in config['floating_ips']:
- if isinstance(floating_ip_config, FloatingIpSettings):
- self.floating_ip_settings.append(floating_ip_config)
- else:
- self.floating_ip_settings.append(FloatingIpSettings(config=floating_ip_config['floating_ip']))
-
- if config.get('vm_boot_timeout'):
- self.vm_boot_timeout = config['vm_boot_timeout']
- else:
- self.vm_boot_timeout = vm_boot_timeout
+ raise Exception(
+ 'Invalid data type for security_group_names attribute')
+ else:
+ self.security_group_names = set()
+
+ self.floating_ip_settings = list()
+ floating_ip_settings = kwargs.get('floating_ips')
+ if not floating_ip_settings:
+ floating_ip_settings = kwargs.get('floating_ip_settings')
+ if floating_ip_settings:
+ for floating_ip_config in floating_ip_settings:
+ if isinstance(floating_ip_config, FloatingIpSettings):
+ self.floating_ip_settings.append(floating_ip_config)
+ else:
+ self.floating_ip_settings.append(FloatingIpSettings(
+ **floating_ip_config['floating_ip']))
- if config.get('vm_delete_timeout'):
- self.vm_delete_timeout = config['vm_delete_timeout']
- else:
- self.vm_delete_timeout = vm_delete_timeout
+ if kwargs.get('vm_boot_timeout'):
+ self.vm_boot_timeout = kwargs['vm_boot_timeout']
+ else:
+ self.vm_boot_timeout = 900
- if config.get('ssh_connect_timeout'):
- self.ssh_connect_timeout = config['ssh_connect_timeout']
- else:
- self.ssh_connect_timeout = ssh_connect_timeout
+ if kwargs.get('vm_delete_timeout'):
+ self.vm_delete_timeout = kwargs['vm_delete_timeout']
+ else:
+ self.vm_delete_timeout = 300
- if config.get('availability_zone'):
- self.availability_zone = config['availability_zone']
- else:
- self.availability_zone = None
+ if kwargs.get('ssh_connect_timeout'):
+ self.ssh_connect_timeout = kwargs['ssh_connect_timeout']
+ else:
+ self.ssh_connect_timeout = 180
+
+ if kwargs.get('availability_zone'):
+ self.availability_zone = kwargs['availability_zone']
else:
- self.name = name
- self.flavor = flavor
- self.port_settings = port_settings
- self.security_group_names = security_group_names
- self.floating_ip_settings = floating_ip_settings
- self.sudo_user = sudo_user
- self.vm_boot_timeout = vm_boot_timeout
- self.vm_delete_timeout = vm_delete_timeout
- self.ssh_connect_timeout = ssh_connect_timeout
- self.availability_zone = availability_zone
- self.userdata = userdata
+ self.availability_zone = None
if not self.name or not self.flavor:
- raise Exception('Instance configuration requires the attributes: name, flavor')
+ raise Exception(
+ 'Instance configuration requires the attributes: name, flavor')
if len(self.port_settings) == 0:
- raise Exception('Instance configuration requires port settings (aka. NICS)')
+ raise Exception(
+ 'Instance configuration requires port settings (aka. NICS)')
class FloatingIpSettings:
"""
Class responsible for holding configuration settings for a floating IP
"""
- def __init__(self, config=None, name=None, port_name=None, router_name=None, subnet_name=None, provisioning=True):
+
+ def __init__(self, **kwargs):
"""
Constructor
- :param config: dict() object containing the configuration settings using the attribute names below as each
- member's the key and overrides any of the other parameters.
:param name: the name of the floating IP
:param port_name: the name of the router to the external network
:param router_name: the name of the router to the external network
- :param subnet_name: the name of the subnet on which to attach the floating IP
- :param provisioning: when true, this floating IP can be used for provisioning
-
- TODO - provisioning flag is a hack as I have only observed a single Floating IPs that actually works on
- an instance. Multiple floating IPs placed on different subnets from the same port are especially troublesome
- as you cannot predict which one will actually connect. For now, it is recommended not to setup multiple
- floating IPs on an instance unless absolutely necessary.
- """
- if config:
- self.name = config.get('name')
- self.port_name = config.get('port_name')
- self.router_name = config.get('router_name')
- self.subnet_name = config.get('subnet_name')
- if config.get('provisioning') is not None:
- self.provisioning = config['provisioning']
- else:
- self.provisioning = provisioning
+ :param subnet_name: the name of the subnet on which to attach the
+ floating IP
+ :param provisioning: when true, this floating IP can be used for
+ provisioning
+
+ TODO - provisioning flag is a hack as I have only observed a single
+ Floating IPs that actually works on an instance. Multiple floating IPs
+ placed on different subnets from the same port are especially
+ troublesome as you cannot predict which one will actually connect.
+ For now, it is recommended not to setup multiple floating IPs on an
+ instance unless absolutely necessary.
+ """
+ self.name = kwargs.get('name')
+ self.port_name = kwargs.get('port_name')
+ self.router_name = kwargs.get('router_name')
+ self.subnet_name = kwargs.get('subnet_name')
+ if kwargs.get('provisioning') is not None:
+ self.provisioning = kwargs['provisioning']
else:
- self.name = name
- self.port_name = port_name
- self.router_name = router_name
- self.subnet_name = subnet_name
- self.provisioning = provisioning
+ self.provisioning = True
if not self.name or not self.port_name or not self.router_name:
- raise Exception('The attributes name, port_name and router_name are required for FloatingIPSettings')
+ raise Exception(
+ 'The attributes name, port_name and router_name are required '
+ 'for FloatingIPSettings')
diff --git a/snaps/openstack/tests/create_instance_tests.py b/snaps/openstack/tests/create_instance_tests.py
index 34cec98..998fe88 100644
--- a/snaps/openstack/tests/create_instance_tests.py
+++ b/snaps/openstack/tests/create_instance_tests.py
@@ -13,25 +13,29 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
-import os
import re
+import shutil
import time
import unittest
import uuid
-import shutil
+import os
+
from snaps import file_utils
-from snaps.openstack.create_instance import VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings
from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
+from snaps.openstack.create_image import OpenStackImage, ImageSettings
+from snaps.openstack.create_instance import (
+ VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings)
from snaps.openstack.create_keypairs import OpenStackKeypair, KeypairSettings
from snaps.openstack.create_network import OpenStackNetwork, PortSettings
from snaps.openstack.create_router import OpenStackRouter
-from snaps.openstack.create_image import OpenStackImage, ImageSettings
-from snaps.openstack.create_security_group import SecurityGroupSettings, OpenStackSecurityGroup, \
- SecurityGroupRuleSettings, Direction, Protocol
+from snaps.openstack.create_security_group import (
+ SecurityGroupSettings, OpenStackSecurityGroup, SecurityGroupRuleSettings,
+ Direction, Protocol)
from snaps.openstack.tests import openstack_tests, validation_utils
+from snaps.openstack.tests.os_source_file_test import (
+ OSIntegrationTestCase, OSComponentTestCase)
from snaps.openstack.utils import nova_utils
-from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase, OSComponentTestCase
__author__ = 'spisarski'
@@ -71,7 +75,8 @@ class VmInstanceSettingsUnitTests(unittest.TestCase):
def test_name_flavor_port_only(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
- settings = VmInstanceSettings(name='foo', flavor='bar', port_settings=[port_settings])
+ settings = VmInstanceSettings(name='foo', flavor='bar',
+ port_settings=[port_settings])
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.flavor)
self.assertEqual(1, len(settings.port_settings))
@@ -87,7 +92,8 @@ class VmInstanceSettingsUnitTests(unittest.TestCase):
def test_config_with_name_flavor_port_only(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
- settings = VmInstanceSettings(config={'name': 'foo', 'flavor': 'bar', 'ports': [port_settings]})
+ settings = VmInstanceSettings(
+ **{'name': 'foo', 'flavor': 'bar', 'ports': [port_settings]})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.flavor)
self.assertEqual(1, len(settings.port_settings))
@@ -103,12 +109,17 @@ class VmInstanceSettingsUnitTests(unittest.TestCase):
def test_all(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
- fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port', router_name='foo-bar-router')
-
- settings = VmInstanceSettings(name='foo', flavor='bar', port_settings=[port_settings],
- security_group_names=['sec_grp_1'], floating_ip_settings=[fip_settings],
- sudo_user='joe', vm_boot_timeout=999, vm_delete_timeout=333,
- ssh_connect_timeout=111, availability_zone='server name')
+ fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port',
+ router_name='foo-bar-router')
+
+ settings = VmInstanceSettings(name='foo', flavor='bar',
+ port_settings=[port_settings],
+ security_group_names=['sec_grp_1'],
+ floating_ip_settings=[fip_settings],
+ sudo_user='joe', vm_boot_timeout=999,
+ vm_delete_timeout=333,
+ ssh_connect_timeout=111,
+ availability_zone='server name')
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.flavor)
self.assertEqual(1, len(settings.port_settings))
@@ -118,8 +129,10 @@ class VmInstanceSettingsUnitTests(unittest.TestCase):
self.assertEqual('sec_grp_1', settings.security_group_names[0])
self.assertEqual(1, len(settings.floating_ip_settings))
self.assertEqual('foo-fip', settings.floating_ip_settings[0].name)
- self.assertEqual('bar-port', settings.floating_ip_settings[0].port_name)
- self.assertEqual('foo-bar-router', settings.floating_ip_settings[0].router_name)
+ self.assertEqual('bar-port',
+ settings.floating_ip_settings[0].port_name)
+ self.assertEqual('foo-bar-router',
+ settings.floating_ip_settings[0].router_name)
self.assertEqual('joe', settings.sudo_user)
self.assertEqual(999, settings.vm_boot_timeout)
self.assertEqual(333, settings.vm_delete_timeout)
@@ -128,13 +141,15 @@ class VmInstanceSettingsUnitTests(unittest.TestCase):
def test_config_all(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
- fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port', router_name='foo-bar-router')
-
- settings = VmInstanceSettings(config={'name': 'foo', 'flavor': 'bar', 'ports': [port_settings],
- 'security_group_names': ['sec_grp_1'],
- 'floating_ips': [fip_settings], 'sudo_user': 'joe',
- 'vm_boot_timeout': 999, 'vm_delete_timeout': 333,
- 'ssh_connect_timeout': 111, 'availability_zone': 'server name'})
+ fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port',
+ router_name='foo-bar-router')
+
+ settings = VmInstanceSettings(
+ **{'name': 'foo', 'flavor': 'bar', 'ports': [port_settings],
+ 'security_group_names': ['sec_grp_1'],
+ 'floating_ips': [fip_settings], 'sudo_user': 'joe',
+ 'vm_boot_timeout': 999, 'vm_delete_timeout': 333,
+ 'ssh_connect_timeout': 111, 'availability_zone': 'server name'})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.flavor)
self.assertEqual(1, len(settings.port_settings))
@@ -143,8 +158,10 @@ class VmInstanceSettingsUnitTests(unittest.TestCase):
self.assertEqual(1, len(settings.security_group_names))
self.assertEqual(1, len(settings.floating_ip_settings))
self.assertEqual('foo-fip', settings.floating_ip_settings[0].name)
- self.assertEqual('bar-port', settings.floating_ip_settings[0].port_name)
- self.assertEqual('foo-bar-router', settings.floating_ip_settings[0].router_name)
+ self.assertEqual('bar-port',
+ settings.floating_ip_settings[0].port_name)
+ self.assertEqual('foo-bar-router',
+ settings.floating_ip_settings[0].router_name)
self.assertEqual('joe', settings.sudo_user)
self.assertEqual(999, settings.vm_boot_timeout)
self.assertEqual(333, settings.vm_delete_timeout)
@@ -163,7 +180,7 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
def test_empty_config(self):
with self.assertRaises(Exception):
- FloatingIpSettings(config=dict())
+ FloatingIpSettings(**dict())
def test_name_only(self):
with self.assertRaises(Exception):
@@ -171,7 +188,7 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
def test_config_with_name_only(self):
with self.assertRaises(Exception):
- FloatingIpSettings(config={'name': 'foo'})
+ FloatingIpSettings(**{'name': 'foo'})
def test_name_port_only(self):
with self.assertRaises(Exception):
@@ -179,7 +196,7 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
def test_config_with_name_port_only(self):
with self.assertRaises(Exception):
- FloatingIpSettings(config={'name': 'foo', 'port_name': 'bar'})
+ FloatingIpSettings(**{'name': 'foo', 'port_name': 'bar'})
def test_name_router_only(self):
with self.assertRaises(Exception):
@@ -187,10 +204,11 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
def test_config_with_name_router_only(self):
with self.assertRaises(Exception):
- FloatingIpSettings(config={'name': 'foo', 'router_name': 'bar'})
+ FloatingIpSettings(**{'name': 'foo', 'router_name': 'bar'})
def test_name_port_router_only(self):
- settings = FloatingIpSettings(name='foo', port_name='foo-port', router_name='bar-router')
+ settings = FloatingIpSettings(name='foo', port_name='foo-port',
+ router_name='bar-router')
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
self.assertEqual('bar-router', settings.router_name)
@@ -198,7 +216,9 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
self.assertTrue(settings.provisioning)
def test_config_with_name_port_router_only(self):
- settings = FloatingIpSettings(config={'name': 'foo', 'port_name': 'foo-port', 'router_name': 'bar-router'})
+ settings = FloatingIpSettings(
+ **{'name': 'foo', 'port_name': 'foo-port',
+ 'router_name': 'bar-router'})
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
self.assertEqual('bar-router', settings.router_name)
@@ -206,8 +226,10 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
self.assertTrue(settings.provisioning)
def test_all(self):
- settings = FloatingIpSettings(name='foo', port_name='foo-port', router_name='bar-router',
- subnet_name='bar-subnet', provisioning=False)
+ settings = FloatingIpSettings(name='foo', port_name='foo-port',
+ router_name='bar-router',
+ subnet_name='bar-subnet',
+ provisioning=False)
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
self.assertEqual('bar-router', settings.router_name)
@@ -215,8 +237,10 @@ class FloatingIpSettingsUnitTests(unittest.TestCase):
self.assertFalse(settings.provisioning)
def test_config_all(self):
- settings = FloatingIpSettings(config={'name': 'foo', 'port_name': 'foo-port', 'router_name': 'bar-router',
- 'subnet_name': 'bar-subnet', 'provisioning': False})
+ settings = FloatingIpSettings(
+ **{'name': 'foo', 'port_name': 'foo-port',
+ 'router_name': 'bar-router', 'subnet_name': 'bar-subnet',
+ 'provisioning': False})
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
self.assertEqual('bar-router', settings.router_name)
@@ -231,7 +255,8 @@ class SimpleHealthCheck(OSIntegrationTestCase):
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file
within OpenStack
"""
super(self.__class__, self).__start__()
@@ -249,25 +274,30 @@ class SimpleHealthCheck(OSIntegrationTestCase):
self.priv_net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-priv-net', subnet_name=guid + '-priv-subnet')
self.port_settings = PortSettings(
- name=self.port_1_name, network_name=self.priv_net_config.network_settings.name)
+ name=self.port_1_name,
+ network_name=self.priv_net_config.network_settings.name)
# Create Image
- # Set the default image settings, then set any custom parameters sent from the app
+ # Set the default image settings, then set any custom parameters sent
+ # from the app
os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
try:
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, self.priv_net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, self.priv_net_config.network_settings)
self.network_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10, vcpus=1, metadata=self.flavor_metadata))
+ FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ vcpus=1, metadata=self.flavor_metadata))
self.flavor_creator.create()
except Exception as e:
self.tearDown()
@@ -281,38 +311,49 @@ class SimpleHealthCheck(OSIntegrationTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message'
+ ' - %s', e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s',
+ e)
super(self.__class__, self).__clean__()
def test_check_vm_ip_dhcp(self):
"""
- Tests the creation of an OpenStack instance with a single port and ensures that it's assigned IP address is
- the actual.
+ Tests the creation of an OpenStack instance with a single port and
+ ensures that it's assigned IP address is the actual.
"""
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm = self.inst_creator.create()
ip = self.inst_creator.get_port_ip(self.port_settings.name)
@@ -327,9 +368,11 @@ class CreateInstanceSimpleTests(OSIntegrationTestCase):
"""
Simple instance creation tests without any other objects
"""
+
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file
within OpenStack
"""
super(self.__class__, self).__start__()
@@ -353,21 +396,25 @@ class CreateInstanceSimpleTests(OSIntegrationTestCase):
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10, vcpus=2, metadata=self.flavor_metadata))
+ FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, net_config.network_settings)
self.network_creator.create()
- self.port_settings = PortSettings(name=guid + '-port',
- network_name=net_config.network_settings.name)
+ self.port_settings = PortSettings(
+ name=guid + '-port',
+ network_name=net_config.network_settings.name)
except Exception as e:
self.tearDown()
@@ -381,46 +428,59 @@ class CreateInstanceSimpleTests(OSIntegrationTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
def test_create_delete_instance(self):
"""
- Tests the creation of an OpenStack instance with a single port with a static IP without a Floating IP.
+ Tests the creation of an OpenStack instance with a single port with a
+ static IP without a Floating IP.
"""
- instance_settings = VmInstanceSettings(name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name,
- port_settings=[self.port_settings])
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings)
+ self.os_creds, instance_settings,
+ 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.assertEqual(1, len(
+ nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
# 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.assertEqual(0, len(
+ nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
# Exception should not be thrown
self.inst_creator.clean()
@@ -433,8 +493,8 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
@@ -463,37 +523,47 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
name=guid + '-image', image_metadata=self.image_metadata)
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, self.pub_net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, self.pub_net_config.network_settings)
self.network_creator.create()
# Create Router
- self.router_creator = OpenStackRouter(self.os_creds, self.pub_net_config.router_settings)
+ self.router_creator = OpenStackRouter(
+ self.os_creds, self.pub_net_config.router_settings)
self.router_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10, vcpus=2, metadata=self.flavor_metadata))
+ FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
self.keypair_creator = OpenStackKeypair(
self.os_creds, KeypairSettings(
- name=self.keypair_name, public_filepath=self.keypair_pub_filepath,
+ name=self.keypair_name,
+ public_filepath=self.keypair_pub_filepath,
private_filepath=self.keypair_priv_filepath))
self.keypair_creator.create()
sec_grp_name = guid + '-sec-grp'
- rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
protocol=Protocol.icmp)
- rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
+ rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
+ protocol=Protocol.tcp,
+ port_range_min=22,
+ port_range_max=22)
self.sec_grp_creator = OpenStackSecurityGroup(
self.os_creds,
- SecurityGroupSettings(name=sec_grp_name, rule_settings=[rule1, rule2]))
+ SecurityGroupSettings(name=sec_grp_name,
+ rule_settings=[rule1, rule2]))
self.sec_grp_creator.create()
except Exception as e:
self.tearDown()
@@ -507,13 +577,17 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
try:
inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message '
+ '- %s', e)
if self.keypair_creator:
try:
self.keypair_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning keypair with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning keypair with message - %s',
+ e)
if os.path.isfile(self.keypair_pub_filepath):
os.remove(self.keypair_pub_filepath)
@@ -525,52 +599,67 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning security group 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning router with message - %s',
+ e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
def test_single_port_static(self):
"""
- Tests the creation of an OpenStack instance with a single port with a static IP without a Floating IP.
+ Tests the creation of an OpenStack instance with a single port with a
+ static IP without a Floating IP.
"""
ip_1 = '10.55.1.100'
-
+ sub_settings = self.pub_net_config.network_settings.subnet_settings
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.pub_net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.pub_net_config.network_settings.subnet_settings[0].name, 'ip': ip_1}])
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name,
+ ip_addrs=[
+ {'subnet_name': sub_settings[0].name, 'ip': ip_1}])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings],
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings],
floating_ip_settings=[FloatingIpSettings(
name=self.floating_ip_name, port_name=self.port_1_name,
router_name=self.pub_net_config.router_settings.name)])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
self.inst_creators.append(inst_creator)
vm_inst = inst_creator.create()
@@ -581,19 +670,24 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
def test_ssh_client_fip_before_active(self):
"""
- Tests the ability to access a VM via SSH and a floating IP when it has been assigned prior to being active.
+ Tests the ability to access a VM via SSH and a floating IP when it has
+ been assigned prior to being active.
"""
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.pub_net_config.network_settings.name)
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name)
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings],
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings],
floating_ip_settings=[FloatingIpSettings(
name=self.floating_ip_name, port_name=self.port_1_name,
router_name=self.pub_net_config.router_settings.name)])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
self.inst_creators.append(inst_creator)
vm_inst = inst_creator.create()
@@ -604,26 +698,32 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
ip = inst_creator.get_port_ip(port_settings.name)
self.assertTrue(check_dhcp_lease(vm_inst, ip))
- inst_creator.add_security_group(self.sec_grp_creator.get_security_group())
+ inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
self.assertEqual(vm_inst, inst_creator.get_vm_inst())
self.assertTrue(validate_ssh_client(inst_creator))
def test_ssh_client_fip_after_active(self):
"""
- Tests the ability to access a VM via SSH and a floating IP when it has been assigned prior to being active.
+ Tests the ability to access a VM via SSH and a floating IP when it has
+ been assigned prior to being active.
"""
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.pub_net_config.network_settings.name)
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name)
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings],
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings],
floating_ip_settings=[FloatingIpSettings(
name=self.floating_ip_name, port_name=self.port_1_name,
router_name=self.pub_net_config.router_settings.name)])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
self.inst_creators.append(inst_creator)
@@ -636,7 +736,8 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
ip = inst_creator.get_port_ip(port_settings.name)
self.assertTrue(check_dhcp_lease(vm_inst, ip))
- inst_creator.add_security_group(self.sec_grp_creator.get_security_group())
+ inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
self.assertEqual(vm_inst, inst_creator.get_vm_inst())
self.assertTrue(validate_ssh_client(inst_creator))
@@ -644,13 +745,14 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase):
class CreateInstancePortManipulationTests(OSIntegrationTestCase):
"""
- Test for the CreateInstance class with a single NIC/Port where mac and IP values are manually set
+ Test for the CreateInstance class with a single NIC/Port where mac and IP
+ values are manually set
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
@@ -674,17 +776,20 @@ class CreateInstancePortManipulationTests(OSIntegrationTestCase):
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, self.net_config.network_settings)
self.network_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10, vcpus=2, metadata=self.flavor_metadata))
+ FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
except Exception as e:
self.tearDown()
@@ -698,147 +803,198 @@ class CreateInstancePortManipulationTests(OSIntegrationTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
def test_set_custom_valid_ip_one_subnet(self):
"""
- Tests the creation of an OpenStack instance with a single port with a static IP on a network with one subnet.
+ Tests the creation of an OpenStack instance with a single port with a
+ static IP on a network with one subnet.
"""
ip = '10.55.0.101'
+ sub_settings = self.net_config.network_settings.subnet_settings
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip}])
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{'subnet_name': sub_settings[0].name, 'ip': ip}])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create(block=True)
self.assertEqual(ip, self.inst_creator.get_port_ip(
- self.port_1_name, subnet_name=self.net_config.network_settings.subnet_settings[0].name))
+ self.port_1_name,
+ subnet_name=self.net_config.network_settings.subnet_settings[
+ 0].name))
def test_set_custom_invalid_ip_one_subnet(self):
"""
- Tests the creation of an OpenStack instance with a single port with a static IP on a network with one subnet.
+ Tests the creation of an OpenStack instance with a single port with a
+ static IP on a network with one subnet.
"""
ip = '10.66.0.101'
+ sub_settings = self.net_config.network_settings.subnet_settings
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip}])
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{'subnet_name': sub_settings[0].name, 'ip': ip}])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
def test_set_custom_valid_mac(self):
"""
- Tests the creation of an OpenStack instance with a single port where the MAC address is assigned.
+ Tests the creation of an OpenStack instance with a single port where
+ the MAC address is assigned.
"""
mac_addr = '0a:1b:2c:3d:4e:5f'
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name, mac_address=mac_addr)
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ mac_address=mac_addr)
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create(block=True)
- self.assertEqual(mac_addr, self.inst_creator.get_port_mac(self.port_1_name))
+ self.assertEqual(mac_addr,
+ self.inst_creator.get_port_mac(self.port_1_name))
def test_set_custom_invalid_mac(self):
"""
- Tests the creation of an OpenStack instance with a single port where an invalid MAC address value is being
+ Tests the creation of an OpenStack instance with a single port where an
+ invalid MAC address value is being
assigned. This should raise an Exception
"""
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name, mac_address='foo')
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ mac_address='foo')
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings)
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
def test_set_custom_mac_and_ip(self):
"""
- Tests the creation of an OpenStack instance with a single port where the IP and MAC address is assigned.
+ Tests the creation of an OpenStack instance with a single port where
+ the IP and MAC address is assigned.
"""
ip = '10.55.0.101'
mac_addr = '0a:1b:2c:3d:4e:5f'
+ sub_settings = self.net_config.network_settings.subnet_settings
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name, mac_address=mac_addr,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip}])
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ mac_address=mac_addr,
+ ip_addrs=[{'subnet_name': sub_settings[0].name, 'ip': ip}])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create(block=True)
self.assertEqual(ip, self.inst_creator.get_port_ip(
- self.port_1_name, subnet_name=self.net_config.network_settings.subnet_settings[0].name))
- self.assertEqual(mac_addr, self.inst_creator.get_port_mac(self.port_1_name))
+ self.port_1_name,
+ subnet_name=self.net_config.network_settings.subnet_settings[
+ 0].name))
+ self.assertEqual(mac_addr,
+ self.inst_creator.get_port_mac(self.port_1_name))
def test_set_allowed_address_pairs(self):
"""
- Tests the creation of an OpenStack instance with a single port where max_allowed_address_pair is set.
+ Tests the creation of an OpenStack instance with a single port where
+ max_allowed_address_pair is set.
"""
ip = '10.55.0.101'
mac_addr = '0a:1b:2c:3d:4e:5f'
pair = {'ip_address': ip, 'mac_address': mac_addr}
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name, allowed_address_pairs=[pair])
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ allowed_address_pairs=[pair])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create(block=True)
port = self.inst_creator.get_port_by_name(port_settings.name)
self.assertIsNotNone(port)
self.assertIsNotNone(port['port'].get('allowed_address_pairs'))
self.assertEqual(1, len(port['port']['allowed_address_pairs']))
- validation_utils.objects_equivalent(pair, port['port']['allowed_address_pairs'][0])
+ validation_utils.objects_equivalent(pair, port['port'][
+ 'allowed_address_pairs'][0])
def test_set_allowed_address_pairs_bad_mac(self):
"""
- Tests the creation of an OpenStack instance with a single port where max_allowed_address_pair is set with an
- invalid MAC address.
+ Tests the creation of an OpenStack instance with a single port where
+ max_allowed_address_pair is set with an invalid MAC address.
"""
ip = '10.55.0.101'
mac_addr = 'foo'
@@ -846,20 +1002,25 @@ class CreateInstancePortManipulationTests(OSIntegrationTestCase):
pairs = set()
pairs.add((ip, mac_addr))
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name, allowed_address_pairs=[pair])
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ allowed_address_pairs=[pair])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
def test_set_allowed_address_pairs_bad_ip(self):
"""
- Tests the creation of an OpenStack instance with a single port where max_allowed_address_pair is set with an
- invalid MAC address.
+ Tests the creation of an OpenStack instance with a single port where
+ max_allowed_address_pair is set with an invalid MAC address.
"""
ip = 'foo'
mac_addr = '0a:1b:2c:3d:4e:5f'
@@ -867,13 +1028,18 @@ class CreateInstancePortManipulationTests(OSIntegrationTestCase):
pairs = set()
pairs.add((ip, mac_addr))
port_settings = PortSettings(
- name=self.port_1_name, network_name=self.net_config.network_settings.name, allowed_address_pairs=[pair])
+ name=self.port_1_name,
+ network_name=self.net_config.network_settings.name,
+ allowed_address_pairs=[pair])
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
@@ -885,8 +1051,8 @@ class CreateInstanceOnComputeHost(OSIntegrationTestCase):
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
@@ -908,17 +1074,20 @@ class CreateInstanceOnComputeHost(OSIntegrationTestCase):
try:
# Create Network
- self.network_creator = OpenStackNetwork(self.admin_os_creds, self.priv_net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.admin_os_creds, self.priv_net_config.network_settings)
self.network_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=512, disk=1, vcpus=1, metadata=self.flavor_metadata))
+ FlavorSettings(name=guid + '-flavor-name', ram=512, disk=1,
+ vcpus=1, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
except Exception as e:
@@ -933,25 +1102,32 @@ class CreateInstanceOnComputeHost(OSIntegrationTestCase):
try:
inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
@@ -968,18 +1144,23 @@ class CreateInstanceOnComputeHost(OSIntegrationTestCase):
for zone in zones:
inst_name = self.vm_inst_name + '-' + zone
ctr += 1
- port_settings = PortSettings(name=self.port_base_name + '-' + str(ctr),
- network_name=self.priv_net_config.network_settings.name)
+ port_settings = PortSettings(
+ name=self.port_base_name + '-' + str(ctr),
+ network_name=self.priv_net_config.network_settings.name)
instance_settings = VmInstanceSettings(
- name=inst_name, flavor=self.flavor_creator.flavor_settings.name, availability_zone=zone,
+ name=inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ availability_zone=zone,
port_settings=[port_settings])
inst_creator = OpenStackVmInstance(
- self.admin_os_creds, instance_settings, self.image_creator.image_settings)
+ self.admin_os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creators.append(inst_creator)
inst_creator.create()
- # Validate instances to ensure they've been deployed to the correct server
+ # Validate instances to ensure they've been deployed to the correct
+ # server
index = 0
for zone in zones:
creator = self.inst_creators[index]
@@ -993,14 +1174,15 @@ class CreateInstanceOnComputeHost(OSIntegrationTestCase):
class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
"""
- Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP and eth1 w/o
+ Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP
+ and eth1 w/o.
These tests require a Centos image
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
@@ -1022,29 +1204,39 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
self.port_2_name = self.guid + '-port-2'
self.floating_ip_name = self.guid + 'fip1'
self.priv_net_config = openstack_tests.get_priv_net_config(
- net_name=self.guid + '-priv-net', subnet_name=self.guid + '-priv-subnet',
- router_name=self.guid + '-priv-router', external_net=self.ext_net_name)
+ net_name=self.guid + '-priv-net',
+ subnet_name=self.guid + '-priv-subnet',
+ router_name=self.guid + '-priv-router',
+ external_net=self.ext_net_name)
self.pub_net_config = openstack_tests.get_pub_net_config(
- net_name=self.guid + '-pub-net', subnet_name=self.guid + '-pub-subnet',
- router_name=self.guid + '-pub-router', external_net=self.ext_net_name)
+ net_name=self.guid + '-pub-net',
+ subnet_name=self.guid + '-pub-subnet',
+ router_name=self.guid + '-pub-router',
+ external_net=self.ext_net_name)
image_name = self.__class__.__name__ + '-' + str(uuid.uuid4())
- os_image_settings = openstack_tests.centos_image_settings(name=image_name, image_metadata=self.image_metadata)
+ os_image_settings = openstack_tests.centos_image_settings(
+ name=image_name, image_metadata=self.image_metadata)
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ 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.pub_net_config.network_settings))
+ self.network_creators.append(OpenStackNetwork(
+ self.os_creds, self.pub_net_config.network_settings))
# Second network is private
- self.network_creators.append(OpenStackNetwork(self.os_creds, self.priv_net_config.network_settings))
+ self.network_creators.append(OpenStackNetwork(
+ self.os_creds, self.priv_net_config.network_settings))
for network_creator in self.network_creators:
network_creator.create()
- self.router_creators.append(OpenStackRouter(self.os_creds, self.pub_net_config.router_settings))
- self.router_creators.append(OpenStackRouter(self.os_creds, self.priv_net_config.router_settings))
+ self.router_creators.append(OpenStackRouter(
+ self.os_creds, self.pub_net_config.router_settings))
+ self.router_creators.append(OpenStackRouter(
+ self.os_creds, self.priv_net_config.router_settings))
# Create Routers
for router_creator in self.router_creators:
@@ -1053,25 +1245,32 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=self.guid + '-flavor-name', ram=512, disk=10, vcpus=2,
+ FlavorSettings(name=self.guid + '-flavor-name', ram=512,
+ disk=10, vcpus=2,
metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Keypair
self.keypair_creator = OpenStackKeypair(
self.os_creds, KeypairSettings(
- name=self.keypair_name, public_filepath=self.keypair_pub_filepath,
+ name=self.keypair_name,
+ public_filepath=self.keypair_pub_filepath,
private_filepath=self.keypair_priv_filepath))
self.keypair_creator.create()
sec_grp_name = self.guid + '-sec-grp'
- rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
protocol=Protocol.icmp)
- rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
+ rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
+ protocol=Protocol.tcp,
+ port_range_min=22,
+ port_range_max=22)
self.sec_grp_creator = OpenStackSecurityGroup(
self.os_creds,
- SecurityGroupSettings(name=sec_grp_name, rule_settings=[rule1, rule2]))
+ SecurityGroupSettings(name=sec_grp_name,
+ rule_settings=[rule1, rule2]))
self.sec_grp_creator.create()
except Exception as e:
self.tearDown()
@@ -1085,13 +1284,17 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message '
+ '- %s', e)
if self.keypair_creator:
try:
self.keypair_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning keypair with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning keypair with message - %s',
+ e)
if os.path.isfile(self.keypair_pub_filepath):
os.remove(self.keypair_pub_filepath)
@@ -1103,42 +1306,55 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
for router_creator in self.router_creators:
try:
router_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning router with message - ' + str(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 - ' + str(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 - ' + str(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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
def test_dual_ports_dhcp(self):
"""
- Tests the creation of an OpenStack instance with a dual ports/NICs with a DHCP assigned IP.
- NOTE: This test and any others that call ansible will most likely fail unless you do one of
- two things:
- 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
- 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
- Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
+ Tests the creation of an OpenStack instance with a dual ports/NICs with
+ a DHCP assigned IP.
+ NOTE: This test and any others that call ansible will most likely fail
+ unless you do one of two things:
+ 1. Have a ~/.ansible.cfg (or alternate means) to
+ set host_key_checking = False
+ 2. Set the following environment variable in your executing shell:
+ ANSIBLE_HOST_KEY_CHECKING=False
+ Should this not be performed, the creation of the host ssh key will
+ cause your ansible calls to fail.
"""
# Create ports/NICs for instance
ports_settings = []
@@ -1151,13 +1367,16 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
# Create instance
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings,
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=ports_settings,
floating_ip_settings=[FloatingIpSettings(
name=self.floating_ip_name, port_name=self.port_1_name,
router_name=self.pub_net_config.router_settings.name)])
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
vm_inst = self.inst_creator.create(block=True)
@@ -1171,28 +1390,32 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase):
self.assertTrue(check_dhcp_lease(vm_inst, ip))
# Add security group to VM
- self.inst_creator.add_security_group(self.sec_grp_creator.get_security_group())
+ self.inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
# Effectively blocks until VM's ssh port has been opened
self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
- # TODO - Refactor config_nics() to return a status that can be validated here.
+ # TODO - Refactor config_nics() to return a status that can be
+ # validated here.
self.inst_creator.config_nics()
# TODO - *** ADD VALIDATION HERE ***
# TODO - Add validation that both floating IPs work
# TODO - Add tests where only one NIC has a floating IP
- # TODO - Add tests where one attempts to place a floating IP on a network/router without an external gateway
+ # TODO - Add tests where one attempts to place a floating IP on a
+ # network/router without an external gateway
class InstanceSecurityGroupTests(OSIntegrationTestCase):
"""
Tests that include, add, and remove security groups from VM instances
"""
+
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
@@ -1208,8 +1431,10 @@ class InstanceSecurityGroupTests(OSIntegrationTestCase):
self.floating_ip_name = self.guid + 'fip1'
net_config = openstack_tests.get_priv_net_config(
- net_name=self.guid + '-pub-net', subnet_name=self.guid + '-pub-subnet',
- router_name=self.guid + '-pub-router', external_net=self.ext_net_name)
+ net_name=self.guid + '-pub-net',
+ subnet_name=self.guid + '-pub-subnet',
+ router_name=self.guid + '-pub-router',
+ external_net=self.ext_net_name)
# Initialize for tearDown()
self.image_creator = None
@@ -1221,22 +1446,26 @@ class InstanceSecurityGroupTests(OSIntegrationTestCase):
try:
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
+ self.image_creator = OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, net_config.network_settings)
self.network_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=self.guid + '-flavor-name', ram=128, disk=10, vcpus=2,
+ FlavorSettings(name=self.guid + '-flavor-name', ram=128,
+ disk=10, vcpus=2,
metadata=self.flavor_metadata))
self.flavor_creator.create()
- self.port_settings = PortSettings(name=self.guid + '-port',
- network_name=net_config.network_settings.name)
+ self.port_settings = PortSettings(
+ name=self.guid + '-port',
+ network_name=net_config.network_settings.name)
except Exception as e:
self.tearDown()
raise e
@@ -1249,31 +1478,40 @@ class InstanceSecurityGroupTests(OSIntegrationTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message -'
+ ' %s', e)
for sec_grp_creator in self.sec_grp_creators:
try:
sec_grp_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning security group with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning security group 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
@@ -1283,26 +1521,33 @@ class InstanceSecurityGroupTests(OSIntegrationTestCase):
"""
# Create instance
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
# Create security group object to add to instance
- sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name', description='hello group')
- sec_grp_creator = OpenStackSecurityGroup(self.os_creds, sec_grp_settings)
+ sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name',
+ description='hello group')
+ sec_grp_creator = OpenStackSecurityGroup(self.os_creds,
+ sec_grp_settings)
sec_grp = sec_grp_creator.create()
self.sec_grp_creators.append(sec_grp_creator)
# Check that group has not been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
# Add security group to instance after activated
self.inst_creator.add_security_group(sec_grp)
# Validate that security group has been added
- self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
def test_add_invalid_security_group(self):
"""
@@ -1310,44 +1555,57 @@ class InstanceSecurityGroupTests(OSIntegrationTestCase):
"""
# Create instance
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
# Create security group object to add to instance
- sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name', description='hello group')
- sec_grp_creator = OpenStackSecurityGroup(self.os_creds, sec_grp_settings)
+ sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name',
+ description='hello group')
+ sec_grp_creator = OpenStackSecurityGroup(self.os_creds,
+ sec_grp_settings)
sec_grp = sec_grp_creator.create()
sec_grp_creator.clean()
self.sec_grp_creators.append(sec_grp_creator)
# Check that group has not been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
# Add security group to instance after activated
self.assertFalse(self.inst_creator.add_security_group(sec_grp))
# Validate that security group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
def test_remove_security_group(self):
"""
- Tests the removal of a security group created before and added to the instance.
+ Tests the removal of a security group created before and added to the
+ instance.
"""
# Create security group object to add to instance
- sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name', description='hello group')
- sec_grp_creator = OpenStackSecurityGroup(self.os_creds, sec_grp_settings)
+ sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name',
+ description='hello group')
+ sec_grp_creator = OpenStackSecurityGroup(self.os_creds,
+ sec_grp_settings)
sec_grp = sec_grp_creator.create()
self.sec_grp_creators.append(sec_grp_creator)
# Create instance
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name,
- security_group_names=[sec_grp_settings.name], port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ security_group_names=[sec_grp_settings.name],
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
@@ -1358,62 +1616,79 @@ class InstanceSecurityGroupTests(OSIntegrationTestCase):
self.assertTrue(self.inst_creator.remove_security_group(sec_grp))
# Validate that security group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
def test_remove_security_group_never_added(self):
"""
- Tests the removal of a security group that was never added in the first place.
+ Tests the removal of a security group that was never added in the first
+ place.
"""
# Create security group object to add to instance
- sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name', description='hello group')
- sec_grp_creator = OpenStackSecurityGroup(self.os_creds, sec_grp_settings)
+ sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name',
+ description='hello group')
+ sec_grp_creator = OpenStackSecurityGroup(self.os_creds,
+ sec_grp_settings)
sec_grp = sec_grp_creator.create()
self.sec_grp_creators.append(sec_grp_creator)
# Create instance
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
# Check that group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
# Add security group to instance after activated
self.assertFalse(self.inst_creator.remove_security_group(sec_grp))
# Validate that security group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
def test_add_same_security_group(self):
"""
- Tests the addition of a security group created before add added to the instance.
+ Tests the addition of a security group created before add added to the
+ instance.
"""
# Create security group object to add to instance
- sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name', description='hello group')
- sec_grp_creator = OpenStackSecurityGroup(self.os_creds, sec_grp_settings)
+ sec_grp_settings = SecurityGroupSettings(name=self.guid + '-name',
+ description='hello group')
+ sec_grp_creator = OpenStackSecurityGroup(self.os_creds,
+ sec_grp_settings)
sec_grp = sec_grp_creator.create()
self.sec_grp_creators.append(sec_grp_creator)
# Create instance
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name,
- security_group_names=[sec_grp_settings.name], port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ security_group_names=[sec_grp_settings.name],
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
# Check that group has been added
- self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
# Add security group to instance after activated
self.assertTrue(self.inst_creator.add_security_group(sec_grp))
# Validate that security group has been added
- self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(), sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
+ sec_grp_settings.name))
def inst_has_sec_grp(vm_inst, sec_grp_name):
@@ -1435,7 +1710,8 @@ def inst_has_sec_grp(vm_inst, sec_grp_name):
def validate_ssh_client(instance_creator):
"""
Returns True if instance_creator returns an SSH client that is valid
- :param instance_creator: the object responsible for creating the VM instance
+ :param instance_creator: the object responsible for creating the VM
+ instance
:return: T/F
"""
ssh_active = instance_creator.vm_ssh_active(block=True)
@@ -1464,8 +1740,8 @@ class CreateInstanceFromThreePartImage(OSIntegrationTestCase):
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
@@ -1491,16 +1767,22 @@ class CreateInstanceFromThreePartImage(OSIntegrationTestCase):
and 'disk_file' in self.image_metadata['cirros']:
metadata = self.image_metadata['cirros']
else:
- metadata = {'disk_url': openstack_tests.CIRROS_DEFAULT_IMAGE_URL,
- 'kernel_url': openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL,
- 'ramdisk_url': openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL}
+ metadata = {
+ 'disk_url': openstack_tests.CIRROS_DEFAULT_IMAGE_URL,
+ 'kernel_url':
+ openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL,
+ 'ramdisk_url':
+ openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL}
image_settings = openstack_tests.cirros_image_settings(
name=self.image_name,
image_metadata=metadata)
- if not image_settings.ramdisk_image_settings or not image_settings.kernel_image_settings:
- logger.warn('3 Part image will not be tested. Image metadata has overridden this functionality')
+ if not image_settings.ramdisk_image_settings or not \
+ image_settings.kernel_image_settings:
+ logger.warn(
+ '3 Part image will not be tested. Image metadata has '
+ 'overridden this functionality')
self.image_creator = OpenStackImage(self.os_creds, image_settings)
self.image_creator.create()
@@ -1508,15 +1790,18 @@ class CreateInstanceFromThreePartImage(OSIntegrationTestCase):
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10, vcpus=2, metadata=self.flavor_metadata))
+ FlavorSettings(name=guid + '-flavor-name', ram=128, disk=10,
+ vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, net_config.network_settings)
self.network_creator.create()
- self.port_settings = PortSettings(name=guid + '-port',
- network_name=net_config.network_settings.name)
+ self.port_settings = PortSettings(
+ name=guid + '-port',
+ network_name=net_config.network_settings.name)
except Exception as e:
self.tearDown()
raise e
@@ -1529,25 +1814,32 @@ class CreateInstanceFromThreePartImage(OSIntegrationTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
super(self.__class__, self).__clean__()
@@ -1555,12 +1847,16 @@ class CreateInstanceFromThreePartImage(OSIntegrationTestCase):
"""
Tests the creation of an OpenStack instance from a 3-part image.
"""
- instance_settings = VmInstanceSettings(name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name,
- port_settings=[self.port_settings])
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
- # The last created image is the main image from which we create the instance
+ # The last created image is the main image from which we create the
+ # instance
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings)
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create()
self.assertIsNotNone(vm_inst)
@@ -1569,14 +1865,15 @@ class CreateInstanceFromThreePartImage(OSIntegrationTestCase):
class CreateInstanceMockOfflineTests(OSComponentTestCase):
"""
- Tests the custom image_metadata that can be set by clients for handling images differently than the
- default behavior of the existing tests primarily for offline testing
+ Tests the custom image_metadata that can be set by clients for handling
+ images differently than the default behavior of the existing tests
+ primarily for offline testing
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
@@ -1595,23 +1892,28 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
self.inst_creator = None
self.priv_net_config = openstack_tests.get_priv_net_config(
- net_name=self.guid + '-priv-net', subnet_name=self.guid + '-priv-subnet')
+ net_name=self.guid + '-priv-net',
+ subnet_name=self.guid + '-priv-subnet')
self.port_settings = PortSettings(
- name=self.port_1_name, network_name=self.priv_net_config.network_settings.name)
+ name=self.port_1_name,
+ network_name=self.priv_net_config.network_settings.name)
try:
# Download image file
- self.image_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_IMAGE_URL, self.tmpDir)
+ self.image_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_IMAGE_URL, self.tmpDir)
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, self.priv_net_config.network_settings)
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, self.priv_net_config.network_settings)
self.network_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.os_creds,
FlavorSettings(
- name=self.guid + '-flavor-name', ram=128, disk=10, vcpus=1))
+ name=self.guid + '-flavor-name', ram=128, disk=10,
+ vcpus=1))
self.flavor_creator.create()
except Exception as e:
self.tearDown()
@@ -1625,43 +1927,54 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning VM instance with message - '
+ '%s', e)
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning network 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 - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning flavor with message - %s',
+ e)
if self.image_creator:
try:
self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + str(e))
+ logger.error(
+ 'Unexpected exception cleaning image with message - %s', e)
if os.path.exists(self.tmpDir) and os.path.isdir(self.tmpDir):
shutil.rmtree(self.tmpDir)
def test_inst_from_file_image_simple_flat(self):
"""
- Creates a VM instance from a locally sourced file image using simply the 'disk_file' attribute vs.
- using the 'config' option which completely overrides all image settings
+ Creates a VM instance from a locally sourced file image using simply
+ the 'disk_file' attribute vs. using the 'config' option which
+ completely overrides all image settings
:return:
"""
metadata = {'disk_file': self.image_file.name}
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=metadata)
self.assertEqual(self.image_file.name, os_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.image_user)
self.assertIsNone(os_image_settings.url)
self.assertFalse(os_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.format)
self.assertIsNone(os_image_settings.kernel_image_settings)
self.assertIsNone(os_image_settings.ramdisk_image_settings)
@@ -1670,27 +1983,34 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
self.image_creator.create()
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
def test_inst_from_file_image_simple_nested(self):
"""
- Creates a VM instance from a locally sourced file image using simply the 'disk_file' attribute under 'cirros'
- vs. using the 'config' option which completely overrides all image settings
+ Creates a VM instance from a locally sourced file image using simply
+ the 'disk_file' attribute under 'cirros' vs. using the 'config' option
+ which completely overrides all image settings
:return:
"""
metadata = {'cirros': {'disk_file': self.image_file.name}}
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=metadata)
self.assertEqual(self.image_file.name, os_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.image_user)
self.assertIsNone(os_image_settings.url)
self.assertFalse(os_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.format)
self.assertIsNone(os_image_settings.kernel_image_settings)
self.assertIsNone(os_image_settings.ramdisk_image_settings)
@@ -1699,57 +2019,78 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
self.image_creator.create()
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
def test_inst_from_existing(self):
"""
- Creates a VM instance from a image creator that has been configured to use an existing image
+ Creates a VM instance from a image creator that has been configured to
+ use an existing image
:return:
"""
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name)
self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
+ image_settings = self.image_creator.image_settings
test_image_creator = OpenStackImage(
- self.os_creds, ImageSettings(name=self.image_creator.image_settings.name,
- image_user=self.image_creator.image_settings.image_user, exists=True))
+ self.os_creds,
+ ImageSettings(name=image_settings.name,
+ image_user=image_settings.image_user,
+ exists=True))
test_image_creator.create()
- self.assertEqual(self.image_creator.get_image().id, test_image_creator.get_image().id)
+ self.assertEqual(self.image_creator.get_image().id,
+ test_image_creator.get_image().id)
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- test_image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ test_image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
def test_inst_from_file_image_complex(self):
"""
- Creates a VM instance from a locally sourced file image by overriding the default settings by using a dict()
- that can be read in by ImageSettings
+ Creates a VM instance from a locally sourced file image by overriding
+ the default settings by using a dict() that can be read in by
+ ImageSettings
:return:
"""
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name)
self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
- metadata = {'cirros': {'config':
- {'name': os_image_settings.name, 'image_user': os_image_settings.image_user,
- 'exists': True}}}
- test_image_settings = openstack_tests.cirros_image_settings(image_metadata=metadata)
+ metadata = {
+ 'cirros':
+ {'config':
+ {'name': os_image_settings.name,
+ 'image_user': os_image_settings.image_user,
+ 'exists': True}}}
+ test_image_settings = openstack_tests.cirros_image_settings(
+ image_metadata=metadata)
test_image = OpenStackImage(self.os_creds, test_image_settings)
test_image.create()
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(self.os_creds,
+ instance_settings,
test_image_settings)
self.inst_creator.create()
@@ -1757,102 +2098,136 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
def test_inst_from_file_3part_image_complex(self):
"""
- Creates a VM instance from a locally sourced file image by overriding the default settings by using a dict()
- that can be read in by ImageSettings
+ Creates a VM instance from a locally sourced file image by overriding
+ the default settings by using a dict() that can be read in by
+ ImageSettings
:return:
"""
- kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
- ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
-
- metadata = {'cirros':
- {'config':
- {'name': self.image_name,
- 'image_user': openstack_tests.CIRROS_USER,
- 'image_file': self.image_file.name,
- 'format': openstack_tests.DEFAULT_IMAGE_FORMAT,
- 'kernel_image_settings':
- {'name': self.image_name + '-kernel',
- 'image_user': openstack_tests.CIRROS_USER,
- 'image_file': kernel_file.name,
- 'format': openstack_tests.DEFAULT_IMAGE_FORMAT},
- 'ramdisk_image_settings':
- {'name': self.image_name + '-ramdisk',
- 'image_user': openstack_tests.CIRROS_USER,
- 'image_file': ramdisk_file.name,
- 'format': openstack_tests.DEFAULT_IMAGE_FORMAT}}}}
-
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ kernel_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+
+ metadata = {
+ 'cirros':
+ {'config':
+ {'name': self.image_name,
+ 'image_user': openstack_tests.CIRROS_USER,
+ 'image_file': self.image_file.name,
+ 'format': openstack_tests.DEFAULT_IMAGE_FORMAT,
+ 'kernel_image_settings':
+ {'name': self.image_name + '-kernel',
+ 'image_user': openstack_tests.CIRROS_USER,
+ 'image_file': kernel_file.name,
+ 'format': openstack_tests.DEFAULT_IMAGE_FORMAT},
+ 'ramdisk_image_settings':
+ {'name': self.image_name + '-ramdisk',
+ 'image_user': openstack_tests.CIRROS_USER,
+ 'image_file': ramdisk_file.name,
+ 'format': openstack_tests.DEFAULT_IMAGE_FORMAT}}}}
+
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=metadata)
self.assertEqual(self.image_name, os_image_settings.name)
self.assertEqual(self.image_file.name, os_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.image_user)
self.assertIsNone(os_image_settings.url)
self.assertFalse(os_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.format)
self.assertIsNotNone(os_image_settings.kernel_image_settings)
- self.assertEqual(self.image_name + '-kernel', os_image_settings.kernel_image_settings.name)
- self.assertEqual(kernel_file.name, os_image_settings.kernel_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.kernel_image_settings.image_user)
+ self.assertEqual(self.image_name + '-kernel',
+ os_image_settings.kernel_image_settings.name)
+ self.assertEqual(kernel_file.name,
+ os_image_settings.kernel_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.kernel_image_settings.image_user)
self.assertIsNone(os_image_settings.kernel_image_settings.url)
self.assertFalse(os_image_settings.kernel_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.kernel_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.kernel_image_settings.format)
self.assertIsNotNone(os_image_settings.ramdisk_image_settings)
- self.assertEqual(self.image_name + '-ramdisk', os_image_settings.ramdisk_image_settings.name)
- self.assertEqual(ramdisk_file.name, os_image_settings.ramdisk_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.ramdisk_image_settings.image_user)
+ self.assertEqual(self.image_name + '-ramdisk',
+ os_image_settings.ramdisk_image_settings.name)
+ self.assertEqual(ramdisk_file.name,
+ os_image_settings.ramdisk_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.ramdisk_image_settings.image_user)
self.assertIsNone(os_image_settings.ramdisk_image_settings.url)
self.assertFalse(os_image_settings.ramdisk_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.ramdisk_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.ramdisk_image_settings.format)
self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings, self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
def test_inst_from_file_3part_image_simple_flat(self):
"""
- Creates a VM instance from a 3-part image locally sourced from file images using simply the 'disk_file',
- 'kernel_file', and 'ramdisk_file' attributes vs. using the 'config' option which completely overrides all
+ Creates a VM instance from a 3-part image locally sourced from file
+ images using simply the 'disk_file', 'kernel_file', and 'ramdisk_file'
+ attributes vs. using the 'config' option which completely overrides all
image settings
:return:
"""
- kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
- ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+ kernel_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
- metadata = {'disk_file': self.image_file.name, 'kernel_file': kernel_file.name,
+ metadata = {'disk_file': self.image_file.name,
+ 'kernel_file': kernel_file.name,
'ramdisk_file': ramdisk_file.name}
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=metadata)
self.assertEqual(self.image_name, os_image_settings.name)
self.assertEqual(self.image_file.name, os_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.image_user)
self.assertIsNone(os_image_settings.url)
self.assertFalse(os_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.format)
self.assertIsNotNone(os_image_settings.kernel_image_settings)
- self.assertEqual(self.image_name + '-kernel', os_image_settings.kernel_image_settings.name)
- self.assertEqual(kernel_file.name, os_image_settings.kernel_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.kernel_image_settings.image_user)
+ self.assertEqual(self.image_name + '-kernel',
+ os_image_settings.kernel_image_settings.name)
+ self.assertEqual(kernel_file.name,
+ os_image_settings.kernel_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.kernel_image_settings.image_user)
self.assertIsNone(os_image_settings.kernel_image_settings.url)
self.assertFalse(os_image_settings.kernel_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.kernel_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.kernel_image_settings.format)
self.assertIsNotNone(os_image_settings.ramdisk_image_settings)
- self.assertEqual(self.image_name + '-ramdisk', os_image_settings.ramdisk_image_settings.name)
- self.assertEqual(ramdisk_file.name, os_image_settings.ramdisk_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.ramdisk_image_settings.image_user)
+ self.assertEqual(self.image_name + '-ramdisk',
+ os_image_settings.ramdisk_image_settings.name)
+ self.assertEqual(ramdisk_file.name,
+ os_image_settings.ramdisk_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.ramdisk_image_settings.image_user)
self.assertIsNone(os_image_settings.ramdisk_image_settings.url)
self.assertFalse(os_image_settings.ramdisk_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.ramdisk_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.ramdisk_image_settings.format)
self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
@@ -1861,50 +2236,68 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
self.assertIsNotNone(self.image_creator.get_ramdisk_image())
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
def test_inst_from_file_3part_image_simple_nested(self):
"""
- Creates a VM instance from a 3-part image locally sourced from file images using simply the 'disk_file',
- 'kernel_file', and 'ramdisk_file' attributes under 'cirros' vs. using the 'config' option which completely
- overrides all image settings
+ Creates a VM instance from a 3-part image locally sourced from file
+ images using simply the 'disk_file', 'kernel_file', and 'ramdisk_file'
+ attributes under 'cirros' vs. using the 'config' option which
+ completely overrides all image settings
:return:
"""
- kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
- ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+ kernel_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
- metadata = {'cirros': {'disk_file': self.image_file.name, 'kernel_file': kernel_file.name,
+ metadata = {'cirros': {'disk_file': self.image_file.name,
+ 'kernel_file': kernel_file.name,
'ramdisk_file': ramdisk_file.name}}
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=metadata)
self.assertEqual(self.image_name, os_image_settings.name)
self.assertEqual(self.image_file.name, os_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.image_user)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.image_user)
self.assertIsNone(os_image_settings.url)
self.assertFalse(os_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.format)
self.assertIsNotNone(os_image_settings.kernel_image_settings)
- self.assertEqual(self.image_name + '-kernel', os_image_settings.kernel_image_settings.name)
- self.assertEqual(kernel_file.name, os_image_settings.kernel_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.kernel_image_settings.image_user)
+ self.assertEqual(self.image_name + '-kernel',
+ os_image_settings.kernel_image_settings.name)
+ self.assertEqual(kernel_file.name,
+ os_image_settings.kernel_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.kernel_image_settings.image_user)
self.assertIsNone(os_image_settings.kernel_image_settings.url)
self.assertFalse(os_image_settings.kernel_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.kernel_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.kernel_image_settings.format)
self.assertIsNotNone(os_image_settings.ramdisk_image_settings)
- self.assertEqual(self.image_name + '-ramdisk', os_image_settings.ramdisk_image_settings.name)
- self.assertEqual(ramdisk_file.name, os_image_settings.ramdisk_image_settings.image_file)
- self.assertEqual(openstack_tests.CIRROS_USER, os_image_settings.ramdisk_image_settings.image_user)
+ self.assertEqual(self.image_name + '-ramdisk',
+ os_image_settings.ramdisk_image_settings.name)
+ self.assertEqual(ramdisk_file.name,
+ os_image_settings.ramdisk_image_settings.image_file)
+ self.assertEqual(openstack_tests.CIRROS_USER,
+ os_image_settings.ramdisk_image_settings.image_user)
self.assertIsNone(os_image_settings.ramdisk_image_settings.url)
self.assertFalse(os_image_settings.ramdisk_image_settings.exists)
- self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT, os_image_settings.ramdisk_image_settings.format)
+ self.assertEqual(openstack_tests.DEFAULT_IMAGE_FORMAT,
+ os_image_settings.ramdisk_image_settings.format)
self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
@@ -1913,9 +2306,12 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
self.assertIsNotNone(self.image_creator.get_ramdisk_image())
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
@@ -1925,26 +2321,37 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
Creates a VM instance from a 3-part image that is existing
:return:
"""
- kernel_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
- ramdisk_file = file_utils.download(openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
+ kernel_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_KERNEL_IMAGE_URL, self.tmpDir)
+ ramdisk_file = file_utils.download(
+ openstack_tests.CIRROS_DEFAULT_RAMDISK_IMAGE_URL, self.tmpDir)
- metadata = {'cirros': {'disk_file': self.image_file.name, 'kernel_file': kernel_file.name,
+ metadata = {'cirros': {'disk_file': self.image_file.name,
+ 'kernel_file': kernel_file.name,
'ramdisk_file': ramdisk_file.name}}
- os_image_settings = openstack_tests.cirros_image_settings(name=self.image_name, image_metadata=metadata)
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.image_name, image_metadata=metadata)
self.image_creator = OpenStackImage(self.os_creds, os_image_settings)
self.image_creator.create()
+ image_settings = self.image_creator.image_settings
test_image_creator = OpenStackImage(
- self.os_creds, ImageSettings(name=self.image_creator.image_settings.name,
- image_user=self.image_creator.image_settings.image_user, exists=True))
+ self.os_creds,
+ ImageSettings(name=image_settings.name,
+ image_user=image_settings.image_user,
+ exists=True))
test_image_creator.create()
- self.assertEqual(self.image_creator.get_image().id, test_image_creator.get_image().id)
+ self.assertEqual(self.image_creator.get_image().id,
+ test_image_creator.get_image().id)
instance_settings = VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
- self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- test_image_creator.image_settings)
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[self.port_settings])
+ self.inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ test_image_creator.image_settings)
self.inst_creator.create()
self.assertTrue(self.inst_creator.vm_active(block=True))
@@ -1953,8 +2360,9 @@ class CreateInstanceMockOfflineTests(OSComponentTestCase):
def check_dhcp_lease(vm, ip, timeout=160):
"""
Returns true if the expected DHCP lease has been acquired
- :param vm:
- :param ip:
+ :param vm: the OpenStack VM instance object
+ :param ip: the IP address to look for
+ :param timeout: how long to query for IP address
:return:
"""
found = False