"""
return self.__vm
- def get_os_vm_server_obj(self):
+ def get_console_output(self):
"""
- Returns the OpenStack server object
- :return: the server object
+ Returns the vm console object for parsing logs
+ :return: the console output object
"""
- return nova_utils.get_latest_server_os_object(self.__nova, self.__vm)
+ return nova_utils.get_server_console_output(self.__nova, self.__vm)
def get_port_ip(self, port_name, subnet_name=None):
"""
logger.warning('Cannot find port with name - ' + port_name)
return None
+ def get_vm_info(self):
+ """
+ Returns a dictionary of a VMs info as returned by OpenStack
+ :return: a dict()
+ """
+ return nova_utils.get_server_info(self.__nova, self.__vm)
+
def config_nics(self):
"""
Responsible for configuring NICs on RPM systems where the instance has
if not self.__vm:
return False
- instance = nova_utils.get_latest_server_os_object(
- self.__nova, self.__vm)
- if not instance:
+ status = nova_utils.get_server_status(self.__nova, self.__vm)
+ if not status:
logger.warning('Cannot find instance with id - ' + self.__vm.id)
return False
- if instance.status == 'ERROR':
+ if status == 'ERROR':
raise Exception('Instance had an error during deployment')
logger.debug(
'Instance status [%s] is - %s', self.instance_settings.name,
- instance.status)
- return instance.status == expected_status_code
+ status)
+ return status == expected_status_code
def vm_ssh_active(self, block=False, poll_interval=POLL_INTERVAL):
"""
self.inst_creator = OpenStackVmInstance(
self.os_creds, instance_settings,
self.image_creator.image_settings)
- vm = self.inst_creator.create()
+ self.inst_creator.create()
ip = self.inst_creator.get_port_ip(self.port_settings.name)
self.assertIsNotNone(ip)
self.assertTrue(self.inst_creator.vm_active(block=True))
- self.assertTrue(check_dhcp_lease(self.nova, vm, ip))
+ self.assertTrue(check_dhcp_lease(self.inst_creator, ip))
class CreateInstanceSimpleTests(OSIntegrationTestCase):
self.assertTrue(inst_creator.vm_active(block=True))
ip = inst_creator.get_port_ip(port_settings.name)
- self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(inst_creator, ip))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
self.assertTrue(inst_creator.vm_active(block=True))
ip = inst_creator.get_port_ip(port_settings.name)
- self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(inst_creator, ip))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
for zone in zones:
creator = self.inst_creators[index]
self.assertTrue(creator.vm_active(block=True))
- vm = creator.get_os_vm_server_obj()
- deployed_zone = vm._info['OS-EXT-AZ:availability_zone']
- deployed_host = vm._info['OS-EXT-SRV-ATTR:host']
+ info = creator.get_vm_info()
+ deployed_zone = info['OS-EXT-AZ:availability_zone']
+ deployed_host = info['OS-EXT-SRV-ATTR:host']
self.assertEqual(zone, deployed_zone + ':' + deployed_host)
index += 1
self.assertTrue(self.inst_creator.vm_active(block=True))
ip = self.inst_creator.get_port_ip(ports_settings[0].name)
- self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(self.inst_creator, ip))
# Add security group to VM
self.inst_creator.add_security_group(
:param nova: the nova client
:param vm_inst: the VmInst domain object
:param sec_grp_name: the name of the security group to validate
- :return:
+ :return: T/F
"""
- vm = nova_utils.get_latest_server_os_object(nova, vm_inst)
- if not hasattr(vm, 'security_groups'):
- return False
-
- found = False
- for sec_grp_dict in vm.security_groups:
- if sec_grp_name in sec_grp_dict['name']:
- found = True
- break
- return found
+ sec_grp_names = nova_utils.get_server_security_group_names(nova, vm_inst)
+ for name in sec_grp_names:
+ if sec_grp_name == name:
+ return True
+ return False
def validate_ssh_client(instance_creator):
self.assertTrue(self.inst_creator.vm_active(block=True))
-def check_dhcp_lease(nova_client, vm_domain, ip, timeout=160):
+def check_dhcp_lease(inst_creator, ip, timeout=160):
"""
Returns true if the expected DHCP lease has been acquired
- :param nova_client: the nova client
- :param vm_domain: the SNAPS VM instance domain object
+ :param inst_creator: the SNAPS OpenStackVmInstance object
:param ip: the IP address to look for
:param timeout: how long to query for IP address
:return:
logger.info("Looking for IP %s in the console log" % ip)
full_log = ''
while timeout > time.time() - start_time:
- vm = nova_utils.get_latest_server_os_object(nova_client, vm_domain)
- output = vm.get_console_output()
+ output = inst_creator.get_console_output()
full_log = full_log + output
if re.search(ip, output):
logger.info('DHCP lease obtained logged in console')
self.assertIsNotNone(retrieved_user)
self.assertEqual(created_user, retrieved_user)
- role = keystone_utils._get_os_role_by_name(self.keystone, 'admin')
+ role = keystone_utils.get_role_by_name(self.keystone, 'admin')
self.assertIsNotNone(role)
os_proj = keystone_utils.get_project(
keystone=self.keystone, project_name=self.os_creds.project_name)
- user_roles = keystone_utils._get_os_roles_by_user(
+ user_roles = keystone_utils.get_roles_by_user(
self.keystone, retrieved_user, os_proj)
self.assertIsNotNone(user_roles)
self.assertEqual(1, len(user_roles))
:param os_creds: the OpenStack credentials used to obtain the Keystone
client if the keystone parameter is None
:param project_name: the name to query
- :return: the ID or None
+ :return: the SNAPS-OO Project domain object or None
"""
if not project_name:
return None
:return: SNAPS-OO Project domain object
"""
if keystone.version == V2_VERSION:
- return keystone.tenants.create(
+ os_project = keystone.tenants.create(
project_settings.name, project_settings.description,
project_settings.enabled)
+ else:
+ os_project = keystone.projects.create(
+ project_settings.name, project_settings.domain,
+ description=project_settings.description,
+ enabled=project_settings.enabled)
- return keystone.projects.create(
- project_settings.name, project_settings.domain,
- description=project_settings.description,
- enabled=project_settings.enabled)
+ return Project(name=os_project.name, project_id=os_project.id)
def delete_project(keystone, project):
domain=user_settings.domain_name, enabled=user_settings.enabled)
for role_name, role_project in user_settings.roles.items():
- os_role = _get_os_role_by_name(keystone, role_name)
+ os_role = get_role_by_name(keystone, role_name)
os_project = get_project(keystone=keystone, project_name=role_project)
if os_role and os_project:
- existing_roles = _get_os_roles_by_user(keystone, os_user,
- os_project)
+ existing_roles = get_roles_by_user(keystone, os_user,
+ os_project)
found = False
for role in existing_roles:
if role.id == os_role.id:
keystone.users.delete(user.id)
-def _get_os_role_by_name(keystone, name):
+def get_role_by_name(keystone, name):
"""
Returns an OpenStack role object of a given name or None if not exists
:param keystone: the keystone client
return Role(name=role.name, role_id=role.id)
-def _get_os_roles_by_user(keystone, user, project):
+def get_roles_by_user(keystone, user, project):
"""
- Returns a list of OpenStack role object associated with a user
+ Returns a list of SNAPS-OO Role domain objects associated with a user
:param keystone: the keystone client
:param user: the OpenStack user object
:param project: the OpenStack project object (only required for v2)
return out
-def __get_os_role_by_id(keystone, role_id):
+def get_role_by_id(keystone, role_id):
"""
Returns an OpenStack role object of a given name or None if not exists
:param keystone: the keystone client
:return:
"""
- os_role = __get_os_role_by_id(keystone, role.id)
+ os_role = get_role_by_id(keystone, role.id)
if keystone.version == V2_VERSION:
keystone.roles.add_user_role(user, os_role, tenant=project)
else:
for network, netInsts in networks.items():
for inst in netInsts:
if inst.get('name') == network_name:
- if project_id and inst.get('project_id') == project_id:
- return {'network': inst}
- else:
- return Network(**inst)
+ return Network(**inst)
return None
Removes an OpenStack port
:param neutron: the client
:param port: the SNAPS-OO Port domain object
- :return:
"""
logger.info('Deleting port with name ' + port.name)
neutron.delete_port(port.id)
Returns the first port object (dictionary) found with a given name
:param neutron: the client
:param port_name: the name of the port to retrieve
- :return:
+ :return: a SNAPS-OO Port domain object
"""
ports = neutron.list_ports(**{'name': port_name})
for port in ports['ports']:
:param neutron: the Neutron client
:param keystone: the Keystone client
:param sec_grp_settings: the security group settings
- :return: the security group object
+ :return: a SNAPS-OO SecurityGroup domain object
"""
logger.info('Creating security group with name - %s',
sec_grp_settings.name)
Returns the first security group object of the given name else None
:param neutron: the client
:param name: the name of security group object to retrieve
+ :return: a SNAPS-OO SecurityGroup domain object or None if not found
"""
logger.info('Retrieving security group with name - ' + name)
Returns the first security group object of the given name else None
:param neutron: the client
:param sec_grp_id: the id of the security group to retrieve
+ :return: a SNAPS-OO SecurityGroup domain object or None if not found
"""
logger.info('Retrieving security group with ID - ' + sec_grp_id)
Creates a security group object in OpenStack
:param neutron: the client
:param sec_grp_rule_settings: the security group rule settings
- :return: the security group object
+ :return: a SNAPS-OO SecurityGroupRule domain object
"""
logger.info('Creating security group to security group - %s',
sec_grp_rule_settings.sec_grp_name)
"""
Retrieves all of the rules for a given security group
:param neutron: the client
- :param sec_grp: the SNAPS SecurityGroup object
+ :param sec_grp: a list of SNAPS SecurityGroupRule domain objects
"""
logger.info('Retrieving security group rules associate with the '
'security group - %s', sec_grp.name)
:param neutron: the client
:param sec_grp: the SNAPS SecurityGroup domain object
:param rule_id: the rule's ID
+ :param sec_grp: a SNAPS SecurityGroupRule domain object
"""
rules = neutron.list_security_group_rules(
**{'security_group_id': sec_grp.id})
Returns a list of external OpenStack network object/dict for all external
networks
:param neutron: the client
- :return: a list of external networks (empty list if none configured)
+ :return: a list of external networks of Type SNAPS-OO domain class Network
"""
out = list()
for network in neutron.list_networks(
**{'router:external': True})['networks']:
- out.append({'network': network})
+ out.append(Network(**network))
return out
Returns a list of servers with a given name
:param nova: the Nova client
:param name: the server name
- :return: the list of servers
+ :return: the list of snaps.domain.VmInst objects
"""
out = list()
servers = nova.servers.list(search_opts={'name': name})
return out
-def get_latest_server_os_object(nova, server):
+def __get_latest_server_os_object(nova, server):
"""
Returns a server with a given id
:param nova: the Nova client
return nova.servers.get(server.id)
+def get_server_status(nova, server):
+ """
+ Returns the a VM instance's status from OpenStack
+ :param nova: the Nova client
+ :param server: the domain VmInst object
+ :return: the VM's string status or None if not founc
+ """
+ server = __get_latest_server_os_object(nova, server)
+ if server:
+ return server.status
+ return None
+
+
+def get_server_console_output(nova, server):
+ """
+ Returns the console object for parsing VM activity
+ :param nova: the Nova client
+ :param server: the domain VmInst object
+ :return: the console output object or None if server object is not found
+ """
+ server = __get_latest_server_os_object(nova, server)
+ if server:
+ return server.get_console_output()
+ return None
+
+
def get_latest_server_object(nova, server):
"""
Returns a server with a given id
:param server: the old server object
:return: the list of servers or None if not found
"""
- server = get_latest_server_os_object(nova, server)
+ server = __get_latest_server_os_object(nova, server)
return VmInst(name=server.name, inst_id=server.id,
networks=server.networks)
+def get_server_security_group_names(nova, server):
+ """
+ Returns a server with a given id
+ :param nova: the Nova client
+ :param server: the old server object
+ :return: the list of security groups associated with a VM
+ """
+ out = list()
+ os_vm_inst = __get_latest_server_os_object(nova, server)
+ for sec_grp_dict in os_vm_inst.security_groups:
+ out.append(sec_grp_dict['name'])
+ return out
+
+
+def get_server_info(nova, server):
+ """
+ Returns a dictionary of a VMs info as returned by OpenStack
+ :param nova: the Nova client
+ :param server: the old server object
+ :return: a dict of the info if VM exists else None
+ """
+ vm = __get_latest_server_os_object(nova, server)
+ if vm:
+ return vm._info
+ return None
+
+
def create_keys(key_size=2048):
"""
Generates public and private keys
:param keys: the keys to save generated by cryptography
:param pub_file_path: the path to the public keys
:param priv_file_path: the path to the private keys
- :return: None
"""
if keys:
if pub_file_path:
nova.servers.delete(vm_inst.id)
-def get_os_flavor(nova, flavor):
+def __get_os_flavor(nova, flavor):
"""
Returns to OpenStack flavor object by name
:param nova: the Nova client
:param flavor: the SNAPS flavor domain object
:return: the SNAPS Flavor domain object
"""
- os_flavor = get_os_flavor(nova, flavor)
+ os_flavor = __get_os_flavor(nova, flavor)
if os_flavor:
return Flavor(
name=os_flavor.name, id=os_flavor.id, ram=os_flavor.ram,
return None
-def get_os_flavor_by_name(nova, name):
+def __get_os_flavor_by_name(nova, name):
"""
Returns to OpenStack flavor object by name
:param nova: the Nova client
:param name: the flavor name to return
:return: the SNAPS flavor domain object or None if not exists
"""
- os_flavor = get_os_flavor_by_name(nova, name)
+ os_flavor = __get_os_flavor_by_name(nova, name)
if os_flavor:
return Flavor(
name=os_flavor.name, id=os_flavor.id, ram=os_flavor.ram,
:param flavor: the SNAPS flavor domain object
:param metadata: the metadata to set
"""
- os_flavor = get_os_flavor(nova, flavor)
+ os_flavor = __get_os_flavor(nova, flavor)
if os_flavor:
os_flavor.set_keys(metadata)
:param floating_ip: FloatingIp domain object
:param ip_addr: the IP to which to bind the floating IP to
"""
- vm = get_latest_server_os_object(nova, vm)
+ vm = __get_latest_server_os_object(nova, vm)
vm.add_floating_ip(floating_ip.ip, ip_addr)
keystone_utils.grant_user_role_to_project(
self.keystone, self.role, self.user, self.project)
- user_roles = keystone_utils._get_os_roles_by_user(
+ user_roles = keystone_utils.get_roles_by_user(
self.keystone, self.user, self.project)
self.assertIsNotNone(user_roles)
self.assertEqual(1, len(user_roles))
ext_networks = neutron_utils.get_external_networks(neutron)
found = False
for network in ext_networks:
- if network['network']['name'] == self.ext_net_name:
+ if network.name == self.ext_net_name:
found = True
break
self.assertTrue(found)
Should this not be performed, the creation of the host ssh key will
cause your ansible calls to fail.
"""
- vm = self.inst_creator.create(block=True)
+ self.inst_creator.create(block=True)
priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
- self.assertTrue(check_dhcp_lease(self.nova, vm, priv_ip))
+ self.assertTrue(check_dhcp_lease(self.inst_creator, priv_ip))
# Apply Security Group
self.inst_creator.add_security_group(
Should this not be performed, the creation of the host ssh key will
cause your ansible calls to fail.
"""
- vm = self.inst_creator.create(block=True)
+ self.inst_creator.create(block=True)
priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
- self.assertTrue(check_dhcp_lease(self.nova, vm, priv_ip))
+ self.assertTrue(check_dhcp_lease(self.inst_creator, priv_ip))
# Apply Security Group
self.inst_creator.add_security_group(
Code Review / snaps.git / commitdiff