diff options
Diffstat (limited to 'nova/tests/unit/db/main/test_migrations.py')
| -rw-r--r-- | nova/tests/unit/db/main/test_migrations.py | 89 |
1 files changed, 45 insertions, 44 deletions
diff --git a/nova/tests/unit/db/main/test_migrations.py b/nova/tests/unit/db/main/test_migrations.py index f5ce3697b3..579888cfd2 100644 --- a/nova/tests/unit/db/main/test_migrations.py +++ b/nova/tests/unit/db/main/test_migrations.py @@ -25,11 +25,11 @@ test will then use that DB and username/password combo to run the tests. Refer to the 'tools/test-setup.sh' for an example of how to configure this. """ +from unittest import mock + from alembic import command as alembic_api from alembic import script as alembic_script import fixtures -from migrate.versioning import api as migrate_api -import mock from oslo_db.sqlalchemy import enginefacade from oslo_db.sqlalchemy import test_fixtures from oslo_db.sqlalchemy import test_migrations @@ -173,47 +173,6 @@ class TestModelsSyncPostgreSQL( FIXTURE = test_fixtures.PostgresqlOpportunisticFixture -class NovaModelsMigrationsLegacySync(NovaModelsMigrationsSync): - """Test that the models match the database after old migrations are run.""" - - def db_sync(self, engine): - # the 'nova.db.migration.db_sync' method will not use the legacy - # sqlalchemy-migrate-based migration flow unless the database is - # already controlled with sqlalchemy-migrate, so we need to manually - # enable version controlling with this tool to test this code path - repository = migration._find_migrate_repo(database='main') - migrate_api.version_control( - engine, repository, migration.MIGRATE_INIT_VERSION['main']) - - # now we can apply migrations as expected and the legacy path will be - # followed - super().db_sync(engine) - - -class TestModelsLegacySyncSQLite( - NovaModelsMigrationsLegacySync, - test_fixtures.OpportunisticDBTestMixin, - base.BaseTestCase, -): - pass - - -class TestModelsLegacySyncMySQL( - NovaModelsMigrationsLegacySync, - test_fixtures.OpportunisticDBTestMixin, - base.BaseTestCase, -): - FIXTURE = test_fixtures.MySQLOpportunisticFixture - - -class TestModelsLegacySyncPostgreSQL( - NovaModelsMigrationsLegacySync, - test_fixtures.OpportunisticDBTestMixin, - base.BaseTestCase, -): - FIXTURE = test_fixtures.PostgresqlOpportunisticFixture - - class NovaMigrationsWalk( test_fixtures.OpportunisticDBTestMixin, test.NoDBTestCase, ): @@ -226,7 +185,7 @@ class NovaMigrationsWalk( super().setUp() self.engine = enginefacade.writer.get_engine() self.config = migration._find_alembic_conf('main') - self.init_version = migration.ALEMBIC_INIT_VERSION['main'] + self.init_version = '8f2f1571d55b' def assertIndexExists(self, connection, table_name, index): self.assertTrue( @@ -240,6 +199,12 @@ class NovaMigrationsWalk( 'Index %s on table %s should not exist' % (index, table_name), ) + def assertColumnExists(self, connection, table_name, column): + self.assertTrue( + oslodbutils.column_exists(connection, table_name, column), + 'Column %s on table %s should exist' % (column, table_name), + ) + def _migrate_up(self, connection, revision): if revision == self.init_version: # no tests for the initial revision alembic_api.upgrade(self.config, revision) @@ -280,6 +245,42 @@ class NovaMigrationsWalk( # no check for the MySQL-specific change + def _check_ccb0fa1a2252(self, connection): + for prefix in ('', 'shadow_'): + table_name = prefix + 'block_device_mapping' + table = oslodbutils.get_table(connection, table_name) + + self.assertColumnExists(connection, table_name, 'encrypted') + self.assertColumnExists( + connection, table_name, 'encryption_secret_uuid') + self.assertColumnExists( + connection, table_name, 'encryption_format') + self.assertColumnExists( + connection, table_name, 'encryption_options') + + # Only check for the expected types if we're using sqlite because + # other databases' types may be different. For example, Boolean + # may be represented as an integer in MySQL + if connection.engine.name != 'sqlite': + return + + self.assertIsInstance(table.c.encrypted.type, sa.types.Boolean) + self.assertIsInstance( + table.c.encryption_secret_uuid.type, sa.types.String) + self.assertIsInstance( + table.c.encryption_format.type, sa.types.String) + self.assertIsInstance( + table.c.encryption_options.type, sa.types.String) + + def _check_960aac0e09ea(self, connection): + self.assertIndexNotExists( + connection, 'console_auth_tokens', + 'console_auth_tokens_token_hash_idx', + ) + self.assertIndexNotExists( + connection, 'instances', 'uuid', + ) + def test_single_base_revision(self): """Ensure we only have a single base revision. |