test_set/cvp-spt/utils/os_client.py

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 novaclient import client as novaclient

import os
import random
import time


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 os.environ.keys():
        INTERFACE = os.environ["OS_ENDPOINT_TYPE"]

    def __init__(self, username=None, password=None,
                 tenant_name=None, auth_url=None, endpoint_type="internalURL",
                 cert=False, domain="Default", **kwargs):
        self.traceback = ""

        self.client_attr_names = [
            "auth",
            "compute",
            "network",
            "volume",
            "image",
        ]
        self.username = username
        self.password = password
        self.tenant_name = tenant_name
        self.project_name = tenant_name
        self.auth_url = auth_url
        self.endpoint_type = endpoint_type
        self.cert = cert
        self.domain = domain
        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,
                          tenant_name=None, auth_url=None, cert=None,
                          domain='Default'):
        if None in (username, password, tenant_name):
            print(username, password, tenant_name)
            msg = ("Missing required credentials for identity client. "
                   "username: {username}, password: {password}, "
                   "tenant_name: {tenant_name}").format(
                       username=username,
                       password=password,
                       tenant_name=tenant_name, )
            raise msg

        if cert and "https" not in auth_url:
            auth_url = auth_url.replace("http", "https")

        if cls.KEYSTONECLIENT_VERSION == (2, 0):
            # auth_url = "{}{}".format(auth_url, "v2.0/")
            auth = keystone_identity.v2.Password(
                username=username,
                password=password,
                auth_url=auth_url,
                tenant_name=tenant_name)
        else:
            auth_url = "{}{}".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=tenant_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,
                        tenant_name=None, auth_url=None, cert=None,
                        domain='Default', **kwargs):
        session = cls._get_auth_session(
            username=username,
            password=password,
            tenant_name=tenant_name,
            auth_url=auth_url,
            cert=cert,
            domain=domain)
        keystone = keystone_client.Client(version=cls.KEYSTONECLIENT_VERSION,
                                          session=session, **kwargs)
        keystone.management_url = auth_url
        return keystone

    @classmethod
    def get_compute_client(cls, username=None, password=None,
                           tenant_name=None, auth_url=None, cert=None,
                           domain='Default', **kwargs):
        session = cls._get_auth_session(
            username=username, password=password, tenant_name=tenant_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,
                           tenant_name=None, auth_url=None, cert=None,
                           domain='Default', **kwargs):
        session = cls._get_auth_session(
            username=username, password=password, tenant_name=tenant_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,
                          tenant_name=None, auth_url=None, cert=None,
                          domain='Default', **kwargs):
        session = cls._get_auth_session(
            username=username, password=password, tenant_name=tenant_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,
                         tenant_name=None, auth_url=None, cert=None,
                         domain='Default', **kwargs):
        session = cls._get_auth_session(
            username=username, password=password, tenant_name=tenant_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.tenant_name, self.auth_url,
                self.cert, self.domain, endpoint_type=self.endpoint_type
            )
        return self._auth

    @property
    def compute(self):
        if self._compute is None:
            self._compute = self.get_compute_client(
                self.username, self.password, self.tenant_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.tenant_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.tenant_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.tenant_name, self.auth_url,
                self.cert, self.domain
            )
        return self._image


