diff options
Diffstat (limited to 'oslo_db/tests/sqlalchemy/test_migrations.py')
| -rw-r--r-- | oslo_db/tests/sqlalchemy/test_migrations.py | 162 |
1 files changed, 0 insertions, 162 deletions
diff --git a/oslo_db/tests/sqlalchemy/test_migrations.py b/oslo_db/tests/sqlalchemy/test_migrations.py index fe15191..275c277 100644 --- a/oslo_db/tests/sqlalchemy/test_migrations.py +++ b/oslo_db/tests/sqlalchemy/test_migrations.py @@ -16,175 +16,13 @@ from unittest import mock -import fixtures -from migrate.versioning import api as versioning_api import sqlalchemy as sa from sqlalchemy import orm -from oslo_db import exception as exc from oslo_db.sqlalchemy import test_migrations as migrate -from oslo_db.tests import base as test_base from oslo_db.tests.sqlalchemy import base as db_test_base -class TestWalkVersions(test_base.BaseTestCase, migrate.WalkVersionsMixin): - migration_api = mock.MagicMock() - REPOSITORY = mock.MagicMock() - engine = mock.MagicMock() - INIT_VERSION = versioning_api.VerNum(4) - - @property - def migrate_engine(self): - return self.engine - - def test_migrate_up(self): - self.migration_api.db_version.return_value = 141 - - self.migrate_up(141) - - self.migration_api.upgrade.assert_called_with( - self.engine, self.REPOSITORY, 141) - self.migration_api.db_version.assert_called_with( - self.engine, self.REPOSITORY) - - @staticmethod - def _fake_upgrade_boom(*args, **kwargs): - raise exc.DBMigrationError("boom") - - def test_migrate_up_fail(self): - version = 141 - self.migration_api.db_version.return_value = version - expected_output = ( - "Failed to migrate to version %(version)s on " - "engine %(engine)s\n" % - {'version': version, 'engine': self.engine}) - - with mock.patch.object( - self.migration_api, - 'upgrade', - side_effect=self._fake_upgrade_boom, - ): - log = self.useFixture(fixtures.FakeLogger()) - self.assertRaises(exc.DBMigrationError, self.migrate_up, version) - self.assertEqual(expected_output, log.output) - - def test_migrate_up_with_data(self): - test_value = {"a": 1, "b": 2} - self.migration_api.db_version.return_value = 141 - self._pre_upgrade_141 = mock.MagicMock() - self._pre_upgrade_141.return_value = test_value - self._check_141 = mock.MagicMock() - - self.migrate_up(141, True) - - self._pre_upgrade_141.assert_called_with(self.engine) - self._check_141.assert_called_with(self.engine, test_value) - - def test_migrate_down(self): - self.migration_api.db_version.return_value = 42 - - self.assertTrue(self.migrate_down(42)) - self.migration_api.db_version.assert_called_with( - self.engine, self.REPOSITORY) - - def test_migrate_down_not_implemented(self): - with mock.patch.object( - self.migration_api, - 'downgrade', - side_effect=NotImplementedError, - ): - self.assertFalse(self.migrate_down(self.engine, 42)) - - def test_migrate_down_with_data(self): - self._post_downgrade_043 = mock.MagicMock() - self.migration_api.db_version.return_value = 42 - - self.migrate_down(42, True) - - self._post_downgrade_043.assert_called_with(self.engine) - - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_up') - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_down') - def test_walk_versions_all_default(self, migrate_up, migrate_down): - self.REPOSITORY.latest = versioning_api.VerNum(20) - self.migration_api.db_version.return_value = self.INIT_VERSION - - self.walk_versions() - - self.migration_api.version_control.assert_called_with( - self.engine, self.REPOSITORY, self.INIT_VERSION) - self.migration_api.db_version.assert_called_with( - self.engine, self.REPOSITORY) - - versions = range(int(self.INIT_VERSION) + 1, - int(self.REPOSITORY.latest) + 1) - upgraded = [mock.call(v, with_data=True) - for v in versions] - self.assertEqual(upgraded, self.migrate_up.call_args_list) - - downgraded = [mock.call(v - 1) for v in reversed(versions)] - self.assertEqual(downgraded, self.migrate_down.call_args_list) - - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_up') - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_down') - def test_walk_versions_all_true(self, migrate_up, migrate_down): - self.REPOSITORY.latest = versioning_api.VerNum(20) - self.migration_api.db_version.return_value = self.INIT_VERSION - - self.walk_versions(snake_walk=True, downgrade=True) - - versions = range(int(self.INIT_VERSION) + 1, - int(self.REPOSITORY.latest) + 1) - upgraded = [] - for v in versions: - upgraded.append(mock.call(v, with_data=True)) - upgraded.append(mock.call(v)) - upgraded.extend([mock.call(v) for v in reversed(versions)]) - self.assertEqual(upgraded, self.migrate_up.call_args_list) - - downgraded_1 = [mock.call(v - 1, with_data=True) for v in versions] - downgraded_2 = [] - for v in reversed(versions): - downgraded_2.append(mock.call(v - 1)) - downgraded_2.append(mock.call(v - 1)) - downgraded = downgraded_1 + downgraded_2 - self.assertEqual(downgraded, self.migrate_down.call_args_list) - - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_up') - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_down') - def test_walk_versions_true_false(self, migrate_up, migrate_down): - self.REPOSITORY.latest = versioning_api.VerNum(20) - self.migration_api.db_version.return_value = self.INIT_VERSION - - self.walk_versions(snake_walk=True, downgrade=False) - - versions = range(int(self.INIT_VERSION) + 1, - int(self.REPOSITORY.latest) + 1) - - upgraded = [] - for v in versions: - upgraded.append(mock.call(v, with_data=True)) - upgraded.append(mock.call(v)) - self.assertEqual(upgraded, self.migrate_up.call_args_list) - - downgraded = [mock.call(v - 1, with_data=True) for v in versions] - self.assertEqual(downgraded, self.migrate_down.call_args_list) - - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_up') - @mock.patch.object(migrate.WalkVersionsMixin, 'migrate_down') - def test_walk_versions_all_false(self, migrate_up, migrate_down): - self.REPOSITORY.latest = versioning_api.VerNum(20) - self.migration_api.db_version.return_value = self.INIT_VERSION - - self.walk_versions(snake_walk=False, downgrade=False) - - versions = range(int(self.INIT_VERSION) + 1, - int(self.REPOSITORY.latest) + 1) - - upgraded = [mock.call(v, with_data=True) for v in versions] - self.assertEqual(upgraded, self.migrate_up.call_args_list) - - class ModelsMigrationSyncMixin(db_test_base._DbTestCase): def setUp(self): |