OpenStack Identity (Keystone)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

4246 lines
188 KiB

# 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 unittest import mock
import uuid
from testtools import matchers
from keystone.common import provider_api
import keystone.conf
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import default_fixtures
CONF = keystone.conf.CONF
PROVIDERS = provider_api.ProviderAPIs
class AssignmentTestHelperMixin(object):
"""Mixin class to aid testing of assignments.
This class supports data driven test plans that enable:
- Creation of initial entities, such as domains, users, groups, projects
and roles
- Creation of assignments referencing the above entities
- A set of input parameters and expected outputs to list_role_assignments
based on the above test data
A test plan is a dict of the form:
test_plan = {
entities: details and number of entities,
group_memberships: group-user entity memberships,
assignments: list of assignments to create,
tests: list of pairs of input params and expected outputs}
An example test plan:
test_plan = {
# First, create the entities required. Entities are specified by
# a dict with the key being the entity type and the value an
# entity specification which can be one of:
#
# - a simple number, e.g. {'users': 3} creates 3 users
# - a dict where more information regarding the contents of the entity
# is required, e.g. {'domains' : {'users : 3}} creates a domain
# with three users
# - a list of entity specifications if multiple are required
#
# The following creates a domain that contains a single user, group and
# project, as well as creating three roles.
'entities': {'domains': {'users': 1, 'groups': 1, 'projects': 1},
'roles': 3},
# If it is required that an existing domain be used for the new
# entities, then the id of that domain can be included in the
# domain dict. For example, if alternatively we wanted to add 3 users
# to the default domain, add a second domain containing 3 projects as
# well as 5 additional empty domains, the entities would be defined as:
#
# 'entities': {'domains': [{'id': DEFAULT_DOMAIN, 'users': 3},
# {'projects': 3}, 5]},
#
# A project hierarchy can be specified within the 'projects' section by
# nesting the 'project' key, for example to create a project with three
# sub-projects you would use:
'projects': {'project': 3}
# A more complex hierarchy can also be defined, for example the
# following would define three projects each containing a
# sub-project, each of which contain a further three sub-projects.
'projects': [{'project': {'project': 3}},
{'project': {'project': 3}},
{'project': {'project': 3}}]
# If the 'roles' entity count is defined as top level key in 'entities'
# dict then these are global roles. If it is placed within the
# 'domain' dict, then they will be domain specific roles. A mix of
# domain specific and global roles are allowed, with the role index
# being calculated in the order they are defined in the 'entities'
# dict.
# A set of implied role specifications. In this case, prior role
# index 0 implies role index 1, and role 1 implies roles 2 and 3.
'roles': [{'role': 0, 'implied_roles': [1]},
{'role': 1, 'implied_roles': [2, 3]}]
# A list of groups and their members. In this case make users with
# index 0 and 1 members of group with index 0. Users and Groups are
# indexed in the order they appear in the 'entities' key above.
'group_memberships': [{'group': 0, 'users': [0, 1]}]
# Next, create assignments between the entities, referencing the
# entities by index, i.e. 'user': 0 refers to user[0]. Entities are
# indexed in the order they appear in the 'entities' key above within
# their entity type.
'assignments': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'project': 0},
{'group': 0, 'role': 2, 'domain': 0},
{'user': 0, 'role': 2, 'project': 0}],
# Finally, define an array of tests where list_role_assignment() is
# called with the given input parameters and the results are then
# confirmed to be as given in 'results'. Again, all entities are
# referenced by index.
'tests': [
{'params': {},
'results': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'project': 0},
{'group': 0, 'role': 2, 'domain': 0},
{'user': 0, 'role': 2, 'project': 0}]},
{'params': {'role': 2},
'results': [{'group': 0, 'role': 2, 'domain': 0},
{'user': 0, 'role': 2, 'project': 0}]}]
# The 'params' key also supports the 'effective',
# 'inherited_to_projects' and 'source_from_group_ids' options to
# list_role_assignments.}
"""
def _handle_project_spec(self, test_data, domain_id, project_spec,
parent_id=None):
"""Handle the creation of a project or hierarchy of projects.
project_spec may either be a count of the number of projects to
create, or it may be a list of the form:
[{'project': project_spec}, {'project': project_spec}, ...]
This method is called recursively to handle the creation of a
hierarchy of projects.
"""
def _create_project(domain_id, parent_id):
new_project = unit.new_project_ref(domain_id=domain_id,
parent_id=parent_id)
new_project = PROVIDERS.resource_api.create_project(
new_project['id'], new_project
)
return new_project
if isinstance(project_spec, list):
for this_spec in project_spec:
self._handle_project_spec(
test_data, domain_id, this_spec, parent_id=parent_id)
elif isinstance(project_spec, dict):
new_proj = _create_project(domain_id, parent_id)
test_data['projects'].append(new_proj)
self._handle_project_spec(
test_data, domain_id, project_spec['project'],
parent_id=new_proj['id'])
else:
for _ in range(project_spec):
test_data['projects'].append(
_create_project(domain_id, parent_id))
def _create_role(self, domain_id=None):
new_role = unit.new_role_ref(domain_id=domain_id)
return PROVIDERS.role_api.create_role(new_role['id'], new_role)
def _handle_domain_spec(self, test_data, domain_spec):
"""Handle the creation of domains and their contents.
domain_spec may either be a count of the number of empty domains to
create, a dict describing the domain contents, or a list of
domain_specs.
In the case when a list is provided, this method calls itself
recursively to handle the list elements.
This method will insert any entities created into test_data
"""
def _create_domain(domain_id=None):
if domain_id is None:
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(
new_domain['id'], new_domain
)
return new_domain
else:
# The test plan specified an existing domain to use
return PROVIDERS.resource_api.get_domain(domain_id)
def _create_entity_in_domain(entity_type, domain_id):
"""Create a user or group entity in the domain."""
if entity_type == 'users':
new_entity = unit.new_user_ref(domain_id=domain_id)
new_entity = PROVIDERS.identity_api.create_user(new_entity)
elif entity_type == 'groups':
new_entity = unit.new_group_ref(domain_id=domain_id)
new_entity = PROVIDERS.identity_api.create_group(new_entity)
elif entity_type == 'roles':
new_entity = self._create_role(domain_id=domain_id)
else:
# Must be a bad test plan
raise exception.NotImplemented()
return new_entity
if isinstance(domain_spec, list):
for x in domain_spec:
self._handle_domain_spec(test_data, x)
elif isinstance(domain_spec, dict):
# If there is a domain ID specified, then use it
the_domain = _create_domain(domain_spec.get('id'))
test_data['domains'].append(the_domain)
for entity_type, value in domain_spec.items():
if entity_type == 'id':
# We already used this above to determine whether to
# use and existing domain
continue
if entity_type == 'projects':
# If it's projects, we need to handle the potential
# specification of a project hierarchy
self._handle_project_spec(
test_data, the_domain['id'], value)
else:
# It's a count of number of entities
for _ in range(value):
test_data[entity_type].append(
_create_entity_in_domain(
entity_type, the_domain['id']))
else:
for _ in range(domain_spec):
test_data['domains'].append(_create_domain())
def create_entities(self, entity_pattern):
"""Create the entities specified in the test plan.
Process the 'entities' key in the test plan, creating the requested
entities. Each created entity will be added to the array of entities
stored in the returned test_data object, e.g.:
test_data['users'] = [user[0], user[1]....]
"""
test_data = {}
for entity in ['users', 'groups', 'domains', 'projects', 'roles']:
test_data[entity] = []
# Create any domains requested and, if specified, any entities within
# those domains
if 'domains' in entity_pattern:
self._handle_domain_spec(test_data, entity_pattern['domains'])
# Create any roles requested
if 'roles' in entity_pattern:
for _ in range(entity_pattern['roles']):
test_data['roles'].append(self._create_role())
return test_data
def _convert_entity_shorthand(self, key, shorthand_data, reference_data):
"""Convert a shorthand entity description into a full ID reference.
In test plan definitions, we allow a shorthand for referencing to an
entity of the form:
'user': 0
which is actually shorthand for:
'user_id': reference_data['users'][0]['id']
This method converts the shorthand version into the full reference.
"""
expanded_key = '%s_id' % key
reference_index = '%ss' % key
index_value = (
reference_data[reference_index][shorthand_data[key]]['id'])
return expanded_key, index_value
def create_implied_roles(self, implied_pattern, test_data):
"""Create the implied roles specified in the test plan."""
for implied_spec in implied_pattern:
# Each implied role specification is a dict of the form:
#
# {'role': 0, 'implied_roles': list of roles}
prior_role = test_data['roles'][implied_spec['role']]['id']
if isinstance(implied_spec['implied_roles'], list):
for this_role in implied_spec['implied_roles']:
implied_role = test_data['roles'][this_role]['id']
PROVIDERS.role_api.create_implied_role(
prior_role, implied_role
)
else:
implied_role = (
test_data['roles'][implied_spec['implied_roles']]['id'])
PROVIDERS.role_api.create_implied_role(
prior_role, implied_role
)
def create_group_memberships(self, group_pattern, test_data):
"""Create the group memberships specified in the test plan."""
for group_spec in group_pattern:
# Each membership specification is a dict of the form:
#
# {'group': 0, 'users': [list of user indexes]}
#
# Add all users in the list to the specified group, first
# converting from index to full entity ID.
group_value = test_data['groups'][group_spec['group']]['id']
for user_index in group_spec['users']:
user_value = test_data['users'][user_index]['id']
PROVIDERS.identity_api.add_user_to_group(
user_value, group_value
)
return test_data
def create_assignments(self, assignment_pattern, test_data):
"""Create the assignments specified in the test plan."""
# First store how many assignments are already in the system,
# so during the tests we can check the number of new assignments
# created.
test_data['initial_assignment_count'] = (
len(PROVIDERS.assignment_api.list_role_assignments()))
# Now create the new assignments in the test plan
for assignment in assignment_pattern:
# Each assignment is a dict of the form:
#
# { 'user': 0, 'project':1, 'role': 6}
#
# where the value of each item is the index into the array of
# entities created earlier.
#
# We process the assignment dict to create the args required to
# make the create_grant() call.
args = {}
for param in assignment:
if param == 'inherited_to_projects':
args[param] = assignment[param]
else:
# Turn 'entity : 0' into 'entity_id = ac6736ba873d'
# where entity in user, group, project or domain
key, value = self._convert_entity_shorthand(
param, assignment, test_data)
args[key] = value
PROVIDERS.assignment_api.create_grant(**args)
return test_data
def execute_assignment_cases(self, test_plan, test_data):
"""Execute the test plan, based on the created test_data."""
def check_results(expected, actual, param_arg_count):
if param_arg_count == 0:
# It was an unfiltered call, so default fixture assignments
# might be polluting our answer - so we take into account
# how many assignments there were before the test.
self.assertEqual(
len(expected) + test_data['initial_assignment_count'],
len(actual))
else:
self.assertThat(actual, matchers.HasLength(len(expected)))
for each_expected in expected:
expected_assignment = {}
for param in each_expected:
if param == 'inherited_to_projects':
expected_assignment[param] = each_expected[param]
elif param == 'indirect':
# We're expecting the result to contain an indirect
# dict with the details how the role came to be placed
# on this entity - so convert the key/value pairs of
# that dict into real entity references.
indirect_term = {}
for indirect_param in each_expected[param]:
key, value = self._convert_entity_shorthand(
indirect_param, each_expected[param],
test_data)
indirect_term[key] = value
expected_assignment[param] = indirect_term
else:
# Convert a simple shorthand entry into a full
# entity reference
key, value = self._convert_entity_shorthand(
param, each_expected, test_data)
expected_assignment[key] = value
self.assertIn(expected_assignment, actual)
def convert_group_ids_sourced_from_list(index_list, reference_data):
value_list = []
for group_index in index_list:
value_list.append(
reference_data['groups'][group_index]['id'])
return value_list
# Go through each test in the array, processing the input params, which
# we build into an args dict, and then call list_role_assignments. Then
# check the results against those specified in the test plan.
for test in test_plan.get('tests', []):
args = {}
for param in test['params']:
if param in ['effective', 'inherited', 'include_subtree']:
# Just pass the value into the args
args[param] = test['params'][param]
elif param == 'source_from_group_ids':
# Convert the list of indexes into a list of IDs
args[param] = convert_group_ids_sourced_from_list(
test['params']['source_from_group_ids'], test_data)
else:
# Turn 'entity : 0' into 'entity_id = ac6736ba873d'
# where entity in user, group, project or domain
key, value = self._convert_entity_shorthand(
param, test['params'], test_data)
args[key] = value
results = PROVIDERS.assignment_api.list_role_assignments(**args)
check_results(test['results'], results, len(args))
def execute_assignment_plan(self, test_plan):
"""Create entities, assignments and execute the test plan.
The standard method to call to create entities and assignments and
execute the tests as specified in the test_plan. The test_data
dict is returned so that, if required, the caller can execute
additional manual tests with the entities and assignments created.
"""
test_data = self.create_entities(test_plan['entities'])
if 'implied_roles' in test_plan:
self.create_implied_roles(test_plan['implied_roles'], test_data)
if 'group_memberships' in test_plan:
self.create_group_memberships(test_plan['group_memberships'],
test_data)
if 'assignments' in test_plan:
test_data = self.create_assignments(test_plan['assignments'],
test_data)
self.execute_assignment_cases(test_plan, test_data)
return test_data
class AssignmentTests(AssignmentTestHelperMixin):
def _get_domain_fixture(self):
domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain['id'], domain)
return domain
def test_project_add_and_remove_user_role(self):
user_ids = PROVIDERS.assignment_api.list_user_ids_for_project(
self.project_bar['id'])
self.assertNotIn(self.user_two['id'], user_ids)
PROVIDERS.assignment_api.add_role_to_user_and_project(
project_id=self.project_bar['id'],
user_id=self.user_two['id'],
role_id=self.role_other['id'])
user_ids = PROVIDERS.assignment_api.list_user_ids_for_project(
self.project_bar['id'])
self.assertIn(self.user_two['id'], user_ids)
PROVIDERS.assignment_api.remove_role_from_user_and_project(
project_id=self.project_bar['id'],
user_id=self.user_two['id'],
role_id=self.role_other['id'])
user_ids = PROVIDERS.assignment_api.list_user_ids_for_project(
self.project_bar['id'])
self.assertNotIn(self.user_two['id'], user_ids)
def test_remove_user_role_not_assigned(self):
# Expect failure if attempt to remove a role that was never assigned to
# the user.
self.assertRaises(exception.RoleNotFound,
PROVIDERS.assignment_api.
remove_role_from_user_and_project,
project_id=self.project_bar['id'],
user_id=self.user_two['id'],
role_id=self.role_other['id'])
def test_list_user_ids_for_project(self):
user_ids = PROVIDERS.assignment_api.list_user_ids_for_project(
self.project_baz['id'])
self.assertEqual(2, len(user_ids))
self.assertIn(self.user_two['id'], user_ids)
self.assertIn(self.user_badguy['id'], user_ids)
def test_list_user_ids_for_project_no_duplicates(self):
# Create user
user_ref = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user_ref = PROVIDERS.identity_api.create_user(user_ref)
# Create project
project_ref = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id)
PROVIDERS.resource_api.create_project(
project_ref['id'], project_ref)
# Create 2 roles and give user each role in project
for i in range(2):
role_ref = unit.new_role_ref()
PROVIDERS.role_api.create_role(role_ref['id'], role_ref)
PROVIDERS.assignment_api.add_role_to_user_and_project(
user_id=user_ref['id'],
project_id=project_ref['id'],
role_id=role_ref['id'])
# Get the list of user_ids in project
user_ids = PROVIDERS.assignment_api.list_user_ids_for_project(
project_ref['id'])
# Ensure the user is only returned once
self.assertEqual(1, len(user_ids))
def test_get_project_user_ids_returns_not_found(self):
self.assertRaises(exception.ProjectNotFound,
PROVIDERS.assignment_api.list_user_ids_for_project,
uuid.uuid4().hex)
def test_list_role_assignments_unfiltered(self):
"""Test unfiltered listing of role assignments."""
test_plan = {
# Create a domain, with a user, group & project
'entities': {'domains': {'users': 1, 'groups': 1, 'projects': 1},
'roles': 3},
# Create a grant of each type (user/group on project/domain)
'assignments': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'project': 0},
{'group': 0, 'role': 2, 'domain': 0},
{'group': 0, 'role': 2, 'project': 0}],
'tests': [
# Check that we get back the 4 assignments
{'params': {},
'results': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'project': 0},
{'group': 0, 'role': 2, 'domain': 0},
{'group': 0, 'role': 2, 'project': 0}]}
]
}
self.execute_assignment_plan(test_plan)
def test_list_role_assignments_filtered_by_role(self):
"""Test listing of role assignments filtered by role ID."""
test_plan = {
# Create a user, group & project in the default domain
'entities': {'domains': {'id': CONF.identity.default_domain_id,
'users': 1, 'groups': 1, 'projects': 1},
'roles': 3},
# Create a grant of each type (user/group on project/domain)
'assignments': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'project': 0},
{'group': 0, 'role': 2, 'domain': 0},
{'group': 0, 'role': 2, 'project': 0}],
'tests': [
# Check that when filtering by role, we only get back those
# that match
{'params': {'role': 2},
'results': [{'group': 0, 'role': 2, 'domain': 0},
{'group': 0, 'role': 2, 'project': 0}]}
]
}
self.execute_assignment_plan(test_plan)
def test_list_group_role_assignment(self):
# When a group role assignment is created and the role assignments are
# listed then the group role assignment is included in the list.
test_plan = {
'entities': {'domains': {'id': CONF.identity.default_domain_id,
'groups': 1, 'projects': 1},
'roles': 1},
'assignments': [{'group': 0, 'role': 0, 'project': 0}],
'tests': [
{'params': {},
'results': [{'group': 0, 'role': 0, 'project': 0}]}
]
}
self.execute_assignment_plan(test_plan)
def test_list_role_assignments_bad_role(self):
assignment_list = PROVIDERS.assignment_api.list_role_assignments(
role_id=uuid.uuid4().hex)
self.assertEqual([], assignment_list)
def test_list_role_assignments_user_not_found(self):
def _user_not_found(value):
raise exception.UserNotFound(user_id=value)
# Note(knikolla): Patch get_user to return UserNotFound
# this simulates the possibility of a user being deleted
# directly in the backend and still having lingering role
# assignments.
with mock.patch.object(PROVIDERS.identity_api, 'get_user',
_user_not_found):
assignment_list = PROVIDERS.assignment_api.list_role_assignments(
include_names=True
)
self.assertNotEqual([], assignment_list)
for assignment in assignment_list:
if 'user_name' in assignment:
# Note(knikolla): In the case of a not found user we
# populate the values with empty strings.
self.assertEqual('', assignment['user_name'])
self.assertEqual('', assignment['user_domain_id'])
self.assertEqual('', assignment['user_domain_name'])
def test_list_role_assignments_group_not_found(self):
def _group_not_found(value):
raise exception.GroupNotFound(group_id=value)
# Setup
# 1) Remove any pre-existing assignments so we control what's there
for a in PROVIDERS.assignment_api.list_role_assignments():
PROVIDERS.assignment_api.delete_grant(**a)
# 2) create a group and 2 users in that group
domain_id = CONF.identity.default_domain_id
group = PROVIDERS.identity_api.create_group(
unit.new_group_ref(domain_id=domain_id))
user1 = PROVIDERS.identity_api.create_user(
unit.new_user_ref(domain_id=domain_id))
user2 = PROVIDERS.identity_api.create_user(
unit.new_user_ref(domain_id=domain_id))
PROVIDERS.identity_api.add_user_to_group(user1['id'], group['id'])
PROVIDERS.identity_api.add_user_to_group(user2['id'], group['id'])
# 3) create a role assignment for the group
PROVIDERS.assignment_api.create_grant(
group_id=group['id'],
domain_id=domain_id,
role_id=default_fixtures.MEMBER_ROLE_ID)
num_assignments = len(PROVIDERS.assignment_api.list_role_assignments())
self.assertEqual(1, num_assignments)
# Patch get_group to return GroupNotFound, allowing us to confirm
# that the exception is handled properly when include_names processing
# attempts to lookup a group that has been deleted in the backend
with mock.patch.object(PROVIDERS.identity_api, 'get_group',
_group_not_found):
assignment_list = PROVIDERS.assignment_api.list_role_assignments(
include_names=True
)
self.assertEqual(num_assignments, len(assignment_list))
for assignment in assignment_list:
includes_group_assignments = False
if 'group_name' in assignment:
includes_group_assignments = True
# Note(knikolla): In the case of a not-found group we
# populate the values with empty strings.
self.assertEqual('', assignment['group_name'])
self.assertEqual('', assignment['group_domain_id'])
self.assertEqual('', assignment['group_domain_name'])
self.assertTrue(includes_group_assignments)
num_effective = len(PROVIDERS.assignment_api.list_role_assignments(
effective=True))
self.assertGreater(num_effective, len(assignment_list))
# Patch list_users_in_group to return GroupNotFound allowing us to
# confirm that the exception is handled properly when effective
# processing attempts to lookup users for a group that has been deleted
# in the backend
with mock.patch.object(PROVIDERS.identity_api, 'list_users_in_group',
_group_not_found):
assignment_list = PROVIDERS.assignment_api.list_role_assignments(
effective=True
)
self.assertGreater(num_effective, len(assignment_list))
# cleanup
PROVIDERS.assignment_api.delete_grant(
group_id=group['id'],
domain_id=domain_id,
role_id=default_fixtures.MEMBER_ROLE_ID)
# TODO(edmondsw) should cleanup users/groups as well, but that raises
# LDAP read-only issues
def test_add_duplicate_role_grant(self):
roles_ref = PROVIDERS.assignment_api.get_roles_for_user_and_project(
self.user_foo['id'], self.project_bar['id'])
self.assertNotIn(self.role_admin['id'], roles_ref)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'], self.project_bar['id'], self.role_admin['id'])
self.assertRaises(
exception.Conflict,
PROVIDERS.assignment_api.add_role_to_user_and_project,
self.user_foo['id'],
self.project_bar['id'],
self.role_admin['id']
)
def test_get_role_by_user_and_project_with_user_in_group(self):
"""Test for get role by user and project, user was added into a group.
Test Plan:
- Create a user, a project & a group, add this user to group
- Create roles and grant them to user and project
- Check the role list get by the user and project was as expected
"""
user_ref = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user_ref = PROVIDERS.identity_api.create_user(user_ref)
project_ref = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id)
PROVIDERS.resource_api.create_project(project_ref['id'], project_ref)
group = unit.new_group_ref(domain_id=CONF.identity.default_domain_id)
group_id = PROVIDERS.identity_api.create_group(group)['id']
PROVIDERS.identity_api.add_user_to_group(user_ref['id'], group_id)
role_ref_list = []
for i in range(2):
role_ref = unit.new_role_ref()
PROVIDERS.role_api.create_role(role_ref['id'], role_ref)
role_ref_list.append(role_ref)
PROVIDERS.assignment_api.add_role_to_user_and_project(
user_id=user_ref['id'],
project_id=project_ref['id'],
role_id=role_ref['id'])
role_list = PROVIDERS.assignment_api.get_roles_for_user_and_project(
user_ref['id'],
project_ref['id'])
self.assertEqual(set([r['id'] for r in role_ref_list]),
set(role_list))
def test_get_role_by_user_and_project(self):
roles_ref = PROVIDERS.assignment_api.get_roles_for_user_and_project(
self.user_foo['id'], self.project_bar['id'])
self.assertNotIn(self.role_admin['id'], roles_ref)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'], self.project_bar['id'], self.role_admin['id'])
roles_ref = PROVIDERS.assignment_api.get_roles_for_user_and_project(
self.user_foo['id'], self.project_bar['id'])
self.assertIn(self.role_admin['id'], roles_ref)
self.assertNotIn(default_fixtures.MEMBER_ROLE_ID, roles_ref)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'],
self.project_bar['id'],
default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.get_roles_for_user_and_project(
self.user_foo['id'], self.project_bar['id'])
self.assertIn(self.role_admin['id'], roles_ref)
self.assertIn(default_fixtures.MEMBER_ROLE_ID, roles_ref)
def test_get_role_by_trustor_and_project(self):
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(new_domain['id'], new_domain)
new_user = unit.new_user_ref(domain_id=new_domain['id'])
new_user = PROVIDERS.identity_api.create_user(new_user)
new_project = unit.new_project_ref(domain_id=new_domain['id'])
PROVIDERS.resource_api.create_project(new_project['id'], new_project)
role = self._create_role(domain_id=new_domain['id'])
# Now create the grants (roles are defined in default_fixtures)
PROVIDERS.assignment_api.create_grant(
user_id=new_user['id'],
project_id=new_project['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
PROVIDERS.assignment_api.create_grant(
user_id=new_user['id'],
domain_id=new_domain['id'],
role_id=role['id'],
inherited_to_projects=True)
roles_ids = PROVIDERS.assignment_api.get_roles_for_trustor_and_project(
new_user['id'], new_project['id'])
self.assertEqual(2, len(roles_ids))
self.assertIn(self.role_member['id'], roles_ids)
self.assertIn(role['id'], roles_ids)
def test_get_roles_for_user_and_domain(self):
"""Test for getting roles for user on a domain.
Test Plan:
- Create a domain, with 2 users
- Check no roles yet exit
- Give user1 two roles on the domain, user2 one role
- Get roles on user1 and the domain - maybe sure we only
get back the 2 roles on user1
- Delete both roles from user1
- Check we get no roles back for user1 on domain
"""
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(new_domain['id'], new_domain)
new_user1 = unit.new_user_ref(domain_id=new_domain['id'])
new_user1 = PROVIDERS.identity_api.create_user(new_user1)
new_user2 = unit.new_user_ref(domain_id=new_domain['id'])
new_user2 = PROVIDERS.identity_api.create_user(new_user2)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=new_user1['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
# Now create the grants (roles are defined in default_fixtures)
PROVIDERS.assignment_api.create_grant(
user_id=new_user1['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
PROVIDERS.assignment_api.create_grant(
user_id=new_user1['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.OTHER_ROLE_ID)
PROVIDERS.assignment_api.create_grant(
user_id=new_user2['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.ADMIN_ROLE_ID)
# Read back the roles for user1 on domain
roles_ids = PROVIDERS.assignment_api.get_roles_for_user_and_domain(
new_user1['id'], new_domain['id'])
self.assertEqual(2, len(roles_ids))
self.assertIn(self.role_member['id'], roles_ids)
self.assertIn(self.role_other['id'], roles_ids)
# Now delete both grants for user1
PROVIDERS.assignment_api.delete_grant(
user_id=new_user1['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
PROVIDERS.assignment_api.delete_grant(
user_id=new_user1['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.OTHER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=new_user1['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
def test_get_roles_for_user_and_domain_returns_not_found(self):
"""Test errors raised when getting roles for user on a domain.
Test Plan:
- Check non-existing user gives UserNotFound
- Check non-existing domain gives DomainNotFound
"""
new_domain = self._get_domain_fixture()
new_user1 = unit.new_user_ref(domain_id=new_domain['id'])
new_user1 = PROVIDERS.identity_api.create_user(new_user1)
self.assertRaises(
exception.UserNotFound,
PROVIDERS.assignment_api.get_roles_for_user_and_domain,
uuid.uuid4().hex,
new_domain['id']
)
self.assertRaises(
exception.DomainNotFound,
PROVIDERS.assignment_api.get_roles_for_user_and_domain,
new_user1['id'],
uuid.uuid4().hex
)
def test_get_roles_for_user_and_project_returns_not_found(self):
self.assertRaises(
exception.UserNotFound,
PROVIDERS.assignment_api.get_roles_for_user_and_project,
uuid.uuid4().hex,
self.project_bar['id']
)
self.assertRaises(
exception.ProjectNotFound,
PROVIDERS.assignment_api.get_roles_for_user_and_project,
self.user_foo['id'],
uuid.uuid4().hex
)
def test_add_role_to_user_and_project_returns_not_found(self):
self.assertRaises(
exception.ProjectNotFound,
PROVIDERS.assignment_api.add_role_to_user_and_project,
self.user_foo['id'],
uuid.uuid4().hex,
self.role_admin['id']
)
self.assertRaises(
exception.RoleNotFound,
PROVIDERS.assignment_api.add_role_to_user_and_project,
self.user_foo['id'],
self.project_bar['id'],
uuid.uuid4().hex
)
def test_add_role_to_user_and_project_no_user(self):
# If add_role_to_user_and_project and the user doesn't exist, then
# no error.
user_id_not_exist = uuid.uuid4().hex
PROVIDERS.assignment_api.add_role_to_user_and_project(
user_id_not_exist, self.project_bar['id'], self.role_admin['id'])
def test_remove_role_from_user_and_project(self):
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'],
self.project_bar['id'],
default_fixtures.MEMBER_ROLE_ID)
PROVIDERS.assignment_api.remove_role_from_user_and_project(
self.user_foo['id'],
self.project_bar['id'],
default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.get_roles_for_user_and_project(
self.user_foo['id'], self.project_bar['id'])
self.assertNotIn(default_fixtures.MEMBER_ROLE_ID, roles_ref)
self.assertRaises(exception.NotFound,
PROVIDERS.assignment_api.
remove_role_from_user_and_project,
self.user_foo['id'],
self.project_bar['id'],
default_fixtures.MEMBER_ROLE_ID)
def test_get_role_grant_by_user_and_project(self):
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=self.user_foo['id'],
project_id=self.project_bar['id'])
self.assertEqual(1, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
user_id=self.user_foo['id'], project_id=self.project_bar['id'],
role_id=self.role_admin['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=self.user_foo['id'],
project_id=self.project_bar['id'])
self.assertIn(self.role_admin['id'],
[role_ref['id'] for role_ref in roles_ref])
PROVIDERS.assignment_api.create_grant(
user_id=self.user_foo['id'],
project_id=self.project_bar['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=self.user_foo['id'],
project_id=self.project_bar['id'])
roles_ref_ids = []
for ref in roles_ref:
roles_ref_ids.append(ref['id'])
self.assertIn(self.role_admin['id'], roles_ref_ids)
self.assertIn(default_fixtures.MEMBER_ROLE_ID, roles_ref_ids)
def test_remove_role_grant_from_user_and_project(self):
PROVIDERS.assignment_api.create_grant(
user_id=self.user_foo['id'],
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=self.user_foo['id'],
project_id=self.project_baz['id'])
self.assertDictEqual(self.role_member, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
user_id=self.user_foo['id'],
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=self.user_foo['id'],
project_id=self.project_baz['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
user_id=self.user_foo['id'],
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_role_assignment_by_project_not_found(self):
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.check_grant_role_id,
user_id=self.user_foo['id'],
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.check_grant_role_id,
group_id=uuid.uuid4().hex,
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_role_assignment_by_domain_not_found(self):
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.check_grant_role_id,
user_id=self.user_foo['id'],
domain_id=CONF.identity.default_domain_id,
role_id=default_fixtures.MEMBER_ROLE_ID)
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.check_grant_role_id,
group_id=uuid.uuid4().hex,
domain_id=CONF.identity.default_domain_id,
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_del_role_assignment_by_project_not_found(self):
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
user_id=self.user_foo['id'],
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
group_id=uuid.uuid4().hex,
project_id=self.project_baz['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_del_role_assignment_by_domain_not_found(self):
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
user_id=self.user_foo['id'],
domain_id=CONF.identity.default_domain_id,
role_id=default_fixtures.MEMBER_ROLE_ID)
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
group_id=uuid.uuid4().hex,
domain_id=CONF.identity.default_domain_id,
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_and_remove_role_grant_by_group_and_project(self):
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(new_domain['id'], new_domain)
new_group = unit.new_group_ref(domain_id=new_domain['id'])
new_group = PROVIDERS.identity_api.create_group(new_group)
new_user = unit.new_user_ref(domain_id=new_domain['id'])
new_user = PROVIDERS.identity_api.create_user(new_user)
PROVIDERS.identity_api.add_user_to_group(
new_user['id'], new_group['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
project_id=self.project_bar['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
group_id=new_group['id'],
project_id=self.project_bar['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
project_id=self.project_bar['id'])
self.assertDictEqual(self.role_member, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
group_id=new_group['id'],
project_id=self.project_bar['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
project_id=self.project_bar['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
group_id=new_group['id'],
project_id=self.project_bar['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_and_remove_role_grant_by_group_and_domain(self):
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(new_domain['id'], new_domain)
new_group = unit.new_group_ref(domain_id=new_domain['id'])
new_group = PROVIDERS.identity_api.create_group(new_group)
new_user = unit.new_user_ref(domain_id=new_domain['id'])
new_user = PROVIDERS.identity_api.create_user(new_user)
PROVIDERS.identity_api.add_user_to_group(
new_user['id'], new_group['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
group_id=new_group['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
domain_id=new_domain['id'])
self.assertDictEqual(self.role_member, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
group_id=new_group['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
group_id=new_group['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_and_remove_correct_role_grant_from_a_mix(self):
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(new_domain['id'], new_domain)
new_project = unit.new_project_ref(domain_id=new_domain['id'])
PROVIDERS.resource_api.create_project(new_project['id'], new_project)
new_group = unit.new_group_ref(domain_id=new_domain['id'])
new_group = PROVIDERS.identity_api.create_group(new_group)
new_group2 = unit.new_group_ref(domain_id=new_domain['id'])
new_group2 = PROVIDERS.identity_api.create_group(new_group2)
new_user = unit.new_user_ref(domain_id=new_domain['id'])
new_user = PROVIDERS.identity_api.create_user(new_user)
new_user2 = unit.new_user_ref(domain_id=new_domain['id'])
new_user2 = PROVIDERS.identity_api.create_user(new_user2)
PROVIDERS.identity_api.add_user_to_group(
new_user['id'], new_group['id']
)
# First check we have no grants
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
# Now add the grant we are going to test for, and some others as
# well just to make sure we get back the right one
PROVIDERS.assignment_api.create_grant(
group_id=new_group['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
PROVIDERS.assignment_api.create_grant(
group_id=new_group2['id'], domain_id=new_domain['id'],
role_id=self.role_admin['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=new_user2['id'], domain_id=new_domain['id'],
role_id=self.role_admin['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=new_group['id'], project_id=new_project['id'],
role_id=self.role_admin['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
domain_id=new_domain['id'])
self.assertDictEqual(self.role_member, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
group_id=new_group['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=new_group['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
group_id=new_group['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_and_remove_role_grant_by_user_and_domain(self):
new_domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(new_domain['id'], new_domain)
new_user = unit.new_user_ref(domain_id=new_domain['id'])
new_user = PROVIDERS.identity_api.create_user(new_user)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=new_user['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
user_id=new_user['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=new_user['id'],
domain_id=new_domain['id'])
self.assertDictEqual(self.role_member, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
user_id=new_user['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=new_user['id'],
domain_id=new_domain['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
user_id=new_user['id'],
domain_id=new_domain['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
def test_get_and_remove_role_grant_by_group_and_cross_domain(self):
group1_domain1_role = unit.new_role_ref()
PROVIDERS.role_api.create_role(
group1_domain1_role['id'], group1_domain1_role
)
group1_domain2_role = unit.new_role_ref()
PROVIDERS.role_api.create_role(
group1_domain2_role['id'], group1_domain2_role
)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
domain2 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain2['id'], domain2)
group1 = unit.new_group_ref(domain_id=domain1['id'])
group1 = PROVIDERS.identity_api.create_group(group1)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain1['id'])
self.assertEqual(0, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain2['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], domain_id=domain1['id'],
role_id=group1_domain1_role['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], domain_id=domain2['id'],
role_id=group1_domain2_role['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain1['id'])
self.assertDictEqual(group1_domain1_role, roles_ref[0])
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain2['id'])
self.assertDictEqual(group1_domain2_role, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
group_id=group1['id'], domain_id=domain2['id'],
role_id=group1_domain2_role['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain2['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
group_id=group1['id'],
domain_id=domain2['id'],
role_id=group1_domain2_role['id'])
def test_get_and_remove_role_grant_by_user_and_cross_domain(self):
user1_domain1_role = unit.new_role_ref()
PROVIDERS.role_api.create_role(
user1_domain1_role['id'], user1_domain1_role
)
user1_domain2_role = unit.new_role_ref()
PROVIDERS.role_api.create_role(
user1_domain2_role['id'], user1_domain2_role
)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
domain2 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain2['id'], domain2)
user1 = unit.new_user_ref(domain_id=domain1['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain1['id'])
self.assertEqual(0, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain2['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], domain_id=domain1['id'],
role_id=user1_domain1_role['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], domain_id=domain2['id'],
role_id=user1_domain2_role['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain1['id'])
self.assertDictEqual(user1_domain1_role, roles_ref[0])
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain2['id'])
self.assertDictEqual(user1_domain2_role, roles_ref[0])
PROVIDERS.assignment_api.delete_grant(
user_id=user1['id'], domain_id=domain2['id'],
role_id=user1_domain2_role['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain2['id'])
self.assertEqual(0, len(roles_ref))
self.assertRaises(exception.RoleAssignmentNotFound,
PROVIDERS.assignment_api.delete_grant,
user_id=user1['id'],
domain_id=domain2['id'],
role_id=user1_domain2_role['id'])
def test_role_grant_by_group_and_cross_domain_project(self):
role1 = unit.new_role_ref()
PROVIDERS.role_api.create_role(role1['id'], role1)
role2 = unit.new_role_ref()
PROVIDERS.role_api.create_role(role2['id'], role2)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
domain2 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain2['id'], domain2)
group1 = unit.new_group_ref(domain_id=domain1['id'])
group1 = PROVIDERS.identity_api.create_group(group1)
project1 = unit.new_project_ref(domain_id=domain2['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
project_id=project1['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role1['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role2['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
project_id=project1['id'])
roles_ref_ids = []
for ref in roles_ref:
roles_ref_ids.append(ref['id'])
self.assertIn(role1['id'], roles_ref_ids)
self.assertIn(role2['id'], roles_ref_ids)
PROVIDERS.assignment_api.delete_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role1['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
project_id=project1['id'])
self.assertEqual(1, len(roles_ref))
self.assertDictEqual(role2, roles_ref[0])
def test_role_grant_by_user_and_cross_domain_project(self):
role1 = unit.new_role_ref()
PROVIDERS.role_api.create_role(role1['id'], role1)
role2 = unit.new_role_ref()
PROVIDERS.role_api.create_role(role2['id'], role2)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
domain2 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain2['id'], domain2)
user1 = unit.new_user_ref(domain_id=domain1['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
project1 = unit.new_project_ref(domain_id=domain2['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=project1['id'], role_id=role1['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=project1['id'], role_id=role2['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
roles_ref_ids = []
for ref in roles_ref:
roles_ref_ids.append(ref['id'])
self.assertIn(role1['id'], roles_ref_ids)
self.assertIn(role2['id'], roles_ref_ids)
PROVIDERS.assignment_api.delete_grant(
user_id=user1['id'], project_id=project1['id'], role_id=role1['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
self.assertEqual(1, len(roles_ref))
self.assertDictEqual(role2, roles_ref[0])
def test_delete_user_grant_no_user(self):
# Can delete a grant where the user doesn't exist.
role = unit.new_role_ref()
role_id = role['id']
PROVIDERS.role_api.create_role(role_id, role)
user_id = uuid.uuid4().hex
PROVIDERS.assignment_api.create_grant(
role_id, user_id=user_id, project_id=self.project_bar['id']
)
PROVIDERS.assignment_api.delete_grant(
role_id, user_id=user_id, project_id=self.project_bar['id']
)
def test_delete_group_grant_no_group(self):
# Can delete a grant where the group doesn't exist.
role = unit.new_role_ref()
role_id = role['id']
PROVIDERS.role_api.create_role(role_id, role)
group_id = uuid.uuid4().hex
PROVIDERS.assignment_api.create_grant(
role_id, group_id=group_id, project_id=self.project_bar['id']
)
PROVIDERS.assignment_api.delete_grant(
role_id, group_id=group_id, project_id=self.project_bar['id']
)
def test_grant_crud_throws_exception_if_invalid_role(self):
"""Ensure RoleNotFound thrown if role does not exist."""
def assert_role_not_found_exception(f, **kwargs):
self.assertRaises(exception.RoleNotFound, f,
role_id=uuid.uuid4().hex, **kwargs)
user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user_resp = PROVIDERS.identity_api.create_user(user)
group = unit.new_group_ref(domain_id=CONF.identity.default_domain_id)
group_resp = PROVIDERS.identity_api.create_group(group)
project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id)
project_resp = PROVIDERS.resource_api.create_project(
project['id'], project
)
for manager_call in [PROVIDERS.assignment_api.create_grant,
PROVIDERS.assignment_api.get_grant]:
assert_role_not_found_exception(
manager_call,
user_id=user_resp['id'], project_id=project_resp['id'])
assert_role_not_found_exception(
manager_call,
group_id=group_resp['id'], project_id=project_resp['id'])
assert_role_not_found_exception(
manager_call,
user_id=user_resp['id'],
domain_id=CONF.identity.default_domain_id)
assert_role_not_found_exception(
manager_call,
group_id=group_resp['id'],
domain_id=CONF.identity.default_domain_id)
assert_role_not_found_exception(
PROVIDERS.assignment_api.delete_grant,
user_id=user_resp['id'], project_id=project_resp['id'])
assert_role_not_found_exception(
PROVIDERS.assignment_api.delete_grant,
group_id=group_resp['id'], project_id=project_resp['id'])
assert_role_not_found_exception(
PROVIDERS.assignment_api.delete_grant,
user_id=user_resp['id'],
domain_id=CONF.identity.default_domain_id)
assert_role_not_found_exception(
PROVIDERS.assignment_api.delete_grant,
group_id=group_resp['id'],
domain_id=CONF.identity.default_domain_id)
def test_multi_role_grant_by_user_group_on_project_domain(self):
role_list = []
for _ in range(10):
role = unit.new_role_ref()
PROVIDERS.role_api.create_role(role['id'], role)
role_list.append(role)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
user1 = unit.new_user_ref(domain_id=domain1['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
group1 = unit.new_group_ref(domain_id=domain1['id'])
group1 = PROVIDERS.identity_api.create_group(group1)
group2 = unit.new_group_ref(domain_id=domain1['id'])
group2 = PROVIDERS.identity_api.create_group(group2)
project1 = unit.new_project_ref(domain_id=domain1['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
PROVIDERS.identity_api.add_user_to_group(
user1['id'], group1['id']
)
PROVIDERS.identity_api.add_user_to_group(
user1['id'], group2['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], domain_id=domain1['id'],
role_id=role_list[0]['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], domain_id=domain1['id'],
role_id=role_list[1]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], domain_id=domain1['id'],
role_id=role_list[2]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], domain_id=domain1['id'],
role_id=role_list[3]['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=project1['id'],
role_id=role_list[4]['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=project1['id'],
role_id=role_list[5]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role_list[6]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role_list[7]['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'], domain_id=domain1['id']
)
self.assertEqual(2, len(roles_ref))
self.assertIn(role_list[0], roles_ref)
self.assertIn(role_list[1], roles_ref)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'], domain_id=domain1['id']
)
self.assertEqual(2, len(roles_ref))
self.assertIn(role_list[2], roles_ref)
self.assertIn(role_list[3], roles_ref)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'], project_id=project1['id']
)
self.assertEqual(2, len(roles_ref))
self.assertIn(role_list[4], roles_ref)
self.assertIn(role_list[5], roles_ref)
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'], project_id=project1['id']
)
self.assertEqual(2, len(roles_ref))
self.assertIn(role_list[6], roles_ref)
self.assertIn(role_list[7], roles_ref)
# Now test the alternate way of getting back lists of grants,
# where user and group roles are combined. These should match
# the above results.
combined_list = (
PROVIDERS.assignment_api.get_roles_for_user_and_project(
user1['id'], project1['id']
)
)
self.assertEqual(4, len(combined_list))
self.assertIn(role_list[4]['id'], combined_list)
self.assertIn(role_list[5]['id'], combined_list)
self.assertIn(role_list[6]['id'], combined_list)
self.assertIn(role_list[7]['id'], combined_list)
combined_role_list = (
PROVIDERS.assignment_api.get_roles_for_user_and_domain(
user1['id'], domain1['id']
)
)
self.assertEqual(4, len(combined_role_list))
self.assertIn(role_list[0]['id'], combined_role_list)
self.assertIn(role_list[1]['id'], combined_role_list)
self.assertIn(role_list[2]['id'], combined_role_list)
self.assertIn(role_list[3]['id'], combined_role_list)
def test_multi_group_grants_on_project_domain(self):
"""Test multiple group roles for user on project and domain.
Test Plan:
- Create 6 roles
- Create a domain, with a project, user and two groups
- Make the user a member of both groups
- Check no roles yet exit
- Assign a role to each user and both groups on both the
project and domain
- Get a list of effective roles for the user on both the
project and domain, checking we get back the correct three
roles
"""
role_list = []
for _ in range(6):
role = unit.new_role_ref()
PROVIDERS.role_api.create_role(role['id'], role)
role_list.append(role)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
user1 = unit.new_user_ref(domain_id=domain1['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
group1 = unit.new_group_ref(domain_id=domain1['id'])
group1 = PROVIDERS.identity_api.create_group(group1)
group2 = unit.new_group_ref(domain_id=domain1['id'])
group2 = PROVIDERS.identity_api.create_group(group2)
project1 = unit.new_project_ref(domain_id=domain1['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
PROVIDERS.identity_api.add_user_to_group(
user1['id'], group1['id']
)
PROVIDERS.identity_api.add_user_to_group(
user1['id'], group2['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
self.assertEqual(0, len(roles_ref))
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], domain_id=domain1['id'],
role_id=role_list[0]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], domain_id=domain1['id'],
role_id=role_list[1]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group2['id'], domain_id=domain1['id'],
role_id=role_list[2]['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=project1['id'],
role_id=role_list[3]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role_list[4]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group2['id'], project_id=project1['id'],
role_id=role_list[5]['id']
)
# Read by the roles, ensuring we get the correct 3 roles for
# both project and domain
combined_list = (
PROVIDERS.assignment_api.get_roles_for_user_and_project(
user1['id'], project1['id']
)
)
self.assertEqual(3, len(combined_list))
self.assertIn(role_list[3]['id'], combined_list)
self.assertIn(role_list[4]['id'], combined_list)
self.assertIn(role_list[5]['id'], combined_list)
combined_role_list = (
PROVIDERS.assignment_api.get_roles_for_user_and_domain(
user1['id'], domain1['id']
)
)
self.assertEqual(3, len(combined_role_list))
self.assertIn(role_list[0]['id'], combined_role_list)
self.assertIn(role_list[1]['id'], combined_role_list)
self.assertIn(role_list[2]['id'], combined_role_list)
def test_delete_role_with_user_and_group_grants(self):
role1 = unit.new_role_ref()
PROVIDERS.role_api.create_role(role1['id'], role1)
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
project1 = unit.new_project_ref(domain_id=domain1['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
user1 = unit.new_user_ref(domain_id=domain1['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
group1 = unit.new_group_ref(domain_id=domain1['id'])
group1 = PROVIDERS.identity_api.create_group(group1)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=project1['id'], role_id=role1['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], domain_id=domain1['id'], role_id=role1['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=role1['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], domain_id=domain1['id'], role_id=role1['id']
)
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
self.assertEqual(1, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
project_id=project1['id'])
self.assertEqual(1, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain1['id'])
self.assertEqual(1, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain1['id'])
self.assertEqual(1, len(roles_ref))
PROVIDERS.role_api.delete_role(role1['id'])
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
project_id=project1['id'])
self.assertEqual(0, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
project_id=project1['id'])
self.assertEqual(0, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
user_id=user1['id'],
domain_id=domain1['id'])
self.assertEqual(0, len(roles_ref))
roles_ref = PROVIDERS.assignment_api.list_grants(
group_id=group1['id'],
domain_id=domain1['id'])
self.assertEqual(0, len(roles_ref))
def test_list_role_assignment_by_domain(self):
"""Test listing of role assignment filtered by domain."""
test_plan = {
# A domain with 3 users, 1 group, a spoiler domain and 2 roles.
'entities': {'domains': [{'users': 3, 'groups': 1}, 1],
'roles': 2},
# Users 1 & 2 are in the group
'group_memberships': [{'group': 0, 'users': [1, 2]}],
# Assign a role for user 0 and the group
'assignments': [{'user': 0, 'role': 0, 'domain': 0},
{'group': 0, 'role': 1, 'domain': 0}],
'tests': [
# List all effective assignments for domain[0].
# Should get one direct user role and user roles for each of
# the users in the group.
{'params': {'domain': 0, 'effective': True},
'results': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 1, 'role': 1, 'domain': 0,
'indirect': {'group': 0}},
{'user': 2, 'role': 1, 'domain': 0,
'indirect': {'group': 0}}
]},
# Using domain[1] should return nothing
{'params': {'domain': 1, 'effective': True},
'results': []},
]
}
self.execute_assignment_plan(test_plan)
def test_list_role_assignment_by_user_with_domain_group_roles(self):
"""Test listing assignments by user, with group roles on a domain."""
test_plan = {
# A domain with 3 users, 3 groups, a spoiler domain
# plus 3 roles.
'entities': {'domains': [{'users': 3, 'groups': 3}, 1],
'roles': 3},
# Users 1 & 2 are in the group 0, User 1 also in group 1
'group_memberships': [{'group': 0, 'users': [0, 1]},
{'group': 1, 'users': [0]}],
'assignments': [{'user': 0, 'role': 0, 'domain': 0},
{'group': 0, 'role': 1, 'domain': 0},
{'group': 1, 'role': 2, 'domain': 0},
# ...and two spoiler assignments
{'user': 1, 'role': 1, 'domain': 0},
{'group': 2, 'role': 2, 'domain': 0}],
'tests': [
# List all effective assignments for user[0].
# Should get one direct user role and a user roles for each of
# groups 0 and 1
{'params': {'user': 0, 'effective': True},
'results': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'domain': 0,
'indirect': {'group': 0}},
{'user': 0, 'role': 2, 'domain': 0,
'indirect': {'group': 1}}
]},
# Adding domain[0] as a filter should return the same data
{'params': {'user': 0, 'domain': 0, 'effective': True},
'results': [{'user': 0, 'role': 0, 'domain': 0},
{'user': 0, 'role': 1, 'domain': 0,
'indirect': {'group': 0}},
{'user': 0, 'role': 2, 'domain': 0,
'indirect': {'group': 1}}
]},
# Using domain[1] should return nothing
{'params': {'user': 0, 'domain': 1, 'effective': True},
'results': []},
# Using user[2] should return nothing
{'params': {'user': 2, 'domain': 0, 'effective': True},
'results': []},
]
}
self.execute_assignment_plan(test_plan)
def test_list_role_assignment_using_sourced_groups(self):
"""Test listing assignments when restricted by source groups."""
test_plan = {
# The default domain with 3 users, 3 groups, 3 projects,
# plus 3 roles.
'entities': {'domains': {'id': CONF.identity.default_domain_id,
'users': 3, 'groups': 3, 'projects': 3},
'roles': 3},
# Users 0 & 1 are in the group 0, User 0 also in group 1
'group_memberships': [{'group': 0, 'users': [0, 1]},
{'group': 1, 'users': [0]}],
# Spread the assignments around - we want to be able to show that
# if sourced by group, assignments from other sources are excluded
'assignments': [{'user': 0, 'role': 0, 'project': 0},
{'group': 0, 'role': 1, 'project': 1},
{'group': 1, 'role': 2, 'project': 0},
{'group': 1, 'role': 2, 'project': 1},
{'user': 2, 'role': 1, 'project': 1},
{'group': 2, 'role': 2, 'project': 2}
],
'tests': [
# List all effective assignments sourced from groups 0 and 1
{'params': {'source_from_group_ids': [0, 1],
'effective': True},
'results': [{'group': 0, 'role': 1, 'project': 1},
{'group': 1, 'role': 2, 'project': 0},
{'group': 1, 'role': 2, 'project': 1}
]},
# Adding a role a filter should further restrict the entries
{'params': {'source_from_group_ids': [0, 1], 'role': 2,
'effective': True},
'results': [{'group': 1, 'role': 2, 'project': 0},
{'group': 1, 'role': 2, 'project': 1}
]},
]
}
self.execute_assignment_plan(test_plan)
def test_list_role_assignment_using_sourced_groups_with_domains(self):
"""Test listing domain assignments when restricted by source groups."""
test_plan = {
# A domain with 3 users, 3 groups, 3 projects, a second domain,
# plus 3 roles.
'entities': {'domains': [{'users': 3, 'groups': 3, 'projects': 3},
1],
'roles': 3},
# Users 0 & 1 are in the group 0, User 0 also in group 1
'group_memberships': [{'group': 0, 'users': [0, 1]},
{'group': 1, 'users': [0]}],
# Spread the assignments around - we want to be able to show that
# if sourced by group, assignments from other sources are excluded
'assignments': [{'user': 0, 'role': 0, 'domain': 0},
{'group': 0, 'role': 1, 'domain': 1},
{'group': 1, 'role': 2, 'project': 0},
{'group': 1, 'role': 2, 'project': 1},
{'user': 2, 'role': 1, 'project': 1},
{'group': 2, 'role': 2, 'project': 2}
],
'tests': [
# List all effective assignments sourced from groups 0 and 1
{'params': {'source_from_group_ids': [0, 1],
'effective': True},
'results': [{'group': 0, 'role': 1, 'domain': 1},
{'group': 1, 'role': 2, 'project': 0},
{'group': 1, 'role': 2, 'project': 1}
]},
# Adding a role a filter should further restrict the entries
{'params': {'source_from_group_ids': [0, 1], 'role': 1,
'effective': True},
'results': [{'group': 0, 'role': 1, 'domain': 1},
]},
]
}
self.execute_assignment_plan(test_plan)
def test_list_role_assignment_fails_with_userid_and_source_groups(self):
"""Show we trap this unsupported internal combination of params."""
group = unit.new_group_ref(domain_id=CONF.identity.default_domain_id)
group = PROVIDERS.identity_api.create_group(group)
self.assertRaises(exception.UnexpectedError,
PROVIDERS.assignment_api.list_role_assignments,
effective=True,
user_id=self.user_foo['id'],
source_from_group_ids=[group['id']])
def test_list_user_project_ids_returns_not_found(self):
self.assertRaises(exception.UserNotFound,
PROVIDERS.assignment_api.list_projects_for_user,
uuid.uuid4().hex)
def test_delete_user_with_project_association(self):
user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user = PROVIDERS.identity_api.create_user(user)
role_member = unit.new_role_ref()
PROVIDERS.role_api.create_role(role_member['id'], role_member)
PROVIDERS.assignment_api.add_role_to_user_and_project(
user['id'], self.project_bar['id'], role_member['id']
)
PROVIDERS.identity_api.delete_user(user['id'])
self.assertRaises(exception.UserNotFound,
PROVIDERS.assignment_api.list_projects_for_user,
user['id'])
def test_delete_user_with_project_roles(self):
user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user = PROVIDERS.identity_api.create_user(user)
PROVIDERS.assignment_api.add_role_to_user_and_project(
user['id'],
self.project_bar['id'],
self.role_member['id'])
PROVIDERS.identity_api.delete_user(user['id'])
self.assertRaises(exception.UserNotFound,
PROVIDERS.assignment_api.list_projects_for_user,
user['id'])
def test_delete_role_returns_not_found(self):
self.assertRaises(exception.RoleNotFound,
PROVIDERS.role_api.delete_role,
uuid.uuid4().hex)
def test_delete_project_with_role_assignments(self):
project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id)
PROVIDERS.resource_api.create_project(project['id'], project)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'],
project['id'],
default_fixtures.MEMBER_ROLE_ID)
PROVIDERS.resource_api.delete_project(project['id'])
self.assertRaises(exception.ProjectNotFound,
PROVIDERS.assignment_api.list_user_ids_for_project,
project['id'])
def test_delete_role_check_role_grant(self):
role = unit.new_role_ref()
alt_role = unit.new_role_ref()
PROVIDERS.role_api.create_role(role['id'], role)
PROVIDERS.role_api.create_role(alt_role['id'], alt_role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'], self.project_bar['id'], role['id'])
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_foo['id'], self.project_bar['id'], alt_role['id'])
PROVIDERS.role_api.delete_role(role['id'])
roles_ref = PROVIDERS.assignment_api.get_roles_for_user_and_project(
self.user_foo['id'], self.project_bar['id'])
self.assertNotIn(role['id'], roles_ref)
self.assertIn(alt_role['id'], roles_ref)
def test_list_projects_for_user(self):
domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain['id'], domain)
user1 = unit.new_user_ref(domain_id=domain['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
user_projects = PROVIDERS.assignment_api.list_projects_for_user(
user1['id']
)
self.assertEqual(0, len(user_projects))
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=self.project_bar['id'],
role_id=self.role_member['id']
)
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=self.project_baz['id'],
role_id=self.role_member['id']
)
user_projects = PROVIDERS.assignment_api.list_projects_for_user(
user1['id']
)
self.assertEqual(2, len(user_projects))
def test_list_projects_for_user_with_grants(self):
# Create two groups each with a role on a different project, and
# make user1 a member of both groups. Both these new projects
# should now be included, along with any direct user grants.
domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain['id'], domain)
user1 = unit.new_user_ref(domain_id=domain['id'])
user1 = PROVIDERS.identity_api.create_user(user1)
group1 = unit.new_group_ref(domain_id=domain['id'])
group1 = PROVIDERS.identity_api.create_group(group1)
group2 = unit.new_group_ref(domain_id=domain['id'])
group2 = PROVIDERS.identity_api.create_group(group2)
project1 = unit.new_project_ref(domain_id=domain['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
project2 = unit.new_project_ref(domain_id=domain['id'])
PROVIDERS.resource_api.create_project(project2['id'], project2)
PROVIDERS.identity_api.add_user_to_group(user1['id'], group1['id'])
PROVIDERS.identity_api.add_user_to_group(user1['id'], group2['id'])
# Create 3 grants, one user grant, the other two as group grants
PROVIDERS.assignment_api.create_grant(
user_id=user1['id'], project_id=self.project_bar['id'],
role_id=self.role_member['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group1['id'], project_id=project1['id'],
role_id=self.role_admin['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group2['id'], project_id=project2['id'],
role_id=self.role_admin['id']
)
user_projects = PROVIDERS.assignment_api.list_projects_for_user(
user1['id']
)
self.assertEqual(3, len(user_projects))
def test_create_grant_no_user(self):
# If call create_grant with a user that doesn't exist, doesn't fail.
PROVIDERS.assignment_api.create_grant(
self.role_other['id'],
user_id=uuid.uuid4().hex,
project_id=self.project_bar['id'])
def test_create_grant_no_group(self):
# If call create_grant with a group that doesn't exist, doesn't fail.
PROVIDERS.assignment_api.create_grant(
self.role_other['id'],
group_id=uuid.uuid4().hex,
project_id=self.project_bar['id'])
def test_delete_group_removes_role_assignments(self):
# When a group is deleted any role assignments for the group are
# removed.
def get_member_assignments():
assignments = PROVIDERS.assignment_api.list_role_assignments()
return ([x for x in assignments if x['role_id'] ==
default_fixtures.MEMBER_ROLE_ID])
orig_member_assignments = get_member_assignments()
# Create a group.
new_group = unit.new_group_ref(
domain_id=CONF.identity.default_domain_id)
new_group = PROVIDERS.identity_api.create_group(new_group)
# Create a project.
new_project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id)
PROVIDERS.resource_api.create_project(new_project['id'], new_project)
# Assign a role to the group.
PROVIDERS.assignment_api.create_grant(
group_id=new_group['id'], project_id=new_project['id'],
role_id=default_fixtures.MEMBER_ROLE_ID)
# Delete the group.
PROVIDERS.identity_api.delete_group(new_group['id'])
# Check that the role assignment for the group is gone
member_assignments = get_member_assignments()
self.assertThat(member_assignments,
matchers.Equals(orig_member_assignments))
def test_get_roles_for_groups_on_domain(self):
"""Test retrieving group domain roles.
Test Plan:
- Create a domain, three groups and three roles
- Assign one an inherited and the others a non-inherited group role
to the domain
- Ensure that only the non-inherited roles are returned on the domain
"""
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
group_list = []
group_id_list = []
role_list = []
for _ in range(3):
group = unit.new_group_ref(domain_id=domain1['id'])
group = PROVIDERS.identity_api.create_group(group)
group_list.append(group)
group_id_list.append(group['id'])
role = unit.new_role_ref()
PROVIDERS.role_api.create_role(role['id'], role)
role_list.append(role)
# Assign the roles - one is inherited
PROVIDERS.assignment_api.create_grant(
group_id=group_list[0]['id'], domain_id=domain1['id'],
role_id=role_list[0]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group_list[1]['id'], domain_id=domain1['id'],
role_id=role_list[1]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group_list[2]['id'], domain_id=domain1['id'],
role_id=role_list[2]['id'], inherited_to_projects=True
)
# Now get the effective roles for the groups on the domain project. We
# shouldn't get back the inherited role.
role_refs = PROVIDERS.assignment_api.get_roles_for_groups(
group_id_list, domain_id=domain1['id'])
self.assertThat(role_refs, matchers.HasLength(2))
self.assertIn(role_list[0], role_refs)
self.assertIn(role_list[1], role_refs)
def test_get_roles_for_groups_on_project(self):
"""Test retrieving group project roles.
Test Plan:
- Create two domains, two projects, six groups and six roles
- Project1 is in Domain1, Project2 is in Domain2
- Domain2/Project2 are spoilers
- Assign a different direct group role to each project as well
as both an inherited and non-inherited role to each domain
- Get the group roles for Project 1 - depending on whether we have
enabled inheritance, we should either get back just the direct role
or both the direct one plus the inherited domain role from Domain 1
"""
domain1 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain1['id'], domain1)
domain2 = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain2['id'], domain2)
project1 = unit.new_project_ref(domain_id=domain1['id'])
PROVIDERS.resource_api.create_project(project1['id'], project1)
project2 = unit.new_project_ref(domain_id=domain2['id'])
PROVIDERS.resource_api.create_project(project2['id'], project2)
group_list = []
group_id_list = []
role_list = []
for _ in range(6):
group = unit.new_group_ref(domain_id=domain1['id'])
group = PROVIDERS.identity_api.create_group(group)
group_list.append(group)
group_id_list.append(group['id'])
role = unit.new_role_ref()
PROVIDERS.role_api.create_role(role['id'], role)
role_list.append(role)
# Assign the roles - one inherited and one non-inherited on Domain1,
# plus one on Project1
PROVIDERS.assignment_api.create_grant(
group_id=group_list[0]['id'], domain_id=domain1['id'],
role_id=role_list[0]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group_list[1]['id'], domain_id=domain1['id'],
role_id=role_list[1]['id'], inherited_to_projects=True
)
PROVIDERS.assignment_api.create_grant(
group_id=group_list[2]['id'], project_id=project1['id'],
role_id=role_list[2]['id']
)
# ...and a duplicate set of spoiler assignments to Domain2/Project2
PROVIDERS.assignment_api.create_grant(
group_id=group_list[3]['id'], domain_id=domain2['id'],
role_id=role_list[3]['id']
)
PROVIDERS.assignment_api.create_grant(
group_id=group_list[4]['id'], domain_id=domain2['id'],
role_id=role_list[4]['id'], inherited_to_projects=True
)
PROVIDERS.assignment_api.create_grant(
group_id=group_list[5]['id'], project_id=project2['id'],
role_id=role_list[5]['id']
)
# With inheritance on, we should also get back the inherited role from
# its owning domain.
role_refs = PROVIDERS.assignment_api.get_roles_for_groups(
group_id_list, project_id=project1['id'])
self.assertThat(role_refs, matchers.HasLength(2))
self.assertIn(role_list[1], role_refs)
self.assertIn(role_list[2], role_refs)
def test_list_domains_for_groups(self):
"""Test retrieving domains for a list of groups.
Test Plan:
- Create three domains, three groups and one role
- Assign a non-inherited group role to two domains, and an inherited
group role to the third
- Ensure only the domains with non-inherited roles are returned
"""
domain_list = []
group_list = []
group_id_list = []
for _ in range(3):
domain = unit.new_domain_ref()
PROVIDERS.resource_api.create_domain(domain['id'], domain)
domain_list.append(domain)
group = unit.new_group_ref(domain_id=domain['id'])
group = PROVIDERS.identity_api.create_group(group)