class OSCliActions(object):
    def __init__(self, os_clients):
        self.os_clients = os_clients

    def get_admin_tenant(self):
        # TODO Keystone v3 doesnt have tenants attribute
        return self.os_clients.auth.projects.find(name="admin")

    # TODO: refactor
    def get_cirros_image(self):
        images_list = list(self.os_clients.image.images.list(name='TestVM'))
        if images_list:
            image = images_list[0]
        else:
            image = self.os_clients.image.images.create(
                name="TestVM",
                disk_format='qcow2',
                container_format='bare')
            with file_cache.get_file(settings.CIRROS_QCOW2_URL) as f:
                self.os_clients.image.images.upload(image.id, f)
        return image

    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 get_external_network(self):
        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"])
        ]
        if networks:
            ext_net = networks[0]
        else:
            ext_net = self.create_fake_external_network()
        return ext_net

    def create_flavor(self, name, ram=256, vcpus=1, disk=2):
        return self.os_clients.compute.flavors.create(name, ram, vcpus, disk)

    def create_sec_group(self, rulesets=None):
        if rulesets is None:
            rulesets = [
                {
                    # ssh
                    'ip_protocol': 'tcp',
                    'from_port': 22,
                    'to_port': 22,
                    'cidr': '0.0.0.0/0',
                },
                {
                    # iperf
                    'ip_protocol': 'tcp',
                    'from_port':5001,
                    'to_port': 5001,
                    'cidr': '0.0.0.0/0',
                },
                {
                    # ping
                    'ip_protocol': 'icmp',
                    'from_port': -1,
                    'to_port': -1,
                    'cidr': '0.0.0.0/0',
                }
            ]
        sg_name = "spt-test-secgroup-{}".format(random.randrange(100, 999))
        sg_desc = sg_name + " SPT"
        secgroup = self.os_clients.compute.security_groups.create(
            sg_name, sg_desc)
        for ruleset in rulesets:
            self.os_clients.compute.security_group_rules.create(
                secgroup.id, **ruleset)
        return secgroup


    def wait(predicate, interval=5, timeout=60, timeout_msg="Waiting timed out"):
        start_time = time.time()
        if not timeout:
            return predicate()
        while not predicate():
            if start_time + timeout < time.time():
                raise exceptions.TimeoutError(timeout_msg)

            seconds_to_sleep = max(
                0,
                min(interval, start_time + timeout - time.time()))
            time.sleep(seconds_to_sleep)

        return timeout + start_time - time.time()

    def create_basic_server(self, image=None, flavor=None, net=None,
                            availability_zone=None, sec_groups=(),
                            keypair=None,
                            wait_timeout=3 * 60):
        os_conn = self.os_clients
        image = image or self.get_cirros_image()
        flavor = flavor or self.get_micro_flavor()
        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)
        # TODO
        #if wait_timeout:
        #    self.wait(
        #        lambda: os_conn.compute.servers.get(server).status == "ACTIVE",
        #        timeout=wait_timeout,
        #        timeout_msg=(
        #            "Create server {!r} failed by timeout. "
        #            "Please, take a look at OpenStack logs".format(server.id)))
        return server

    def create_network(self, tenant_id):
        net_name = "spt-test-net-{}".format(random.randrange(100, 999))
        net_body = {
            'network': {
                'name': net_name,
                'tenant_id': tenant_id
            }
        }
        net = self.os_clients.network.create_network(net_body)['network']
        return net
        #yield net
        #self.os_clients.network.delete_network(net['id'])

    def create_subnet(self, net, tenant_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',
                'tenant_id': tenant_id
            }
        }
        subnet = self.os_clients.network.create_subnet(subnet_body)['subnet']
        return subnet
        #yield subnet
        #self.os_clients.network.delete_subnet(subnet['id'])

    def create_router(self, ext_net, tenant_id):
        name = 'spt-test-router-{}'.format(random.randrange(100, 999))
        router_body = {
            'router': {
                'name': name,
                'external_gateway_info': {
                    'network_id': ext_net['id']
                },
                'tenant_id': tenant_id
            }
        }
        router = self.os_clients.network.create_router(router_body)['router']
        return router
        #yield router
        #self.os_clients.network.delete_router(router['id'])

    def create_network_resources(self):
        tenant_id = self.get_admin_tenant().id
        ext_net = self.get_external_network()
        net = self.create_network(tenant_id)
        subnet = self.create_subnet(net, tenant_id)
        #router = self.create_router(ext_net, tenant_id)
        #self.os_clients.network.add_interface_router(
        #    router['id'], {'subnet_id': subnet['id']})

        private_net_id = net['id']
        # floating_ip_pool = ext_net['id']

        return net
        #yield private_net_id, floating_ip_pool
        #yield private_net_id
        #
        #self.os_clients.network.remove_interface_router(
        #     router['id'], {'subnet_id': subnet['id']})
        #self.os_clients.network.remove_gateway_router(router['id'])