diff options
Diffstat (limited to 'oslo_db/tests/sqlalchemy/test_migrate_cli.py')
| -rw-r--r-- | oslo_db/tests/sqlalchemy/test_migrate_cli.py | 222 |
1 files changed, 222 insertions, 0 deletions
diff --git a/oslo_db/tests/sqlalchemy/test_migrate_cli.py b/oslo_db/tests/sqlalchemy/test_migrate_cli.py new file mode 100644 index 0000000..c1ab53c --- /dev/null +++ b/oslo_db/tests/sqlalchemy/test_migrate_cli.py @@ -0,0 +1,222 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +from oslotest import base as test_base + +from oslo_db.sqlalchemy.migration_cli import ext_alembic +from oslo_db.sqlalchemy.migration_cli import ext_migrate +from oslo_db.sqlalchemy.migration_cli import manager + + +class MockWithCmp(mock.MagicMock): + + order = 0 + + def __init__(self, *args, **kwargs): + super(MockWithCmp, self).__init__(*args, **kwargs) + + self.__lt__ = lambda self, other: self.order < other.order + + +@mock.patch(('oslo_db.sqlalchemy.migration_cli.' + 'ext_alembic.alembic.command')) +class TestAlembicExtension(test_base.BaseTestCase): + + def setUp(self): + self.migration_config = {'alembic_ini_path': '.', + 'db_url': 'sqlite://'} + self.alembic = ext_alembic.AlembicExtension(self.migration_config) + super(TestAlembicExtension, self).setUp() + + def test_check_enabled_true(self, command): + """Check enabled returns True + + Verifies that enabled returns True on non empty + alembic_ini_path conf variable + """ + self.assertTrue(self.alembic.enabled) + + def test_check_enabled_false(self, command): + """Check enabled returns False + + Verifies enabled returns False on empty alembic_ini_path variable + """ + self.migration_config['alembic_ini_path'] = '' + alembic = ext_alembic.AlembicExtension(self.migration_config) + self.assertFalse(alembic.enabled) + + def test_upgrade_none(self, command): + self.alembic.upgrade(None) + command.upgrade.assert_called_once_with(self.alembic.config, 'head') + + def test_upgrade_normal(self, command): + self.alembic.upgrade('131daa') + command.upgrade.assert_called_once_with(self.alembic.config, '131daa') + + def test_downgrade_none(self, command): + self.alembic.downgrade(None) + command.downgrade.assert_called_once_with(self.alembic.config, 'base') + + def test_downgrade_int(self, command): + self.alembic.downgrade(111) + command.downgrade.assert_called_once_with(self.alembic.config, 'base') + + def test_downgrade_normal(self, command): + self.alembic.downgrade('131daa') + command.downgrade.assert_called_once_with( + self.alembic.config, '131daa') + + def test_revision(self, command): + self.alembic.revision(message='test', autogenerate=True) + command.revision.assert_called_once_with( + self.alembic.config, message='test', autogenerate=True) + + def test_stamp(self, command): + self.alembic.stamp('stamp') + command.stamp.assert_called_once_with( + self.alembic.config, revision='stamp') + + def test_version(self, command): + version = self.alembic.version() + self.assertIsNone(version) + + +@mock.patch(('oslo_db.sqlalchemy.migration_cli.' + 'ext_migrate.migration')) +class TestMigrateExtension(test_base.BaseTestCase): + + def setUp(self): + self.migration_config = {'migration_repo_path': '.', + 'db_url': 'sqlite://'} + self.migrate = ext_migrate.MigrateExtension(self.migration_config) + super(TestMigrateExtension, self).setUp() + + def test_check_enabled_true(self, migration): + self.assertTrue(self.migrate.enabled) + + def test_check_enabled_false(self, migration): + self.migration_config['migration_repo_path'] = '' + migrate = ext_migrate.MigrateExtension(self.migration_config) + self.assertFalse(migrate.enabled) + + def test_upgrade_head(self, migration): + self.migrate.upgrade('head') + migration.db_sync.assert_called_once_with( + self.migrate.engine, self.migrate.repository, None, init_version=0) + + def test_upgrade_normal(self, migration): + self.migrate.upgrade(111) + migration.db_sync.assert_called_once_with( + mock.ANY, self.migrate.repository, 111, init_version=0) + + def test_downgrade_init_version_from_base(self, migration): + self.migrate.downgrade('base') + migration.db_sync.assert_called_once_with( + self.migrate.engine, self.migrate.repository, mock.ANY, + init_version=mock.ANY) + + def test_downgrade_init_version_from_none(self, migration): + self.migrate.downgrade(None) + migration.db_sync.assert_called_once_with( + self.migrate.engine, self.migrate.repository, mock.ANY, + init_version=mock.ANY) + + def test_downgrade_normal(self, migration): + self.migrate.downgrade(101) + migration.db_sync.assert_called_once_with( + self.migrate.engine, self.migrate.repository, 101, init_version=0) + + def test_version(self, migration): + self.migrate.version() + migration.db_version.assert_called_once_with( + self.migrate.engine, self.migrate.repository, init_version=0) + + def test_change_init_version(self, migration): + self.migration_config['init_version'] = 101 + migrate = ext_migrate.MigrateExtension(self.migration_config) + migrate.downgrade(None) + migration.db_sync.assert_called_once_with( + migrate.engine, + self.migrate.repository, + self.migration_config['init_version'], + init_version=self.migration_config['init_version']) + + +class TestMigrationManager(test_base.BaseTestCase): + + def setUp(self): + self.migration_config = {'alembic_ini_path': '.', + 'migrate_repo_path': '.', + 'db_url': 'sqlite://'} + self.migration_manager = manager.MigrationManager( + self.migration_config) + self.ext = mock.Mock() + self.migration_manager._manager.extensions = [self.ext] + super(TestMigrationManager, self).setUp() + + def test_manager_update(self): + self.migration_manager.upgrade('head') + self.ext.obj.upgrade.assert_called_once_with('head') + + def test_manager_update_revision_none(self): + self.migration_manager.upgrade(None) + self.ext.obj.upgrade.assert_called_once_with(None) + + def test_downgrade_normal_revision(self): + self.migration_manager.downgrade('111abcd') + self.ext.obj.downgrade.assert_called_once_with('111abcd') + + def test_version(self): + self.migration_manager.version() + self.ext.obj.version.assert_called_once_with() + + def test_revision_message_autogenerate(self): + self.migration_manager.revision('test', True) + self.ext.obj.revision.assert_called_once_with('test', True) + + def test_revision_only_message(self): + self.migration_manager.revision('test', False) + self.ext.obj.revision.assert_called_once_with('test', False) + + def test_stamp(self): + self.migration_manager.stamp('stamp') + self.ext.obj.stamp.assert_called_once_with('stamp') + + +class TestMigrationRightOrder(test_base.BaseTestCase): + + def setUp(self): + self.migration_config = {'alembic_ini_path': '.', + 'migrate_repo_path': '.', + 'db_url': 'sqlite://'} + self.migration_manager = manager.MigrationManager( + self.migration_config) + self.first_ext = MockWithCmp() + self.first_ext.obj.order = 1 + self.first_ext.obj.upgrade.return_value = 100 + self.first_ext.obj.downgrade.return_value = 0 + self.second_ext = MockWithCmp() + self.second_ext.obj.order = 2 + self.second_ext.obj.upgrade.return_value = 200 + self.second_ext.obj.downgrade.return_value = 100 + self.migration_manager._manager.extensions = [self.first_ext, + self.second_ext] + super(TestMigrationRightOrder, self).setUp() + + def test_upgrade_right_order(self): + results = self.migration_manager.upgrade(None) + self.assertEqual(results, [100, 200]) + + def test_downgrade_right_order(self): + results = self.migration_manager.downgrade(None) + self.assertEqual(results, [100, 0]) |