summaryrefslogtreecommitdiff
path: root/tests/test_mysql.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2019-01-06 12:37:53 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2019-01-06 13:22:59 -0500
commitb6e3fae3f51e46028fa33e0bbc62fb2bce08b963 (patch)
tree8f41e9a70317c2181c57c0ef2bbb4680f14b9978 /tests/test_mysql.py
parent16c905b5f2a24ee7bdb766f9b60ce68e44b7dab0 (diff)
downloadalembic-b6e3fae3f51e46028fa33e0bbc62fb2bce08b963.tar.gz
pure black run + flake8
run black -l 79 against source code, set up for full flake8 testing. Change-Id: I4108e1274d49894b9898ec5bd3a1147933a473d7
Diffstat (limited to 'tests/test_mysql.py')
-rw-r--r--tests/test_mysql.py319
1 files changed, 156 insertions, 163 deletions
diff --git a/tests/test_mysql.py b/tests/test_mysql.py
index dd872f7..68746ba 100644
--- a/tests/test_mysql.py
+++ b/tests/test_mysql.py
@@ -7,53 +7,68 @@ from alembic import op, util
from alembic.testing import eq_, assert_raises_message
from alembic.testing.fixtures import capture_context_buffer, op_fixture
-from alembic.testing.env import staging_env, _no_sql_testing_config, \
- three_rev_fixture, clear_staging_env
+from alembic.testing.env import (
+ staging_env,
+ _no_sql_testing_config,
+ three_rev_fixture,
+ clear_staging_env,
+)
from alembic.migration import MigrationContext
class MySQLOpTest(TestBase):
-
def test_rename_column(self):
- context = op_fixture('mysql')
+ 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'
+ "t1", "c1", new_column_name="c2", existing_type=Integer
)
+ context.assert_("ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL")
def test_rename_column_quotes_needed_one(self):
- context = op_fixture('mysql')
- op.alter_column('MyTable', 'ColumnOne', new_column_name="ColumnTwo",
- existing_type=Integer)
+ context = op_fixture("mysql")
+ op.alter_column(
+ "MyTable",
+ "ColumnOne",
+ new_column_name="ColumnTwo",
+ existing_type=Integer,
+ )
context.assert_(
- 'ALTER TABLE `MyTable` CHANGE `ColumnOne` `ColumnTwo` INTEGER NULL'
+ "ALTER TABLE `MyTable` CHANGE `ColumnOne` `ColumnTwo` INTEGER NULL"
)
def test_rename_column_quotes_needed_two(self):
- context = op_fixture('mysql')
- op.alter_column('my table', 'column one', new_column_name="column two",
- existing_type=Integer)
+ context = op_fixture("mysql")
+ op.alter_column(
+ "my table",
+ "column one",
+ new_column_name="column two",
+ existing_type=Integer,
+ )
context.assert_(
- 'ALTER TABLE `my table` CHANGE `column one` '
- '`column two` INTEGER NULL'
+ "ALTER TABLE `my table` CHANGE `column one` "
+ "`column two` INTEGER NULL"
)
def test_rename_column_serv_default(self):
- context = op_fixture('mysql')
+ 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'"
+ "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')
+ context = op_fixture("mysql")
op.alter_column(
- 't1', 'c1', existing_type=Integer,
- server_default=func.utc_thing(func.current_timestamp()))
+ "t1",
+ "c1",
+ existing_type=Integer,
+ 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_(
@@ -62,184 +77,183 @@ class MySQLOpTest(TestBase):
)
def test_rename_column_autoincrement(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.alter_column(
- 't1', 'c1', new_column_name="c2", existing_type=Integer,
- existing_autoincrement=True)
+ "t1",
+ "c1",
+ new_column_name="c2",
+ existing_type=Integer,
+ existing_autoincrement=True,
+ )
context.assert_(
- 'ALTER TABLE t1 CHANGE c1 c2 INTEGER NULL AUTO_INCREMENT'
+ "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', existing_type=Integer,
- autoincrement=True)
- context.assert_(
- 'ALTER TABLE t1 MODIFY c1 INTEGER NULL AUTO_INCREMENT'
- )
+ context = op_fixture("mysql")
+ op.alter_column("t1", "c1", existing_type=Integer, autoincrement=True)
+ context.assert_("ALTER TABLE t1 MODIFY c1 INTEGER NULL AUTO_INCREMENT")
def test_col_remove_autoincrement(self):
- context = op_fixture('mysql')
- op.alter_column('t1', 'c1', existing_type=Integer,
- existing_autoincrement=True,
- autoincrement=False)
- context.assert_(
- 'ALTER TABLE t1 MODIFY c1 INTEGER NULL'
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ existing_type=Integer,
+ existing_autoincrement=True,
+ autoincrement=False,
)
+ context.assert_("ALTER TABLE t1 MODIFY c1 INTEGER NULL")
def test_col_dont_remove_server_default(self):
- context = op_fixture('mysql')
- op.alter_column('t1', 'c1', existing_type=Integer,
- existing_server_default='1',
- server_default=False)
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ existing_type=Integer,
+ existing_server_default="1",
+ server_default=False,
+ )
context.assert_()
def test_alter_column_drop_default(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.alter_column("t", "c", existing_type=Integer, server_default=None)
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c DROP DEFAULT'
- )
+ context.assert_("ALTER TABLE t ALTER COLUMN c DROP DEFAULT")
def test_alter_column_remove_schematype(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.alter_column(
- "t", "c",
+ "t",
+ "c",
type_=Integer,
existing_type=Boolean(create_constraint=True, name="ck1"),
- server_default=None)
- context.assert_(
- 'ALTER TABLE t MODIFY c INTEGER NULL'
+ server_default=None,
)
+ context.assert_("ALTER TABLE t MODIFY c INTEGER NULL")
def test_alter_column_modify_default(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
# notice we dont need the existing type on this one...
- op.alter_column("t", "c", server_default='1')
- context.assert_(
- "ALTER TABLE t ALTER COLUMN c SET DEFAULT '1'"
- )
+ op.alter_column("t", "c", server_default="1")
+ context.assert_("ALTER TABLE t ALTER COLUMN c SET DEFAULT '1'")
def test_col_not_nullable(self):
- context = op_fixture('mysql')
- op.alter_column('t1', 'c1', nullable=False, existing_type=Integer)
- context.assert_(
- 'ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL'
- )
+ context = op_fixture("mysql")
+ op.alter_column("t1", "c1", nullable=False, existing_type=Integer)
+ context.assert_("ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL")
def test_col_not_nullable_existing_serv_default(self):
- context = op_fixture('mysql')
- op.alter_column('t1', 'c1', nullable=False, existing_type=Integer,
- existing_server_default='5')
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ nullable=False,
+ existing_type=Integer,
+ existing_server_default="5",
+ )
context.assert_(
"ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL DEFAULT '5'"
)
def test_col_nullable(self):
- context = op_fixture('mysql')
- op.alter_column('t1', 'c1', nullable=True, existing_type=Integer)
- context.assert_(
- 'ALTER TABLE t1 MODIFY c1 INTEGER NULL'
- )
+ context = op_fixture("mysql")
+ op.alter_column("t1", "c1", nullable=True, existing_type=Integer)
+ context.assert_("ALTER TABLE t1 MODIFY c1 INTEGER NULL")
def test_col_multi_alter(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.alter_column(
- 't1', 'c1', nullable=False, server_default="q", type_=Integer)
+ "t1", "c1", nullable=False, server_default="q", type_=Integer
+ )
context.assert_(
"ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL DEFAULT 'q'"
)
def test_alter_column_multi_alter_w_drop_default(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.alter_column(
- 't1', 'c1', nullable=False, server_default=None, type_=Integer)
- context.assert_(
- "ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL"
+ "t1", "c1", nullable=False, server_default=None, type_=Integer
)
+ context.assert_("ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL")
def test_col_alter_type_required(self):
- op_fixture('mysql')
+ op_fixture("mysql")
assert_raises_message(
util.CommandError,
"MySQL CHANGE/MODIFY COLUMN operations require the existing type.",
- op.alter_column, 't1', 'c1', nullable=False, server_default="q"
+ op.alter_column,
+ "t1",
+ "c1",
+ nullable=False,
+ server_default="q",
)
def test_drop_fk(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.drop_constraint("f1", "t1", "foreignkey")
- context.assert_(
- "ALTER TABLE t1 DROP FOREIGN KEY f1"
- )
+ context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
def test_drop_fk_quoted(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.drop_constraint("MyFk", "MyTable", "foreignkey")
- context.assert_(
- "ALTER TABLE `MyTable` DROP FOREIGN KEY `MyFk`"
- )
+ context.assert_("ALTER TABLE `MyTable` DROP FOREIGN KEY `MyFk`")
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"
- )
+ 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')
+ context = op_fixture("mysql")
op.drop_constraint("f1", "t1", "unique")
- context.assert_(
- "ALTER TABLE t1 DROP INDEX f1"
- )
+ context.assert_("ALTER TABLE t1 DROP INDEX f1")
def test_drop_unique_quoted(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.drop_constraint("MyUnique", "MyTable", "unique")
- context.assert_(
- "ALTER TABLE `MyTable` DROP INDEX `MyUnique`"
- )
+ context.assert_("ALTER TABLE `MyTable` DROP INDEX `MyUnique`")
def test_drop_check(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.drop_constraint("f1", "t1", "check")
- context.assert_(
- "ALTER TABLE t1 DROP CONSTRAINT f1"
- )
+ context.assert_("ALTER TABLE t1 DROP CONSTRAINT f1")
def test_drop_check_quoted(self):
- context = op_fixture('mysql')
+ context = op_fixture("mysql")
op.drop_constraint("MyCheck", "MyTable", "check")
- context.assert_(
- "ALTER TABLE `MyTable` DROP CONSTRAINT `MyCheck`"
- )
+ context.assert_("ALTER TABLE `MyTable` DROP CONSTRAINT `MyCheck`")
def test_drop_unknown(self):
- op_fixture('mysql')
+ op_fixture("mysql")
assert_raises_message(
TypeError,
"'type' can be one of 'check', 'foreignkey', "
"'primary', 'unique', None",
- op.drop_constraint, "f1", "t1", "typo"
+ op.drop_constraint,
+ "f1",
+ "t1",
+ "typo",
)
def test_drop_generic_constraint(self):
- op_fixture('mysql')
+ op_fixture("mysql")
assert_raises_message(
NotImplementedError,
"No generic 'DROP CONSTRAINT' in MySQL - please "
"specify constraint type",
- op.drop_constraint, "f1", "t1"
+ op.drop_constraint,
+ "f1",
+ "t1",
)
class MySQLDefaultCompareTest(TestBase):
- __only_on__ = 'mysql'
+ __only_on__ = "mysql"
__backend__ = True
- __requires__ = 'mysql_timestamp_reflection',
+ __requires__ = ("mysql_timestamp_reflection",)
@classmethod
def setup_class(cls):
@@ -247,17 +261,14 @@ class MySQLDefaultCompareTest(TestBase):
staging_env()
context = MigrationContext.configure(
connection=cls.bind.connect(),
- opts={
- 'compare_type': True,
- 'compare_server_default': True
- }
+ opts={"compare_type": True, "compare_server_default": True},
)
connection = context.bind
cls.autogen_context = {
- 'imports': set(),
- 'connection': connection,
- 'dialect': connection.dialect,
- 'context': context
+ "imports": set(),
+ "connection": connection,
+ "dialect": connection.dialect,
+ "context": context,
}
@classmethod
@@ -277,64 +288,46 @@ class MySQLDefaultCompareTest(TestBase):
alternate = txt
expected = False
t = Table(
- "test", self.metadata,
+ "test",
+ self.metadata,
Column(
- "somecol", type_,
- server_default=text(txt) if txt else None
- )
+ "somecol", type_, server_default=text(txt) if txt else None
+ ),
+ )
+ t2 = Table(
+ "test",
+ MetaData(),
+ Column("somecol", type_, server_default=text(alternate)),
)
- 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
- ):
+ 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)
refl = Table(t1.name, MetaData())
insp.reflecttable(refl, None)
- ctx = self.autogen_context['context']
+ ctx = self.autogen_context["context"]
return ctx.impl.compare_server_default(
- refl.c[cols[0]['name']],
- col,
- rendered,
- cols[0]['default'])
+ refl.c[cols[0]["name"]], col, rendered, cols[0]["default"]
+ )
def test_compare_timestamp_current_timestamp(self):
- self._compare_default_roundtrip(
- TIMESTAMP(),
- "CURRENT_TIMESTAMP",
- )
+ self._compare_default_roundtrip(TIMESTAMP(), "CURRENT_TIMESTAMP")
def test_compare_timestamp_current_timestamp_diff(self):
- self._compare_default_roundtrip(
- TIMESTAMP(),
- None, "CURRENT_TIMESTAMP",
- )
+ self._compare_default_roundtrip(TIMESTAMP(), None, "CURRENT_TIMESTAMP")
def test_compare_integer_same(self):
- self._compare_default_roundtrip(
- Integer(), "5"
- )
+ self._compare_default_roundtrip(Integer(), "5")
def test_compare_integer_diff(self):
- self._compare_default_roundtrip(
- Integer(), "5", "7"
- )
+ self._compare_default_roundtrip(Integer(), "5", "7")
def test_compare_boolean_same(self):
- self._compare_default_roundtrip(
- Boolean(), "1"
- )
+ self._compare_default_roundtrip(Boolean(), "1")
def test_compare_boolean_diff(self):
- self._compare_default_roundtrip(
- Boolean(), "1", "0"
- )
+ self._compare_default_roundtrip(Boolean(), "1", "0")