diff options
Diffstat (limited to 'tests/test_sql_upgrade.py')
| -rw-r--r-- | tests/test_sql_upgrade.py | 146 |
1 files changed, 89 insertions, 57 deletions
diff --git a/tests/test_sql_upgrade.py b/tests/test_sql_upgrade.py index e6d6162b..9540c4cd 100644 --- a/tests/test_sql_upgrade.py +++ b/tests/test_sql_upgrade.py @@ -45,8 +45,7 @@ CONF = config.CONF DEFAULT_DOMAIN_ID = CONF.identity.default_domain_id -class SqlUpgradeTests(test.TestCase): - +class SqlMigrateBase(test.TestCase): def initialize_sql(self): self.metadata = sqlalchemy.MetaData() self.metadata.bind = self.engine @@ -55,12 +54,15 @@ class SqlUpgradeTests(test.TestCase): test.testsdir('test_overrides.conf'), test.testsdir('backend_sql.conf')] - #override this to sepcify the complete list of configuration files + #override this to specify the complete list of configuration files def config_files(self): return self._config_file_list + def repo_package(self): + return None + def setUp(self): - super(SqlUpgradeTests, self).setUp() + super(SqlMigrateBase, self).setUp() self.config(self.config_files()) self.base = sql.Base() @@ -71,7 +73,7 @@ class SqlUpgradeTests(test.TestCase): autocommit=False) self.initialize_sql() - self.repo_path = migration._find_migrate_repo() + self.repo_path = migration.find_migrate_repo(self.repo_package()) self.schema = versioning_api.ControlledSchema.create( self.engine, self.repo_path, 0) @@ -85,7 +87,64 @@ class SqlUpgradeTests(test.TestCase): autoload=True) self.downgrade(0) table.drop(self.engine, checkfirst=True) - super(SqlUpgradeTests, self).tearDown() + super(SqlMigrateBase, self).tearDown() + + def select_table(self, name): + table = sqlalchemy.Table(name, + self.metadata, + autoload=True) + s = sqlalchemy.select([table]) + return s + + def assertTableExists(self, table_name): + try: + self.select_table(table_name) + except sqlalchemy.exc.NoSuchTableError: + raise AssertionError('Table "%s" does not exist' % table_name) + + def assertTableDoesNotExist(self, table_name): + """Asserts that a given table exists cannot be selected by name.""" + # Switch to a different metadata otherwise you might still + # detect renamed or dropped tables + try: + temp_metadata = sqlalchemy.MetaData() + temp_metadata.bind = self.engine + sqlalchemy.Table(table_name, temp_metadata, autoload=True) + except sqlalchemy.exc.NoSuchTableError: + pass + else: + raise AssertionError('Table "%s" already exists' % table_name) + + def upgrade(self, *args, **kwargs): + self._migrate(*args, **kwargs) + + def downgrade(self, *args, **kwargs): + self._migrate(*args, downgrade=True, **kwargs) + + def _migrate(self, version, repository=None, downgrade=False, + current_schema=None): + repository = repository or self.repo_path + err = '' + version = versioning_api._migrate_version(self.schema, + version, + not downgrade, + err) + if not current_schema: + current_schema = self.schema + changeset = current_schema.changeset(version) + for ver, change in changeset: + self.schema.runchange(ver, change, changeset.step) + self.assertEqual(self.schema.version, version) + + def assertTableColumns(self, table_name, expected_cols): + """Asserts that the table contains the expected set of columns.""" + self.initialize_sql() + table = self.select_table(table_name) + actual_cols = [col.name for col in table.columns] + self.assertEqual(expected_cols, actual_cols, '%s table' % table_name) + + +class SqlUpgradeTests(SqlMigrateBase): def test_blank_db_to_start(self): self.assertTableDoesNotExist('user') @@ -108,13 +167,6 @@ class SqlUpgradeTests(test.TestCase): self.downgrade(x - 1) self.upgrade(x) - def assertTableColumns(self, table_name, expected_cols): - """Asserts that the table contains the expected set of columns.""" - self.initialize_sql() - table = self.select_table(table_name) - actual_cols = [col.name for col in table.columns] - self.assertEqual(expected_cols, actual_cols, '%s table' % table_name) - def test_upgrade_add_initial_tables(self): self.upgrade(1) self.assertTableColumns("user", ["id", "name", "extra"]) @@ -594,6 +646,22 @@ class SqlUpgradeTests(test.TestCase): 'data': json.dumps({"roles": [role2['id']]})} session.execute(metadata_table.insert().values(role_grant)) + # Create another user to test the case where member_role_id is already + # assigned. + user2 = {'id': uuid.uuid4().hex, + 'name': uuid.uuid4().hex, + 'domain_id': domain['id'], + 'password': uuid.uuid4().hex, + 'enabled': True, + 'extra': json.dumps({})} + session.execute(user_table.insert().values(user2)) + + # Grant CONF.member_role_id to User2 + role_grant = {'user_id': user2['id'], + 'tenant_id': project['id'], + 'data': json.dumps({"roles": [CONF.member_role_id]})} + session.execute(metadata_table.insert().values(role_grant)) + session.commit() self.upgrade(17) @@ -635,6 +703,14 @@ class SqlUpgradeTests(test.TestCase): 'data': data} cmd = user_project_metadata_table.insert().values(role_grant) self.engine.execute(cmd) + + # Create another conflicting user-project for User2 + data = json.dumps({"roles": [role2['id']]}) + role_grant = {'user_id': user2['id'], + 'project_id': project['id'], + 'data': data} + cmd = user_project_metadata_table.insert().values(role_grant) + self.engine.execute(cmd) # End Scaffolding session.commit() @@ -1278,50 +1354,6 @@ class SqlUpgradeTests(test.TestCase): 'extra': json.dumps(extra)}) self.engine.execute(ins) - def select_table(self, name): - table = sqlalchemy.Table(name, - self.metadata, - autoload=True) - s = sqlalchemy.select([table]) - return s - - def assertTableExists(self, table_name): - try: - self.select_table(table_name) - except sqlalchemy.exc.NoSuchTableError: - raise AssertionError('Table "%s" does not exist' % table_name) - - def assertTableDoesNotExist(self, table_name): - """Asserts that a given table exists cannot be selected by name.""" - # Switch to a different metadata otherwise you might still - # detect renamed or dropped tables - try: - temp_metadata = sqlalchemy.MetaData() - temp_metadata.bind = self.engine - sqlalchemy.Table(table_name, temp_metadata, autoload=True) - except sqlalchemy.exc.NoSuchTableError: - pass - else: - raise AssertionError('Table "%s" already exists' % table_name) - - def upgrade(self, *args, **kwargs): - self._migrate(*args, **kwargs) - - def downgrade(self, *args, **kwargs): - self._migrate(*args, downgrade=True, **kwargs) - - def _migrate(self, version, repository=None, downgrade=False): - repository = repository or self.repo_path - err = '' - version = versioning_api._migrate_version(self.schema, - version, - not downgrade, - err) - changeset = self.schema.changeset(version) - for ver, change in changeset: - self.schema.runchange(ver, change, changeset.step) - self.assertEqual(self.schema.version, version) - def _mysql_check_all_tables_innodb(self): database = self.engine.url.database |