tempest/common/isolated_creds.py

# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2013 IBM Corp.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import keystoneclient.v2_0.client

from tempest import clients
from tempest.common.utils.data_utils import rand_name
from tempest import config
from tempest import exceptions
from tempest.openstack.common import log as logging

LOG = logging.getLogger(__name__)


class IsolatedCreds(object):

    def __init__(self, name, tempest_client=True, interface='json',
                 password='pass'):
        self.isolated_creds = {}
        self.name = name
        self.config = config.TempestConfig()
        self.tempest_client = tempest_client
        self.interface = interface
        self.password = password
        self.admin_client = self._get_identity_admin_client()

    def _get_keystone_client(self):
        username = self.config.identity.admin_username
        password = self.config.identity.admin_password
        tenant_name = self.config.identity.admin_tenant_name
        auth_url = self.config.identity.uri
        dscv = self.config.identity.disable_ssl_certificate_validation
        return keystoneclient.v2_0.client.Client(username=username,
                                                 password=password,
                                                 tenant_name=tenant_name,
                                                 auth_url=auth_url,
                                                 insecure=dscv)

    def _get_identity_admin_client(self):
        """
        Returns an instance of the Identity Admin API client
        """
        if self.tempest_client:
            os = clients.AdminManager(interface=self.interface)
            admin_client = os.identity_client
        else:
            admin_client = self._get_keystone_client()
        return admin_client

    def _create_tenant(self, name, description):
        if self.tempest_client:
            resp, tenant = self.admin_client.create_tenant(
                name=name, description=description)
        else:
            tenant = self.admin_client.tenants.create(name,
                                                      description=description)
        return tenant

    def _get_tenant_by_name(self, name):
        if self.tempest_client:
            resp, tenant = self.admin_client.get_tenant_by_name(name)
        else:
            tenants = self.admin_client.tenants.list()
            for ten in tenants:
                if ten['name'] == name:
                    tenant = ten
            raise exceptions.NotFound('No such tenant')
        return tenant

    def _create_user(self, username, password, tenant, email):
        if self.tempest_client:
            resp, user = self.admin_client.create_user(username, password,
                                                       tenant['id'], email)
        else:
            user = self.admin_client.users.create(username, password, email,
                                                  tenant_id=tenant.id)
        return user

    def _get_user(self, tenant, username):
        if self.tempest_client:
            resp, user = self.admin_client.get_user_by_username(tenant['id'],
                                                                username)
        else:
            user = self.admin_client.users.get(username)
        return user

    def _list_roles(self):
        if self.tempest_client:
            resp, roles = self.admin_client.list_roles()
        else:
            roles = self.admin_client.roles.list()
        return roles

    def _assign_user_role(self, tenant, user, role):
        if self.tempest_client:
            self.admin_client.assign_user_role(tenant, user, role)
        else:
            self.admin_client.roles.add_user_role(user, role, tenant=tenant)

    def _delete_user(self, user):
        if self.tempest_client:
            self.admin_client.delete_user(user)
        else:
            self.admin_client.users.delete(user)

    def _delete_tenant(self, tenant):
        if self.tempest_client:
            self.admin_client.delete_tenant(tenant)
        else:
            self.admin_client.tenants.delete(tenant)

    def _create_creds(self, suffix=None, admin=False):
        rand_name_root = rand_name(self.name)
        if suffix:
            rand_name_root += suffix
        tenant_name = rand_name_root + "-tenant"
        tenant_desc = tenant_name + "-desc"
        rand_name_root = rand_name(self.name)
        tenant = self._create_tenant(name=tenant_name,
                                     description=tenant_desc)
        if suffix:
            rand_name_root += suffix
        username = rand_name_root + "-user"
        email = rand_name_root + "@example.com"
        user = self._create_user(username, self.password,
                                 tenant, email)
        if admin:
            role = None
            try:
                roles = self._list_roles()
                if self.tempest_client:
                    role = next(r for r in roles if r['name'] == 'admin')
                else:
                    role = next(r for r in roles if r.name == 'admin')
            except StopIteration:
                msg = "No admin role found"
                raise exceptions.NotFound(msg)
            if self.tempest_client:
                self._assign_user_role(tenant['id'], user['id'], role['id'])
            else:
                self._assign_user_role(tenant.id, user.id, role.id)
        return user, tenant

    def _get_cred_names(self, user, tenant):
        if self.tempest_client:
            username = user.get('name')
            tenant_name = tenant.get('name')
        else:
            username = user.name
            tenant_name = tenant.name
        return username, tenant_name

    def get_primary_tenant(self):
        return self.isolated_creds.get('primary')[1]

    def get_primary_user(self):
        return self.isolated_creds.get('primary')[0]

    def get_alt_tenant(self):
        return self.isolated_creds.get('alt')[1]

    def get_alt_user(self):
        return self.isolated_creds.get('alt')[0]

    def get_admin_tenant(self):
        return self.isolated_creds.get('admin')[1]

    def get_admin_user(self):
        return self.isolated_creds.get('admin')[0]

    def get_primary_creds(self):
        if self.isolated_creds.get('primary'):
            user, tenant = self.isolated_creds['primary']
            username, tenant_name = self._get_cred_names(user, tenant)
        else:
            user, tenant = self._create_creds()
            username, tenant_name = self._get_cred_names(user, tenant)
            self.isolated_creds['primary'] = (user, tenant)
            LOG.info("Aquired isolated creds:\n user: %s, tenant: %s"
                     % (username, tenant_name))
        return username, tenant_name, self.password

    def get_admin_creds(self):
        if self.isolated_creds.get('admin'):
            user, tenant = self.isolated_creds['admin']
            username, tenant_name = self._get_cred_names(user, tenant)
        else:
            user, tenant = self._create_creds(admin=True)
            username, tenant_name = self._get_cred_names(user, tenant)
            self.isolated_creds['admin'] = (user, tenant)
            LOG.info("Aquired admin isolated creds:\n user: %s, tenant: %s"
                     % (username, tenant_name))
            return username, tenant_name, self.password

    def get_alt_creds(self):
        if self.isolated_creds.get('alt'):
            user, tenant = self.isolated_creds['alt']
            username, tenant_name = self._get_cred_names(user, tenant)
        else:
            user, tenant = self._create_creds()
            username, tenant_name = self._get_cred_names(user, tenant)
            self.isolated_creds['alt'] = (user, tenant)
            LOG.info("Aquired alt isolated creds:\n user: %s, tenant: %s"
                     % (username, tenant_name))
        return username, tenant_name, self.password

    def clear_isolated_creds(self):
        if not self.isolated_creds:
            return
        for cred in self.isolated_creds:
            user, tenant = self.isolated_creds.get(cred)
            try:
                if self.tempest_client:
                    self._delete_user(user['id'])
                else:
                    self._delete_user(user.id)
            except exceptions.NotFound:
                if self.tempest_client:
                    name = user['name']
                else:
                    name = user.name
                LOG.warn("user with name: %s not found for delete" % name)
                pass
            try:
                if self.tempest_client:
                    self._delete_tenant(tenant['id'])
                else:
                    self._delete_tenant(tenant.id)
            except exceptions.NotFound:
                if self.tempest_client:
                    name = tenant['name']
                else:
                    name = tenant.name
                LOG.warn("tenant with name: %s not found for delete" % name)
                pass