diff options
author | Jim Rollenhagen <jim@jimrollenhagen.com> | 2019-09-26 09:56:42 -0400 |
---|---|---|
committer | Jim Rollenhagen <jim@jimrollenhagen.com> | 2019-09-26 09:56:42 -0400 |
commit | 52672a64cc0cab4ea14a4a756fce850eb03315e3 (patch) | |
tree | a86024e4e6141aa8983c750f751c58d924f5b11a /migrate/tests/versioning/test_genmodel.py | |
parent | 8acab2cd75a5b23ac162e49c8e4fb1e3f958352a (diff) | |
download | sqlalchemy-migrate-master.tar.gz |
Diffstat (limited to 'migrate/tests/versioning/test_genmodel.py')
-rw-r--r-- | migrate/tests/versioning/test_genmodel.py | 214 |
1 files changed, 0 insertions, 214 deletions
diff --git a/migrate/tests/versioning/test_genmodel.py b/migrate/tests/versioning/test_genmodel.py deleted file mode 100644 index f800826..0000000 --- a/migrate/tests/versioning/test_genmodel.py +++ /dev/null @@ -1,214 +0,0 @@ -# -*- coding: utf-8 -*- - -import os - -import six -import sqlalchemy -from sqlalchemy import * - -from migrate.versioning import genmodel, schemadiff -from migrate.changeset import schema - -from migrate.tests import fixture - - -class TestSchemaDiff(fixture.DB): - table_name = 'tmp_schemadiff' - level = fixture.DB.CONNECT - - def _setup(self, url): - super(TestSchemaDiff, self)._setup(url) - self.meta = MetaData(self.engine) - self.meta.reflect() - self.meta.drop_all() # in case junk tables are lying around in the test database - self.meta = MetaData(self.engine) - self.meta.reflect() # needed if we just deleted some tables - self.table = Table(self.table_name, self.meta, - Column('id',Integer(), primary_key=True), - Column('name', UnicodeText()), - Column('data', UnicodeText()), - ) - - def _teardown(self): - if self.table.exists(): - self.meta = MetaData(self.engine) - self.meta.reflect() - self.meta.drop_all() - super(TestSchemaDiff, self)._teardown() - - def _applyLatestModel(self): - diff = schemadiff.getDiffOfModelAgainstDatabase(self.meta, self.engine, excludeTables=['migrate_version']) - genmodel.ModelGenerator(diff,self.engine).runB2A() - - # NOTE(mriedem): DB2 handles UnicodeText as LONG VARGRAPHIC - # so the schema diffs on the columns don't work with this test. - @fixture.usedb(not_supported='ibm_db_sa') - def test_functional(self): - def assertDiff(isDiff, tablesMissingInDatabase, tablesMissingInModel, tablesWithDiff): - diff = schemadiff.getDiffOfModelAgainstDatabase(self.meta, self.engine, excludeTables=['migrate_version']) - self.assertEqual( - (diff.tables_missing_from_B, - diff.tables_missing_from_A, - list(diff.tables_different.keys()), - bool(diff)), - (tablesMissingInDatabase, - tablesMissingInModel, - tablesWithDiff, - isDiff) - ) - - # Model is defined but database is empty. - assertDiff(True, [self.table_name], [], []) - - # Check Python upgrade and downgrade of database from updated model. - diff = schemadiff.getDiffOfModelAgainstDatabase(self.meta, self.engine, excludeTables=['migrate_version']) - decls, upgradeCommands, downgradeCommands = genmodel.ModelGenerator(diff,self.engine).genB2AMigration() - - # Feature test for a recent SQLa feature; - # expect different output in that case. - if repr(String()) == 'String()': - self.assertEqualIgnoreWhitespace(decls, ''' - from migrate.changeset import schema - pre_meta = MetaData() - post_meta = MetaData() - tmp_schemadiff = Table('tmp_schemadiff', post_meta, - Column('id', Integer, primary_key=True, nullable=False), - Column('name', UnicodeText), - Column('data', UnicodeText), - ) - ''') - else: - self.assertEqualIgnoreWhitespace(decls, ''' - from migrate.changeset import schema - pre_meta = MetaData() - post_meta = MetaData() - tmp_schemadiff = Table('tmp_schemadiff', post_meta, - Column('id', Integer, primary_key=True, nullable=False), - Column('name', UnicodeText(length=None)), - Column('data', UnicodeText(length=None)), - ) - ''') - - # Create table in database, now model should match database. - self._applyLatestModel() - assertDiff(False, [], [], []) - - # Check Python code gen from database. - diff = schemadiff.getDiffOfModelAgainstDatabase(MetaData(), self.engine, excludeTables=['migrate_version']) - src = genmodel.ModelGenerator(diff,self.engine).genBDefinition() - - namespace = {} - six.exec_(src, namespace) - - c1 = Table('tmp_schemadiff', self.meta, autoload=True).c - c2 = namespace['tmp_schemadiff'].c - self.compare_columns_equal(c1, c2, ['type']) - # TODO: get rid of ignoring type - - if not self.engine.name == 'oracle': - # Add data, later we'll make sure it's still present. - result = self.engine.execute(self.table.insert(), id=1, name=u'mydata') - dataId = result.inserted_primary_key[0] - - # Modify table in model (by removing it and adding it back to model) - # Drop column data, add columns data2 and data3. - self.meta.remove(self.table) - self.table = Table(self.table_name,self.meta, - Column('id',Integer(),primary_key=True), - Column('name',UnicodeText(length=None)), - Column('data2',Integer(),nullable=True), - Column('data3',Integer(),nullable=True), - ) - assertDiff(True, [], [], [self.table_name]) - - # Apply latest model changes and find no more diffs. - self._applyLatestModel() - assertDiff(False, [], [], []) - - # Drop column data3, add data4 - self.meta.remove(self.table) - self.table = Table(self.table_name,self.meta, - Column('id',Integer(),primary_key=True), - Column('name',UnicodeText(length=None)), - Column('data2',Integer(),nullable=True), - Column('data4',Float(),nullable=True), - ) - assertDiff(True, [], [], [self.table_name]) - - diff = schemadiff.getDiffOfModelAgainstDatabase( - self.meta, self.engine, excludeTables=['migrate_version']) - decls, upgradeCommands, downgradeCommands = genmodel.ModelGenerator(diff,self.engine).genB2AMigration(indent='') - - # decls have changed since genBDefinition - six.exec_(decls, namespace) - # migration commands expect a namespace containing migrate_engine - namespace['migrate_engine'] = self.engine - # run the migration up and down - six.exec_(upgradeCommands, namespace) - assertDiff(False, [], [], []) - - six.exec_(decls, namespace) - six.exec_(downgradeCommands, namespace) - assertDiff(True, [], [], [self.table_name]) - - six.exec_(decls, namespace) - six.exec_(upgradeCommands, namespace) - assertDiff(False, [], [], []) - - if not self.engine.name == 'oracle': - # Make sure data is still present. - result = self.engine.execute(self.table.select(self.table.c.id==dataId)) - rows = result.fetchall() - self.assertEqual(len(rows), 1) - self.assertEqual(rows[0].name, 'mydata') - - # Add data, later we'll make sure it's still present. - result = self.engine.execute(self.table.insert(), id=2, name=u'mydata2', data2=123) - dataId2 = result.inserted_primary_key[0] - - # Change column type in model. - self.meta.remove(self.table) - self.table = Table(self.table_name,self.meta, - Column('id',Integer(),primary_key=True), - Column('name',UnicodeText(length=None)), - Column('data2',String(255),nullable=True), - ) - - # XXX test type diff - return - - assertDiff(True, [], [], [self.table_name]) - - # Apply latest model changes and find no more diffs. - self._applyLatestModel() - assertDiff(False, [], [], []) - - if not self.engine.name == 'oracle': - # Make sure data is still present. - result = self.engine.execute(self.table.select(self.table.c.id==dataId2)) - rows = result.fetchall() - self.assertEqual(len(rows), 1) - self.assertEqual(rows[0].name, 'mydata2') - self.assertEqual(rows[0].data2, '123') - - # Delete data, since we're about to make a required column. - # Not even using sqlalchemy.PassiveDefault helps because we're doing explicit column select. - self.engine.execute(self.table.delete(), id=dataId) - - if not self.engine.name == 'firebird': - # Change column nullable in model. - self.meta.remove(self.table) - self.table = Table(self.table_name,self.meta, - Column('id',Integer(),primary_key=True), - Column('name',UnicodeText(length=None)), - Column('data2',String(255),nullable=False), - ) - assertDiff(True, [], [], [self.table_name]) # TODO test nullable diff - - # Apply latest model changes and find no more diffs. - self._applyLatestModel() - assertDiff(False, [], [], []) - - # Remove table from model. - self.meta.remove(self.table) - assertDiff(True, [], [self.table_name], []) |