openstack
/
keystone
Code Issues Proposed changes

Stop calling deprecated assignment manager methods 

Several methods are deprecated in assignment.Manager since they've
been moved to the role and resource managers, but there were still
several places where the old deprecated methods were being called.

The places where the deprecated methods are called (except in the
case of testing deprecated support) are changed to use the new
location.

Closes-Bug: 1482660
Change-Id: Ic88401a6d0458aa227c15ab93a5b6a36436da33d
This commit is contained in:
Brant Knudson 2015-08-07 08:19:06 -05:00
parent f7bf1c0281
commit 1a615fb5e2
7 changed files with 91 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
keystone/assignment/core.py
View File

@ -158,7 +158,8 @@ class Manager(manager.Manager):
except (exception.MetadataNotFound, exception.NotImplemented):
pass
# As well inherited roles from parent projects
for p in self.list_project_parents(project_ref['id']):
for p in self.resource_api.list_project_parents(
project_ref['id']):
p_roles = self.list_grants(
user_id=user_id, project_id=p['id'],
inherited_to_projects=True)
@ -210,7 +211,7 @@ class Manager(manager.Manager):
return self._roles_from_role_dicts(
metadata_ref.get('roles', {}), False)
self.get_domain(domain_id)
self.resource_api.get_domain(domain_id)
user_role_list = _get_user_domain_roles(user_id, domain_id)
group_role_list = _get_group_domain_roles(user_id, domain_id)
# Use set() to process the list to remove any duplicates

16
keystone/auth/plugins/mapped.py
View File

