Merge "tests: Remove 'ModelsMigrationsSync.check_foreign_keys'"

This commit is contained in:
Zuul 2021-07-28 18:34:36 +00:00 committed by Gerrit Code Review
commit a3ee5b9942
3 changed files with 38 additions and 300 deletions

View File

@ -15,7 +15,6 @@
# under the License.
import abc
import collections
import functools
import logging
import pprint
@ -121,30 +120,6 @@ class WalkVersionsMixin(object, metaclass=abc.ABCMeta):
"""
pass
def _walk_versions(self, snake_walk=False, downgrade=True):
"""Check if migration upgrades and downgrades successfully.
DEPRECATED: this function is deprecated and will be removed from
oslo.db in a few releases. Please use walk_versions() method instead.
"""
self.walk_versions(snake_walk, downgrade)
def _migrate_down(self, version, with_data=False):
"""Migrate down to a previous version of the db.
DEPRECATED: this function is deprecated and will be removed from
oslo.db in a few releases. Please use migrate_down() method instead.
"""
return self.migrate_down(version, with_data)
def _migrate_up(self, version, with_data=False):
"""Migrate up to a new version of the db.
DEPRECATED: this function is deprecated and will be removed from
oslo.db in a few releases. Please use migrate_up() method instead.
"""
self.migrate_up(version, with_data)
def walk_versions(self, snake_walk=False, downgrade=True):
"""Check if migration upgrades and downgrades successfully.
@ -478,77 +453,6 @@ class ModelsMigrationsSync(object, metaclass=abc.ABCMeta):
isinstance(meta_def.arg, expr.False_) and insp_def == "0"
)
FKInfo = collections.namedtuple('fk_info', ['constrained_columns',
'referred_table',
'referred_columns'])
def check_foreign_keys(self, metadata, bind):
"""Compare foreign keys between model and db table.
:returns: a list that contains information about:
* should be a new key added or removed existing,
* name of that key,
* source table,
* referred table,
* constrained columns,
* referred columns
Output::
[('drop_key',
'testtbl_fk_check_fkey',
'testtbl',
fk_info(constrained_columns=('fk_check',),
referred_table='table',
referred_columns=('fk_check',)))]
DEPRECATED: this function is deprecated and will be removed from
oslo.db in a few releases. Alembic autogenerate.compare_metadata()
now includes foreign key comparison directly.
"""
diff = []
insp = sqlalchemy.engine.reflection.Inspector.from_engine(bind)
# Get all tables from db
db_tables = insp.get_table_names()
# Get all tables from models
model_tables = metadata.tables
for table in db_tables:
if table not in model_tables:
continue
# Get all necessary information about key of current table from db
fk_db = dict((self._get_fk_info_from_db(i), i['name'])
for i in insp.get_foreign_keys(table))
fk_db_set = set(fk_db.keys())
# Get all necessary information about key of current table from
# models
fk_models = dict((self._get_fk_info_from_model(fk), fk)
for fk in model_tables[table].foreign_keys)
fk_models_set = set(fk_models.keys())
for key in (fk_db_set - fk_models_set):
diff.append(('drop_key', fk_db[key], table, key))
LOG.info(("Detected removed foreign key %(fk)r on "
"table %(table)r"), {'fk': fk_db[key],
'table': table})
for key in (fk_models_set - fk_db_set):
diff.append(('add_key', fk_models[key], table, key))
LOG.info((
"Detected added foreign key for column %(fk)r on table "
"%(table)r"), {'fk': fk_models[key].column.name,
'table': table})
return diff
def _get_fk_info_from_db(self, fk):
return self.FKInfo(tuple(fk['constrained_columns']),
fk['referred_table'],
tuple(fk['referred_columns']))
def _get_fk_info_from_model(self, fk):
return self.FKInfo((fk.parent.name,), fk.column.table.name,
(fk.column.name,))
def filter_metadata_diff(self, diff):
"""Filter changes before assert in test_models_sync().

View File

@ -13,6 +13,7 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from unittest import mock
import fixtures
@ -53,13 +54,16 @@ class TestWalkVersions(test.BaseTestCase, migrate.WalkVersionsMixin):
def test_migrate_up_fail(self):
version = 141
self.migration_api.db_version.return_value = version
expected_output = (u"Failed to migrate to version %(version)s on "
"engine %(engine)s\n" %
{'version': version, 'engine': self.engine})
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):
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)
@ -84,9 +88,11 @@ class TestWalkVersions(test.BaseTestCase, migrate.WalkVersionsMixin):
self.engine, self.REPOSITORY)
def test_migrate_down_not_implemented(self):
with mock.patch.object(self.migration_api,
'downgrade',
side_effect=NotImplementedError):
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):
@ -259,8 +265,7 @@ class ModelsMigrationSyncMixin(test_base._DbTestCase):
def include_object(self, object_, name, type_, reflected, compare_to):
if type_ == 'table':
return name == 'testtbl'
else:
return True
return True
def _test_models_not_sync_filtered(self):
self.metadata_migrations.clear()
@ -359,9 +364,11 @@ class ModelsMigrationSyncMixin(test_base._DbTestCase):
self.assertIn('variant', msg)
class ModelsMigrationsSyncMysql(ModelsMigrationSyncMixin,
migrate.ModelsMigrationsSync,
test_base._MySQLOpportunisticTestCase):
class ModelsMigrationsSyncMySQL(
ModelsMigrationSyncMixin,
migrate.ModelsMigrationsSync,
test_base._MySQLOpportunisticTestCase,
):
def test_models_not_sync(self):
self._test_models_not_sync()
@ -370,199 +377,14 @@ class ModelsMigrationsSyncMysql(ModelsMigrationSyncMixin,
self._test_models_not_sync_filtered()
class ModelsMigrationsSyncPsql(ModelsMigrationSyncMixin,
migrate.ModelsMigrationsSync,
test_base._PostgreSQLOpportunisticTestCase):
class ModelsMigrationsSyncPostgreSQL(
ModelsMigrationSyncMixin,
migrate.ModelsMigrationsSync,
test_base._PostgreSQLOpportunisticTestCase,
):
def test_models_not_sync(self):
self._test_models_not_sync()
def test_models_not_sync_filtered(self):
self._test_models_not_sync_filtered()
class TestOldCheckForeignKeys(test_base._DbTestCase):
def setUp(self):
super(TestOldCheckForeignKeys, self).setUp()
test = self
class MigrateSync(migrate.ModelsMigrationsSync):
def get_engine(self):
return test.engine
def get_metadata(self):
return test.metadata
def db_sync(self):
raise NotImplementedError()
self.migration_sync = MigrateSync()
def _fk_added_fixture(self):
self.metadata = sa.MetaData()
self.metadata_migrations = sa.MetaData()
sa.Table(
'testtbl_one', self.metadata,
sa.Column('id', sa.Integer, primary_key=True),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_two', self.metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('tone_id', sa.Integer),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_one', self.metadata_migrations,
sa.Column('id', sa.Integer, primary_key=True),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_two', self.metadata_migrations,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column(
'tone_id', sa.Integer,
sa.ForeignKey('testtbl_one.id', name="tone_id_fk")),
mysql_engine='InnoDB'
)
def _fk_removed_fixture(self):
self.metadata = sa.MetaData()
self.metadata_migrations = sa.MetaData()
sa.Table(
'testtbl_one', self.metadata,
sa.Column('id', sa.Integer, primary_key=True),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_two', self.metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column(
'tone_id', sa.Integer,
sa.ForeignKey('testtbl_one.id', name="tone_id_fk")),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_one', self.metadata_migrations,
sa.Column('id', sa.Integer, primary_key=True),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_two', self.metadata_migrations,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('tone_id', sa.Integer),
mysql_engine='InnoDB'
)
def _fk_no_change_fixture(self):
self.metadata = sa.MetaData()
self.metadata_migrations = sa.MetaData()
sa.Table(
'testtbl_one', self.metadata,
sa.Column('id', sa.Integer, primary_key=True),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_two', self.metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column(
'tone_id', sa.Integer,
sa.ForeignKey('testtbl_one.id', name="tone_id_fk")),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_one', self.metadata_migrations,
sa.Column('id', sa.Integer, primary_key=True),
mysql_engine='InnoDB'
)
sa.Table(
'testtbl_two', self.metadata_migrations,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column(
'tone_id', sa.Integer,
sa.ForeignKey('testtbl_one.id', name="tone_id_fk")),
mysql_engine='InnoDB'
)
def _run_test(self):
self.metadata.create_all(bind=self.engine)
return self.migration_sync.check_foreign_keys(
self.metadata_migrations, self.engine)
def _compare_diffs(self, diffs, compare_to):
diffs = [
(
cmd,
fk._get_colspec() if isinstance(fk, sa.ForeignKey)
else "tone_id_fk" if fk is None # sqlite workaround
else fk,
tname, fk_info
)
for cmd, fk, tname, fk_info in diffs
]
self.assertEqual(compare_to, diffs)
def test_fk_added(self):
self._fk_added_fixture()
diffs = self._run_test()
self._compare_diffs(
diffs,
[(
'add_key',
'testtbl_one.id',
'testtbl_two',
self.migration_sync.FKInfo(
constrained_columns=('tone_id',),
referred_table='testtbl_one',
referred_columns=('id',))
)]
)
def test_fk_removed(self):
self._fk_removed_fixture()
diffs = self._run_test()
self._compare_diffs(
diffs,
[(
'drop_key',
"tone_id_fk",
'testtbl_two',
self.migration_sync.FKInfo(
constrained_columns=('tone_id',),
referred_table='testtbl_one',
referred_columns=('id',))
)]
)
def test_fk_no_change(self):
self._fk_no_change_fixture()
diffs = self._run_test()
self._compare_diffs(
diffs,
[])
class PGTestOldCheckForeignKeys(
TestOldCheckForeignKeys, test_base._PostgreSQLOpportunisticTestCase):
pass
class MySQLTestOldCheckForeignKeys(
TestOldCheckForeignKeys, test_base._MySQLOpportunisticTestCase):
pass

View File

@ -0,0 +1,12 @@
---
upgrade:
- |
The ``check_foreign_keys`` helper of the
``oslo_db.sqlalchemy.test_migrations.ModelsMigrationsSync`` base test class
has been removed. This was deprecated in 1.4.1 as alembic now supports this
capability.
- The ``_walk_versions``, ``_migrate_down``, and ``_migrate_up`` methods of
the ``oslo_db.sqlalchemy.test_migrations.ModelsMigrationsSync`` base test
class have been removed. These were deprecated in 0.5.0 in favour of their
non-private equivalents, ``walk_versions``, ``migrate_down``, and
``migrate_up`` respectively.