| import logging |
| import os |
| import random |
| import sys |
| import time |
| |
| from cinderclient import client as cinder_client |
| from glanceclient import client as glance_client |
| from keystoneauth1 import identity as keystone_identity |
| from keystoneauth1 import session as keystone_session |
| from keystoneclient.v3 import client as keystone_client |
| from neutronclient.v2_0 import client as neutron_client |
| from neutronclient import common as neutron_common |
| from novaclient import client as novaclient |
| |
| import utils |
| |
| logger = logging.getLogger(__name__) |
| |
| |
| class OfficialClientManager(object): |
| """Manager that provides access to the official python clients for |
| calling various OpenStack APIs. |
| """ |
| |
| CINDERCLIENT_VERSION = 3 |
| GLANCECLIENT_VERSION = 2 |
| KEYSTONECLIENT_VERSION = 3 |
| NEUTRONCLIENT_VERSION = 2 |
| NOVACLIENT_VERSION = 2 |
| INTERFACE = 'admin' |
| if "OS_ENDPOINT_TYPE" in list(os.environ.keys()): |
| INTERFACE = os.environ["OS_ENDPOINT_TYPE"] |
| |
| def __init__(self, username=None, password=None, |
| project_name=None, auth_url=None, endpoint_type="internalURL", |
| interface=INTERFACE, |
| cert=False, domain="Default", **kwargs): |
| self.traceback = "" |
| |
| self.client_attr_names = [ |
| "auth", |
| "compute", |
| "network", |
| "volume", |
| "image", |
| ] |
| self.username = username |
| self.password = password |
| self.project_name = project_name |
| self.auth_url = auth_url |
| self.endpoint_type = endpoint_type |
| self.cert = cert |
| self.domain = domain |
| self.interface = interface |
| self.kwargs = kwargs |
| |
| # Lazy clients |
| self._auth = None |
| self._compute = None |
| self._network = None |
| self._volume = None |
| self._image = None |
| |
| @classmethod |
| def _get_auth_session(cls, username=None, password=None, |
| project_name=None, auth_url=None, cert=None, |
| domain='Default'): |
| if None in (username, password, project_name): |
| sys.stdout.write((username, password, project_name)) |
| msg = ("Missing required credentials for identity client. " |
| "username: {username}, password: {password}, " |
| "project_name: {project_name}").format( |
| username=username, |
| password=password, |
| project_name=project_name |
| ) |
| raise msg |
| |
| if cert and "https" not in auth_url: |
| auth_url = auth_url.replace("http", "https") |
| |
| if "v2" in auth_url: |
| raise BaseException("Keystone v2 is deprecated since OpenStack" |
| "Queens release. So current OS_AUTH_URL {} " |
| "is not valid. Please use Keystone v3." |
| "".format(auth_url)) |
| else: |
| auth_url = auth_url if ("v3" in auth_url) else "{}{}".format( |
| auth_url, "/v3") |
| auth = keystone_identity.v3.Password( |
| auth_url=auth_url, |
| user_domain_name=domain, |
| username=username, |
| password=password, |
| project_domain_name=domain, |
| project_name=project_name) |
| |
| auth_session = keystone_session.Session(auth=auth, verify=cert) |
| # auth_session.get_auth_headers() |
| return auth_session |
| |
| @classmethod |
| def get_auth_client(cls, username=None, password=None, |
| project_name=None, auth_url=None, cert=None, |
| domain='Default', os_interface=None, **kwargs): |
| session = cls._get_auth_session( |
| username=username, |
| password=password, |
| project_name=project_name, |
| auth_url=auth_url, |
| cert=cert, |
| domain=domain) |
| endpoint_filter = {"interface": os_interface} |
| keystone = keystone_client.Client(version=cls.KEYSTONECLIENT_VERSION, |
| session=session, **kwargs, |
| interface=os_interface, |
| endpoint_filter=endpoint_filter) |
| keystone.management_url = auth_url |
| return keystone |
| |
| @classmethod |
| def get_compute_client(cls, username=None, password=None, |
| project_name=None, auth_url=None, cert=None, |
| domain='Default', **kwargs): |
| session = cls._get_auth_session( |
| username=username, password=password, project_name=project_name, |
| auth_url=auth_url, cert=cert, domain=domain) |
| service_type = 'compute' |
| compute_client = novaclient.Client( |
| version=cls.NOVACLIENT_VERSION, session=session, |
| service_type=service_type, os_cache=False, **kwargs) |
| return compute_client |
| |
| @classmethod |
| def get_network_client(cls, username=None, password=None, |
| project_name=None, auth_url=None, cert=None, |
| domain='Default', **kwargs): |
| session = cls._get_auth_session( |
| username=username, password=password, project_name=project_name, |
| auth_url=auth_url, cert=cert, domain=domain) |
| service_type = 'network' |
| return neutron_client.Client( |
| service_type=service_type, session=session, |
| interface=cls.INTERFACE, **kwargs) |
| |
| @classmethod |
| def get_volume_client(cls, username=None, password=None, |
| project_name=None, auth_url=None, cert=None, |
| domain='Default', **kwargs): |
| session = cls._get_auth_session( |
| username=username, password=password, project_name=project_name, |
| auth_url=auth_url, cert=cert, domain=domain) |
| service_type = 'volume' |
| return cinder_client.Client( |
| version=cls.CINDERCLIENT_VERSION, |
| service_type=service_type, |
| interface=cls.INTERFACE, |
| session=session, **kwargs) |
| |
| @classmethod |
| def get_image_client(cls, username=None, password=None, |
| project_name=None, auth_url=None, cert=None, |
| domain='Default', **kwargs): |
| session = cls._get_auth_session( |
| username=username, password=password, project_name=project_name, |
| auth_url=auth_url, cert=cert, domain=domain) |
| service_type = 'image' |
| return glance_client.Client( |
| version=cls.GLANCECLIENT_VERSION, |
| service_type=service_type, |
| session=session, interface=cls.INTERFACE, |
| **kwargs) |
| |
| @property |
| def auth(self): |
| if self._auth is None: |
| self._auth = self.get_auth_client( |
| self.username, self.password, self.project_name, self.auth_url, |
| self.cert, self.domain, endpoint_type=self.endpoint_type, |
| os_interface=self.interface |
| ) |
| return self._auth |
| |
| @property |
| def compute(self): |
| if self._compute is None: |
| self._compute = self.get_compute_client( |
| self.username, self.password, self.project_name, self.auth_url, |
| self.cert, self.domain, endpoint_type=self.endpoint_type |
| ) |
| return self._compute |
| |
| @property |
| def network(self): |
| if self._network is None: |
| self._network = self.get_network_client( |
| self.username, self.password, self.project_name, self.auth_url, |
| self.cert, self.domain, endpoint_type=self.endpoint_type |
| ) |
| return self._network |
| |
| @property |
| def volume(self): |
| if self._volume is None: |
| self._volume = self.get_volume_client( |
| self.username, self.password, self.project_name, self.auth_url, |
| self.cert, self.domain, endpoint_type=self.endpoint_type |
| ) |
| return self._volume |
| |
| @property |
| def image(self): |
| |
| if self._image is None: |
| self._image = self.get_image_client( |
| self.username, self.password, self.project_name, self.auth_url, |
| self.cert, self.domain |
| ) |
| return self._image |
| |
| |
| class OSCliActions(object): |
| def __init__(self, os_clients): |
| self.os_clients = os_clients |
| self.create_fake_ext_net = False |
| |
| def get_project_by_name(self, name): |
| return self.os_clients.auth.projects.find( |
| name=name, domain_id=self.os_clients.auth.user_domain_id) |
| |
| def get_internal_network(self): |
| networks = [ |
| net for net in self.os_clients.network.list_networks()["networks"] |
| if net["admin_state_up"] and not net["router:external"] and |
| len(net["subnets"]) |
| ] |
| if networks: |
| net = networks[0] |
| else: |
| net = self.create_network_resources() |
| return net |
| |
| def create_fake_external_network(self): |
| logger.info( |
| "Could not find any external network, creating a fake one...") |
| net_name = "spt-ext-net-{}".format(random.randrange(100, 999)) |
| net_body = {"network": {"name": net_name, |
| "router:external": True, |
| "provider:network_type": "local"}} |
| try: |
| ext_net = \ |
| self.os_clients.network.create_network(net_body)['network'] |
| logger.info("Created a fake external net {}".format(net_name)) |
| except Exception as e: |
| # in case 'local' net type is absent, create with default type |
| net_body["network"].pop('provider:network_type', None) |
| ext_net = \ |
| self.os_clients.network.create_network(net_body)['network'] |
| subnet_name = "spt-ext-subnet-{}".format(random.randrange(100, 999)) |
| subnet_body = { |
| "subnet": { |
| "name": subnet_name, |
| "network_id": ext_net["id"], |
| "ip_version": 4, |
| "cidr": "10.255.255.0/24", |
| "allocation_pools": [{"start": "10.255.255.100", |
| "end": "10.255.255.200"}] |
| } |
| } |
| self.os_clients.network.create_subnet(subnet_body) |
| self.create_fake_ext_net = True |
| return ext_net |
| |
| def get_external_network(self): |
| config = utils.get_configuration() |
| ext_net = config.get('external_network') or '' |
| if not ext_net: |
| networks = [ |
| net for net in |
| self.os_clients.network.list_networks()["networks"] |
| if net["admin_state_up"] and net["router:external"] and |
| len(net["subnets"]) |
| ] |
| else: |
| networks = [net for net in |
| self.os_clients.network.list_networks()["networks"] |
| if net["name"] == ext_net] |
| |
| if networks: |
| ext_net = networks[0] |
| logger.info("Using external net '{}'".format(ext_net["name"])) |
| else: |
| ext_net = self.create_fake_external_network() |
| return ext_net |
| |
| def create_flavor(self, name, ram=256, vcpus=1, disk=2): |
| logger.info("Creating a flavor {}".format(name)) |
| return self.os_clients.compute.flavors.create(name, ram, vcpus, disk) |
| |
| def create_sec_group(self, rulesets=None): |
| if rulesets is None: |
| rulesets = [ |
| { |
| # ssh |
| 'protocol': 'tcp', |
| 'port_range_max': 22, |
| 'port_range_min': 22, |
| 'remote_ip_prefix': '0.0.0.0/0', |
| 'direction': 'ingress' |
| }, |
| { |
| # iperf3 |
| 'protocol': 'tcp', |
| 'port_range_max': 5201, |
| 'port_range_min': 5201, |
| 'remote_ip_prefix': '0.0.0.0/0', |
| 'direction': 'ingress' |
| }, |
| { |
| # iperf |
| 'protocol': 'tcp', |
| 'port_range_max': 5001, |
| 'port_range_min': 5001, |
| 'remote_ip_prefix': '0.0.0.0/0', |
| 'direction': 'ingress' |
| }, |
| { |
| # ping |
| 'protocol': 'icmp', |
| 'remote_ip_prefix': '0.0.0.0/0', |
| 'direction': 'ingress' |
| } |
| ] |
| sg_name = "spt-test-secgroup-{}".format(random.randrange(100, 999)) |
| sg_desc = sg_name + " SPT" |
| body = {"security_group": {"name": sg_name, "description": sg_desc}} |
| secgroup = self.os_clients.network.create_security_group(body=body) |
| |
| rule_body_teplate = {"security_group_rule": {}} |
| for ruleset in rulesets: |
| rule_body_teplate["security_group_rule"] = ruleset |
| rule_body_teplate["security_group_rule"]["security_group_id"] = \ |
| secgroup['security_group']['id'] |
| self.os_clients.network.create_security_group_rule( |
| body=rule_body_teplate) |
| logger.info("Created a security group {}".format(sg_name)) |
| return secgroup['security_group'] |
| |
| def create_basic_server(self, image=None, flavor=None, net=None, |
| availability_zone=None, sec_groups=(), |
| keypair=None): |
| os_conn = self.os_clients |
| net = net or self.get_internal_network() |
| kwargs = {} |
| if sec_groups: |
| kwargs['security_groups'] = sec_groups |
| server = os_conn.compute.servers.create( |
| "spt-test-server-{}".format(random.randrange(100, 999)), |
| image, flavor, nics=[{"net-id": net["id"]}], |
| availability_zone=availability_zone, key_name=keypair, **kwargs) |
| |
| return server |
| |
| def get_vm(self, vm_id): |
| os_conn = self.os_clients |
| try: |
| vm = os_conn.compute.servers.find(id=vm_id) |
| except Exception as e: |
| raise Exception( |
| "Could not get the VM \"{}\": {}".format( |
| vm_id, e)) |
| return vm |
| |
| def check_vm_is_active(self, vm_uuid, retry_delay=5, timeout=500): |
| vm = None |
| timeout_reached = False |
| start_time = time.time() |
| expected_state = 'ACTIVE' |
| while not timeout_reached: |
| vm = self.get_vm(vm_uuid) |
| if vm.status == expected_state: |
| logger.info( |
| "VM {} is in {} status.".format(vm_uuid, vm.status)) |
| break |
| if vm.status == 'ERROR': |
| break |
| time.sleep(retry_delay) |
| timeout_reached = (time.time() - start_time) > timeout |
| if vm.status != expected_state: |
| logger.info("VM {} is in {} status.".format(vm_uuid, vm.status)) |
| raise TimeoutError( |
| "VM {vm_uuid} on is expected to be in '{expected_state}' " |
| "state, but is in '{actual}' state instead.".format( |
| vm_uuid=vm_uuid, expected_state=expected_state, |
| actual=vm.status)) |
| |
| def create_network(self, project_id): |
| net_name = "spt-test-net-{}".format(random.randrange(100, 999)) |
| config = utils.get_configuration() |
| mtu = config.get('custom_mtu') or 'default' |
| net_body = { |
| 'network': { |
| 'name': net_name, |
| 'project_id': project_id |
| } |
| } |
| # in TF2011 we cannot set MTU while creating the net: 400 Bad request |
| # error can happen with 'mtu' field, so we update mtu of ports later |
| if (mtu != 'default') and (not self.is_cloud_tf()): |
| try: |
| net_body['network']['mtu'] = int(mtu) |
| except ValueError as e: |
| raise ValueError("MTU value '{}' is not correct. " |
| "Must be an integer at 'custom_mtu' in " |
| "global_config.yaml.\n{}".format(mtu, e)) |
| net = self.os_clients.network.create_network(net_body)['network'] |
| # WA for TF because the network object does not have 'mtu' field by |
| # default, so this blocked running tests at TF envs with default MTU |
| if 'mtu' not in net: |
| net['mtu'] = None |
| logger.info("Created internal network {} in {} project".format( |
| net_name, project_id)) |
| return net |
| |
| def create_subnet(self, net, project_id, cidr=None): |
| subnet_name = "spt-test-subnet-{}".format(random.randrange(100, 999)) |
| subnet_body = { |
| 'subnet': { |
| "name": subnet_name, |
| 'network_id': net['id'], |
| 'ip_version': 4, |
| 'cidr': cidr if cidr else '10.1.7.0/24', |
| 'project_id': project_id |
| } |
| } |
| subnet = self.os_clients.network.create_subnet(subnet_body)['subnet'] |
| logger.info("Created subnet {} in {} project".format( |
| subnet_name, project_id)) |
| return subnet |
| |
| def create_router(self, ext_net, project_id): |
| name = 'spt-test-router-{}'.format(random.randrange(100, 999)) |
| router_body = { |
| 'router': { |
| 'name': name, |
| 'external_gateway_info': { |
| 'network_id': ext_net['id'] |
| }, |
| 'project_id': project_id |
| } |
| } |
| logger.info("Created a router {} in {} project".format( |
| name, project_id)) |
| router = self.os_clients.network.create_router(router_body)['router'] |
| return router |
| |
| def create_network_resources(self, project="admin", cidr=None): |
| project_id = self.get_project_by_name(project).id |
| self.get_external_network() |
| net = self.create_network(project_id) |
| self.create_subnet(net, project_id, cidr) |
| return net |
| |
| def list_nova_computes(self): |
| nova_services = self.os_clients.compute.hosts.list() |
| computes_list = [h for h in nova_services if h.service == "compute"] |
| return computes_list |
| |
| def create_floating_ip(self, floating_net_id): |
| fip = self.os_clients.network.create_floatingip({"floatingip": { |
| "floating_network_id": floating_net_id}}) |
| return fip['floatingip'] |
| |
| def delete_floating_ip(self, floatingip_id): |
| try: |
| return self.os_clients.network.delete_floatingip(floatingip_id) |
| except neutron_common.exceptions.NotFound as e: |
| msg = "Could not delete a Floating IP, UUID {}. Error: {}" \ |
| "".format(floatingip_id, e) |
| logger.info(msg) |
| |
| def create_project(self): |
| project_name = "spt-test-project-{}".format(random.randrange(100, 999)) |
| project = self.os_clients.auth.projects.create( |
| name=project_name, domain=self.os_clients.domain, |
| description="Mirantis SPT test project") |
| logger.info("Created a project {}, uuid: {}".format( |
| project.name, project.id)) |
| return project |
| |
| def add_roles_to_user_in_project(self, project_id, username='admin', |
| domain='default', roles=None): |
| user_id = [ |
| user.id for user in self.os_clients.auth.users.list() |
| if (user.name == username) and (user.domain_id == domain)][0] |
| if roles is None: |
| roles = ["admin", "member", "creator"] |
| for role in roles: |
| try: |
| role_id = self.os_clients.auth.roles.list(name=role)[0].id |
| self.os_clients.auth.roles.grant( |
| role=role_id, user=user_id, project=project_id) |
| except Exception as e: |
| continue |
| logger.info("Added admin user to {} project".format(project_id)) |
| |
| def is_project_empty(self, project_id): |
| sec_groups = [i for i in self.os_clients.network.list_security_groups( |
| tenant_id=project_id)['security_groups'] if i['name'] != 'default'] |
| servers = self.os_clients.compute.servers.list( |
| search_opts={'project_id': project_id}) |
| nets = self.os_clients.network.list_networks( |
| project_id=project_id)["networks"] |
| subnets = self.os_clients.network.list_subnets( |
| project_id=project_id)["subnets"] |
| ports = self.os_clients.network.list_ports( |
| project_id=project_id)["ports"] |
| routers = self.os_clients.network.list_routers( |
| project_id=project_id)["routers"] |
| resources = [*sec_groups, *servers, *nets, *subnets, *ports, *routers] |
| return not bool(resources) |
| |
| def is_cloud_tf(self): |
| # Detect the TF cloud by assuming it does not have any neutron |
| # agents (404 in response) |
| try: |
| self.os_clients.network.list_agents() |
| except neutron_common.exceptions.NotFound: |
| logger.info("MOS TF cloud is detected.") |
| return True |
| return False |
| |
| def update_network_port_with_custom_mtu(self, vm_uuid, custom_mtu): |
| port_uuid = self.os_clients.network.list_ports( |
| device_id=vm_uuid).get("ports")[0]["id"] |
| body = {"port": {"extra_dhcp_opts": [ |
| {"opt_name": "interface-mtu", "opt_value": str(custom_mtu)}]}} |
| try: |
| self.os_clients.network.update_port(port_uuid, body) |
| except Exception as e: |
| raise Exception("Could not set custom MTU by updating the port. " |
| "See detailed error: {}".format(e)) |
| logger.info("The port {} is updated with custom MTU {}." |
| "".format(port_uuid, custom_mtu)) |
| |
| def get_flavor_id_by_name(self, name): |
| flavors = [flavor for flavor in self.os_clients.compute.flavors.list()] |
| flavor_id = [f.id for f in flavors if f.name == name] |
| if not flavor_id: |
| return None |
| return str(flavor_id[0]) |