cleanup dependent data upon user/tenant deletion

fixes bug 974199
fixes bug 973243

* upon deletion of tenant also delete user tenant relations
* upon deletion of tenant or user also delete corresponding metadata
* add foreign keys in metadata to ensure consistency

see also: https://bugs.launchpad.net/keystone/+bug/959294/comments/16

Change-Id: I264714fe82b727e3e0f5273bcb781a580a3f3826

AUTHORS

@@ -39,6 +39,7 @@ gholt <gholt@brim.net>
 Guang Yee <guang.yee@hp.com>
 Hengqing Hu <hudayou@hotmail.com>
 Ionuț Arțăriși <iartarisi@suse.cz>
+J. Daniel Schmidt <jdsn@suse.de>
 jabdul <abdulkader.j@hcl.com>
 James E. Blair <corvus@gnu.org>
 James E. Blair <james.blair@rackspace.com>

keystone/identity/backends/sql.py

@@ -330,13 +330,20 @@ def delete_user(self, user_id):
         membership_refs = session.query(UserTenantMembership)\
                           .filter_by(user_id=user_id)\
                           .all()
+        metadata_refs = session.query(Metadata)\
+                        .filter_by(user_id=user_id)\
+                        .all()
 
         with session.begin():
             if membership_refs:
                 for membership_ref in membership_refs:
                     session.delete(membership_ref)
                     session.flush()
 
+            if metadata_refs:
+                for metadata_ref in metadata_refs:
+                    session.delete(metadata_ref)
+
             session.delete(user_ref)
             session.flush()
 
@@ -367,7 +374,21 @@ def update_tenant(self, tenant_id, tenant):
     def delete_tenant(self, tenant_id):
         session = self.get_session()
         tenant_ref = session.query(Tenant).filter_by(id=tenant_id).first()
+        membership_refs = session.query(UserTenantMembership)\
+                          .filter_by(tenant_id=tenant_id)\
+                          .all()
+        metadata_refs = session.query(Metadata)\
+                        .filter_by(tenant_id=tenant_id)\
+                        .all()
+
         with session.begin():
+            if membership_refs:
+                for membership_ref in membership_refs:
+                    session.delete(membership_ref)
+            if metadata_refs:
+                for metadata_ref in metadata_refs:
+                    session.delete(metadata_ref)
+
             session.delete(tenant_ref)
             session.flush()
 

tests/test_backend_sql.py

@@ -48,6 +48,41 @@ def test_delete_user_with_tenant_association(self):
         tenants = self.identity_api.get_tenants_for_user(user['id'])
         self.assertEquals(tenants, [])
 
+    def test_delete_tenant_with_user_association(self):
+        user = {'id': 'fake',
+                'name': 'fakeuser',
+                'password': 'passwd'}
+        self.identity_api.create_user('fake', user)
+        self.identity_api.add_user_to_tenant(self.tenant_bar['id'],
+                                             user['id'])
+        self.identity_api.delete_tenant(self.tenant_bar['id'])
+        tenants = self.identity_api.get_tenants_for_user(user['id'])
+        self.assertEquals(tenants, [])
+
+    def test_delete_user_with_metadata(self):
+        user = {'id': 'fake',
+                'name': 'fakeuser',
+                'password': 'passwd'}
+        self.identity_api.create_user('fake', user)
+        self.identity_api.create_metadata(user['id'],
+                                          self.tenant_bar['id'],
+                                          {'extra': 'extra'})
+        self.identity_api.delete_user(user['id'])
+        metadata = self.identity_api.get_metadata(user['id'], self.tenant_bar['id'])
+        self.assertEquals(metadata, {})
+
+    def test_delete_tenant_with_metadata(self):
+        user = {'id': 'fake',
+                'name': 'fakeuser',
+                'password': 'passwd'}
+        self.identity_api.create_user('fake', user)
+        self.identity_api.create_metadata(user['id'],
+                                          self.tenant_bar['id'],
+                                          {'extra': 'extra'})
+        self.identity_api.delete_tenant(self.tenant_bar['id'])
+        metadata = self.identity_api.get_metadata(user['id'], self.tenant_bar['id'])
+        self.assertEquals(metadata, {})
+
 
 class SqlToken(test.TestCase, test_backend.TokenTests):
     def setUp(self):