@ -32,8 +32,8 @@ LOG = log.getLogger(__name__)
METHOD_NAME = 'mapped'
@dependency.requires('assignment_api', 'federation_api', 'identity_api',
'token_provider_api')
@dependency.requires('federation_api', 'identity_api',
'resource_api', 'token_provider_api')
class Mapped(auth.AuthMethodHandler):
def _get_token_ref(self, auth_payload):
@ -65,7 +65,7 @@ class Mapped(auth.AuthMethodHandler):
self.token_provider_api)
else:
handle_unscoped_token(context, auth_payload, auth_context,
self.assignment_api, self.federation_api,
self.resource_api, self.federation_api,
self.identity_api)
@ -105,7 +105,7 @@ def handle_scoped_token(context, auth_payload, auth_context, token_ref,
def handle_unscoped_token(context, auth_payload, auth_context,
assignment_api, federation_api, identity_api):
resource_api, federation_api, identity_api):
def is_ephemeral_user(mapped_properties):
return mapped_properties['user']['type'] == utils.UserType.EPHEMERAL
@ -140,7 +140,7 @@ def handle_unscoped_token(context, auth_payload, auth_context,
try:
mapped_properties, mapping_id = apply_mapping_filter(
identity_provider, protocol, assertion, assignment_api,
identity_provider, protocol, assertion, resource_api,
federation_api, identity_api)
if is_ephemeral_user(mapped_properties):
@ -180,7 +180,7 @@ def extract_assertion_data(context):
def apply_mapping_filter(identity_provider, protocol, assertion,
assignment_api, federation_api, identity_api):
resource_api, federation_api, identity_api):
idp = federation_api.get_idp(identity_provider)
utils.validate_idp(idp, protocol, assertion)
@ -191,7 +191,7 @@ def apply_mapping_filter(identity_provider, protocol, assertion,
# groups identified by name/domain twice.
# NOTE(marek-denis): Groups are translated from name/domain to their
# corresponding ids in the auth plugin, as we need information what
# ``mapping_id`` was used as well as idenity_api and assignment_api
# ``mapping_id`` was used as well as idenity_api and resource_api
# objects.
group_ids = mapped_properties['group_ids']
utils.validate_groups_in_backend(group_ids,
@ -200,7 +200,7 @@ def apply_mapping_filter(identity_provider, protocol, assertion,
group_ids.extend(
utils.transform_to_group_ids(
mapped_properties['group_names'], mapping_id,
identity_api, assignment_api))
identity_api, resource_api))
mapped_properties['group_ids'] = list(set(group_ids))
return mapped_properties, mapping_id

6
keystone/contrib/federation/utils.py
View File

@ -287,7 +287,7 @@ def validate_groups(group_ids, mapping_id, identity_api):
# TODO(marek-denis): Optimize this function, so the number of calls to the
# backend are minimized.
def transform_to_group_ids(group_names, mapping_id,
identity_api, assignment_api):
identity_api, resource_api):
"""Transform groups identitified by name/domain to their ids
Function accepts list of groups identified by a name and domain giving
@ -318,7 +318,7 @@ def transform_to_group_ids(group_names, mapping_id,
:type mapping_id: str
:param identity_api: identity_api object
:param assignment_api: assignment_api object
:param resource_api: resource manager object
:returns: generator object with group ids
@ -339,7 +339,7 @@ def transform_to_group_ids(group_names, mapping_id,
"""
domain_id = (domain.get('id') or
assignment_api.get_domain_by_name(
resource_api.get_domain_by_name(
domain.get('name')).get('id'))
return domain_id

44
keystone/tests/unit/common/test_notifications.py
View File

@ -321,7 +321,7 @@ class NotificationsForEntities(BaseNotificationTest):
def test_create_project(self):
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
self._assert_last_note(
project_ref['id'], CREATED_OPERATION, 'project')
self._assert_last_audit(project_ref['id'], CREATED_OPERATION,
@ -368,8 +368,8 @@ class NotificationsForEntities(BaseNotificationTest):
def test_delete_project(self):
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.assignment_api.delete_project(project_ref['id'])
self.resource_api.create_project(project_ref['id'], project_ref)
self.resource_api.delete_project(project_ref['id'])
self._assert_last_note(
project_ref['id'], DELETED_OPERATION, 'project')
self._assert_last_audit(project_ref['id'], DELETED_OPERATION,
@ -400,19 +400,19 @@ class NotificationsForEntities(BaseNotificationTest):
def test_update_domain(self):
domain_ref = self.new_domain_ref()
self.assignment_api.create_domain(domain_ref['id'], domain_ref)
self.resource_api.create_domain(domain_ref['id'], domain_ref)
domain_ref['description'] = uuid.uuid4().hex
self.assignment_api.update_domain(domain_ref['id'], domain_ref)
self.resource_api.update_domain(domain_ref['id'], domain_ref)
self._assert_last_note(domain_ref['id'], UPDATED_OPERATION, 'domain')
self._assert_last_audit(domain_ref['id'], UPDATED_OPERATION, 'domain',
cadftaxonomy.SECURITY_DOMAIN)
def test_delete_domain(self):
domain_ref = self.new_domain_ref()
self.assignment_api.create_domain(domain_ref['id'], domain_ref)
self.resource_api.create_domain(domain_ref['id'], domain_ref)
domain_ref['enabled'] = False
self.assignment_api.update_domain(domain_ref['id'], domain_ref)
self.assignment_api.delete_domain(domain_ref['id'])
self.resource_api.update_domain(domain_ref['id'], domain_ref)
self.resource_api.delete_domain(domain_ref['id'])
self._assert_last_note(domain_ref['id'], DELETED_OPERATION, 'domain')
self._assert_last_audit(domain_ref['id'], DELETED_OPERATION, 'domain',
cadftaxonomy.SECURITY_DOMAIN)
@ -539,19 +539,19 @@ class NotificationsForEntities(BaseNotificationTest):
def test_disable_domain(self):
domain_ref = self.new_domain_ref()
self.assignment_api.create_domain(domain_ref['id'], domain_ref)
self.resource_api.create_domain(domain_ref['id'], domain_ref)
domain_ref['enabled'] = False
self.assignment_api.update_domain(domain_ref['id'], domain_ref)
self.resource_api.update_domain(domain_ref['id'], domain_ref)
self._assert_notify_sent(domain_ref['id'], 'disabled', 'domain',
public=False)
def test_disable_of_disabled_domain_does_not_notify(self):
domain_ref = self.new_domain_ref()
domain_ref['enabled'] = False
self.assignment_api.create_domain(domain_ref['id'], domain_ref)
self.resource_api.create_domain(domain_ref['id'], domain_ref)
# The domain_ref above is not changed during the create process. We
# can use the same ref to perform the update.
self.assignment_api.update_domain(domain_ref['id'], domain_ref)
self.resource_api.update_domain(domain_ref['id'], domain_ref)
self._assert_notify_not_sent(domain_ref['id'], 'disabled', 'domain',
public=False)
@ -565,8 +565,8 @@ class NotificationsForEntities(BaseNotificationTest):
def test_update_project(self):
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.assignment_api.update_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
self.resource_api.update_project(project_ref['id'], project_ref)
self._assert_notify_sent(
project_ref['id'], UPDATED_OPERATION, 'project', public=True)
self._assert_last_audit(project_ref['id'], UPDATED_OPERATION,
@ -574,27 +574,27 @@ class NotificationsForEntities(BaseNotificationTest):
def test_disable_project(self):
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
project_ref['enabled'] = False
self.assignment_api.update_project(project_ref['id'], project_ref)
self.resource_api.update_project(project_ref['id'], project_ref)
self._assert_notify_sent(project_ref['id'], 'disabled', 'project',
public=False)
def test_disable_of_disabled_project_does_not_notify(self):
project_ref = self.new_project_ref(domain_id=self.domain_id)
project_ref['enabled'] = False
self.assignment_api.create_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
# The project_ref above is not changed during the create process. We
# can use the same ref to perform the update.
self.assignment_api.update_project(project_ref['id'], project_ref)
self.resource_api.update_project(project_ref['id'], project_ref)
self._assert_notify_not_sent(project_ref['id'], 'disabled', 'project',
public=False)
def test_update_project_does_not_send_disable(self):
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
project_ref['enabled'] = True
self.assignment_api.update_project(project_ref['id'], project_ref)
self.resource_api.update_project(project_ref['id'], project_ref)
self._assert_last_note(
project_ref['id'], UPDATED_OPERATION, 'project')
self._assert_notify_not_sent(project_ref['id'], 'disabled', 'project')
@ -662,7 +662,7 @@ class TestEventCallbacks(test_v3.RestfulTestCase):
def test_notification_received(self):
callback = register_callback(CREATED_OPERATION, 'project')
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
self.assertTrue(callback.called)
def test_notification_method_not_callable(self):
@ -707,7 +707,7 @@ class TestEventCallbacks(test_v3.RestfulTestCase):
Foo()
project_ref = self.new_project_ref(domain_id=self.domain_id)
self.assignment_api.create_project(project_ref['id'], project_ref)
self.resource_api.create_project(project_ref['id'], project_ref)
self.assertEqual([True], callback_called)
def test_invalid_event_callbacks(self):

30
keystone/tests/unit/filtering.py
View File

@ -42,19 +42,49 @@ class FilterTests(object):
self.assertTrue(found)
def _create_entity(self, entity_type):
"""Find the create_<entity_type> method.
Searches through the [identity_api, resource_api, assignment_api]
managers for a method called create_<entity_type> and returns the first
one.
"""
f = getattr(self.identity_api, 'create_%s' % entity_type, None)
if f is None:
f = getattr(self.resource_api, 'create_%s' % entity_type, None)
if f is None:
f = getattr(self.assignment_api, 'create_%s' % entity_type)
return f
def _delete_entity(self, entity_type):
"""Find the delete_<entity_type> method.
Searches through the [identity_api, resource_api, assignment_api]
managers for a method called delete_<entity_type> and returns the first
one.
"""
f = getattr(self.identity_api, 'delete_%s' % entity_type, None)
if f is None:
f = getattr(self.resource_api, 'delete_%s' % entity_type, None)
if f is None:
f = getattr(self.assignment_api, 'delete_%s' % entity_type)
return f
def _list_entities(self, entity_type):
"""Find the list_<entity_type> method.
Searches through the [identity_api, resource_api, assignment_api]
managers for a method called list_<entity_type> and returns the first
one.
"""
f = getattr(self.identity_api, 'list_%ss' % entity_type, None)
if f is None:
f = getattr(self.resource_api, 'list_%ss' % entity_type, None)
if f is None:
f = getattr(self.assignment_api, 'list_%ss' % entity_type)
return f

42
keystone/tests/unit/test_v3_assignment.py
View File

@ -422,26 +422,26 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
for domain in create_domains():
self.assertRaises(
AssertionError, self.assignment_api.create_domain,
AssertionError, self.resource_api.create_domain,
domain['id'], domain)
self.assertRaises(
AssertionError, self.assignment_api.update_domain,
AssertionError, self.resource_api.update_domain,
domain['id'], domain)
self.assertRaises(
exception.DomainNotFound, self.assignment_api.delete_domain,
exception.DomainNotFound, self.resource_api.delete_domain,
domain['id'])
# swap 'name' with 'id' and try again, expecting the request to
# gracefully fail
domain['id'], domain['name'] = domain['name'], domain['id']
self.assertRaises(
AssertionError, self.assignment_api.create_domain,
AssertionError, self.resource_api.create_domain,
domain['id'], domain)
self.assertRaises(
AssertionError, self.assignment_api.update_domain,
AssertionError, self.resource_api.update_domain,
domain['id'], domain)
self.assertRaises(
exception.DomainNotFound, self.assignment_api.delete_domain,
exception.DomainNotFound, self.resource_api.delete_domain,
domain['id'])
def test_forbid_operations_on_defined_federated_domain(self):
@ -457,13 +457,13 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
domain = self.new_domain_ref()
domain['name'] = non_default_name
self.assertRaises(AssertionError,
self.assignment_api.create_domain,
self.resource_api.create_domain,
domain['id'], domain)
self.assertRaises(exception.DomainNotFound,
self.assignment_api.delete_domain,
self.resource_api.delete_domain,
domain['id'])
self.assertRaises(AssertionError,
self.assignment_api.update_domain,
self.resource_api.update_domain,
domain['id'], domain)
def test_set_federated_domain_when_config_empty(self):
@ -479,25 +479,25 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
domain = self.new_domain_ref()
domain['id'] = federated_name
self.assertRaises(AssertionError,
self.assignment_api.create_domain,
self.resource_api.create_domain,
domain['id'], domain)
self.assertRaises(exception.DomainNotFound,
self.assignment_api.delete_domain,
self.resource_api.delete_domain,
domain['id'])
self.assertRaises(AssertionError,
self.assignment_api.update_domain,
self.resource_api.update_domain,
domain['id'], domain)
# swap id with name
domain['id'], domain['name'] = domain['name'], domain['id']
self.assertRaises(AssertionError,
self.assignment_api.create_domain,
self.resource_api.create_domain,
domain['id'], domain)
self.assertRaises(exception.DomainNotFound,
self.assignment_api.delete_domain,
self.resource_api.delete_domain,
domain['id'])
self.assertRaises(AssertionError,
self.assignment_api.update_domain,
self.resource_api.update_domain,
domain['id'], domain)
# Project CRUD tests
@ -1761,7 +1761,7 @@ class RoleAssignmentBaseTestCase(test_v3.RestfulTestCase,
for i in range(breadth):
subprojects.append(self.new_project_ref(
domain_id=self.domain_id, parent_id=parent_id))
self.assignment_api.create_project(subprojects[-1]['id'],
self.resource_api.create_project(subprojects[-1]['id'],
subprojects[-1])
new_parent = subprojects[random.randint(0, breadth - 1)]
@ -1772,12 +1772,12 @@ class RoleAssignmentBaseTestCase(test_v3.RestfulTestCase,
# Create a domain
self.domain = self.new_domain_ref()
self.domain_id = self.domain['id']
self.assignment_api.create_domain(self.domain_id, self.domain)
self.resource_api.create_domain(self.domain_id, self.domain)
# Create a project hierarchy
self.project = self.new_project_ref(domain_id=self.domain_id)
self.project_id = self.project['id']
self.assignment_api.create_project(self.project_id, self.project)
self.resource_api.create_project(self.project_id, self.project)
# Create a random project hierarchy
create_project_hierarchy(self.project_id,
@ -1810,7 +1810,7 @@ class RoleAssignmentBaseTestCase(test_v3.RestfulTestCase,
# Create a role
self.role = self.new_role_ref()
self.role_id = self.role['id']
self.assignment_api.create_role(self.role_id, self.role)
self.role_api.create_role(self.role_id, self.role)
# Set default user and group to be used on tests
self.default_user_id = self.user_ids[0]
@ -2106,11 +2106,11 @@ class RoleAssignmentEffectiveTestCase(RoleAssignmentInheritedTestCase):
project_ids = [None]
if filters.get('domain_id'):
project_ids = [project['id'] for project in
self.assignment_api.list_projects_in_domain(
self.resource_api.list_projects_in_domain(
filters.pop('domain_id'))]
else:
project_ids = [project['id'] for project in
self.assignment_api.list_projects_in_subtree(
self.resource_api.list_projects_in_subtree(
self.project_id)]
# Compute expected role assignments

6
keystone/tests/unit/test_v3_auth.py
View File

@ -420,7 +420,7 @@ class AllowRescopeScopedTokenDisabledTests(test_v3.RestfulTestCase):
def test_rescoped_domain_token_disabled(self):
self.domainA = self.new_domain_ref()
self.assignment_api.create_domain(self.domainA['id'], self.domainA)
self.resource_api.create_domain(self.domainA['id'], self.domainA)
self.assignment_api.create_grant(self.role['id'],
user_id=self.user['id'],
domain_id=self.domainA['id'])
@ -2827,7 +2827,7 @@ class TestTrustRedelegation(test_v3.RestfulTestCase):
def test_roles_subset(self):
# Build second role
role = self.new_role_ref()
self.assignment_api.create_role(role['id'], role)
self.role_api.create_role(role['id'], role)
# assign a new role to the user
self.assignment_api.create_grant(role_id=role['id'],
user_id=self.user_id,
@ -2895,7 +2895,7 @@ class TestTrustRedelegation(test_v3.RestfulTestCase):
# Build second trust with a role not in parent's roles
role = self.new_role_ref()
self.assignment_api.create_role(role['id'], role)
self.role_api.create_role(role['id'], role)
# assign a new role to the user
self.assignment_api.create_grant(role_id=role['id'],
user_id=self.user_id,