summaryrefslogtreecommitdiff
path: root/tests/test_autogen_composition.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_autogen_composition.py')
-rw-r--r--tests/test_autogen_composition.py328
1 files changed, 328 insertions, 0 deletions
diff --git a/tests/test_autogen_composition.py b/tests/test_autogen_composition.py
new file mode 100644
index 0000000..b1717ab
--- /dev/null
+++ b/tests/test_autogen_composition.py
@@ -0,0 +1,328 @@
+import re
+
+from alembic import autogenerate
+from alembic.migration import MigrationContext
+from alembic.testing import TestBase
+from alembic.testing import eq_
+
+from ._autogen_fixtures import AutogenTest, ModelOne, _default_include_object
+
+
+class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
+ __only_on__ = 'sqlite'
+
+ def test_render_nothing(self):
+ context = MigrationContext.configure(
+ connection=self.bind.connect(),
+ opts={
+ 'compare_type': True,
+ 'compare_server_default': True,
+ 'target_metadata': self.m1,
+ 'upgrade_token': "upgrades",
+ 'downgrade_token': "downgrades",
+ }
+ )
+ template_args = {}
+ autogenerate._render_migration_diffs(context, template_args, set())
+
+ eq_(re.sub(r"u'", "'", template_args['upgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ pass
+ ### end Alembic commands ###""")
+ eq_(re.sub(r"u'", "'", template_args['downgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ pass
+ ### end Alembic commands ###""")
+
+ def test_render_nothing_batch(self):
+ context = MigrationContext.configure(
+ connection=self.bind.connect(),
+ opts={
+ 'compare_type': True,
+ 'compare_server_default': True,
+ 'target_metadata': self.m1,
+ 'upgrade_token': "upgrades",
+ 'downgrade_token': "downgrades",
+ 'alembic_module_prefix': 'op.',
+ 'sqlalchemy_module_prefix': 'sa.',
+ 'render_as_batch': True,
+ 'include_symbol': lambda name, schema: False
+ }
+ )
+ template_args = {}
+ autogenerate._render_migration_diffs(
+ context, template_args, set(),
+
+ )
+ eq_(re.sub(r"u'", "'", template_args['upgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ pass
+ ### end Alembic commands ###""")
+ eq_(re.sub(r"u'", "'", template_args['downgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ pass
+ ### end Alembic commands ###""")
+
+ def test_render_diffs_standard(self):
+ """test a full render including indentation"""
+
+ template_args = {}
+ autogenerate._render_migration_diffs(
+ self.context, template_args, set())
+ eq_(re.sub(r"u'", "'", template_args['upgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ op.create_table('item',
+ sa.Column('id', sa.Integer(), nullable=False),
+ sa.Column('description', sa.String(length=100), nullable=True),
+ sa.Column('order_id', sa.Integer(), nullable=True),
+ sa.CheckConstraint('len(description) > 5'),
+ sa.ForeignKeyConstraint(['order_id'], ['order.order_id'], ),
+ sa.PrimaryKeyConstraint('id')
+ )
+ op.drop_table('extra')
+ op.add_column('address', sa.Column('street', sa.String(length=50), \
+nullable=True))
+ op.create_unique_constraint('uq_email', 'address', ['email_address'])
+ op.add_column('order', sa.Column('user_id', sa.Integer(), nullable=True))
+ op.alter_column('order', 'amount',
+ existing_type=sa.NUMERIC(precision=8, scale=2),
+ type_=sa.Numeric(precision=10, scale=2),
+ nullable=True,
+ existing_server_default=sa.text('0'))
+ op.create_foreign_key(None, 'order', 'user', ['user_id'], ['id'])
+ op.alter_column('user', 'a1',
+ existing_type=sa.TEXT(),
+ server_default='x',
+ existing_nullable=True)
+ op.alter_column('user', 'name',
+ existing_type=sa.VARCHAR(length=50),
+ nullable=False)
+ op.drop_index('pw_idx', table_name='user')
+ op.drop_column('user', 'pw')
+ ### end Alembic commands ###""")
+
+ eq_(re.sub(r"u'", "'", template_args['downgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), \
+nullable=True))
+ op.create_index('pw_idx', 'user', ['pw'], unique=False)
+ op.alter_column('user', 'name',
+ existing_type=sa.VARCHAR(length=50),
+ nullable=True)
+ op.alter_column('user', 'a1',
+ existing_type=sa.TEXT(),
+ server_default=None,
+ existing_nullable=True)
+ op.drop_constraint(None, 'order', type_='foreignkey')
+ op.alter_column('order', 'amount',
+ existing_type=sa.Numeric(precision=10, scale=2),
+ type_=sa.NUMERIC(precision=8, scale=2),
+ nullable=False,
+ existing_server_default=sa.text('0'))
+ op.drop_column('order', 'user_id')
+ op.drop_constraint('uq_email', 'address', type_='unique')
+ op.drop_column('address', 'street')
+ op.create_table('extra',
+ sa.Column('x', sa.CHAR(), nullable=True),
+ sa.Column('uid', sa.INTEGER(), nullable=True),
+ sa.ForeignKeyConstraint(['uid'], ['user.id'], )
+ )
+ op.drop_table('item')
+ ### end Alembic commands ###""")
+
+ def test_render_diffs_batch(self):
+ """test a full render in batch mode including indentation"""
+
+ template_args = {}
+ self.context.opts['render_as_batch'] = True
+ autogenerate._render_migration_diffs(
+ self.context, template_args, set())
+
+ eq_(re.sub(r"u'", "'", template_args['upgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ op.create_table('item',
+ sa.Column('id', sa.Integer(), nullable=False),
+ sa.Column('description', sa.String(length=100), nullable=True),
+ sa.Column('order_id', sa.Integer(), nullable=True),
+ sa.CheckConstraint('len(description) > 5'),
+ sa.ForeignKeyConstraint(['order_id'], ['order.order_id'], ),
+ sa.PrimaryKeyConstraint('id')
+ )
+ op.drop_table('extra')
+ with op.batch_alter_table('address', schema=None) as batch_op:
+ batch_op.add_column(sa.Column('street', sa.String(length=50), nullable=True))
+ batch_op.create_unique_constraint('uq_email', ['email_address'])
+
+ with op.batch_alter_table('order', schema=None) as batch_op:
+ batch_op.add_column(sa.Column('user_id', sa.Integer(), nullable=True))
+ batch_op.alter_column('amount',
+ existing_type=sa.NUMERIC(precision=8, scale=2),
+ type_=sa.Numeric(precision=10, scale=2),
+ nullable=True,
+ existing_server_default=sa.text('0'))
+ batch_op.create_foreign_key(None, 'order', 'user', ['user_id'], ['id'])
+
+ with op.batch_alter_table('user', schema=None) as batch_op:
+ batch_op.alter_column('a1',
+ existing_type=sa.TEXT(),
+ server_default='x',
+ existing_nullable=True)
+ batch_op.alter_column('name',
+ existing_type=sa.VARCHAR(length=50),
+ nullable=False)
+ batch_op.drop_index('pw_idx')
+ batch_op.drop_column('pw')
+
+ ### end Alembic commands ###""")
+
+ eq_(re.sub(r"u'", "'", template_args['downgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ with op.batch_alter_table('user', schema=None) as batch_op:
+ batch_op.add_column(sa.Column('pw', sa.VARCHAR(length=50), nullable=True))
+ batch_op.create_index('pw_idx', ['pw'], unique=False)
+ batch_op.alter_column('name',
+ existing_type=sa.VARCHAR(length=50),
+ nullable=True)
+ batch_op.alter_column('a1',
+ existing_type=sa.TEXT(),
+ server_default=None,
+ existing_nullable=True)
+
+ with op.batch_alter_table('order', schema=None) as batch_op:
+ batch_op.drop_constraint(None, type_='foreignkey')
+ batch_op.alter_column('amount',
+ existing_type=sa.Numeric(precision=10, scale=2),
+ type_=sa.NUMERIC(precision=8, scale=2),
+ nullable=False,
+ existing_server_default=sa.text('0'))
+ batch_op.drop_column('user_id')
+
+ with op.batch_alter_table('address', schema=None) as batch_op:
+ batch_op.drop_constraint('uq_email', type_='unique')
+ batch_op.drop_column('street')
+
+ op.create_table('extra',
+ sa.Column('x', sa.CHAR(), nullable=True),
+ sa.Column('uid', sa.INTEGER(), nullable=True),
+ sa.ForeignKeyConstraint(['uid'], ['user.id'], )
+ )
+ op.drop_table('item')
+ ### end Alembic commands ###""")
+
+
+class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
+ __only_on__ = 'postgresql'
+ schema = "test_schema"
+
+ def test_render_nothing(self):
+ context = MigrationContext.configure(
+ connection=self.bind.connect(),
+ opts={
+ 'compare_type': True,
+ 'compare_server_default': True,
+ 'target_metadata': self.m1,
+ 'upgrade_token': "upgrades",
+ 'downgrade_token': "downgrades",
+ 'alembic_module_prefix': 'op.',
+ 'sqlalchemy_module_prefix': 'sa.',
+ 'include_symbol': lambda name, schema: False
+ }
+ )
+ template_args = {}
+ autogenerate._render_migration_diffs(
+ context, template_args, set(),
+
+ )
+ eq_(re.sub(r"u'", "'", template_args['upgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ pass
+ ### end Alembic commands ###""")
+ eq_(re.sub(r"u'", "'", template_args['downgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ pass
+ ### end Alembic commands ###""")
+
+ def test_render_diffs_extras(self):
+ """test a full render including indentation (include and schema)"""
+
+ template_args = {}
+ self.context.opts.update({
+ 'include_object': _default_include_object,
+ 'include_schemas': True
+ })
+ autogenerate._render_migration_diffs(
+ self.context, template_args, set()
+ )
+
+ eq_(re.sub(r"u'", "'", template_args['upgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ op.create_table('item',
+ sa.Column('id', sa.Integer(), nullable=False),
+ sa.Column('description', sa.String(length=100), nullable=True),
+ sa.Column('order_id', sa.Integer(), nullable=True),
+ sa.CheckConstraint('len(description) > 5'),
+ sa.ForeignKeyConstraint(['order_id'], ['%(schema)s.order.order_id'], ),
+ sa.PrimaryKeyConstraint('id'),
+ schema='%(schema)s'
+ )
+ op.drop_table('extra', schema='%(schema)s')
+ op.add_column('address', sa.Column('street', sa.String(length=50), \
+nullable=True), schema='%(schema)s')
+ op.create_unique_constraint('uq_email', 'address', ['email_address'], \
+schema='test_schema')
+ op.add_column('order', sa.Column('user_id', sa.Integer(), nullable=True), \
+schema='%(schema)s')
+ op.alter_column('order', 'amount',
+ existing_type=sa.NUMERIC(precision=8, scale=2),
+ type_=sa.Numeric(precision=10, scale=2),
+ nullable=True,
+ existing_server_default=sa.text('0'),
+ schema='%(schema)s')
+ op.create_foreign_key(None, 'order', 'user', ['user_id'], ['id'], \
+source_schema='%(schema)s', referent_schema='%(schema)s')
+ op.alter_column('user', 'a1',
+ existing_type=sa.TEXT(),
+ server_default='x',
+ existing_nullable=True,
+ schema='%(schema)s')
+ op.alter_column('user', 'name',
+ existing_type=sa.VARCHAR(length=50),
+ nullable=False,
+ schema='%(schema)s')
+ op.drop_index('pw_idx', table_name='user', schema='test_schema')
+ op.drop_column('user', 'pw', schema='%(schema)s')
+ ### end Alembic commands ###""" % {"schema": self.schema})
+
+ eq_(re.sub(r"u'", "'", template_args['downgrades']),
+ """### commands auto generated by Alembic - please adjust! ###
+ op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), \
+autoincrement=False, nullable=True), schema='%(schema)s')
+ op.create_index('pw_idx', 'user', ['pw'], unique=False, schema='%(schema)s')
+ op.alter_column('user', 'name',
+ existing_type=sa.VARCHAR(length=50),
+ nullable=True,
+ schema='%(schema)s')
+ op.alter_column('user', 'a1',
+ existing_type=sa.TEXT(),
+ server_default=None,
+ existing_nullable=True,
+ schema='%(schema)s')
+ op.drop_constraint(None, 'order', schema='%(schema)s', type_='foreignkey')
+ op.alter_column('order', 'amount',
+ existing_type=sa.Numeric(precision=10, scale=2),
+ type_=sa.NUMERIC(precision=8, scale=2),
+ nullable=False,
+ existing_server_default=sa.text('0'),
+ schema='%(schema)s')
+ op.drop_column('order', 'user_id', schema='%(schema)s')
+ op.drop_constraint('uq_email', 'address', schema='test_schema', type_='unique')
+ op.drop_column('address', 'street', schema='%(schema)s')
+ op.create_table('extra',
+ sa.Column('x', sa.CHAR(length=1), autoincrement=False, nullable=True),
+ sa.Column('uid', sa.INTEGER(), autoincrement=False, nullable=True),
+ sa.ForeignKeyConstraint(['uid'], ['%(schema)s.user.id'], \
+name='extra_uid_fkey'),
+ schema='%(schema)s'
+ )
+ op.drop_table('item', schema='%(schema)s')
+ ### end Alembic commands ###""" % {"schema": self.schema})