From: Mike Bayer Date: Wed, 27 Nov 2013 00:05:03 +0000 (-0500) Subject: - rework mysql tests to have suites X-Git-Tag: rel_0_6_1~2 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=bc829880d6cccaebada4ac3836486af368b4e9b0;p=thirdparty%2Fsqlalchemy%2Falembic.git - rework mysql tests to have suites - add a test to verify pullreq #5 - check if a TIMESTAMP column with CURRENT_TIMESTAMP as default successfully compares for equality/inequality. test passes without the pullreq, so no failure case has been demonstrated. --- diff --git a/tests/test_mysql.py b/tests/test_mysql.py index 8ffa2a5b..7f3a6313 100644 --- a/tests/test_mysql.py +++ b/tests/test_mysql.py @@ -1,124 +1,203 @@ from sqlalchemy import Integer, func - +from unittest import TestCase +from sqlalchemy import TIMESTAMP, MetaData, Table, Column, text +from sqlalchemy.engine.reflection import Inspector from alembic import op, util -from . import op_fixture, assert_raises_message - -def test_rename_column(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer) - context.assert_( - 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL' - ) - -def test_rename_column_serv_default(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, - existing_server_default="q") - context.assert_( - "ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL DEFAULT 'q'" - ) - -def test_rename_column_serv_compiled_default(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, - existing_server_default=func.utc_thing(func.current_timestamp())) - # this is not a valid MySQL default but the point is to just - # test SQL expression rendering - context.assert_( - "ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL DEFAULT utc_thing(CURRENT_TIMESTAMP)" - ) - -def test_rename_column_autoincrement(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, - existing_autoincrement=True) - context.assert_( - 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL AUTO_INCREMENT' - ) - -def test_col_add_autoincrement(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, - autoincrement=True) - context.assert_( - 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL AUTO_INCREMENT' - ) - -def test_col_remove_autoincrement(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, - existing_autoincrement=True, - autoincrement=False) - context.assert_( - 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL' - ) - -def test_col_nullable(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', nullable=False, existing_type=Integer) - context.assert_( - 'ALTER TABLE t1 CHANGE c1 c1 INTEGER NOT NULL' - ) - -def test_col_multi_alter(): - context = op_fixture('mysql') - op.alter_column('t1', 'c1', nullable=False, server_default="q", type_=Integer) - context.assert_( - "ALTER TABLE t1 CHANGE c1 c1 INTEGER NOT NULL DEFAULT 'q'" - ) - - -def test_col_alter_type_required(): - context = op_fixture('mysql') - assert_raises_message( - util.CommandError, - "All MySQL ALTER COLUMN operations require the existing type.", - op.alter_column, 't1', 'c1', nullable=False, server_default="q" - ) - -def test_drop_fk(): - context = op_fixture('mysql') - op.drop_constraint("f1", "t1", "foreignkey") - context.assert_( - "ALTER TABLE t1 DROP FOREIGN KEY f1" - ) - -def test_drop_constraint_primary(): - context = op_fixture('mysql') - op.drop_constraint('primary', 't1', type_='primary') - context.assert_( - "ALTER TABLE t1 DROP PRIMARY KEY " - ) - -def test_drop_unique(): - context = op_fixture('mysql') - op.drop_constraint("f1", "t1", "unique") - context.assert_( - "ALTER TABLE t1 DROP INDEX f1" - ) - -def test_drop_check(): - context = op_fixture('mysql') - assert_raises_message( - NotImplementedError, - "MySQL does not support CHECK constraints.", - op.drop_constraint, "f1", "t1", "check" - ) - -def test_drop_unknown(): - context = op_fixture('mysql') - assert_raises_message( - TypeError, - "'type' can be one of 'check', 'foreignkey', " - "'primary', 'unique', None", - op.drop_constraint, "f1", "t1", "typo" - ) - -def test_drop_generic_constraint(): - context = op_fixture('mysql') - assert_raises_message( - NotImplementedError, - "No generic 'DROP CONSTRAINT' in MySQL - please " - "specify constraint type", - op.drop_constraint, "f1", "t1" - ) +from . import op_fixture, assert_raises_message, db_for_dialect, \ + staging_env, clear_staging_env +from alembic.migration import MigrationContext + +class MySQLOpTest(TestCase): + def test_rename_column(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer) + context.assert_( + 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL' + ) + + def test_rename_column_serv_default(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, + existing_server_default="q") + context.assert_( + "ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL DEFAULT 'q'" + ) + + def test_rename_column_serv_compiled_default(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, + existing_server_default=func.utc_thing(func.current_timestamp())) + # this is not a valid MySQL default but the point is to just + # test SQL expression rendering + context.assert_( + "ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL DEFAULT utc_thing(CURRENT_TIMESTAMP)" + ) + + def test_rename_column_autoincrement(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, + existing_autoincrement=True) + context.assert_( + 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL AUTO_INCREMENT' + ) + + def test_col_add_autoincrement(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, + autoincrement=True) + context.assert_( + 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL AUTO_INCREMENT' + ) + + def test_col_remove_autoincrement(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', new_column_name="c2", existing_type=Integer, + existing_autoincrement=True, + autoincrement=False) + context.assert_( + 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL' + ) + + def test_col_nullable(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', nullable=False, existing_type=Integer) + context.assert_( + 'ALTER TABLE t1 CHANGE c1 c1 INTEGER NOT NULL' + ) + + def test_col_multi_alter(self): + context = op_fixture('mysql') + op.alter_column('t1', 'c1', nullable=False, server_default="q", type_=Integer) + context.assert_( + "ALTER TABLE t1 CHANGE c1 c1 INTEGER NOT NULL DEFAULT 'q'" + ) + + + def test_col_alter_type_required(self): + op_fixture('mysql') + assert_raises_message( + util.CommandError, + "All MySQL ALTER COLUMN operations require the existing type.", + op.alter_column, 't1', 'c1', nullable=False, server_default="q" + ) + + def test_drop_fk(self): + context = op_fixture('mysql') + op.drop_constraint("f1", "t1", "foreignkey") + context.assert_( + "ALTER TABLE t1 DROP FOREIGN KEY f1" + ) + + def test_drop_constraint_primary(self): + context = op_fixture('mysql') + op.drop_constraint('primary', 't1', type_='primary') + context.assert_( + "ALTER TABLE t1 DROP PRIMARY KEY " + ) + + def test_drop_unique(self): + context = op_fixture('mysql') + op.drop_constraint("f1", "t1", "unique") + context.assert_( + "ALTER TABLE t1 DROP INDEX f1" + ) + + def test_drop_check(self): + op_fixture('mysql') + assert_raises_message( + NotImplementedError, + "MySQL does not support CHECK constraints.", + op.drop_constraint, "f1", "t1", "check" + ) + + def test_drop_unknown(self): + op_fixture('mysql') + assert_raises_message( + TypeError, + "'type' can be one of 'check', 'foreignkey', " + "'primary', 'unique', None", + op.drop_constraint, "f1", "t1", "typo" + ) + + def test_drop_generic_constraint(self): + op_fixture('mysql') + assert_raises_message( + NotImplementedError, + "No generic 'DROP CONSTRAINT' in MySQL - please " + "specify constraint type", + op.drop_constraint, "f1", "t1" + ) + +class MySQLDefaultCompareTest(TestCase): + @classmethod + def setup_class(cls): + cls.bind = db_for_dialect("mysql") + staging_env() + context = MigrationContext.configure( + connection=cls.bind.connect(), + opts={ + 'compare_type': True, + 'compare_server_default': True + } + ) + connection = context.bind + cls.autogen_context = { + 'imports': set(), + 'connection': connection, + 'dialect': connection.dialect, + 'context': context + } + + @classmethod + def teardown_class(cls): + clear_staging_env() + + def setUp(self): + self.metadata = MetaData(self.bind) + + def tearDown(self): + self.metadata.drop_all() + + def _compare_default_roundtrip(self, type_, txt, alternate=None): + if alternate: + expected = True + else: + alternate = txt + expected = False + t = Table("test", self.metadata, + Column("somecol", type_, server_default=text(txt) if txt else None) + ) + t2 = Table("test", MetaData(), + Column("somecol", type_, server_default=text(alternate)) + ) + assert self._compare_default( + t, t2, t2.c.somecol, alternate + ) is expected + + def _compare_default( + self, + t1, t2, col, + rendered + ): + t1.create(self.bind) + insp = Inspector.from_engine(self.bind) + cols = insp.get_columns(t1.name) + ctx = self.autogen_context['context'] + return ctx.impl.compare_server_default( + None, + col, + rendered, + cols[0]['default']) + + def test_compare_timestamp_current_timestamp(self): + self._compare_default_roundtrip( + TIMESTAMP(), + "CURRENT_TIMESTAMP", + ) + + def test_compare_timestamp_current_timestamp_diff(self): + self._compare_default_roundtrip( + TIMESTAMP(), + None, "CURRENT_TIMESTAMP", + ) +