openstack/rally-openstack
Code Issues Proposed changes

Merge "[Services] Introduce Identity Service"

Browse Source
This commit is contained in:
Jenkins 2016-11-23 08:42:18 +00:00 committed by Gerrit Code Review
commit 3f83a02bf0
20 changed files with 1747 additions and 169 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
rally/plugins/openstack/cleanup/resources.py
View File

@ -27,8 +27,8 @@ from rally.plugins.openstack.scenarios.cinder import utils as cinder_utils
from rally.plugins.openstack.scenarios.fuel import utils as futils
from rally.plugins.openstack.scenarios.keystone import utils as kutils
from rally.plugins.openstack.scenarios.nova import utils as nova_utils
from rally.plugins.openstack.services.identity import identity
from rally.plugins.openstack.wrappers import glance as glance_wrapper
from rally.plugins.openstack.wrappers import keystone as keystone_wrapper
from rally.task import utils as task_utils
CONF = cfg.CONF
@ -890,7 +890,7 @@ _keystone_order = get_order(9000)
class KeystoneMixin(SynchronizedDeletion):
def _manager(self):
return keystone_wrapper.wrap(getattr(self.admin, self._service)())
return identity.Identity(self.admin)
def delete(self):
delete_method = getattr(self._manager(), "delete_%s" % self._resource)

16
rally/plugins/openstack/context/keystone/roles.py
View File

@ -21,7 +21,7 @@ from rally.common import logging
from rally import consts
from rally import exceptions
from rally import osclients
from rally.plugins.openstack.wrappers import keystone
from rally.plugins.openstack.services.identity import identity
from rally.task import context
LOG = logging.getLogger(__name__)
@ -63,8 +63,8 @@ class RoleGenerator(context.Context):
:param context_role: name of existing role.
"""
client = keystone.wrap(osclients.Clients(self.credential).keystone())
default_roles = client.list_roles()
keystone = identity.Identity(osclients.Clients(self.credential))
default_roles = keystone.list_roles()
for def_role in default_roles:
if str(def_role.name) == context_role:
return def_role
@ -76,8 +76,10 @@ class RoleGenerator(context.Context):
role_id, user_id, project_id = args
if "client" not in cache:
clients = osclients.Clients(self.credential)
cache["client"] = keystone.wrap(clients.keystone())
getattr(cache["client"], func_name)(role_id, user_id, project_id)
cache["client"] = identity.Identity(clients)
getattr(cache["client"], func_name)(role_id=role_id,
user_id=user_id,
project_id=project_id)
return consume
@logging.log_task_wrapper(LOG.info, _("Enter context: `roles`"))
@ -109,9 +111,9 @@ class RoleGenerator(context.Context):
def publish(queue):
for role_id in self.context["roles"]:
LOG.debug("Removing role %s from all users" % (role_id))
LOG.debug("Removing role %s from all users" % role_id)
for user in self.context["users"]:
args = (role_id, user["id"], user["tenant_id"])
queue.append(args)
broker.run(publish, self._get_consumer("remove_role"), threads)
broker.run(publish, self._get_consumer("revoke_role"), threads)

27
rally/plugins/openstack/context/keystone/users.py
View File

@ -27,7 +27,7 @@ from rally.common import utils as rutils
from rally import consts
from rally import exceptions
from rally import osclients
from rally.plugins.openstack.wrappers import keystone
from rally.plugins.openstack.services.identity import identity
from rally.plugins.openstack.wrappers import network
from rally.task import context
from rally.task import utils
@ -214,9 +214,9 @@ class UserGenerator(UserContextMixin, context.Context):
domain, task_id, i = args
if "client" not in cache:
clients = osclients.Clients(self.credential)
cache["client"] = keystone.wrap(clients.keystone())
tenant = cache["client"].create_project(
self.generate_random_name(), domain)
cache["client"] = identity.Identity(
clients, name_generator=self.generate_random_name)
tenant = cache["client"].create_project(domain_name=domain)
tenant_dict = {"id": tenant.id, "name": tenant.name, "users": []}
tenants.append(tenant_dict)
@ -249,17 +249,18 @@ class UserGenerator(UserContextMixin, context.Context):
username, password, project_dom, user_dom, tenant_id = args
if "client" not in cache:
clients = osclients.Clients(self.credential)
cache["client"] = keystone.wrap(clients.keystone())
cache["client"] = identity.Identity(
clients, name_generator=self.generate_random_name)
client = cache["client"]
user = client.create_user(
username, password,
"%s@email.me" % username,
tenant_id, user_dom,
default_role=default_role)
user = client.create_user(username, password=password,
email="%s@email.me" % username,
project_id=tenant_id,
domain_name=user_dom,
default_role=default_role)
user_credential = objects.Credential(
client.auth_url, user.name, password,
self.credential.auth_url, user.name, password,
self.context["tenants"][tenant_id]["name"],
consts.EndpointPermission.USER, client.region_name,
consts.EndpointPermission.USER, self.credential.region_name,
project_domain_name=project_dom, user_domain_name=user_dom,
endpoint_type=self.credential.endpoint_type,
https_insecure=self.credential.insecure,
@ -276,7 +277,7 @@ class UserGenerator(UserContextMixin, context.Context):
def consume(cache, resource_id):
if "client" not in cache:
clients = osclients.Clients(self.credential)
cache["client"] = keystone.wrap(clients.keystone())
cache["client"] = identity.Identity(clients)
getattr(cache["client"], func_name)(resource_id)
return consume

32
rally/plugins/openstack/scenarios/keystone/basic.py
View File

@ -13,15 +13,16 @@
# License for the specific language governing permissions and limitations
# under the License.
"""
Benchmark scenarios for Keystone.
"""
from rally.common import logging
from rally.plugins.openstack import scenario
from rally.plugins.openstack.scenarios.keystone import utils as kutils
from rally.task import validation
"""Benchmark scenarios for Keystone."""
@validation.required_openstack(admin=True)
@scenario.configure(context={"admin_cleanup": ["keystone"]},
name="KeystoneBasic.create_user")
@ -176,9 +177,11 @@ class AddAndRemoveUserRole(kutils.KeystoneScenario):
"""Create a user role add to a user and disassociate."""
tenant_id = self.context["tenant"]["id"]
user_id = self.context["user"]["id"]
role = self._role_create()
self._role_add(user_id, role, tenant_id)
self._role_remove(user_id, role, tenant_id)
role = self.admin_keystone.create_role()
self.admin_keystone.add_role(role_id=role.id, user_id=user_id,
project_id=tenant_id)
self.admin_keystone.revoke_role(role.id, user_id=user_id,
project_id=tenant_id)
@validation.required_openstack(admin=True)
@ -188,8 +191,8 @@ class CreateAndDeleteRole(kutils.KeystoneScenario):
def run(self):
"""Create a user role and delete it."""
role = self._role_create()
self._role_delete(role.id)
role = self.admin_keystone.create_role()
self.admin_keystone.delete_role(role.id)
@validation.required_openstack(admin=True, users=True)
@ -201,9 +204,10 @@ class CreateAddAndListUserRoles(kutils.KeystoneScenario):
"""Create user role, add it and list user roles for given user."""
tenant_id = self.context["tenant"]["id"]
user_id = self.context["user"]["id"]
role = self._role_create()
self._role_add(user_id, role, tenant_id)
self._list_roles_for_user(user_id, tenant_id)
role = self.admin_keystone.create_role()
self.admin_keystone.add_role(user_id=user_id, role_id=role.id,
project_id=tenant_id)
self.admin_keystone.list_roles(user_id=user_id, project_id=tenant_id)
@validation.required_openstack(admin=True)
@ -228,7 +232,7 @@ class GetEntities(kutils.KeystoneScenario):
"""
tenant = self._tenant_create()
user = self._user_create()
role = self._role_create()
role = self.admin_keystone.create_role()
self._get_tenant(tenant.id)
self._get_user(user.id)
self._get_role(role.id)
@ -344,5 +348,5 @@ class CreateAndGetRole(kutils.KeystoneScenario):
:param kwargs: Optional additional arguments for roles creation
"""
role = self._role_create(**kwargs)
self._get_role(role.id)
role = self.admin_keystone.create_role(**kwargs)
self.admin_keystone.get_role(role.id)

12
rally/plugins/openstack/scenarios/keystone/utils.py
View File

@ -16,6 +16,7 @@
import uuid
from rally.plugins.openstack import scenario
from rally.plugins.openstack.services.identity import identity
from rally.plugins.openstack.wrappers import keystone as keystone_wrapper
from rally.task import atomic
@ -23,6 +24,17 @@ from rally.task import atomic
class KeystoneScenario(scenario.OpenStackScenario):
"""Base class for Keystone scenarios with basic atomic actions."""
def __init__(self, context=None, admin_clients=None, clients=None):
super(KeystoneScenario, self).__init__(context, admin_clients, clients)
if hasattr(self, "_admin_clients"):
self.admin_keystone = identity.Identity(
self._admin_clients, name_generator=self.generate_random_name,
atomic_inst=self.atomic_actions())
if hasattr(self, "_clients"):
self.keystone = identity.Identity(
self._clients, name_generator=self.generate_random_name,
atomic_inst=self.atomic_actions())
@atomic.action_timer("keystone.create_user")
def _user_create(self, email=None, **kwargs):
"""Creates keystone user with random name.

0
rally/plugins/openstack/services/identity/__init__.py Normal file
View File

137
rally/plugins/openstack/services/identity/identity.py Normal file
View File

