Merge "Allow users to clean up role assignments"

keystone/assignment/controllers.py

@@ -16,6 +16,7 @@
 """Workflow Logic the Assignment service."""
 
 import copy
+import functools
 import uuid
 
 import six
@@ -490,7 +491,8 @@ class RoleV3(controller.V3Controller):
 
     def _check_grant_protection(self, context, protection, role_id=None,
                                 user_id=None, group_id=None,
-                                domain_id=None, project_id=None):
+                                domain_id=None, project_id=None,
+                                allow_no_user=False):
         """Check protection for role grant APIs.
 
         The policy rule might want to inspect attributes of any of the entities
@@ -502,7 +504,11 @@ class RoleV3(controller.V3Controller):
         if role_id:
             ref['role'] = self.assignment_api.get_role(role_id)
         if user_id:
-            ref['user'] = self.identity_api.get_user(user_id)
+            try:
+                ref['user'] = self.identity_api.get_user(user_id)
+            except exception.UserNotFound:
+                if not allow_no_user:
+                    raise
         else:
             ref['group'] = self.identity_api.get_group(group_id)
 
@@ -547,7 +553,11 @@ class RoleV3(controller.V3Controller):
             role_id, user_id, group_id, domain_id, project_id,
             self._check_if_inherited(context))
 
-    @controller.protected(callback=_check_grant_protection)
+    # NOTE(lbragstad): This will allow users to clean up role assignments
+    # from the backend in the event the user was removed prior to the role
+    # assignment being removed.
+    @controller.protected(callback=functools.partial(
+        _check_grant_protection, allow_no_user=True))
     def revoke_grant(self, context, role_id, user_id=None,
                      group_id=None, domain_id=None, project_id=None):
         """Revokes a role from user/group on either a domain or project."""

keystone/tests/test_v3_identity.py

@@ -941,6 +941,45 @@ class IdentityTestCase(test_v3.RestfulTestCase):
         self.delete('/roles/%(role_id)s' % {
             'role_id': self.role_id})
 
+    def _create_new_user_and_assign_role_on_project(self):
+        """Create a new user and assign user a role on a project."""
+        # Create a new user
+        new_user = self.new_user_ref(domain_id=self.domain_id)
+        user_ref = self.identity_api.create_user(new_user)
+        # Assign the user a role on the project
+        collection_url = (
+            '/projects/%(project_id)s/users/%(user_id)s/roles' % {
+                'project_id': self.project_id,
+                'user_id': user_ref['id']})
+        member_url = ('%(collection_url)s/%(role_id)s' % {
+            'collection_url': collection_url,
+            'role_id': self.role_id})
+        self.put(member_url, expected_status=204)
+        # Check the user has the role assigned
+        self.head(member_url, expected_status=204)
+        return member_url, user_ref
+
+    # TODO(lbragstad): Move this test to tests/test_v3_assignment.py
+    def test_delete_user_before_removing_role_assignment_succeeds(self):
+        """Call ``DELETE`` on the user before the role assignment."""
+        member_url, user = self._create_new_user_and_assign_role_on_project()
+        # Delete the user from identity backend
+        self.identity_api.driver.delete_user(user['id'])
+        # Clean up the role assignment
+        self.delete(member_url, expected_status=204)
+        # Make sure the role is gone
+        self.head(member_url, expected_status=404)
+
+    # TODO(lbragstad): Move this test to tests/test_v3_assignment.py
+    def test_delete_user_and_check_role_assignment_fails(self):
+        """Call ``DELETE`` on the user and check the role assignment."""
+        member_url, user = self._create_new_user_and_assign_role_on_project()
+        # Delete the user from identity backend
+        self.identity_api.driver.delete_user(user['id'])
+        # We should get a 404 when looking for the user in the identity
+        # backend because we're not performing a delete operation on the role.
+        self.head(member_url, expected_status=404)
+
     def test_crud_user_project_role_grants(self):
         collection_url = (
             '/projects/%(project_id)s/users/%(user_id)s/roles' % {