diff options
Diffstat (limited to 'keystone-moon/keystone/tests/unit/test_backend.py')
| -rw-r--r-- | keystone-moon/keystone/tests/unit/test_backend.py | 1019 |
1 files changed, 841 insertions, 178 deletions
diff --git a/keystone-moon/keystone/tests/unit/test_backend.py b/keystone-moon/keystone/tests/unit/test_backend.py index 45b8e0b0..d3b51edd 100644 --- a/keystone-moon/keystone/tests/unit/test_backend.py +++ b/keystone-moon/keystone/tests/unit/test_backend.py @@ -28,7 +28,7 @@ from testtools import matchers from keystone.catalog import core from keystone.common import driver_hints from keystone import exception -from keystone.tests import unit as tests +from keystone.tests import unit from keystone.tests.unit import default_fixtures from keystone.tests.unit import filtering from keystone.tests.unit import utils as test_utils @@ -40,7 +40,379 @@ DEFAULT_DOMAIN_ID = CONF.identity.default_domain_id NULL_OBJECT = object() -class IdentityTests(object): +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}}] + + # 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' and + # 'inherited_to_projects' 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 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, + 'domain_id': domain_id, 'parent_id': parent_id} + new_project = self.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 _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 = {'id': uuid.uuid4().hex, + 'name': uuid.uuid4().hex} + self.resource_api.create_domain(new_domain['id'], + new_domain) + return new_domain + else: + # The test plan specified an existing domain to use + return self.resource_api.get_domain(domain_id) + + def _create_entity_in_domain(entity_type, domain_id): + """Create a user or group entity in the domain.""" + + new_entity = {'name': uuid.uuid4().hex, 'domain_id': domain_id} + if entity_type == 'users': + new_entity = self.identity_api.create_user(new_entity) + elif entity_type == 'groups': + new_entity = self.identity_api.create_group(new_entity) + 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]....] + + """ + def _create_role(): + new_role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex} + return self.role_api.create_role(new_role['id'], new_role) + + 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(_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_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'] + self.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(self.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 + self.assignment_api.create_grant(**args) + return test_data + + def execute_assignment_tests(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) + + # 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']: + # Just pass the value into the args + args[param] = test['params'][param] + 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 = self.assignment_api.list_role_assignments(**args) + check_results(test['results'], results, len(args)) + + def execute_assignment_test_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 '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_tests(test_plan, test_data) + return test_data + + +class IdentityTests(AssignmentTestHelperMixin): def _get_domain_fixture(self): domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex} self.resource_api.create_domain(domain['id'], domain) @@ -227,7 +599,7 @@ class IdentityTests(object): self.user_foo.pop('password') self.assertDictEqual(user_ref, self.user_foo) - @tests.skip_if_cache_disabled('identity') + @unit.skip_if_cache_disabled('identity') def test_cache_layer_get_user(self): user = { 'name': uuid.uuid4().hex.lower(), @@ -279,7 +651,7 @@ class IdentityTests(object): self.user_foo.pop('password') self.assertDictEqual(user_ref, self.user_foo) - @tests.skip_if_cache_disabled('identity') + @unit.skip_if_cache_disabled('identity') def test_cache_layer_get_user_by_name(self): user = { 'name': uuid.uuid4().hex.lower(), @@ -506,168 +878,83 @@ class IdentityTests(object): 'fake2') def test_list_role_assignments_unfiltered(self): - """Test unfiltered listing of role assignments. - - Test Plan: - - - Create a domain, with a user, group & project - - Find how many role assignments already exist (from default - fixtures) - - Create a grant of each type (user/group on project/domain) - - Check the number of assignments has gone up by 4 and that - the entries we added are in the list returned - - Check that if we list assignments by role_id, then we get back - assignments that only contain that role. - - """ - new_domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex} - self.resource_api.create_domain(new_domain['id'], new_domain) - new_user = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex, - 'enabled': True, 'domain_id': new_domain['id']} - new_user = self.identity_api.create_user(new_user) - new_group = {'domain_id': new_domain['id'], 'name': uuid.uuid4().hex} - new_group = self.identity_api.create_group(new_group) - new_project = {'id': uuid.uuid4().hex, - 'name': uuid.uuid4().hex, - 'domain_id': new_domain['id']} - self.resource_api.create_project(new_project['id'], new_project) - - # First check how many role grants already exist - existing_assignments = len(self.assignment_api.list_role_assignments()) - - # Now create the grants (roles are defined in default_fixtures) - self.assignment_api.create_grant(user_id=new_user['id'], - domain_id=new_domain['id'], - role_id='member') - self.assignment_api.create_grant(user_id=new_user['id'], - project_id=new_project['id'], - role_id='other') - self.assignment_api.create_grant(group_id=new_group['id'], - domain_id=new_domain['id'], - role_id='admin') - self.assignment_api.create_grant(group_id=new_group['id'], - project_id=new_project['id'], - role_id='admin') - - # Read back the full list of assignments - check it is gone up by 4 - assignment_list = self.assignment_api.list_role_assignments() - self.assertEqual(existing_assignments + 4, len(assignment_list)) - - # Now check that each of our four new entries are in the list - self.assertIn( - {'user_id': new_user['id'], 'domain_id': new_domain['id'], - 'role_id': 'member'}, - assignment_list) - self.assertIn( - {'user_id': new_user['id'], 'project_id': new_project['id'], - 'role_id': 'other'}, - assignment_list) - self.assertIn( - {'group_id': new_group['id'], 'domain_id': new_domain['id'], - 'role_id': 'admin'}, - assignment_list) - self.assertIn( - {'group_id': new_group['id'], 'project_id': new_project['id'], - 'role_id': 'admin'}, - assignment_list) + """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_test_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 - - Find how many role assignments already exist (from default - fixtures) - - Create a grant of each type (user/group on project/domain) - - Check that if we list assignments by role_id, then we get back - assignments that only contain that role. - - """ - new_user = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex, - 'enabled': True, 'domain_id': DEFAULT_DOMAIN_ID} - new_user = self.identity_api.create_user(new_user) - new_group = {'domain_id': DEFAULT_DOMAIN_ID, 'name': uuid.uuid4().hex} - new_group = self.identity_api.create_group(new_group) - new_project = {'id': uuid.uuid4().hex, - 'name': uuid.uuid4().hex, - 'domain_id': DEFAULT_DOMAIN_ID} - self.resource_api.create_project(new_project['id'], new_project) - - # First check how many role grants already exist - existing_assignments_for_role = len( - self.assignment_api.list_role_assignments_for_role( - role_id='admin')) - - # Now create the grants (roles are defined in default_fixtures) - self.assignment_api.create_grant(user_id=new_user['id'], - domain_id=DEFAULT_DOMAIN_ID, - role_id='member') - self.assignment_api.create_grant(user_id=new_user['id'], - project_id=new_project['id'], - role_id='other') - self.assignment_api.create_grant(group_id=new_group['id'], - domain_id=DEFAULT_DOMAIN_ID, - role_id='admin') - self.assignment_api.create_grant(group_id=new_group['id'], - project_id=new_project['id'], - role_id='admin') + """Test listing of role assignments filtered by role ID.""" + + test_plan = { + # Create a user, group & project in the default domain + 'entities': {'domains': {'id': 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}]} + ] + } + test_data = self.execute_assignment_test_plan(test_plan) - # Read back the list of assignments for just the admin role, checking - # this only goes up by two. + # Also test that list_role_assignments_for_role() gives the same answer assignment_list = self.assignment_api.list_role_assignments_for_role( - role_id='admin') - self.assertEqual(existing_assignments_for_role + 2, - len(assignment_list)) + role_id=test_data['roles'][2]['id']) + self.assertThat(assignment_list, matchers.HasLength(2)) # Now check that each of our two new entries are in the list self.assertIn( - {'group_id': new_group['id'], 'domain_id': DEFAULT_DOMAIN_ID, - 'role_id': 'admin'}, + {'group_id': test_data['groups'][0]['id'], + 'domain_id': DEFAULT_DOMAIN_ID, + 'role_id': test_data['roles'][2]['id']}, assignment_list) self.assertIn( - {'group_id': new_group['id'], 'project_id': new_project['id'], - 'role_id': 'admin'}, + {'group_id': test_data['groups'][0]['id'], + 'project_id': test_data['projects'][0]['id'], + 'role_id': test_data['roles'][2]['id']}, assignment_list) 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. - MEMBER_ROLE_ID = 'member' - - def get_member_assignments(): - assignments = self.assignment_api.list_role_assignments() - return [x for x in assignments if x['role_id'] == MEMBER_ROLE_ID] - - orig_member_assignments = get_member_assignments() - - # Create a group. - new_group = { - 'domain_id': DEFAULT_DOMAIN_ID, - 'name': self.getUniqueString(prefix='tlgra')} - new_group = self.identity_api.create_group(new_group) - - # Create a project. - new_project = { - 'id': uuid.uuid4().hex, - 'name': self.getUniqueString(prefix='tlgra'), - 'domain_id': DEFAULT_DOMAIN_ID} - self.resource_api.create_project(new_project['id'], new_project) - - # Assign a role to the group. - self.assignment_api.create_grant( - group_id=new_group['id'], project_id=new_project['id'], - role_id=MEMBER_ROLE_ID) - - # List role assignments - new_member_assignments = get_member_assignments() - - expected_member_assignments = orig_member_assignments + [{ - 'group_id': new_group['id'], 'project_id': new_project['id'], - 'role_id': MEMBER_ROLE_ID}] - self.assertItemsEqual(expected_member_assignments, - new_member_assignments) + test_plan = { + 'entities': {'domains': {'id': 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_test_plan(test_plan) def test_list_role_assignments_bad_role(self): assignment_list = self.assignment_api.list_role_assignments_for_role( @@ -1695,6 +1982,82 @@ class IdentityTests(object): self.identity_api.delete_group(group1['id']) self.identity_api.get_user(user1['id']) + 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_test_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_test_plan(test_plan) + def test_delete_domain_with_user_group_project_links(self): # TODO(chungg):add test case once expected behaviour defined pass @@ -2162,7 +2525,7 @@ class IdentityTests(object): self.assertIn(self.tenant_mtu['id'], project_ids) self.assertIn(self.tenant_service['id'], project_ids) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support def test_list_projects_for_alternate_domain(self): domain1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex} self.resource_api.create_domain(domain1['id'], domain1) @@ -2219,7 +2582,7 @@ class IdentityTests(object): return projects - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support def test_create_domain_with_project_api(self): project_id = uuid.uuid4().hex project = {'id': project_id, @@ -2233,7 +2596,7 @@ class IdentityTests(object): self.assertTrue(ref['is_domain']) self.assertEqual(DEFAULT_DOMAIN_ID, ref['domain_id']) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support @test_utils.wip('waiting for projects acting as domains implementation') def test_is_domain_sub_project_has_parent_domain_id(self): project = {'id': uuid.uuid4().hex, @@ -2258,7 +2621,7 @@ class IdentityTests(object): self.assertEqual(project['id'], ref['parent_id']) self.assertEqual(project['id'], ref['domain_id']) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support @test_utils.wip('waiting for projects acting as domains implementation') def test_delete_domain_with_project_api(self): project_id = uuid.uuid4().hex @@ -2283,7 +2646,7 @@ class IdentityTests(object): # Successfuly delete the project self.resource_api.delete_project(project['id']) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support @test_utils.wip('waiting for projects acting as domains implementation') def test_create_domain_under_regular_project_hierarchy_fails(self): # Creating a regular project hierarchy. Projects acting as domains @@ -2303,7 +2666,7 @@ class IdentityTests(object): self.resource_api.create_project, project['id'], project) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support @test_utils.wip('waiting for projects acting as domains implementation') def test_create_project_under_domain_hierarchy(self): projects_hierarchy = self._create_projects_hierarchy(is_domain=True) @@ -2917,7 +3280,7 @@ class IdentityTests(object): uuid.uuid4().hex, DEFAULT_DOMAIN_ID) - @tests.skip_if_cache_disabled('identity') + @unit.skip_if_cache_disabled('identity') def test_cache_layer_group_crud(self): group = {'domain_id': DEFAULT_DOMAIN_ID, 'name': uuid.uuid4().hex} group = self.identity_api.create_group(group) @@ -2992,7 +3355,7 @@ class IdentityTests(object): group1['id'], group1) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support def test_project_crud(self): domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, 'enabled': True} @@ -3093,7 +3456,7 @@ class IdentityTests(object): project['id'], project) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support def test_create_leaf_project_with_different_domain(self): root_project = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, @@ -3308,7 +3671,7 @@ class IdentityTests(object): self.resource_api.get_domain, domain['id']) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support def test_create_domain_case_sensitivity(self): # create a ref with a lowercase name ref = { @@ -3443,8 +3806,8 @@ class IdentityTests(object): user_projects = self.assignment_api.list_projects_for_user(user1['id']) self.assertEqual(3, len(user_projects)) - @tests.skip_if_cache_disabled('resource') - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_cache_disabled('resource') + @unit.skip_if_no_multiple_domains_support def test_domain_rename_invalidates_get_domain_by_name_cache(self): domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, 'enabled': True} @@ -3458,7 +3821,7 @@ class IdentityTests(object): self.resource_api.get_domain_by_name, domain_name) - @tests.skip_if_cache_disabled('resource') + @unit.skip_if_cache_disabled('resource') def test_cache_layer_domain_crud(self): domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, 'enabled': True} @@ -3513,8 +3876,8 @@ class IdentityTests(object): self.resource_api.get_domain, domain_id) - @tests.skip_if_cache_disabled('resource') - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_cache_disabled('resource') + @unit.skip_if_no_multiple_domains_support def test_project_rename_invalidates_get_project_by_name_cache(self): domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, 'enabled': True} @@ -3533,8 +3896,8 @@ class IdentityTests(object): project_name, domain['id']) - @tests.skip_if_cache_disabled('resource') - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_cache_disabled('resource') + @unit.skip_if_no_multiple_domains_support def test_cache_layer_project_crud(self): domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex, 'enabled': True} @@ -3646,7 +4009,7 @@ class IdentityTests(object): group_id=uuid.uuid4().hex, project_id=self.tenant_bar['id']) - @tests.skip_if_no_multiple_domains_support + @unit.skip_if_no_multiple_domains_support def test_get_default_domain_by_name(self): domain_name = 'default' @@ -4308,7 +4671,7 @@ class TokenTests(object): def test_list_revoked_tokens_for_multiple_tokens(self): self.check_list_revoked_tokens([self.delete_token() - for x in six.moves.range(2)]) + for x in range(2)]) def test_flush_expired_token(self): token_id = uuid.uuid4().hex @@ -4339,7 +4702,7 @@ class TokenTests(object): self.assertEqual(1, len(tokens)) self.assertIn(token_id, tokens) - @tests.skip_if_cache_disabled('token') + @unit.skip_if_cache_disabled('token') def test_revocation_list_cache(self): expire_time = timeutils.utcnow() + datetime.timedelta(minutes=10) token_id = uuid.uuid4().hex @@ -4769,7 +5132,7 @@ class CatalogTests(object): for region in regions: self.assertEqual(parent_id, region['parent_region_id']) - @tests.skip_if_cache_disabled('catalog') + @unit.skip_if_cache_disabled('catalog') def test_cache_layer_region_crud(self): region_id = uuid.uuid4().hex new_region = { @@ -4797,7 +5160,7 @@ class CatalogTests(object): self.assertRaises(exception.RegionNotFound, self.catalog_api.get_region, region_id) - @tests.skip_if_cache_disabled('catalog') + @unit.skip_if_cache_disabled('catalog') def test_invalidate_cache_when_updating_region(self): region_id = uuid.uuid4().hex new_region = { @@ -4875,7 +5238,7 @@ class CatalogTests(object): region_two['id'], {'parent_region_id': region_four['id']}) - @mock.patch.object(core.Driver, + @mock.patch.object(core.CatalogDriverV8, "_ensure_no_circle_in_hierarchical_regions") def test_circular_regions_can_be_deleted(self, mock_ensure_on_circle): # turn off the enforcement so that cycles can be created for the test @@ -5003,7 +5366,7 @@ class CatalogTests(object): self.catalog_api.delete_service(unrelated_service1['id']) self.catalog_api.delete_service(unrelated_service2['id']) - @tests.skip_if_cache_disabled('catalog') + @unit.skip_if_cache_disabled('catalog') def test_cache_layer_service_crud(self): service_id = uuid.uuid4().hex new_service = { @@ -5040,7 +5403,7 @@ class CatalogTests(object): self.catalog_api.get_service, service_id) - @tests.skip_if_cache_disabled('catalog') + @unit.skip_if_cache_disabled('catalog') def test_invalidate_cache_when_updating_service(self): service_id = uuid.uuid4().hex new_service = { @@ -5351,7 +5714,7 @@ class CatalogTests(object): endpoint_ids = [x['id'] for x in catalog[0]['endpoints']] self.assertEqual([enabled_endpoint_ref['id']], endpoint_ids) - @tests.skip_if_cache_disabled('catalog') + @unit.skip_if_cache_disabled('catalog') def test_invalidate_cache_when_updating_endpoint(self): service = { 'id': uuid.uuid4().hex, @@ -5468,7 +5831,71 @@ class PolicyTests(object): uuid.uuid4().hex) -class InheritanceTests(object): +class InheritanceTests(AssignmentTestHelperMixin): + + def test_role_assignments_user_domain_to_project_inheritance(self): + test_plan = { + 'entities': {'domains': {'users': 2, 'projects': 1}, + 'roles': 3}, + 'assignments': [{'user': 0, 'role': 0, 'domain': 0}, + {'user': 0, 'role': 1, 'project': 0}, + {'user': 0, 'role': 2, 'domain': 0, + 'inherited_to_projects': True}, + {'user': 1, 'role': 1, 'project': 0}], + 'tests': [ + # List all direct assignments for user[0] + {'params': {'user': 0}, + 'results': [{'user': 0, 'role': 0, 'domain': 0}, + {'user': 0, 'role': 1, 'project': 0}, + {'user': 0, 'role': 2, 'domain': 0, + 'inherited_to_projects': 'projects'}]}, + # Now the effective ones - so the domain role should turn into + # a project role + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'domain': 0}, + {'user': 0, 'role': 1, 'project': 0}, + {'user': 0, 'role': 2, 'project': 0, + 'indirect': {'domain': 0}}]}, + # Narrow down to effective roles for user[0] and project[0] + {'params': {'user': 0, 'project': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 1, 'project': 0}, + {'user': 0, 'role': 2, 'project': 0, + 'indirect': {'domain': 0}}]} + ] + } + self.config_fixture.config(group='os_inherit', enabled=True) + self.execute_assignment_test_plan(test_plan) + + def test_inherited_role_assignments_excluded_if_os_inherit_false(self): + test_plan = { + 'entities': {'domains': {'users': 2, 'groups': 1, 'projects': 1}, + 'roles': 4}, + 'group_memberships': [{'group': 0, 'users': [0]}], + 'assignments': [{'user': 0, 'role': 0, 'domain': 0}, + {'user': 0, 'role': 1, 'project': 0}, + {'user': 0, 'role': 2, 'domain': 0, + 'inherited_to_projects': True}, + {'user': 1, 'role': 1, 'project': 0}, + {'group': 0, 'role': 3, 'project': 0}], + 'tests': [ + # List all direct assignments for user[0], since os-inherit is + # disabled, we should not see the inherited role + {'params': {'user': 0}, + 'results': [{'user': 0, 'role': 0, 'domain': 0}, + {'user': 0, 'role': 1, 'project': 0}]}, + # Same in effective mode - inherited roles should not be + # included or expanded...but the group role should now + # turn up as a user role, since group expansion is not + # part of os-inherit. + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'domain': 0}, + {'user': 0, 'role': 1, 'project': 0}, + {'user': 0, 'role': 3, 'project': 0, + 'indirect': {'group': 0}}]}, + ] + } + self.config_fixture.config(group='os_inherit', enabled=False) + self.execute_assignment_test_plan(test_plan) def _test_crud_inherited_and_direct_assignment(self, **kwargs): """Tests inherited and direct assignments for the actor and target @@ -5616,6 +6043,42 @@ class InheritanceTests(object): self.assertEqual(1, len(combined_role_list)) self.assertIn(role_list[1]['id'], combined_role_list) + # TODO(henry-nash): The test above uses get_roles_for_user_and_project + # and get_roles_for_user_and_domain, which will, in a subsequent patch, + # be re-implemeted to simply call list_role_assignments (see blueprint + # remove-role-metadata). + # + # The test plan below therefore mirrors this test, to ensure that + # list_role_assignments works the same. Once get_roles_for_user_and + # project/domain have been re-implemented then the manual tests above + # can be refactored to simply ensure it gives the same answers. + test_plan = { + # A domain with a user & project, plus 3 roles. + 'entities': {'domains': {'users': 1, 'projects': 1}, + 'roles': 3}, + 'assignments': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 1, 'domain': 0}, + {'user': 0, 'role': 2, 'domain': 0, + 'inherited_to_projects': True}], + 'tests': [ + # List all effective assignments for user[0] on project[0]. + # Should get one direct role and one inherited role. + {'params': {'user': 0, 'project': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 2, 'project': 0, + 'indirect': {'domain': 0}}]}, + # Ensure effective mode on the domain does not list the + # inherited role on that domain + {'params': {'user': 0, 'domain': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 1, 'domain': 0}]}, + # Ensure non-inherited mode also only returns the non-inherited + # role on the domain + {'params': {'user': 0, 'domain': 0, 'inherited': False}, + 'results': [{'user': 0, 'role': 1, 'domain': 0}]}, + ] + } + self.execute_assignment_test_plan(test_plan) + def test_inherited_role_grants_for_group(self): """Test inherited group roles. @@ -5699,6 +6162,42 @@ class InheritanceTests(object): self.assertIn(role_list[2]['id'], combined_list) self.assertIn(role_list[3]['id'], combined_list) + # TODO(henry-nash): The test above uses get_roles_for_user_and_project + # which will, in a subsequent patch, be re-implemeted to simply call + # list_role_assignments (see blueprint remove-role-metadata). + # + # The test plan below therefore mirrors this test, to ensure that + # list_role_assignments works the same. Once + # get_roles_for_user_and_project has been re-implemented then the + # manual tests above can be refactored to simply ensure it gives + # the same answers. + test_plan = { + # A domain with a user and project, 2 groups, plus 4 roles. + 'entities': {'domains': {'users': 1, 'projects': 1, 'groups': 2}, + 'roles': 4}, + 'group_memberships': [{'group': 0, 'users': [0]}, + {'group': 1, 'users': [0]}], + 'assignments': [{'user': 0, 'role': 0, 'project': 0}, + {'group': 0, 'role': 1, 'domain': 0}, + {'group': 1, 'role': 2, 'domain': 0, + 'inherited_to_projects': True}, + {'group': 1, 'role': 3, 'domain': 0, + 'inherited_to_projects': True}], + 'tests': [ + # List all effective assignments for user[0] on project[0]. + # Should get one direct role and both inherited roles, but + # not the direct one on domain[0], even though user[0] is + # in group[0]. + {'params': {'user': 0, 'project': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 2, 'project': 0, + 'indirect': {'domain': 0, 'group': 1}}, + {'user': 0, 'role': 3, 'project': 0, + 'indirect': {'domain': 0, 'group': 1}}]} + ] + } + self.execute_assignment_test_plan(test_plan) + def test_list_projects_for_user_with_inherited_grants(self): """Test inherited user roles. @@ -5738,6 +6237,37 @@ class InheritanceTests(object): user_projects = self.assignment_api.list_projects_for_user(user1['id']) self.assertEqual(3, len(user_projects)) + # TODO(henry-nash): The test above uses list_projects_for_user + # which may, in a subsequent patch, be re-implemeted to call + # list_role_assignments and then report only the distinct projects. + # + # The test plan below therefore mirrors this test, to ensure that + # list_role_assignments works the same. Once list_projects_for_user + # has been re-implemented then the manual tests above can be + # refactored. + test_plan = { + # A domain with 1 project, plus a second domain with 2 projects, + # as well as a user. Also, create 2 roles. + 'entities': {'domains': [{'projects': 1}, + {'users': 1, 'projects': 2}], + 'roles': 2}, + 'assignments': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 1, 'domain': 1, + 'inherited_to_projects': True}], + 'tests': [ + # List all effective assignments for user[0] + # Should get one direct role plus one inherited role for each + # project in domain + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 1, 'project': 1, + 'indirect': {'domain': 1}}, + {'user': 0, 'role': 1, 'project': 2, + 'indirect': {'domain': 1}}]} + ] + } + self.execute_assignment_test_plan(test_plan) + def test_list_projects_for_user_with_inherited_user_project_grants(self): """Test inherited role assignments for users on nested projects. @@ -5798,6 +6328,50 @@ class InheritanceTests(object): self.assertEqual(1, len(user_projects)) self.assertIn(root_project, user_projects) + # TODO(henry-nash): The test above uses list_projects_for_user + # which may, in a subsequent patch, be re-implemeted to call + # list_role_assignments and then report only the distinct projects. + # + # The test plan below therefore mirrors this test, to ensure that + # list_role_assignments works the same. Once list_projects_for_user + # has been re-implemented then the manual tests above can be + # refactored. + test_plan = { + # A domain with a project and sub-project, plus a user. + # Also, create 2 roles. + 'entities': { + 'domains': {'id': DEFAULT_DOMAIN_ID, 'users': 1, + 'projects': {'project': 1}}, + 'roles': 2}, + # A direct role and an inherited role on the parent + 'assignments': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 1, 'project': 0, + 'inherited_to_projects': True}], + 'tests': [ + # List all effective assignments for user[0] - should get back + # one direct role plus one inherited role. + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 1, 'project': 1, + 'indirect': {'project': 0}}]} + ] + } + + test_plan_with_os_inherit_disabled = { + 'tests': [ + # List all effective assignments for user[0] - should only get + # back the one direct role. + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0}]} + ] + } + self.config_fixture.config(group='os_inherit', enabled=True) + test_data = self.execute_assignment_test_plan(test_plan) + self.config_fixture.config(group='os_inherit', enabled=False) + # Pass the existing test data in to allow execution of 2nd test plan + self.execute_assignment_tests( + test_plan_with_os_inherit_disabled, test_data) + def test_list_projects_for_user_with_inherited_group_grants(self): """Test inherited group roles. @@ -5862,6 +6436,48 @@ class InheritanceTests(object): user_projects = self.assignment_api.list_projects_for_user(user1['id']) self.assertEqual(5, len(user_projects)) + # TODO(henry-nash): The test above uses list_projects_for_user + # which may, in a subsequent patch, be re-implemeted to call + # list_role_assignments and then report only the distinct projects. + # + # The test plan below therefore mirrors this test, to ensure that + # list_role_assignments works the same. Once list_projects_for_user + # has been re-implemented then the manual tests above can be + # refactored. + test_plan = { + # A domain with a 1 project, plus a second domain with 2 projects, + # as well as a user & group and a 3rd domain with 2 projects. + # Also, created 2 roles. + 'entities': {'domains': [{'projects': 1}, + {'users': 1, 'groups': 1, 'projects': 2}, + {'projects': 2}], + 'roles': 2}, + 'group_memberships': [{'group': 0, 'users': [0]}], + 'assignments': [{'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 0, 'project': 3}, + {'user': 0, 'role': 1, 'domain': 1, + 'inherited_to_projects': True}, + {'user': 0, 'role': 1, 'domain': 2, + 'inherited_to_projects': True}], + 'tests': [ + # List all effective assignments for user[0] + # Should get back both direct roles plus roles on both projects + # from each domain. Duplicates should not be fitered out. + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 3}, + {'user': 0, 'role': 0, 'project': 0}, + {'user': 0, 'role': 1, 'project': 1, + 'indirect': {'domain': 1}}, + {'user': 0, 'role': 1, 'project': 2, + 'indirect': {'domain': 1}}, + {'user': 0, 'role': 1, 'project': 3, + 'indirect': {'domain': 2}}, + {'user': 0, 'role': 1, 'project': 4, + 'indirect': {'domain': 2}}]} + ] + } + self.execute_assignment_test_plan(test_plan) + def test_list_projects_for_user_with_inherited_group_project_grants(self): """Test inherited role assignments for groups on nested projects. @@ -5925,6 +6541,53 @@ class InheritanceTests(object): self.assertEqual(1, len(user_projects)) self.assertIn(root_project, user_projects) + # TODO(henry-nash): The test above uses list_projects_for_user + # which may, in a subsequent patch, be re-implemeted to call + # list_role_assignments and then report only the distinct projects. + # + # The test plan below therefore mirrors this test, to ensure that + # list_role_assignments works the same. Once list_projects_for_user + # has been re-implemented then the manual tests above can be + # refactored. + test_plan = { + # A domain with a project ans sub-project, plus a user. + # Also, create 2 roles. + 'entities': { + 'domains': {'id': DEFAULT_DOMAIN_ID, 'users': 1, 'groups': 1, + 'projects': {'project': 1}}, + 'roles': 2}, + 'group_memberships': [{'group': 0, 'users': [0]}], + # A direct role and an inherited role on the parent + 'assignments': [{'group': 0, 'role': 0, 'project': 0}, + {'group': 0, 'role': 1, 'project': 0, + 'inherited_to_projects': True}], + 'tests': [ + # List all effective assignments for user[0] - should get back + # one direct role plus one inherited role. + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0, + 'indirect': {'group': 0}}, + {'user': 0, 'role': 1, 'project': 1, + 'indirect': {'group': 0, 'project': 0}}]} + ] + } + + test_plan_with_os_inherit_disabled = { + 'tests': [ + # List all effective assignments for user[0] - should only get + # back the one direct role. + {'params': {'user': 0, 'effective': True}, + 'results': [{'user': 0, 'role': 0, 'project': 0, + 'indirect': {'group': 0}}]} + ] + } + self.config_fixture.config(group='os_inherit', enabled=True) + test_data = self.execute_assignment_test_plan(test_plan) + self.config_fixture.config(group='os_inherit', enabled=False) + # Pass the existing test data in to allow execution of 2nd test plan + self.execute_assignment_tests( + test_plan_with_os_inherit_disabled, test_data) + class FilterTests(filtering.FilterTests): def test_list_entities_filtered(self): |