@ -0,0 +1,137 @@
# All Rights Reserved.
#
# 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 collections
from rally.plugins.openstack import service
Project = collections.namedtuple("Project", ["id", "name", "domain_id"])
User = collections.namedtuple("User",
["id", "name", "project_id", "domain_id"])
Service = collections.namedtuple("Service", ["id", "name"])
Role = collections.namedtuple("Role", ["id", "name"])
class Identity(service.UnifiedOpenStackService):
@classmethod
def is_applicable(cls, clients):
cloud_version = clients.keystone().version.split(".")[0][1:]
return cloud_version == cls._meta_get("impl")._meta_get("version")
def create_project(self, project_name=None, domain_name="Default"):
"""Creates new project/tenant and return project object.
:param project_name: Name of project to be created.
:param domain_name: Name or id of domain where to create project, for
those service implementations that don't support
domains you should use None or 'Default' value.
"""
return self._impl.create_project(project_name,
domain_name=domain_name)
def delete_project(self, project_id):
"""Deletes project."""
return self._impl.delete_project(project_id)
def list_projects(self):
"""List all projects."""
return self._impl.list_projects()
def create_user(self, username, password, email=None, project_id=None,
domain_name="Default", default_role="member"):
"""Create user.
:param username: name of user
:param password: user password
:param email: user's email
:param project_id: user's default project
:param domain_name: Name or id of domain where to create user, for
those service implementations that don't support
domains you should use None or 'Default' value.
:param default_role: Name of role, for implementations that don't
support domains this argument must be None or
'member'.
"""
return self._impl.create_user(username=username,
password=password,
email=email,
project_id=project_id,
domain_name=domain_name,
default_role=default_role)
def delete_user(self, user_id):
"""Deletes user by its id."""
self._impl.delete_user(user_id)
def list_users(self):
"""List all users."""
return self._impl.list_users()
def delete_service(self, service_id):
"""Deletes service."""
self._impl.delete_service(service_id)
def list_services(self):
"""List all services."""
return self._impl.list_services()
def create_role(self, name=None, domain_name="Default"):
"""Create role with specific name
:param name: role name
:param domain_name: Name or id of domain where to create role, for
those service implementations that don't support
domains you should use None or 'Default' value.
"""
return self._impl.create_role(name=name, domain_name=domain_name)
def add_role(self, role_id, user_id, project_id):
"""Add role to user."""
return self._impl.add_role(role_id=role_id, user_id=user_id,
project_id=project_id)
def delete_role(self, role_id):
"""Deletes role."""
self._impl.delete_role(role_id)
def revoke_role(self, role_id, user_id, project_id):
"""Revokes a role from a user."""
return self._impl.revoke_role(role_id=role_id, user_id=user_id,
project_id=project_id)
def list_roles(self, user_id=None, project_id=None, domain_name=None):
"""List all roles.
:param user_id: filter in role grants for the specified user on a
resource. Domain or project must be specified.
:param project_id: filter in role grants on the specified project.
user_id should be specified
:param domain_name: filter in role grants on the specified domain.
user_id should be specified
"""
return self._impl.list_roles(user_id=user_id, project_id=project_id,
domain_name=domain_name)
def get_role(self, role_id):
"""Get role."""
return self._impl.get_role(role_id)
@staticmethod
def _unify_service(service):
return Service(id=service.id, name=service.name)
@staticmethod
def _unify_role(role):
return Role(id=role.id, name=role.name)

208
rally/plugins/openstack/services/identity/keystone_v2.py Normal file
View File

@ -0,0 +1,208 @@
# All Rights Reserved.
#
# 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.
from rally.plugins.openstack import service
from rally.plugins.openstack.services.identity import identity
from rally.task import atomic
@service.service("keystone", service_type="identity", version="2")
class KeystoneV2Service(service.Service):
@atomic.action_timer("keystone_v2.create_tenant")
def create_tenant(self, tenant_name=None):
tenant_name = tenant_name or self.generate_random_name()
return self._clients.keystone("2").tenants.create(tenant_name)
@atomic.action_timer("keystone_v2.delete_tenant")
def delete_tenant(self, tenant_id):
return self._clients.keystone("2").tenants.delete(tenant_id)
@atomic.action_timer("keystone_v2.list_tenants")
def list_tenants(self):
return self._clients.keystone("2").tenants.list()
@atomic.action_timer("keystone_v2.create_user")
def create_user(self, username, password, email=None, tenant_id=None):
return self._clients.keystone("2").users.create(name=username,
password=password,
email=email,
tenant_id=tenant_id)
@atomic.action_timer("keystone_v2.list_users")
def list_users(self):
return self._clients.keystone("2").users.list()
@atomic.action_timer("keystone_v2.delete_user")
def delete_user(self, user_id):
"""Deletes user by its id."""
self._clients.keystone("2").users.delete(user_id)
@atomic.action_timer("keystone_v2.delete_service")
def delete_service(self, service_id):
"""Deletes service."""
self._clients.keystone("2").services.delete(service_id)
@atomic.action_timer("keystone.list_services")
def list_services(self):
"""List all services."""
return self._clients.keystone("2").services.list()
@atomic.action_timer("keystone_v2.create_role")
def create_role(self, name=None):
name = name or self.generate_random_name()
return self._clients.keystone("2").roles.create(name)
@atomic.action_timer("keystone_v2.add_role")
def add_role(self, role_id, user_id, tenant_id):
return self._clients.keystone("2").roles.add_user_role(
user=user_id, role=role_id, tenant=tenant_id)
@atomic.action_timer("keystone.delete_role")
def delete_role(self, role_id):
"""Deletes role."""
self._clients.keystone("2").roles.delete(role_id)
@atomic.action_timer("keystone_v2.list_roles")
def list_roles(self):
"""List all roles."""
return self._clients.keystone("2").roles.list()
@atomic.action_timer("keystone_v2.list_roles_for_user")
def list_roles_for_user(self, user_id, tenant_id=None):
return self._clients.keystone("2").roles.roles_for_user(
user_id, tenant_id)
@atomic.action_timer("keystone_v2.revoke_role")
def revoke_role(self, role_id, user_id, tenant_id):
self._clients.keystone("2").roles.remove_user_role(user=user_id,
role=role_id,
tenant=tenant_id)
@atomic.action_timer("keystone_v2.get_role")
def get_role(self, role_id):
"""Get role."""
return self._clients.keystone("2").roles.get(role_id)
@service.compat_layer(KeystoneV2Service)
class UnifiedKeystoneV2Service(identity.Identity):
"""Compatibility layer for Keystone V2."""
@staticmethod
def _check_domain(domain_name):
if domain_name.lower() != "default":
raise NotImplementedError("Domain functionality not implemented "
"in Keystone v2")
@staticmethod
def _unify_tenant(tenant):
return identity.Project(id=tenant.id, name=tenant.name,
domain_id="default")
@staticmethod
def _unify_user(user):
return identity.User(id=user.id, name=user.name,
project_id=getattr(user, "tenantId", None),
domain_id="default")
def create_project(self, project_name=None, domain_name="Default"):
"""Creates new project/tenant and return project object.
:param project_name: Name of project to be created.
:param domain_name: Restricted for Keystone V2. Should not be set or
"Default" is expected.
"""
self._check_domain(domain_name)
tenant = self._impl.create_tenant(project_name)
return self._unify_tenant(tenant)
def delete_project(self, project_id):
"""Deletes project."""
return self._impl.delete_tenant(project_id)
def list_projects(self):
"""List all projects."""
return [self._unify_tenant(t) for t in self._impl.list_tenants()]
def create_user(self, username, password, email=None, project_id=None,
domain_name="Default", default_role="member"):
"""Create user.
:param username: name of user
:param password: user password
:param email: user's email
:param project_id: user's default project
:param domain_name: Restricted for Keystone V2. Should not be set or
"Default" is expected.
:param default_role: Restricted for Keystone V2. Should not be set or
"member" is expected.
"""
self._check_domain(domain_name)
user = self._impl.create_user(username=username,
password=password,
email=email,
tenant_id=project_id)
return self._unify_user(user)
def delete_user(self, user_id):
"""Deletes user by its id."""
return self._impl.delete_user(user_id)
def list_users(self):
"""List all users."""
return [self._unify_user(u) for u in self._impl.list_users()]
def delete_service(self, service_id):
"""Deletes service."""
return self._impl.delete_service(service_id)
def list_services(self):
"""List all services."""
return [self._unify_service(s) for s in self._impl.list_services()]
def create_role(self, name=None, domain_name="Default"):
"""Add role to user."""
self._check_domain(domain_name)
return self._unify_role(self._impl.create_role(name))
def add_role(self, role_id, user_id, project_id):
"""Add role to user."""
return self._unify_role(self._impl.add_role(
role_id=role_id, user_id=user_id, tenant_id=project_id))
def delete_role(self, role_id):
"""Deletes role."""
return self._impl.delete_role(role_id)
def revoke_role(self, role_id, user_id, project_id):
"""Revokes a role from a user."""
return self._impl.revoke_role(role_id=role_id, user_id=user_id,
tenant_id=project_id)
def list_roles(self, user_id=None, project_id=None, domain_name=None):
"""List all roles."""
if domain_name:
raise NotImplementedError("Domain functionality not implemented "
"in Keystone v2")
if user_id:
roles = self._impl.list_roles_for_user(user_id,
tenant_id=project_id)
else:
roles = self._impl.list_roles()
return [self._unify_role(role) for role in roles]
def get_role(self, role_id):
"""Get role."""
return self._unify_role(self._impl.get_role(role_id))

242
rally/plugins/openstack/services/identity/keystone_v3.py Normal file
View File

