Migrate roles from metadata to user_project_metadata

migration 017 did not migrate existing roles from the metadata table
that was created in 001. Adding a migration (20) that compares the roles
in the metadata table (if any) and joins them to the new
user_project_metadata role that matches the user and tenant from the old
table. Also adding subsequent tests to check both of the issues above.
bug 1131087

Change-Id: I00ea6043d949c9c358827e25f05c63515fe5dea8

1 files changed, 132 insertions, 4 deletions

diff --git a/tests/test_sql_upgrade.py b/tests/test_sql_upgrade.py
index 62c69cc9..1ba9b419 100644
--- a/tests/test_sql_upgrade.py
+++ b/tests/test_sql_upgrade.py
@@ -35,7 +35,6 @@ import sqlalchemy
 from keystone.common import sql
 from keystone.common.sql import migration
 from keystone import config
-from keystone import exception
 from keystone import test
 
 import default_fixtures
@@ -440,6 +439,137 @@ class SqlUpgradeTests(test.TestCase):
         self.assertTableColumns('user_domain_metadata',
                                 ['user_id', 'domain_id', 'data'])
 
+    def test_metadata_table_migration(self):
+        # Scaffolding
+        session = self.Session()
+
+        self.upgrade(16)
+        domain_table = sqlalchemy.Table('domain', self.metadata, autoload=True)
+        user_table = sqlalchemy.Table('user', self.metadata, autoload=True)
+        role_table = sqlalchemy.Table('role', self.metadata, autoload=True)
+        project_table = sqlalchemy.Table(
+            'project', self.metadata, autoload=True)
+        metadata_table = sqlalchemy.Table(
+            'metadata', self.metadata, autoload=True)
+
+        # Create a Domain
+        domain = {'id': uuid.uuid4().hex,
+                  'name': uuid.uuid4().hex,
+                  'enabled': True}
+        self.engine.execute(domain_table.insert().values(domain))
+
+        # Create a Project
+        project = {'id': uuid.uuid4().hex,
+                   'name': uuid.uuid4().hex,
+                   'domain_id': domain['id'],
+                   'extra': "{}"}
+        self.engine.execute(project_table.insert().values(project))
+
+        # Create another Project
+        project2 = {'id': uuid.uuid4().hex,
+                    'name': uuid.uuid4().hex,
+                    'domain_id': domain['id'],
+                    'extra': "{}"}
+        self.engine.execute(project_table.insert().values(project2))
+
+        # Create a User
+        user = {'id': uuid.uuid4().hex,
+                'name': uuid.uuid4().hex,
+                'domain_id': domain['id'],
+                'password': uuid.uuid4().hex,
+                'enabled': True,
+                'extra': json.dumps({})}
+        self.engine.execute(user_table.insert().values(user))
+
+        # Create a Role
+        role = {'id': uuid.uuid4().hex,
+                'name': uuid.uuid4().hex}
+        self.engine.execute(role_table.insert().values(role))
+
+        # And another role
+        role2 = {'id': uuid.uuid4().hex,
+                 'name': uuid.uuid4().hex}
+        self.engine.execute(role_table.insert().values(role2))
+
+        # Grant Role to User
+        role_grant = {'user_id': user['id'],
+                      'tenant_id': project['id'],
+                      'data': json.dumps({"roles": [role['id']]})}
+        self.engine.execute(metadata_table.insert().values(role_grant))
+
+        role_grant = {'user_id': user['id'],
+                      'tenant_id': project2['id'],
+                      'data': json.dumps({"roles": [role2['id']]})}
+        self.engine.execute(metadata_table.insert().values(role_grant))
+
+        session.commit()
+
+        self.upgrade(17)
+
+        user_project_metadata_table = sqlalchemy.Table(
+            'user_project_metadata', self.metadata, autoload=True)
+
+        # Test user in project has role
+        r = session.execute('select data from metadata where user_id="%s"'
+                            'and tenant_id="%s"' %
+                            (user['id'], project['id']))
+        test_project1 = json.loads(r.fetchone()['data'])
+        self.assertEqual(len(test_project1['roles']), 1)
+        self.assertIn(role['id'], test_project1['roles'])
+
+        # Test user in project2 has role2
+        r = session.execute('select data from metadata where user_id="%s"'
+                            ' and tenant_id="%s"' %
+                            (user['id'], project2['id']))
+        test_project2 = json.loads(r.fetchone()['data'])
+        self.assertEqual(len(test_project2['roles']), 1)
+        self.assertIn(role2['id'], test_project2['roles'])
+
+        # Test for user in project has role in user_project_metadata
+        # Migration 17 does not properly migrate this data, so this should
+        # be None.
+        r = session.execute('select data from user_project_metadata where '
+                            'user_id="%s" and project_id="%s"' %
+                            (user['id'], project['id']))
+        self.assertIsNone(r.fetchone())
+
+        # Create a conflicting user-project in user_project_metadata with
+        # a different role
+        data = json.dumps({"roles": [role2['id']]})
+        role_grant = {'user_id': user['id'],
+                      'project_id': project['id'],
+                      'data': data}
+        cmd = user_project_metadata_table.insert().values(role_grant)
+        self.engine.execute(cmd)
+        # End Scaffolding
+
+        # Migrate to 20
+        self.upgrade(20)
+
+        # The user-project pairs should have all roles from the previous
+        # metadata table in addition to any roles currently in
+        # user_project_metadata
+        r = session.execute('select data from user_project_metadata '
+                            'where user_id="%s" and project_id="%s"' %
+                            (user['id'], project['id']))
+        role_ids = json.loads(r.fetchone()['data'])['roles']
+        self.assertEqual(len(role_ids), 3)
+        self.assertIn(CONF.member_role_id, role_ids)
+        self.assertIn(role['id'], role_ids)
+        self.assertIn(role2['id'], role_ids)
+
+        # pairs that only existed in old metadata table should be in
+        # user_project_metadata
+        r = session.execute('select data from user_project_metadata where '
+                            'user_id="%s" and project_id="%s"' %
+                            (user['id'], project2['id']))
+        role_ids = json.loads(r.fetchone()['data'])['roles']
+        self.assertEqual(len(role_ids), 2)
+        self.assertIn(CONF.member_role_id, role_ids)
+        self.assertIn(role2['id'], role_ids)
+
+        self.assertTableDoesNotExist('metadata')
+
     def test_upgrade_default_roles(self):
         def count_member_roles():
             session = self.Session()
@@ -673,9 +803,7 @@ class SqlUpgradeTests(test.TestCase):
         try:
             temp_metadata = sqlalchemy.MetaData()
             temp_metadata.bind = self.engine
-            table = sqlalchemy.Table(table_name,
-                                     temp_metadata,
-                                     autoload=True)
+            sqlalchemy.Table(table_name, temp_metadata, autoload=True)
         except sqlalchemy.exc.NoSuchTableError:
             pass
         else: