openstack/deb-python-oslo.db
Code Issues Proposed changes

Sanity check after migration

Browse Source 
Added a sanity_check call after the migration, so if a table in the
migration scripts is created not correctly the exception arise
before to go in production.

Closes-Bug: #1428065

Change-Id: Ib91f2471fa389a6072e8c2aa302504a6c068aa0a
This commit is contained in:
Marco Fargetta
2015-03-04 11:22:05 +01:00
committed by Roman Podoliaka
parent 9c852c63cd
commit 9e004bdbc9
3 changed files with 11 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
oslo_db/sqlalchemy/migration.py
View File

@@ -76,10 +76,14 @@ def db_sync(engine, abs_path, version=None, init_version=0, sanity_check=True):
if sanity_check:
_db_schema_sanity_check(engine)
if version is None or version > current_version:
return versioning_api.upgrade(engine, repository, version)
migration = versioning_api.upgrade(engine, repository, version)
else:
return versioning_api.downgrade(engine, repository,
version)
migration = versioning_api.downgrade(engine, repository,
version)
if sanity_check:
_db_schema_sanity_check(engine)
return migration
def _db_schema_sanity_check(engine):

3
oslo_db/tests/old_import_api/sqlalchemy/test_migration_common.py
View File

@@ -158,7 +158,8 @@ class TestMigrationCommon(test_base.DbTestCase):
mock_find_repo.return_value = self.return_value
migration.db_sync(self.engine, self.path, self.test_version)
mock_sanity.assert_called_once_with(self.engine)
self.assertEqual([mock.call(self.engine), mock.call(self.engine)],
mock_sanity.call_args_list)
def test_db_sync_sanity_skipped(self):
with test_utils.nested(

3
oslo_db/tests/sqlalchemy/test_migration_common.py
View File

@@ -174,7 +174,8 @@ class TestMigrationCommon(test_base.DbTestCase):
mock_find_repo.return_value = self.return_value
migration.db_sync(self.engine, self.path, self.test_version)
mock_sanity.assert_called_once_with(self.engine)
self.assertEqual([mock.call(self.engine), mock.call(self.engine)],
mock_sanity.call_args_list)
def test_db_sync_sanity_skipped(self):
with test_utils.nested(