@ -0,0 +1,242 @@
# All Rights Reserved.
#
# 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.
from rally.common import logging
from rally import exceptions
from rally.plugins.openstack import service
from rally.plugins.openstack.services.identity import identity
from rally.task import atomic
LOG = logging.getLogger(__name__)
@service.service("keystone", service_type="identity", version="3")
class KeystoneV3Service(service.Service):
def _get_domain_id(self, domain_name_or_id):
from keystoneclient import exceptions as kc_exceptions
try:
# First try to find domain by ID
return self._clients.keystone("3").domains.get(
domain_name_or_id).id
except kc_exceptions.NotFound:
# Domain not found by ID, try to find it by name
domains = self._clients.keystone("3").domains.list(
name=domain_name_or_id)
if domains:
return domains[0].id
# Domain not found by name
raise exceptions.GetResourceNotFound(
resource="KeystoneDomain(%s)" % domain_name_or_id)
@atomic.action_timer("keystone_v3.create_project")
def create_project(self, project_name=None, domain_name="Default"):
project_name = project_name or self.generate_random_name()
domain_id = self._get_domain_id(domain_name)
return self._clients.keystone("3").projects.create(name=project_name,
domain=domain_id)
@atomic.action_timer("keystone_v3.delete_project")
def delete_project(self, project_id):
self._clients.keystone("3").projects.delete(project_id)
@atomic.action_timer("keystone_v3.list_projects")
def list_projects(self):
return self._clients.keystone("3").projects.list()
@atomic.action_timer("keystone_v3.create_user")
def create_user(self, username, password, email=None, project_id=None,
domain_name="Default", default_role="member"):
"""Create user.
:param username: name of user
:param password: user password
:param email: user'ss email
:param project_id: user's default project
:param domain_name: Name or id of domain where to create project.
:param default_role: user's default role
"""
domain_id = self._get_domain_id(domain_name)
user = self._clients.keystone("3").users.create(
name=username, password=password, default_project=project_id,
email=email, domain=domain_id)
for role in self.list_roles():
if default_role in role.name.lower():
self.add_role(role_id=role.id,
user_id=user.id,
project_id=project_id)
break
else:
LOG.warning("Unable to set %s role to created user." %
default_role)
return user
@atomic.action_timer("keystone_v3.list_users")
def list_users(self):
return self._clients.keystone("3").users.list()
@atomic.action_timer("keystone_v3.delete_user")
def delete_user(self, user_id):
"""Deletes user by its id."""
self._clients.keystone("3").users.delete(user_id)
@atomic.action_timer("keystone_v3.delete_service")
def delete_service(self, service_id):
"""Deletes service."""
self._clients.keystone("3").services.delete(service_id)
@atomic.action_timer("keystone_v3.list_services")
def list_services(self):
"""List all services."""
return self._clients.keystone("3").services.list()
@atomic.action_timer("keystone_v3.create_role")
def create_role(self, name=None, domain_name="Default"):
domain_id = self._get_domain_id(domain_name)
name = name or self.generate_random_name()
return self._clients.keystone("3").roles.create(name, domain=domain_id)
@atomic.action_timer("keystone_v3.add_role")
def add_role(self, role_id, user_id, project_id):
return self._clients.keystone("3").roles.grant(role=role_id,
user=user_id,
project=project_id)
@atomic.action_timer("keystone_v3.delete_role")
def delete_role(self, role_id):
"""Deletes role."""
self._clients.keystone("3").roles.delete(role_id)
@atomic.action_timer("keystone_v3.list_roles")
def list_roles(self, user_id=None, project_id=None, domain_name=None):
"""List all roles."""
domain_id = None
if domain_name:
domain_id = self._get_domain_id(domain_name)
return self._clients.keystone("3").roles.list(user=user_id,
project=project_id,
domain=domain_id)
@atomic.action_timer("keystone_v3.revoke_role")
def revoke_role(self, role_id, user_id, project_id):
self._clients.keystone("3").roles.revoke(role=role_id,
user=user_id,
project=project_id)
@atomic.action_timer("keystone_v3.get_role")
def get_role(self, role_id):
"""Get role."""
return self._clients.keystone("3").roles.get(role_id)
@atomic.action_timer("keystone_v3.create_domain")
def create_domain(self, name, description=None, enabled=True):
return self._clients.keystone("3").domains.create(
name, description=description, enabled=enabled)
@service.compat_layer(KeystoneV3Service)
class UnifiedKeystoneV3Service(identity.Identity):
@staticmethod
def _unify_project(project):
return identity.Project(id=project.id, name=project.name,
domain_id=project.domain_id)
@staticmethod
def _unify_user(user):
# When user has default_project_id that is None user.default_project_id
# will raise AttributeError
project_id = getattr(user, "default_project_id", None)
return identity.User(id=user.id, name=user.name, project_id=project_id,
domain_id=user.domain_id)
def create_project(self, project_name=None, domain_name="Default"):
"""Creates new project/tenant and return project object.
:param project_name: Name of project to be created.
:param domain_name: Name or id of domain where to create project,
"""
project = self._impl.create_project(project_name)
return self._unify_project(project)
def delete_project(self, project_id):
"""Deletes project."""
return self._impl.delete_project(project_id)
def list_projects(self):
"""List all projects."""
return [self._unify_project(p) for p in self._impl.list_projects()]
def create_user(self, username, password, email=None, project_id=None,
domain_name="Default", default_role="member"):
"""Create user.
:param username: name of user
:param password: user password
:param email: user's email
:param project_id: user's default project
:param domain_name: Name or id of domain where to create project,
:param default_role: Name of default user's role
"""
return self._unify_user(self._impl.create_user(
username=username, password=password, email=email,
project_id=project_id, domain_name=domain_name,
default_role=default_role))
def delete_user(self, user_id):
"""Deletes user by its id."""
return self._impl.delete_user(user_id)
def list_users(self):
"""List all users."""
return [self._unify_user(u) for u in self._impl.list_users()]
def delete_service(self, service_id):
"""Deletes service."""
return self._impl.delete_service(service_id)
def list_services(self):
"""List all services."""
return [self._unify_service(s) for s in self._impl.list_services()]
def create_role(self, name=None, domain_name="Default"):
"""Add role to user."""
return self._unify_role(self._impl.create_role(
name, domain_name=domain_name))
def add_role(self, role_id, user_id, project_id):
"""Add role to user."""
return self._unify_role(self._impl.add_role(
role_id=role_id, user_id=user_id, project_id=project_id))
def delete_role(self, role_id):
"""Deletes role."""
return self._impl.delete_role(role_id)
def revoke_role(self, role_id, user_id, project_id):
"""Revokes a role from a user."""
return self._impl.revoke_role(role_id=role_id, user_id=user_id,
project_id=project_id)
def list_roles(self, user_id=None, project_id=None, domain_name=None):
"""List all roles."""
return [self._unify_role(role) for role in self._impl.list_roles(
user_id=user_id, project_id=project_id, domain_name=domain_name)]
def get_role(self, role_id):
"""Get role."""
return self._unify_role(self._impl.get_role(role_id))

10
rally/plugins/openstack/wrappers/keystone.py
View File

@ -36,6 +36,12 @@ class KeystoneWrapper(object):
def __init__(self, client):
self.client = client
LOG.warning(
"Class %s is deprecated since Rally 0.8.0 and will be removed "
"soon. Use "
"rally.plugins.openstack.services.identity.identity.Identity "
"instead." % self.__class__)
def __getattr__(self, attr_name):
return getattr(self.client, attr_name)
@ -253,6 +259,10 @@ class KeystoneV3Wrapper(KeystoneWrapper):
def wrap(client):
"""Returns keystone wrapper based on keystone client version."""
LOG.warning("Method wrap from %s and whole Keystone wrappers are "
"deprecated since Rally 0.8.0 and will be removed soon. Use "
"rally.plugins.openstack.services.identity.identity.Identity "
"instead." % __file__)
if client.version == "v2.0":
return KeystoneV2Wrapper(client)

2
tests/unit/fakes.py
View File

@ -1605,7 +1605,7 @@ class FakeClients(object):
"fake_password",
"fake_tenant_name")
def keystone(self):
def keystone(self, version=None):
if not self._keystone:
self._keystone = FakeKeystoneClient()
return self._keystone

34
tests/unit/plugins/openstack/cleanup/test_resources.py
View File

@ -590,30 +590,30 @@ class KeystoneMixinTestCase(test.TestCase):
kmixin._service = "keystone"
return kmixin
@mock.patch("%s.keystone_wrapper.wrap" % BASE)
def test_manager(self, mock_wrap):
@mock.patch("%s.identity" % BASE)
def test_manager(self, mock_identity):
keystone_mixin = self.get_keystone_mixin()
keystone_mixin.admin = mock.MagicMock()
self.assertEqual(mock_wrap.return_value, keystone_mixin._manager())
mock_wrap.assert_called_once_with(
keystone_mixin.admin.keystone.return_value)
self.assertEqual(mock_identity.Identity.return_value,
keystone_mixin._manager())
mock_identity.Identity.assert_called_once_with(
keystone_mixin.admin)
@mock.patch("%s.keystone_wrapper.wrap" % BASE)
def test_delete(self, mock_wrap):
@mock.patch("%s.identity" % BASE)
def test_delete(self, mock_identity):
keystone_mixin = self.get_keystone_mixin()
keystone_mixin._resource = "some_resource"
keystone_mixin.id = lambda: "id_a"
keystone_mixin.admin = mock.MagicMock()
keystone_mixin.delete()
mock_wrap.assert_called_once_with(
keystone_mixin.admin.keystone.return_value)
mock_wrap().delete_some_resource.assert_called_once_with("id_a")
mock_identity.Identity.assert_called_once_with(keystone_mixin.admin)
identity_service = mock_identity.Identity.return_value
identity_service.delete_some_resource.assert_called_once_with("id_a")
@mock.patch(
"rally.common.utils.name_matches_object")
@mock.patch("%s.keystone_wrapper.wrap" % BASE)
def test_list(self, mock_wrap, mock_name_matches_object):
@mock.patch("rally.common.utils.name_matches_object")
@mock.patch("%s.identity" % BASE)
def test_list(self, mock_identity, mock_name_matches_object):
keystone_mixin = self.get_keystone_mixin()
keystone_mixin._resource = "fake_resource"
keystone_mixin.admin = mock.MagicMock()
@ -623,10 +623,12 @@ class KeystoneMixinTestCase(test.TestCase):
mock.MagicMock(name="foo3")]
mock_name_matches_object.side_effect = [True, True, False]
mock_wrap().list_fake_resources.return_value = result
identity_service = mock_identity.Identity.return_value
identity_service.list_fake_resources.return_value = result
self.assertSequenceEqual(result[:2], keystone_mixin.list())
mock_wrap().list_fake_resources.assert_called_once_with()
identity_service.list_fake_resources.assert_called_once_with()
mock_name_matches_object.assert_has_calls(
[mock.call(r.name, kutils.KeystoneScenario) for r in result])

24
tests/unit/plugins/openstack/context/keystone/test_roles.py
View File

@ -91,10 +91,10 @@ class RoleGeneratorTestCase(test.TestCase):
ctx.credential = mock.MagicMock()
ctx.cleanup()
calls = [
mock.call("u1", "r1", tenant="t1"),
mock.call("u2", "r1", tenant="t2"),
mock.call("u1", "r2", tenant="t1"),
mock.call("u2", "r2", tenant="t2")
mock.call(user="u1", role="r1", tenant="t1"),
mock.call(user="u2", role="r1", tenant="t2"),
mock.call(user="u1", role="r2", tenant="t1"),
mock.call(user="u2", role="r2", tenant="t2")
]
fc.keystone().roles.remove_user_role.assert_has_calls(calls,
@ -113,10 +113,10 @@ class RoleGeneratorTestCase(test.TestCase):
ctx.setup()
ctx.credential = mock.MagicMock()
calls = [
mock.call("u1", "r1", tenant="t1"),
mock.call("u2", "r1", tenant="t2"),
mock.call("u1", "r2", tenant="t1"),
mock.call("u2", "r2", tenant="t2")
mock.call(user="u1", role="r1", tenant="t1"),
mock.call(user="u2", role="r1", tenant="t2"),
mock.call(user="u1", role="r2", tenant="t1"),
mock.call(user="u2", role="r2", tenant="t2")
]
fc.keystone().roles.add_user_role.assert_has_calls(calls,
any_order=True)
@ -132,10 +132,10 @@ class RoleGeneratorTestCase(test.TestCase):
self.assertEqual(4, fc.keystone().roles.add_user_role.call_count)
self.assertEqual(4, fc.keystone().roles.remove_user_role.call_count)
calls = [
mock.call("u1", "r1", tenant="t1"),
mock.call("u2", "r1", tenant="t2"),
mock.call("u1", "r2", tenant="t1"),
mock.call("u2", "r2", tenant="t2")
mock.call(user="u1", role="r1", tenant="t1"),
mock.call(user="u2", role="r1", tenant="t2"),
mock.call(user="u1", role="r2", tenant="t1"),
mock.call(user="u2", role="r2", tenant="t2")
]
fc.keystone().roles.remove_user_role.assert_has_calls(calls,
any_order=True)

63
tests/unit/plugins/openstack/context/keystone/test_users.py
View File

@ -288,8 +288,8 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
"nova-network")
nova_admin.networks.disassociate.assert_called_once_with(networks[0])
@mock.patch("%s.keystone" % CTX)
def test__create_tenants(self, mock_keystone):
@mock.patch("%s.identity" % CTX)
def test__create_tenants(self, mock_identity):
self.context["config"]["users"]["tenants"] = 1
user_generator = users.UserGenerator(self.context)
tenants = user_generator._create_tenants()
@ -297,8 +297,8 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
id, tenant = tenants.popitem()
self.assertIn("name", tenant)
@mock.patch("%s.keystone" % CTX)
def test__create_users(self, mock_keystone):
@mock.patch("%s.identity" % CTX)
def test__create_users(self, mock_identity):
self.context["config"]["users"]["users_per_tenant"] = 2
user_generator = users.UserGenerator(self.context)
user_generator.context["tenants"] = {"t1": {"id": "t1", "name": "t1"},
@ -309,26 +309,26 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
self.assertIn("id", user)
self.assertIn("credential", user)
@mock.patch("%s.keystone" % CTX)
def test__delete_tenants(self, mock_keystone):
@mock.patch("%s.identity" % CTX)
def test__delete_tenants(self, mock_identity):
user_generator = users.UserGenerator(self.context)
user_generator.context["tenants"] = {"t1": {"id": "t1", "name": "t1"},
"t2": {"id": "t2", "name": "t2"}}
user_generator._delete_tenants()
self.assertEqual(len(user_generator.context["tenants"]), 0)
@mock.patch("%s.keystone" % CTX)
def test__delete_tenants_failure(self, mock_keystone):
wrapped_keystone = mock_keystone.wrap.return_value
wrapped_keystone.delete_project.side_effect = Exception()
@mock.patch("%s.identity" % CTX)
def test__delete_tenants_failure(self, mock_identity):
identity_service = mock_identity.Identity.return_value
identity_service.delete_project.side_effect = Exception()
user_generator = users.UserGenerator(self.context)
user_generator.context["tenants"] = {"t1": {"id": "t1", "name": "t1"},
"t2": {"id": "t2", "name": "t2"}}
user_generator._delete_tenants()
self.assertEqual(len(user_generator.context["tenants"]), 0)
@mock.patch("%s.keystone" % CTX)
def test__delete_users(self, mock_keystone):
@mock.patch("%s.identity" % CTX)
def test__delete_users(self, mock_identity):
user_generator = users.UserGenerator(self.context)
user1 = mock.MagicMock()
user2 = mock.MagicMock()
@ -336,10 +336,10 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
user_generator._delete_users()
self.assertEqual(len(user_generator.context["users"]), 0)
@mock.patch("%s.keystone" % CTX)
def test__delete_users_failure(self, mock_keystone):
wrapped_keystone = mock_keystone.wrap.return_value
wrapped_keystone.delete_user.side_effect = Exception()
@mock.patch("%s.identity" % CTX)
def test__delete_users_failure(self, mock_identity):
identity_service = mock_identity.Identity.return_value
identity_service.delete_user.side_effect = Exception()
user_generator = users.UserGenerator(self.context)
user1 = mock.MagicMock()
user2 = mock.MagicMock()
@ -347,10 +347,8 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
user_generator._delete_users()
self.assertEqual(len(user_generator.context["users"]), 0)
@mock.patch("%s.keystone" % CTX)
def test_setup_and_cleanup(self, mock_keystone):
wrapped_keystone = mock.MagicMock()
mock_keystone.wrap.return_value = wrapped_keystone
@mock.patch("%s.identity" % CTX)
def test_setup_and_cleanup(self, mock_identity):
with users.UserGenerator(self.context) as ctx:
ctx.setup()
@ -365,11 +363,11 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
self.assertEqual(len(ctx.context["tenants"]), 0)
@mock.patch("rally.common.broker.LOG.warning")
@mock.patch("%s.keystone" % CTX)
@mock.patch("%s.identity" % CTX)
def test_setup_and_cleanup_with_error_during_create_user(
self, mock_keystone, mock_log_warning):
wrapped_keystone = mock_keystone.wrap.return_value
wrapped_keystone.create_user.side_effect = Exception
self, mock_identity, mock_log_warning):
identity_service = mock_identity.Identity.return_value
identity_service.create_user.side_effect = Exception()
with users.UserGenerator(self.context) as ctx:
self.assertRaises(exceptions.ContextSetupFailure, ctx.setup)
mock_log_warning.assert_called_with(
@ -378,10 +376,9 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
# Ensure that tenants get deleted anyway
self.assertEqual(0, len(ctx.context["tenants"]))
@mock.patch("%s.keystone" % CTX)
def test_users_and_tenants_in_context(self, mock_keystone):
wrapped_keystone = mock.MagicMock()
mock_keystone.wrap.return_value = wrapped_keystone
@mock.patch("%s.identity" % CTX)
def test_users_and_tenants_in_context(self, mock_identity):
identity_service = mock_identity.Identity.return_value
credential = objects.Credential("foo_url", "foo", "foo_pass",
https_insecure=True,
@ -395,7 +392,7 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
credential_dict = credential.to_dict(False)
user_list = [mock.MagicMock(id="id_%d" % i)
for i in range(self.users_num)]
wrapped_keystone.create_user.side_effect = user_list
identity_service.create_user.side_effect = user_list
with users.UserGenerator(tmp_context) as ctx:
ctx.generate_random_name = mock.Mock()
@ -430,8 +427,8 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
self.assertEqual(user["id"], orig_user.id)
self.assertEqual(user["tenant_id"], tenant_id)
@mock.patch("%s.keystone" % CTX)
def test_users_contains_correct_endpoint_type(self, mock_keystone):
@mock.patch("%s.identity" % CTX)
def test_users_contains_correct_endpoint_type(self, mock_identity):
credential = objects.Credential(
"foo_url", "foo", "foo_pass",
endpoint_type=consts.EndpointType.INTERNAL)
@ -453,8 +450,8 @@ class UserGeneratorTestCase(test.ScenarioTestCase):
for user in users_:
self.assertEqual("internal", user["credential"].endpoint_type)
@mock.patch("%s.keystone" % CTX)
def test_users_contains_default_endpoint_type(self, mock_keystone):
@mock.patch("%s.identity" % CTX)
def test_users_contains_default_endpoint_type(self, mock_identity):
credential = objects.Credential("foo_url", "foo", "foo_pass")
config = {
"config": {

143
tests/unit/plugins/openstack/scenarios/keystone/test_basic.py
View File

@ -13,18 +13,23 @@
# License for the specific language governing permissions and limitations
# under the License.
import ddt
import mock
from rally.plugins.openstack.scenarios.keystone import basic
from tests.unit import test
@ddt.ddt
class KeystoneBasicTestCase(test.ScenarioTestCase):
@staticmethod
def _get_context():
context = test.get_test_context()
def get_test_context(self):
context = super(KeystoneBasicTestCase, self).get_test_context()
context.update({
"admin": {
"id": "fake_user_id",
"credential": mock.MagicMock()
},
"user": {
"id": "fake_user_id",
"credential": mock.MagicMock()
@ -34,6 +39,13 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
})
return context
def setUp(self):
super(KeystoneBasicTestCase, self).setUp()
patch = mock.patch(
"rally.plugins.openstack.services.identity.identity.Identity")
self.addCleanup(patch.stop)
self.mock_identity = patch.start()
def test_create_user(self):
scenario = basic.CreateUser(self.context)
scenario._user_create = mock.MagicMock()
@ -70,7 +82,7 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
def test_user_authenticate_and_validate_token(self):
fake_token = mock.MagicMock()
context = self._get_context()
context = self.context
scenario = basic.AuthenticateUserAndValidateToken(context)
fake_user = context["user"]["credential"].username
@ -120,71 +132,73 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
scenario._list_tenants.assert_called_with()
def test_assign_and_remove_user_role(self):
context = self._get_context()
scenario = basic.AddAndRemoveUserRole(context)
fake_tenant = context["tenant"]["id"]
fake_user = context["user"]["id"]
fake_tenant = self.context["tenant"]["id"]
fake_user = self.context["user"]["id"]
fake_role = mock.MagicMock()
scenario._tenant_create = mock.MagicMock(return_value=fake_tenant)
scenario._user_create = mock.MagicMock(return_value=fake_user)
scenario._role_create = mock.MagicMock(return_value=fake_role)
scenario._role_add = mock.MagicMock()
scenario._role_remove = mock.MagicMock()
self.mock_identity.return_value.create_role.return_value = fake_role
scenario = basic.AddAndRemoveUserRole(self.context)
scenario.run()
scenario._role_create.assert_called_once_with()
scenario._role_add.assert_called_once_with(fake_user,
fake_role,
fake_tenant)
scenario._role_remove.assert_called_once_with(fake_user,
fake_role,
fake_tenant)
self.mock_identity.return_value.create_role.assert_called_once_with()
self.mock_identity.return_value.add_role.assert_called_once_with(
role_id=fake_role.id, user_id=fake_user, project_id=fake_tenant)
self.mock_identity.return_value.revoke_role.assert_called_once_with(
fake_role.id, user_id=fake_user, project_id=fake_tenant)
def test_create_and_delete_role(self):
scenario = basic.CreateAndDeleteRole(self.context)
fake_role = mock.MagicMock()
scenario._role_create = mock.MagicMock(return_value=fake_role)
scenario._role_delete = mock.MagicMock()
self.mock_identity.return_value.create_role.return_value = fake_role
scenario = basic.CreateAndDeleteRole(self.context)
scenario.run()
scenario._role_create.assert_called_once_with()
scenario._role_delete.assert_called_once_with(fake_role.id)
self.mock_identity.return_value.create_role.assert_called_once_with()
self.mock_identity.return_value.delete_role.assert_called_once_with(
fake_role.id)
def test_create_and_get_role(self):
scenario = basic.CreateAndGetRole(self.context)
fake_role = mock.MagicMock()
scenario._role_create = mock.MagicMock(return_value=fake_role)
scenario._get_role = mock.MagicMock()
self.mock_identity.return_value.create_role.return_value = fake_role
scenario = basic.CreateAndGetRole(self.context)
scenario.run()
scenario._role_create.assert_called_once_with()
scenario._get_role.assert_called_once_with(fake_role.id)
self.mock_identity.return_value.create_role.assert_called_once_with()
self.mock_identity.return_value.get_role.assert_called_once_with(
fake_role.id)
def test_create_and_list_user_roles(self):
context = self._get_context()
scenario = basic.CreateAddAndListUserRoles(context)
fake_tenant = context["tenant"]["id"]
fake_user = context["user"]["id"]
scenario = basic.CreateAddAndListUserRoles(self.context)
fake_tenant = self.context["tenant"]["id"]
fake_user = self.context["user"]["id"]
fake_role = mock.MagicMock()
scenario._tenant_create = mock.MagicMock(return_value=fake_tenant)
scenario._user_create = mock.MagicMock(return_value=fake_user)
scenario._role_create = mock.MagicMock(return_value=fake_role)
scenario._role_add = mock.MagicMock()
scenario._list_roles_for_user = mock.MagicMock()
scenario.run()
scenario._role_create.assert_called_once_with()
scenario._role_add.assert_called_once_with(fake_user,
fake_role, fake_tenant)
scenario._list_roles_for_user.assert_called_once_with(fake_user,
fake_tenant)
self.mock_identity.return_value.create_role.return_value = fake_role
def _test_get_entities(self, service_name="keystone"):
scenario = basic.GetEntities(self.context)
scenario.run()
self.mock_identity.return_value.create_role.assert_called_once_with()
self.mock_identity.return_value.add_role.assert_called_once_with(
user_id=fake_user, role_id=fake_role.id, project_id=fake_tenant)
self.mock_identity.return_value.list_roles.assert_called_once_with(
user_id=fake_user, project_id=fake_tenant)
@ddt.data(None, "keystone", "fooservice")
def test_get_entities(self, service_name):
fake_tenant = mock.MagicMock()
fake_user = mock.MagicMock()
fake_role = mock.MagicMock()
fake_service = mock.MagicMock()
self.mock_identity.return_value.create_role.return_value = fake_role
scenario = basic.GetEntities(self.context)
scenario._tenant_create = mock.MagicMock(return_value=fake_tenant)
scenario._user_create = mock.MagicMock(return_value=fake_user)
scenario._role_create = mock.MagicMock(return_value=fake_role)
scenario._service_create = mock.MagicMock(return_value=fake_service)
scenario._get_tenant = mock.MagicMock(return_value=fake_tenant)
@ -198,7 +212,7 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
scenario._tenant_create.assert_called_once_with()
scenario._user_create.assert_called_once_with()
scenario._role_create.assert_called_once_with()
self.mock_identity.return_value.create_role.assert_called_once_with()
scenario._get_tenant.assert_called_once_with(fake_tenant.id)
scenario._get_user.assert_called_once_with(fake_user.id)
@ -212,15 +226,6 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
self.assertFalse(scenario._service_create.called)
scenario._get_service.assert_called_once_with(fake_service.id)
def test_get_entities(self):
self._test_get_entities()
def test_get_entities_with_service_name(self):
self._test_get_entities(service_name="fooservice")
def test_get_entities_create_service(self):
self._test_get_entities(service_name=None)
def test_create_and_delete_service(self):
scenario = basic.CreateAndDeleteService(self.context)
service_type = "test_service_type"
@ -271,7 +276,7 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
scenario._list_services.assert_called_once_with()
def test_create_and_list_ec2credentials(self):
context = self._get_context()
context = self.context
scenario = basic.CreateAndListEc2Credentials(context)
scenario._create_ec2credentials = mock.MagicMock()
scenario._list_ec2credentials = mock.MagicMock()
@ -282,7 +287,7 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
def test_create_and_delete_ec2credential(self):
fake_creds = mock.MagicMock()
context = self._get_context()
context = self.context
scenario = basic.CreateAndDeleteEc2Credential(context)
scenario._create_ec2credentials = mock.MagicMock(
return_value=fake_creds)
@ -294,20 +299,18 @@ class KeystoneBasicTestCase(test.ScenarioTestCase):
"fake_user_id", fake_creds.access)
def test_add_and_remove_user_role(self):
context = self._get_context()
context = self.context
tenant_id = context["tenant"]["id"]
user_id = context["user"]["id"]
scenario = basic.AddAndRemoveUserRole(context)
fake_role = mock.MagicMock()
scenario._role_create = mock.MagicMock(return_value=fake_role)
scenario._role_add = mock.MagicMock()
scenario._role_remove = mock.MagicMock()
self.mock_identity.return_value.create_role.return_value = fake_role
scenario = basic.AddAndRemoveUserRole(context)
scenario.run()
scenario._role_create.assert_called_once_with()
scenario._role_add.assert_called_once_with(
user_id, fake_role, tenant_id)
scenario._role_remove.assert_called_once_with(
user_id, fake_role, tenant_id)
self.mock_identity.return_value.create_role.assert_called_once_with()
self.mock_identity.return_value.add_role.assert_called_once_with(
role_id=fake_role.id, user_id=user_id, project_id=tenant_id)
self.mock_identity.return_value.revoke_role.assert_called_once_with(
fake_role.id, user_id=user_id, project_id=tenant_id)

0
tests/unit/plugins/openstack/services/identity/__init__.py Normal file
View File

169
tests/unit/plugins/openstack/services/identity/test_identity.py Normal file
View File

@ -0,0 +1,169 @@
# Copyright 2014: Mirantis Inc.
# All Rights Reserved.
#
# 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 ddt
import mock
from rally.plugins.openstack.services.identity import identity
from tests.unit import test
@ddt.ddt
class IdentityTestCase(test.TestCase):
def setUp(self):
super(IdentityTestCase, self).setUp()
self.clients = mock.MagicMock()
def get_service_with_fake_impl(self):
path = "rally.plugins.openstack.services.identity.identity"
with mock.patch("%s.Identity.discover_impl" % path) as mock_discover:
mock_discover.return_value = mock.MagicMock(), None
service = identity.Identity(self.clients)
return service
def test_create_project(self):
service = self.get_service_with_fake_impl()
project_name = "name"
domain_name = "domain"
service.create_project(project_name, domain_name=domain_name)
service._impl.create_project.assert_called_once_with(
project_name, domain_name=domain_name)
def test_delete_project(self):
service = self.get_service_with_fake_impl()
project = "id"
service.delete_project(project)
service._impl.delete_project.assert_called_once_with(project)
def test_list_projects(self):
service = self.get_service_with_fake_impl()
service.list_projects()
service._impl.list_projects.assert_called_once_with()
def test_create_user(self):
service = self.get_service_with_fake_impl()
username = "username"
password = "password"
email = "email"
project_id = "project_id"
domain_name = "domain_name"
service.create_user(username=username, password=password, email=email,
project_id=project_id, domain_name=domain_name)
service._impl.create_user.assert_called_once_with(
username=username, password=password, email=email,
project_id=project_id, domain_name=domain_name,
default_role="member")
def test_delete_user(self):
service = self.get_service_with_fake_impl()
user_id = "fake_id"
service.delete_user(user_id)
service._impl.delete_user.assert_called_once_with(user_id)
def test_list_users(self):
service = self.get_service_with_fake_impl()
service.list_users()
service._impl.list_users.assert_called_once_with()
def test_delete_service(self):
service = self.get_service_with_fake_impl()
service_id = "id"
service.delete_service(service_id)
service._impl.delete_service.assert_called_once_with(service_id)
def test_list_services(self):
service = self.get_service_with_fake_impl()
service.list_services()
service._impl.list_services.assert_called_once_with()
def test_create_role(self):
service = self.get_service_with_fake_impl()
name = "name"
service.create_role(name)
service._impl.create_role.assert_called_once_with(
name=name, domain_name="Default")
def test_add_role(self):
service = self.get_service_with_fake_impl()
role_id = "id"
user_id = "user_id"
project_id = "project_id"
service.add_role(role_id, user_id=user_id, project_id=project_id)
service._impl.add_role.assert_called_once_with(role_id=role_id,
user_id=user_id,
project_id=project_id)
def test_delete_role(self):
service = self.get_service_with_fake_impl()
role = "id"
service.delete_role(role)
service._impl.delete_role.assert_called_once_with(role)
def test_revoke_role(self):
service = self.get_service_with_fake_impl()
role_id = "id"
user_id = "user_id"
project_id = "project_id"
service.revoke_role(role_id, user_id=user_id, project_id=project_id)
service._impl.revoke_role.assert_called_once_with(
role_id=role_id, user_id=user_id, project_id=project_id)
@ddt.data((None, None, None), ("user_id", "project_id", "domain"))
def test_list_roles(self, params):
user, project, domain = params
service = self.get_service_with_fake_impl()
service.list_roles(user_id=user, project_id=project,
domain_name=domain)
service._impl.list_roles.assert_called_once_with(user_id=user,
project_id=project,
domain_name=domain)
def test_get_role(self):
service = self.get_service_with_fake_impl()
role = "id"
service.get_role(role)
service._impl.get_role.assert_called_once_with(role)
def test__unify_service(self):
class SomeFakeService(object):
id = 123123123123123
name = "asdfasdfasdfasdfadf"
other_var = "asdfasdfasdfasdfasdfasdfasdf"
service = self.get_service_with_fake_impl()._unify_service(
SomeFakeService())
self.assertIsInstance(service, identity.Service)
self.assertEqual(SomeFakeService.id, service.id)
self.assertEqual(SomeFakeService.name, service.name)
def test__unify_role(self):
class SomeFakeRole(object):
id = 123123123123123
name = "asdfasdfasdfasdfadf"
other_var = "asdfasdfasdfasdfasdfasdfasdf"
role = self.get_service_with_fake_impl()._unify_role(
SomeFakeRole())
self.assertIsInstance(role, identity.Role)
self.assertEqual(SomeFakeRole.id, role.id)
self.assertEqual(SomeFakeRole.name, role.name)

364
tests/unit/plugins/openstack/services/identity/test_keystone_v2.py Normal file
View File

@ -0,0 +1,364 @@
# Copyright 2014: Mirantis Inc.
# All Rights Reserved.
#
# 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 uuid
import mock
from rally.plugins.openstack.services.identity import identity
from rally.plugins.openstack.services.identity import keystone_v2
from tests.unit import test
PATH = "rally.plugins.openstack.services.identity.keystone_v2"
class KeystoneV2ServiceTestCase(test.TestCase):
def setUp(self):
super(KeystoneV2ServiceTestCase, self).setUp()
self.clients = mock.MagicMock()
self.kc = self.clients.keystone.return_value
self.service = keystone_v2.KeystoneV2Service(self.clients)
def test_create_tenant(self):
name = "name"
tenant = self.service.create_tenant(name)
self.assertEqual(tenant, self.kc.tenants.create.return_value)
self.kc.tenants.create.assert_called_once_with(name)
def test_delete_tenant(self):
tenant_id = "fake_id"
self.service.delete_tenant(tenant_id)
self.kc.tenants.delete.assert_called_once_with(tenant_id)
def test_list_tenants(self):
self.assertEqual(self.kc.tenants.list.return_value,
self.service.list_tenants())
self.kc.tenants.list.assert_called_once_with()
def test_create_user(self):
name = "name"
password = "passwd"
email = "rally@example.com"
tenant_id = "project"
user = self.service.create_user(name, password=password, email=email,
tenant_id=tenant_id)
self.assertEqual(user, self.kc.users.create.return_value)
self.kc.users.create.assert_called_once_with(
name=name, password=password, email=email, tenant_id=tenant_id)
def test_list_users(self):
self.assertEqual(self.kc.users.list.return_value,
self.service.list_users())
self.kc.users.list.assert_called_once_with()
def test_delete_user(self):
user_id = "fake_id"
self.service.delete_user(user_id)
self.kc.users.delete.assert_called_once_with(user_id)
def test_delete_service(self):
service_id = "fake_id"
self.service.delete_service(service_id)
self.kc.services.delete.assert_called_once_with(service_id)
def test_list_services(self):
self.assertEqual(self.kc.services.list.return_value,
self.service.list_services())
self.kc.services.list.assert_called_once_with()
def test_create_role(self):
name = "some"
self.service.create_role(name)
self.kc.roles.create.assert_called_once_with(name)
def test_add_role(self):
role_id = "fake_id"
user_id = "user_id"
tenant_id = "tenant_id"
self.service.add_role(role_id, user_id=user_id, tenant_id=tenant_id)
self.kc.roles.add_user_role.assert_called_once_with(
user=user_id, role=role_id, tenant=tenant_id)
def test_delete_role(self):
role_id = "fake_id"
self.service.delete_role(role_id)
self.kc.roles.delete.assert_called_once_with(role_id)
def test_list_roles(self):
self.assertEqual(self.kc.roles.list.return_value,
self.service.list_roles())
self.kc.roles.list.assert_called_once_with()
def test_list_roles_for_user(self):
user_id = "user_id"
tenant_id = "tenant_id"
self.assertEqual(self.kc.roles.roles_for_user.return_value,
self.service.list_roles_for_user(user_id,
tenant_id=tenant_id))
self.kc.roles.roles_for_user.assert_called_once_with(user_id,
tenant_id)
def test_revoke_role(self):
role_id = "fake_id"
user_id = "user_id"
tenant_id = "tenant_id"
self.service.revoke_role(role_id, user_id=user_id,
tenant_id=tenant_id)
self.kc.roles.remove_user_role.assert_called_once_with(
user=user_id, role=role_id, tenant=tenant_id)
def test_get_role(self):
role_id = "fake_id"
self.service.get_role(role_id)
self.kc.roles.get.assert_called_once_with(role_id)
class UnifiedKeystoneV2ServiceTestCase(test.TestCase):
def setUp(self):
super(UnifiedKeystoneV2ServiceTestCase, self).setUp()
self.clients = mock.MagicMock()
self.service = keystone_v2.UnifiedKeystoneV2Service(self.clients)
self.service._impl = mock.MagicMock()
def test_init_identity_service(self):
self.clients.keystone.return_value.version = "v2.0"
self.assertIsInstance(identity.Identity(self.clients)._impl,
keystone_v2.UnifiedKeystoneV2Service)
def test__check_domain(self):
self.service._check_domain("Default")
self.service._check_domain("default")
self.assertRaises(NotImplementedError, self.service._check_domain,
"non-default")
def test__unify_tenant(self):
class KeystoneV2Tenant(object):
def __init__(self, domain_id="domain_id"):
self.id = str(uuid.uuid4())
self.name = str(uuid.uuid4())
self.domain_id = domain_id
tenant = KeystoneV2Tenant()
project = self.service._unify_tenant(tenant)
self.assertIsInstance(project, identity.Project)
self.assertEqual(tenant.id, project.id)
self.assertEqual(tenant.name, project.name)
self.assertEqual("default", project.domain_id)
self.assertNotEqual(tenant.domain_id, project.domain_id)
def test__unify_user(self):
class KeystoneV2User(object):
def __init__(self, tenantId=None):
self.id = str(uuid.uuid4())
self.name = str(uuid.uuid4())
if tenantId is not None:
self.tenantId = tenantId
user = KeystoneV2User()
unified_user = self.service._unify_user(user)
self.assertIsInstance(unified_user, identity.User)
self.assertEqual(user.id, unified_user.id)
self.assertEqual(user.name, unified_user.name)
self.assertEqual("default", unified_user.domain_id)
self.assertIsNone(unified_user.project_id)
tenant_id = "tenant_id"
user = KeystoneV2User(tenantId=tenant_id)
unified_user = self.service._unify_user(user)
self.assertIsInstance(unified_user, identity.User)
self.assertEqual(user.id, unified_user.id)
self.assertEqual(user.name, unified_user.name)
self.assertEqual("default", unified_user.domain_id)
self.assertEqual(tenant_id, unified_user.project_id)
@mock.patch("%s.UnifiedKeystoneV2Service._check_domain" % PATH)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_tenant" % PATH)
def test_create_project(
self, mock_unified_keystone_v2_service__unify_tenant,
mock_unified_keystone_v2_service__check_domain):
mock_unify_tenant = mock_unified_keystone_v2_service__unify_tenant
mock_check_domain = mock_unified_keystone_v2_service__check_domain
name = "name"
self.assertEqual(mock_unify_tenant.return_value,
self.service.create_project(name))
mock_check_domain.assert_called_once_with("Default")
mock_unify_tenant.assert_called_once_with(
self.service._impl.create_tenant.return_value)
self.service._impl.create_tenant.assert_called_once_with(name)
def test_delete_project(self):
tenant_id = "fake_id"
self.service.delete_project(tenant_id)
self.service._impl.delete_tenant.assert_called_once_with(tenant_id)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_tenant" % PATH)
def test_list_projects(self,
mock_unified_keystone_v2_service__unify_tenant):
mock_unify_tenant = mock_unified_keystone_v2_service__unify_tenant
tenants = [mock.MagicMock()]
self.service._impl.list_tenants.return_value = tenants
self.assertEqual([mock_unify_tenant.return_value],
self.service.list_projects())
mock_unify_tenant.assert_called_once_with(tenants[0])
@mock.patch("%s.UnifiedKeystoneV2Service._check_domain" % PATH)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_user" % PATH)
def test_create_user(self, mock_unified_keystone_v2_service__unify_user,
mock_unified_keystone_v2_service__check_domain):
mock_check_domain = mock_unified_keystone_v2_service__check_domain
mock_unify_user = mock_unified_keystone_v2_service__unify_user
name = "name"
password = "passwd"
email = "rally@example.com"
tenant_id = "project"
self.assertEqual(mock_unify_user.return_value,
self.service.create_user(name, password=password,
email=email,
project_id=tenant_id))
mock_check_domain.assert_called_once_with("Default")
mock_unify_user.assert_called_once_with(
self.service._impl.create_user.return_value)
self.service._impl.create_user.assert_called_once_with(
username=name, password=password, email=email, tenant_id=tenant_id)
def test_delete_user(self):
user_id = "fake_id"
self.service.delete_user(user_id)
self.service._impl.delete_user.assert_called_once_with(user_id)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_user" % PATH)
def test_list_users(self, mock_unified_keystone_v2_service__unify_user):
mock_unify_user = mock_unified_keystone_v2_service__unify_user
users = [mock.MagicMock()]
self.service._impl.list_users.return_value = users
self.assertEqual([mock_unify_user.return_value],
self.service.list_users())
mock_unify_user.assert_called_once_with(users[0])
def test_delete_service(self):
service_id = "fake_id"
self.service.delete_service(service_id)
self.service._impl.delete_service.assert_called_once_with(service_id)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_service" % PATH)
def test_list_services(self,
mock_unified_keystone_v2_service__unify_service):
mock_unify_service = mock_unified_keystone_v2_service__unify_service
services = [mock.MagicMock()]
self.service._impl.list_services.return_value = services
self.assertEqual([mock_unify_service.return_value],
self.service.list_services())
mock_unify_service.assert_called_once_with(services[0])
@mock.patch("%s.UnifiedKeystoneV2Service._unify_role" % PATH)
def test_create_role(self, mock_unified_keystone_v2_service__unify_role):
mock_unify_role = mock_unified_keystone_v2_service__unify_role
name = "some"
self.assertEqual(mock_unify_role.return_value,
self.service.create_role(name))
self.service._impl.create_role.assert_called_once_with(name)
mock_unify_role.assert_called_once_with(
self.service._impl.create_role.return_value)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_role" % PATH)
def test_add_role(self, mock_unified_keystone_v2_service__unify_role):
mock_unify_role = mock_unified_keystone_v2_service__unify_role
role_id = "fake_id"
user_id = "user_id"
project_id = "user_id"
self.assertEqual(mock_unify_role.return_value,
self.service.add_role(role_id, user_id=user_id,
project_id=project_id))
self.service._impl.add_role.assert_called_once_with(
user_id=user_id, role_id=role_id, tenant_id=project_id)
mock_unify_role.assert_called_once_with(
self.service._impl.add_role.return_value)
def test_delete_role(self):
role_id = "fake_id"
self.service.delete_role(role_id)
self.service._impl.delete_role.assert_called_once_with(role_id)
def test_revoke_role(self):
role_id = "fake_id"
user_id = "user_id"
project_id = "user_id"
self.service.revoke_role(role_id, user_id=user_id,
project_id=project_id)
self.service._impl.revoke_role.assert_called_once_with(
user_id=user_id, role_id=role_id, tenant_id=project_id)
@mock.patch("%s.UnifiedKeystoneV2Service._unify_role" % PATH)
def test_list_roles(self, mock_unified_keystone_v2_service__unify_role):
mock_unify_role = mock_unified_keystone_v2_service__unify_role
roles = [mock.MagicMock()]
another_roles = [mock.MagicMock()]
self.service._impl.list_roles.return_value = roles
self.service._impl.list_roles_for_user.return_value = another_roles
# case 1
self.assertEqual([mock_unify_role.return_value],
self.service.list_roles())
self.service._impl.list_roles.assert_called_once_with()
mock_unify_role.assert_called_once_with(roles[0])
self.assertFalse(self.service._impl.list_roles_for_user.called)
self.service._impl.list_roles.reset_mock()
mock_unify_role.reset_mock()
# case 2
user = "user"
project = "project"
self.assertEqual([mock_unify_role.return_value],
self.service.list_roles(user_id=user,
project_id=project))
self.service._impl.list_roles_for_user.assert_called_once_with(
user, tenant_id=project)
self.assertFalse(self.service._impl.list_roles.called)
mock_unify_role.assert_called_once_with(another_roles[0])
# case 3
self.assertRaises(NotImplementedError, self.service.list_roles,
domain_name="some")
def test_get_role(self):
role_id = "fake_id"
self.service.get_role(role_id)
self.service._impl.get_role.assert_called_once_with(role_id)

424
tests/unit/plugins/openstack/services/identity/test_keystone_v3.py Normal file
View File

@ -0,0 +1,424 @@
# Copyright 2014: Mirantis Inc.
# All Rights Reserved.
#
# 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 uuid
import mock
from rally import exceptions
from rally.plugins.openstack.services.identity import identity
from rally.plugins.openstack.services.identity import keystone_v3
from tests.unit import test
PATH = "rally.plugins.openstack.services.identity.keystone_v3"
class KeystoneV3ServiceTestCase(test.TestCase):
def setUp(self):
super(KeystoneV3ServiceTestCase, self).setUp()
self.clients = mock.MagicMock()
self.kc = self.clients.keystone.return_value
self.service = keystone_v3.KeystoneV3Service(self.clients)
def test__get_domain_id_not_found(self):
from keystoneclient import exceptions as kc_exceptions
self.kc.domains.get.side_effect = kc_exceptions.NotFound
self.kc.domains.list.return_value = []
domain_name_or_id = "some"
self.assertRaises(exceptions.GetResourceNotFound,
self.service._get_domain_id, domain_name_or_id)
self.kc.domains.get.assert_called_once_with(domain_name_or_id)
self.kc.domains.list.assert_called_once_with(name=domain_name_or_id)
def test__get_domain_id_find_by_name(self):
from keystoneclient import exceptions as kc_exceptions
self.kc.domains.get.side_effect = kc_exceptions.NotFound
domain = mock.MagicMock()
self.kc.domains.list.return_value = [domain]
domain_name_or_id = "some"
self.assertEqual(domain.id,
self.service._get_domain_id(domain_name_or_id))
self.kc.domains.get.assert_called_once_with(domain_name_or_id)
self.kc.domains.list.assert_called_once_with(name=domain_name_or_id)
def test__get_domain_id_find_by_id(self):
domain = mock.MagicMock()
self.kc.domains.get.return_value = domain
domain_name_or_id = "some"
self.assertEqual(domain.id,
self.service._get_domain_id(domain_name_or_id))
self.kc.domains.get.assert_called_once_with(domain_name_or_id)
self.assertFalse(self.kc.domains.list.called)
@mock.patch("%s.KeystoneV3Service._get_domain_id" % PATH)
def test_create_project(self, mock__get_domain_id):
name = "name"
domain_name = "domain"
domain_id = "id"
mock__get_domain_id.return_value = domain_id
project = self.service.create_project(name, domain_name=domain_name)
mock__get_domain_id.assert_called_once_with(domain_name)
self.assertEqual(project, self.kc.projects.create.return_value)
self.kc.projects.create.assert_called_once_with(name=name,
domain=domain_id)
def test_delete_project(self):
project_id = "fake_id"
self.service.delete_project(project_id)
self.kc.projects.delete.assert_called_once_with(project_id)
def test_list_projects(self):
self.assertEqual(self.kc.projects.list.return_value,
self.service.list_projects())
self.kc.projects.list.assert_called_once_with()
@mock.patch("%s.LOG" % PATH)
@mock.patch("%s.KeystoneV3Service._get_domain_id" % PATH)
def test_create_user(self, mock__get_domain_id, mock_log):
name = "name"
password = "passwd"
email = "rally@example.com"
project_id = "project"
domain_name = "domain"
self.service.list_roles = mock.MagicMock(return_value=[])
user = self.service.create_user(name, password=password, email=email,
project_id=project_id,
domain_name=domain_name)
self.assertEqual(user, self.kc.users.create.return_value)
self.kc.users.create.assert_called_once_with(
name=name, password=password, email=email,
default_project=project_id,
domain=mock__get_domain_id.return_value)
self.assertTrue(mock_log.warning.called)
@mock.patch("%s.LOG" % PATH)
@mock.patch("%s.KeystoneV3Service._get_domain_id" % PATH)
def test_create_user_and_add_role(
self, mock_keystone_v3_service__get_domain_id, mock_log):
mock__get_domain_id = mock_keystone_v3_service__get_domain_id
name = "name"
password = "passwd"
email = "rally@example.com"
project_id = "project"
domain_name = "domain"
class Role(object):
def __init__(self, name):
self.name = name
self.id = str(uuid.uuid4())
self.service.list_roles = mock.MagicMock(
return_value=[Role("admin"), Role("member")])
self.service.add_role = mock.MagicMock()
user = self.service.create_user(name, password=password, email=email,
project_id=project_id,
domain_name=domain_name)
self.assertEqual(user, self.kc.users.create.return_value)
self.kc.users.create.assert_called_once_with(
name=name, password=password, email=email,
default_project=project_id,
domain=mock__get_domain_id.return_value)
self.assertFalse(mock_log.warning.called)
self.service.add_role.assert_called_once_with(
role_id=self.service.list_roles.return_value[1].id,
user_id=user.id,
project_id=project_id)
def test_list_users(self):
self.assertEqual(self.kc.users.list.return_value,
self.service.list_users())
self.kc.users.list.assert_called_once_with()
def test_delete_user(self):
user_id = "fake_id"
self.service.delete_user(user_id)
self.kc.users.delete.assert_called_once_with(user_id)
def test_delete_service(self):
service_id = "fake_id"
self.service.delete_service(service_id)
self.kc.services.delete.assert_called_once_with(service_id)
def test_list_services(self):
self.assertEqual(self.kc.services.list.return_value,
self.service.list_services())
self.kc.services.list.assert_called_once_with()
@mock.patch("%s.KeystoneV3Service._get_domain_id" % PATH)
def test_create_role(self, mock__get_domain_id):
domain_name = "domain"
name = "some"
user = self.service.create_role(name, domain_name=domain_name)
self.assertEqual(user, self.kc.roles.create.return_value)
self.kc.roles.create.assert_called_once_with(
name, domain=mock__get_domain_id.return_value)
def test_add_role(self):
role_id = "fake_id"
user_id = "user_id"
project_id = "project_id"
self.service.add_role(role_id, user_id=user_id, project_id=project_id)
self.kc.roles.grant.assert_called_once_with(
user=user_id, role=role_id, project=project_id)
def test_delete_role(self):
role_id = "fake_id"
self.service.delete_role(role_id)
self.kc.roles.delete.assert_called_once_with(role_id)
def test_list_roles(self):
self.assertEqual(self.kc.roles.list.return_value,
self.service.list_roles())
self.kc.roles.list.assert_called_once_with(project=None, user=None,
domain=None)
def test_revoke_role(self):
role_id = "fake_id"
user_id = "user_id"
project_id = "tenant_id"
self.service.revoke_role(role_id, user_id=user_id,
project_id=project_id)
self.kc.roles.revoke.assert_called_once_with(
user=user_id, role=role_id, project=project_id)
def test_get_role(self):
role_id = "fake_id"
self.service.get_role(role_id)
self.kc.roles.get.assert_called_once_with(role_id)
def test_create_domain(self):
name = "some_domain"
descr = "descr"
enabled = False
self.service.create_domain(name, description=descr, enabled=enabled)
self.kc.domains.create.assert_called_once_with(
name, description=descr, enabled=enabled)
class UnifiedKeystoneV3ServiceTestCase(test.TestCase):
def setUp(self):
super(UnifiedKeystoneV3ServiceTestCase, self).setUp()
self.clients = mock.MagicMock()
self.service = keystone_v3.UnifiedKeystoneV3Service(self.clients)
self.service._impl = mock.MagicMock()
def test_init_identity_service(self):
self.clients.keystone.return_value.version = "v3"
self.assertIsInstance(identity.Identity(self.clients)._impl,
keystone_v3.UnifiedKeystoneV3Service)
def test__unify_project(self):
class KeystoneV3Project(object):
def __init__(self):
self.id = str(uuid.uuid4())
self.name = str(uuid.uuid4())
self.domain_id = str(uuid.uuid4())
project = KeystoneV3Project()
unified_project = self.service._unify_project(project)
self.assertIsInstance(unified_project, identity.Project)
self.assertEqual(project.id, unified_project.id)
self.assertEqual(project.name, unified_project.name)
self.assertEqual(project.domain_id, unified_project.domain_id)
self.assertEqual(project.domain_id, unified_project.domain_id)
def test__unify_user(self):
class KeystoneV3User(object):
def __init__(self, project_id=None):
self.id = str(uuid.uuid4())
self.name = str(uuid.uuid4())
self.domain_id = str(uuid.uuid4())
if project_id is not None:
self.default_project_id = project_id
user = KeystoneV3User()
unified_user = self.service._unify_user(user)
self.assertIsInstance(unified_user, identity.User)
self.assertEqual(user.id, unified_user.id)
self.assertEqual(user.name, unified_user.name)
self.assertEqual(user.domain_id, unified_user.domain_id)
self.assertIsNone(unified_user.project_id)
project_id = "tenant_id"
user = KeystoneV3User(project_id=project_id)
unified_user = self.service._unify_user(user)
self.assertIsInstance(unified_user, identity.User)
self.assertEqual(user.id, unified_user.id)
self.assertEqual(user.name, unified_user.name)
self.assertEqual(user.domain_id, unified_user.domain_id)
self.assertEqual(project_id, unified_user.project_id)
@mock.patch("%s.UnifiedKeystoneV3Service._unify_project" % PATH)
def test_create_project(self,
mock_unified_keystone_v3_service__unify_project):
mock_unify_project = mock_unified_keystone_v3_service__unify_project
name = "name"
self.assertEqual(mock_unify_project.return_value,
self.service.create_project(name))
mock_unify_project.assert_called_once_with(
self.service._impl.create_project.return_value)
self.service._impl.create_project.assert_called_once_with(name)
def test_delete_project(self):
project_id = "fake_id"
self.service.delete_project(project_id)
self.service._impl.delete_project.assert_called_once_with(project_id)
@mock.patch("%s.UnifiedKeystoneV3Service._unify_project" % PATH)
def test_list_projects(self,
mock_unified_keystone_v3_service__unify_project):
mock_unify_project = mock_unified_keystone_v3_service__unify_project
projects = [mock.MagicMock()]
self.service._impl.list_projects.return_value = projects
self.assertEqual([mock_unify_project.return_value],
self.service.list_projects())
mock_unify_project.assert_called_once_with(projects[0])
@mock.patch("%s.UnifiedKeystoneV3Service._unify_user" % PATH)
def test_create_user(self, mock_unified_keystone_v3_service__unify_user):
mock_unify_user = mock_unified_keystone_v3_service__unify_user
name = "name"
password = "passwd"
email = "rally@example.com"
project_id = "project"
domain_name = "domain"
default_role = "role"
self.assertEqual(mock_unify_user.return_value,
self.service.create_user(name, password=password,
email=email,
project_id=project_id,
domain_name=domain_name,
default_role=default_role))
mock_unify_user.assert_called_once_with(
self.service._impl.create_user.return_value)
self.service._impl.create_user.assert_called_once_with(
username=name, password=password, email=email,
project_id=project_id, domain_name=domain_name,
default_role=default_role)
def test_delete_user(self):
user_id = "fake_id"
self.service.delete_user(user_id)
self.service._impl.delete_user.assert_called_once_with(user_id)
@mock.patch("%s.UnifiedKeystoneV3Service._unify_user" % PATH)
def test_list_users(self, mock_unified_keystone_v3_service__unify_user):
mock_unify_user = mock_unified_keystone_v3_service__unify_user
users = [mock.MagicMock()]
self.service._impl.list_users.return_value = users
self.assertEqual([mock_unify_user.return_value],
self.service.list_users())
mock_unify_user.assert_called_once_with(users[0])
def test_delete_service(self):
service_id = "fake_id"
self.service.delete_service(service_id)
self.service._impl.delete_service.assert_called_once_with(service_id)
@mock.patch("%s.UnifiedKeystoneV3Service._unify_service" % PATH)
def test_list_services(self,
mock_unified_keystone_v3_service__unify_service):
mock_unify_service = mock_unified_keystone_v3_service__unify_service
services = [mock.MagicMock()]
self.service._impl.list_services.return_value = services
self.assertEqual([mock_unify_service.return_value],
self.service.list_services())
mock_unify_service.assert_called_once_with(services[0])
@mock.patch("%s.UnifiedKeystoneV3Service._unify_role" % PATH)
def test_add_role(self, mock_unified_keystone_v3_service__unify_role):
mock_unify_role = mock_unified_keystone_v3_service__unify_role
role_id = "fake_id"
user_id = "user_id"
project_id = "user_id"
self.assertEqual(mock_unify_role.return_value,
self.service.add_role(role_id, user_id=user_id,
project_id=project_id))
self.service._impl.add_role.assert_called_once_with(
user_id=user_id, role_id=role_id, project_id=project_id)
mock_unify_role.assert_called_once_with(
self.service._impl.add_role.return_value)
def test_delete_role(self):
role_id = "fake_id"
self.service.delete_role(role_id)
self.service._impl.delete_role.assert_called_once_with(role_id)
def test_revoke_role(self):
role_id = "fake_id"
user_id = "user_id"
project_id = "user_id"
self.service.revoke_role(role_id, user_id=user_id,
project_id=project_id)
self.service._impl.revoke_role.assert_called_once_with(
user_id=user_id, role_id=role_id, project_id=project_id)
@mock.patch("%s.UnifiedKeystoneV3Service._unify_role" % PATH)
def test_list_roles(self, mock_unified_keystone_v3_service__unify_role):
mock_unify_role = mock_unified_keystone_v3_service__unify_role
roles = [mock.MagicMock()]
self.service._impl.list_roles.return_value = roles
self.assertEqual([mock_unify_role.return_value],
self.service.list_roles())
mock_unify_role.assert_called_once_with(roles[0])
def test_get_role(self):
role_id = "fake_id"
self.service.get_role(role_id)
self.service._impl.get_role.assert_called_once_with(role_id)

5
tests/unit/test.py
View File

@ -111,6 +111,9 @@ class ScenarioTestCase(TestCase):
mock.Mock(side_effect=self.admin_clients))
]
def get_test_context(self):
return get_test_context()
def setUp(self):
super(ScenarioTestCase, self).setUp()
if self.patch_benchmark_utils:
@ -139,7 +142,7 @@ class ScenarioTestCase(TestCase):
for patcher in self._client_mocks:
patcher.start()
self.context = get_test_context()
self.context = self.get_test_context()
def tearDown(self):
for patcher in self._client_mocks: