summaryrefslogtreecommitdiff
path: root/tests/test_autogen_diffs.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_autogen_diffs.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_autogen_diffs.py')
-rw-r--r--tests/test_autogen_diffs.py1016
1 files changed, 548 insertions, 468 deletions
diff --git a/tests/test_autogen_diffs.py b/tests/test_autogen_diffs.py
index 38e06b6..af2e2c2 100644
--- a/tests/test_autogen_diffs.py
+++ b/tests/test_autogen_diffs.py
@@ -1,10 +1,30 @@
import sys
-from sqlalchemy import MetaData, Column, Table, Integer, String, Text, \
- Numeric, CHAR, ForeignKey, INTEGER, Index, UniqueConstraint, \
- TypeDecorator, CheckConstraint, text, PrimaryKeyConstraint, \
- ForeignKeyConstraint, VARCHAR, DECIMAL, DateTime, BigInteger, BIGINT, \
- SmallInteger
+from sqlalchemy import (
+ MetaData,
+ Column,
+ Table,
+ Integer,
+ String,
+ Text,
+ Numeric,
+ CHAR,
+ ForeignKey,
+ INTEGER,
+ Index,
+ UniqueConstraint,
+ TypeDecorator,
+ CheckConstraint,
+ text,
+ PrimaryKeyConstraint,
+ ForeignKeyConstraint,
+ VARCHAR,
+ DECIMAL,
+ DateTime,
+ BigInteger,
+ BIGINT,
+ SmallInteger,
+)
from sqlalchemy.dialects import sqlite
from sqlalchemy.types import NULLTYPE, VARBINARY
from sqlalchemy.engine.reflection import Inspector
@@ -20,62 +40,41 @@ from alembic.testing import eq_, is_, is_not_
from alembic.util import CommandError
from ._autogen_fixtures import AutogenTest, AutogenFixtureTest
-py3k = sys.version_info >= (3, )
+py3k = sys.version_info >= (3,)
class AutogenCrossSchemaTest(AutogenTest, TestBase):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
__backend__ = True
@classmethod
def _get_db_schema(cls):
m = MetaData()
- Table('t1', m,
- Column('x', Integer)
- )
- Table('t2', m,
- Column('y', Integer),
- schema=config.test_schema
- )
- Table('t6', m,
- Column('u', Integer)
- )
- Table('t7', m,
- Column('v', Integer),
- schema=config.test_schema
- )
+ Table("t1", m, Column("x", Integer))
+ Table("t2", m, Column("y", Integer), schema=config.test_schema)
+ Table("t6", m, Column("u", Integer))
+ Table("t7", m, Column("v", Integer), schema=config.test_schema)
return m
@classmethod
def _get_model_schema(cls):
m = MetaData()
- Table('t3', m,
- Column('q', Integer)
- )
- Table('t4', m,
- Column('z', Integer),
- schema=config.test_schema
- )
- Table('t6', m,
- Column('u', Integer)
- )
- Table('t7', m,
- Column('v', Integer),
- schema=config.test_schema
- )
+ Table("t3", m, Column("q", Integer))
+ Table("t4", m, Column("z", Integer), schema=config.test_schema)
+ Table("t6", m, Column("u", Integer))
+ Table("t7", m, Column("v", Integer), schema=config.test_schema)
return m
def test_default_schema_omitted_upgrade(self):
-
def include_object(obj, name, type_, reflected, compare_to):
if type_ == "table":
return name == "t3"
else:
return True
+
self._update_context(
- object_filters=include_object,
- include_schemas=True,
+ object_filters=include_object, include_schemas=True
)
uo = ops.UpgradeOps(ops=[])
autogenerate._produce_net_changes(self.autogen_context, uo)
@@ -85,7 +84,6 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
eq_(diffs[0][1].schema, None)
def test_alt_schema_included_upgrade(self):
-
def include_object(obj, name, type_, reflected, compare_to):
if type_ == "table":
return name == "t4"
@@ -93,8 +91,7 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
return True
self._update_context(
- object_filters=include_object,
- include_schemas=True,
+ object_filters=include_object, include_schemas=True
)
uo = ops.UpgradeOps(ops=[])
autogenerate._produce_net_changes(self.autogen_context, uo)
@@ -109,9 +106,9 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
return name == "t1"
else:
return True
+
self._update_context(
- object_filters=include_object,
- include_schemas=True,
+ object_filters=include_object, include_schemas=True
)
uo = ops.UpgradeOps(ops=[])
autogenerate._produce_net_changes(self.autogen_context, uo)
@@ -121,15 +118,14 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
eq_(diffs[0][1].schema, None)
def test_alt_schema_included_downgrade(self):
-
def include_object(obj, name, type_, reflected, compare_to):
if type_ == "table":
return name == "t2"
else:
return True
+
self._update_context(
- object_filters=include_object,
- include_schemas=True,
+ object_filters=include_object, include_schemas=True
)
uo = ops.UpgradeOps(ops=[])
autogenerate._produce_net_changes(self.autogen_context, uo)
@@ -139,7 +135,7 @@ class AutogenCrossSchemaTest(AutogenTest, TestBase):
class AutogenDefaultSchemaTest(AutogenFixtureTest, TestBase):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
__backend__ = True
def test_uses_explcit_schema_in_default_one(self):
@@ -149,8 +145,8 @@ class AutogenDefaultSchemaTest(AutogenFixtureTest, TestBase):
m1 = MetaData()
m2 = MetaData()
- Table('a', m1, Column('x', String(50)))
- Table('a', m2, Column('x', String(50)), schema=default_schema)
+ Table("a", m1, Column("x", String(50)))
+ Table("a", m2, Column("x", String(50)), schema=default_schema)
diffs = self._fixture(m1, m2, include_schemas=True)
eq_(diffs, [])
@@ -162,15 +158,15 @@ class AutogenDefaultSchemaTest(AutogenFixtureTest, TestBase):
m1 = MetaData()
m2 = MetaData()
- Table('a', m1, Column('x', String(50)))
- Table('a', m2, Column('x', String(50)), schema=default_schema)
- Table('a', m2, Column('y', String(50)), schema="test_schema")
+ Table("a", m1, Column("x", String(50)))
+ Table("a", m2, Column("x", String(50)), schema=default_schema)
+ Table("a", m2, Column("y", String(50)), schema="test_schema")
diffs = self._fixture(m1, m2, include_schemas=True)
eq_(len(diffs), 1)
eq_(diffs[0][0], "add_table")
eq_(diffs[0][1].schema, "test_schema")
- eq_(diffs[0][1].c.keys(), ['y'])
+ eq_(diffs[0][1].c.keys(), ["y"])
def test_uses_explcit_schema_in_default_three(self):
@@ -179,20 +175,20 @@ class AutogenDefaultSchemaTest(AutogenFixtureTest, TestBase):
m1 = MetaData()
m2 = MetaData()
- Table('a', m1, Column('y', String(50)), schema="test_schema")
+ Table("a", m1, Column("y", String(50)), schema="test_schema")
- Table('a', m2, Column('x', String(50)), schema=default_schema)
- Table('a', m2, Column('y', String(50)), schema="test_schema")
+ Table("a", m2, Column("x", String(50)), schema=default_schema)
+ Table("a", m2, Column("y", String(50)), schema="test_schema")
diffs = self._fixture(m1, m2, include_schemas=True)
eq_(len(diffs), 1)
eq_(diffs[0][0], "add_table")
eq_(diffs[0][1].schema, default_schema)
- eq_(diffs[0][1].c.keys(), ['x'])
+ eq_(diffs[0][1].c.keys(), ["x"])
class AutogenDefaultSchemaIsNoneTest(AutogenFixtureTest, TestBase):
- __only_on__ = 'sqlite'
+ __only_on__ = "sqlite"
def setUp(self):
super(AutogenDefaultSchemaIsNoneTest, self).setUp()
@@ -205,23 +201,23 @@ class AutogenDefaultSchemaIsNoneTest(AutogenFixtureTest, TestBase):
m1 = MetaData()
m2 = MetaData()
- Table('a', m1, Column('x', String(50)))
- Table('a', m2, Column('x', String(50)))
+ Table("a", m1, Column("x", String(50)))
+ Table("a", m2, Column("x", String(50)))
def _include_object(obj, name, type_, reflected, compare_to):
if type_ == "table":
- return name in 'a' and obj.schema != 'main'
+ return name in "a" and obj.schema != "main"
else:
return True
diffs = self._fixture(
- m1, m2, include_schemas=True,
- object_filters=_include_object)
+ m1, m2, include_schemas=True, object_filters=_include_object
+ )
eq_(len(diffs), 0)
class ModelOne(object):
- __requires__ = ('unique_constraint_reflection', )
+ __requires__ = ("unique_constraint_reflection",)
schema = None
@@ -231,30 +227,42 @@ class ModelOne(object):
m = MetaData(schema=schema)
- Table('user', m,
- Column('id', Integer, primary_key=True),
- Column('name', String(50)),
- Column('a1', Text),
- Column("pw", String(50)),
- Index('pw_idx', 'pw')
- )
-
- Table('address', m,
- Column('id', Integer, primary_key=True),
- Column('email_address', String(100), nullable=False),
- )
-
- Table('order', m,
- Column('order_id', Integer, primary_key=True),
- Column("amount", Numeric(8, 2), nullable=False,
- server_default=text("0")),
- CheckConstraint('amount >= 0', name='ck_order_amount')
- )
-
- Table('extra', m,
- Column("x", CHAR),
- Column('uid', Integer, ForeignKey('user.id'))
- )
+ Table(
+ "user",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50)),
+ Column("a1", Text),
+ Column("pw", String(50)),
+ Index("pw_idx", "pw"),
+ )
+
+ Table(
+ "address",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("email_address", String(100), nullable=False),
+ )
+
+ Table(
+ "order",
+ m,
+ Column("order_id", Integer, primary_key=True),
+ Column(
+ "amount",
+ Numeric(8, 2),
+ nullable=False,
+ server_default=text("0"),
+ ),
+ CheckConstraint("amount >= 0", name="ck_order_amount"),
+ )
+
+ Table(
+ "extra",
+ m,
+ Column("x", CHAR),
+ Column("uid", Integer, ForeignKey("user.id")),
+ )
return m
@@ -264,38 +272,50 @@ class ModelOne(object):
m = MetaData(schema=schema)
- Table('user', m,
- Column('id', Integer, primary_key=True),
- Column('name', String(50), nullable=False),
- Column('a1', Text, server_default="x")
- )
-
- Table('address', m,
- Column('id', Integer, primary_key=True),
- Column('email_address', String(100), nullable=False),
- Column('street', String(50)),
- UniqueConstraint('email_address', name="uq_email")
- )
-
- Table('order', m,
- Column('order_id', Integer, primary_key=True),
- Column('amount', Numeric(10, 2), nullable=True,
- server_default=text("0")),
- Column('user_id', Integer, ForeignKey('user.id')),
- CheckConstraint('amount > -1', name='ck_order_amount'),
- )
-
- Table('item', m,
- Column('id', Integer, primary_key=True),
- Column('description', String(100)),
- Column('order_id', Integer, ForeignKey('order.order_id')),
- CheckConstraint('len(description) > 5')
- )
+ Table(
+ "user",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50), nullable=False),
+ Column("a1", Text, server_default="x"),
+ )
+
+ Table(
+ "address",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("email_address", String(100), nullable=False),
+ Column("street", String(50)),
+ UniqueConstraint("email_address", name="uq_email"),
+ )
+
+ Table(
+ "order",
+ m,
+ Column("order_id", Integer, primary_key=True),
+ Column(
+ "amount",
+ Numeric(10, 2),
+ nullable=True,
+ server_default=text("0"),
+ ),
+ Column("user_id", Integer, ForeignKey("user.id")),
+ CheckConstraint("amount > -1", name="ck_order_amount"),
+ )
+
+ Table(
+ "item",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("description", String(100)),
+ Column("order_id", Integer, ForeignKey("order.order_id")),
+ CheckConstraint("len(description) > 5"),
+ )
return m
class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
- __only_on__ = 'sqlite'
+ __only_on__ = "sqlite"
def test_diffs(self):
"""test generation of diff rules"""
@@ -304,23 +324,18 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
uo = ops.UpgradeOps(ops=[])
ctx = self.autogen_context
- autogenerate._produce_net_changes(
- ctx, uo
- )
+ autogenerate._produce_net_changes(ctx, uo)
diffs = uo.as_diffs()
- eq_(
- diffs[0],
- ('add_table', metadata.tables['item'])
- )
+ eq_(diffs[0], ("add_table", metadata.tables["item"]))
- eq_(diffs[1][0], 'remove_table')
+ eq_(diffs[1][0], "remove_table")
eq_(diffs[1][1].name, "extra")
eq_(diffs[2][0], "add_column")
eq_(diffs[2][1], None)
eq_(diffs[2][2], "address")
- eq_(diffs[2][3], metadata.tables['address'].c.street)
+ eq_(diffs[2][3], metadata.tables["address"].c.street)
eq_(diffs[3][0], "add_constraint")
eq_(diffs[3][1].name, "uq_email")
@@ -328,7 +343,7 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
eq_(diffs[4][0], "add_column")
eq_(diffs[4][1], None)
eq_(diffs[4][2], "order")
- eq_(diffs[4][3], metadata.tables['order'].c.user_id)
+ eq_(diffs[4][3], metadata.tables["order"].c.user_id)
eq_(diffs[5][0][0], "modify_type")
eq_(diffs[5][0][1], None)
@@ -338,9 +353,7 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
eq_(repr(diffs[5][0][6]), "Numeric(precision=10, scale=2)")
self._assert_fk_diff(
- diffs[6], "add_fk",
- "order", ["user_id"],
- "user", ["id"]
+ diffs[6], "add_fk", "order", ["user_id"], "user", ["id"]
)
eq_(diffs[7][0][0], "modify_default")
@@ -349,45 +362,47 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
eq_(diffs[7][0][3], "a1")
eq_(diffs[7][0][6].arg, "x")
- eq_(diffs[8][0][0], 'modify_nullable')
+ eq_(diffs[8][0][0], "modify_nullable")
eq_(diffs[8][0][5], True)
eq_(diffs[8][0][6], False)
- eq_(diffs[9][0], 'remove_index')
- eq_(diffs[9][1].name, 'pw_idx')
+ eq_(diffs[9][0], "remove_index")
+ eq_(diffs[9][1].name, "pw_idx")
- eq_(diffs[10][0], 'remove_column')
- eq_(diffs[10][3].name, 'pw')
- eq_(diffs[10][3].table.name, 'user')
- assert isinstance(
- diffs[10][3].type, String
- )
+ eq_(diffs[10][0], "remove_column")
+ eq_(diffs[10][3].name, "pw")
+ eq_(diffs[10][3].table.name, "user")
+ assert isinstance(diffs[10][3].type, String)
def test_include_symbol(self):
diffs = []
def include_symbol(name, schema=None):
- return name in ('address', 'order')
+ return name in ("address", "order")
context = MigrationContext.configure(
connection=self.bind.connect(),
opts={
- 'compare_type': True,
- 'compare_server_default': True,
- 'target_metadata': self.m2,
- 'include_symbol': include_symbol,
- }
+ "compare_type": True,
+ "compare_server_default": True,
+ "target_metadata": self.m2,
+ "include_symbol": include_symbol,
+ },
)
diffs = autogenerate.compare_metadata(
- context, context.opts['target_metadata'])
+ context, context.opts["target_metadata"]
+ )
- alter_cols = set([
- d[2] for d in self._flatten_diffs(diffs)
- if d[0].startswith('modify')
- ])
- eq_(alter_cols, set(['order']))
+ alter_cols = set(
+ [
+ d[2]
+ for d in self._flatten_diffs(diffs)
+ if d[0].startswith("modify")
+ ]
+ )
+ eq_(alter_cols, set(["order"]))
def test_include_object(self):
def include_object(obj, name, type_, reflected, compare_to):
@@ -410,33 +425,46 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
context = MigrationContext.configure(
connection=self.bind.connect(),
opts={
- 'compare_type': True,
- 'compare_server_default': True,
- 'target_metadata': self.m2,
- 'include_object': include_object,
- }
+ "compare_type": True,
+ "compare_server_default": True,
+ "target_metadata": self.m2,
+ "include_object": include_object,
+ },
)
diffs = autogenerate.compare_metadata(
- context, context.opts['target_metadata'])
+ context, context.opts["target_metadata"]
+ )
- alter_cols = set([
- d[2] for d in self._flatten_diffs(diffs)
- if d[0].startswith('modify')
- ]).union(
- d[3].name for d in self._flatten_diffs(diffs)
- if d[0] == 'add_column'
- ).union(
- d[1].name for d in self._flatten_diffs(diffs)
- if d[0] == 'add_table'
+ alter_cols = (
+ set(
+ [
+ d[2]
+ for d in self._flatten_diffs(diffs)
+ if d[0].startswith("modify")
+ ]
+ )
+ .union(
+ d[3].name
+ for d in self._flatten_diffs(diffs)
+ if d[0] == "add_column"
+ )
+ .union(
+ d[1].name
+ for d in self._flatten_diffs(diffs)
+ if d[0] == "add_table"
+ )
)
- eq_(alter_cols, set(['user_id', 'order', 'user']))
+ eq_(alter_cols, set(["user_id", "order", "user"]))
def test_skip_null_type_comparison_reflected(self):
ac = ops.AlterColumnOp("sometable", "somecol")
autogenerate.compare._compare_type(
- self.autogen_context, ac,
- None, "sometable", "somecol",
+ self.autogen_context,
+ ac,
+ None,
+ "sometable",
+ "somecol",
Column("somecol", NULLTYPE),
Column("somecol", Integer()),
)
@@ -446,8 +474,11 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
def test_skip_null_type_comparison_local(self):
ac = ops.AlterColumnOp("sometable", "somecol")
autogenerate.compare._compare_type(
- self.autogen_context, ac,
- None, "sometable", "somecol",
+ self.autogen_context,
+ ac,
+ None,
+ "sometable",
+ "somecol",
Column("somecol", Integer()),
Column("somecol", NULLTYPE),
)
@@ -463,8 +494,11 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
ac = ops.AlterColumnOp("sometable", "somecol")
autogenerate.compare._compare_type(
- self.autogen_context, ac,
- None, "sometable", "somecol",
+ self.autogen_context,
+ ac,
+ None,
+ "sometable",
+ "somecol",
Column("somecol", INTEGER()),
Column("somecol", MyType()),
)
@@ -473,56 +507,63 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
ac = ops.AlterColumnOp("sometable", "somecol")
autogenerate.compare._compare_type(
- self.autogen_context, ac,
- None, "sometable", "somecol",
+ self.autogen_context,
+ ac,
+ None,
+ "sometable",
+ "somecol",
Column("somecol", String()),
Column("somecol", MyType()),
)
diff = ac.to_diff_tuple()
- eq_(
- diff[0][0:4],
- ('modify_type', None, 'sometable', 'somecol')
- )
+ eq_(diff[0][0:4], ("modify_type", None, "sometable", "somecol"))
def test_affinity_typedec(self):
class MyType(TypeDecorator):
impl = CHAR
def load_dialect_impl(self, dialect):
- if dialect.name == 'sqlite':
+ if dialect.name == "sqlite":
return dialect.type_descriptor(Integer())
else:
return dialect.type_descriptor(CHAR(32))
- uo = ops.AlterColumnOp('sometable', 'somecol')
+ uo = ops.AlterColumnOp("sometable", "somecol")
autogenerate.compare._compare_type(
- self.autogen_context, uo,
- None, "sometable", "somecol",
+ self.autogen_context,
+ uo,
+ None,
+ "sometable",
+ "somecol",
Column("somecol", Integer, nullable=True),
- Column("somecol", MyType())
+ Column("somecol", MyType()),
)
assert not uo.has_changes()
def test_dont_barf_on_already_reflected(self):
from sqlalchemy.util import OrderedSet
+
inspector = Inspector.from_engine(self.bind)
uo = ops.UpgradeOps(ops=[])
autogenerate.compare._compare_tables(
- OrderedSet([(None, 'extra'), (None, 'user')]),
- OrderedSet(), inspector,
- uo, self.autogen_context
+ OrderedSet([(None, "extra"), (None, "user")]),
+ OrderedSet(),
+ inspector,
+ uo,
+ self.autogen_context,
)
eq_(
[(rec[0], rec[1].name) for rec in uo.as_diffs()],
[
- ('remove_table', 'extra'),
- ('remove_index', 'pw_idx'),
- ('remove_table', 'user'), ]
+ ("remove_table", "extra"),
+ ("remove_index", "pw_idx"),
+ ("remove_table", "user"),
+ ],
)
class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
__backend__ = True
schema = "test_schema"
@@ -531,26 +572,21 @@ class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
metadata = self.m2
- self._update_context(
- include_schemas=True,
- )
+ self._update_context(include_schemas=True)
uo = ops.UpgradeOps(ops=[])
autogenerate._produce_net_changes(self.autogen_context, uo)
diffs = uo.as_diffs()
- eq_(
- diffs[0],
- ('add_table', metadata.tables['%s.item' % self.schema])
- )
+ eq_(diffs[0], ("add_table", metadata.tables["%s.item" % self.schema]))
- eq_(diffs[1][0], 'remove_table')
+ eq_(diffs[1][0], "remove_table")
eq_(diffs[1][1].name, "extra")
eq_(diffs[2][0], "add_column")
eq_(diffs[2][1], self.schema)
eq_(diffs[2][2], "address")
- eq_(diffs[2][3], metadata.tables['%s.address' % self.schema].c.street)
+ eq_(diffs[2][3], metadata.tables["%s.address" % self.schema].c.street)
eq_(diffs[3][0], "add_constraint")
eq_(diffs[3][1].name, "uq_email")
@@ -558,7 +594,7 @@ class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
eq_(diffs[4][0], "add_column")
eq_(diffs[4][1], self.schema)
eq_(diffs[4][2], "order")
- eq_(diffs[4][3], metadata.tables['%s.order' % self.schema].c.user_id)
+ eq_(diffs[4][3], metadata.tables["%s.order" % self.schema].c.user_id)
eq_(diffs[5][0][0], "modify_type")
eq_(diffs[5][0][1], self.schema)
@@ -568,10 +604,13 @@ class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
eq_(repr(diffs[5][0][6]), "Numeric(precision=10, scale=2)")
self._assert_fk_diff(
- diffs[6], "add_fk",
- "order", ["user_id"],
- "user", ["id"],
- source_schema=config.test_schema
+ diffs[6],
+ "add_fk",
+ "order",
+ ["user_id"],
+ "user",
+ ["id"],
+ source_schema=config.test_schema,
)
eq_(diffs[7][0][0], "modify_default")
@@ -580,15 +619,15 @@ class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
eq_(diffs[7][0][3], "a1")
eq_(diffs[7][0][6].arg, "x")
- eq_(diffs[8][0][0], 'modify_nullable')
+ eq_(diffs[8][0][0], "modify_nullable")
eq_(diffs[8][0][5], True)
eq_(diffs[8][0][6], False)
- eq_(diffs[9][0], 'remove_index')
- eq_(diffs[9][1].name, 'pw_idx')
+ eq_(diffs[9][0], "remove_index")
+ eq_(diffs[9][1].name, "pw_idx")
- eq_(diffs[10][0], 'remove_column')
- eq_(diffs[10][3].name, 'pw')
+ eq_(diffs[10][0], "remove_column")
+ eq_(diffs[10][3].name, "pw")
class CompareTypeSpecificityTest(TestBase):
@@ -597,10 +636,10 @@ class CompareTypeSpecificityTest(TestBase):
from sqlalchemy.engine import default
return impl.DefaultImpl(
- default.DefaultDialect(), None, False, True, None, {})
+ default.DefaultDialect(), None, False, True, None, {}
+ )
def test_typedec_to_nonstandard(self):
-
class PasswordType(TypeDecorator):
impl = VARBINARY
@@ -608,7 +647,7 @@ class CompareTypeSpecificityTest(TestBase):
return PasswordType(self.impl.length)
def load_dialect_impl(self, dialect):
- if dialect.name == 'default':
+ if dialect.name == "default":
impl = sqlite.NUMERIC(self.length)
else:
impl = VARBINARY(self.length)
@@ -616,8 +655,8 @@ class CompareTypeSpecificityTest(TestBase):
impl = self._fixture()
impl.compare_type(
- Column('x', sqlite.NUMERIC(50)),
- Column('x', PasswordType(50)))
+ Column("x", sqlite.NUMERIC(50)), Column("x", PasswordType(50))
+ )
def test_string(self):
t1 = String(30)
@@ -626,9 +665,9 @@ class CompareTypeSpecificityTest(TestBase):
t4 = Integer
impl = self._fixture()
- is_(impl.compare_type(Column('x', t3), Column('x', t1)), False)
- is_(impl.compare_type(Column('x', t3), Column('x', t2)), True)
- is_(impl.compare_type(Column('x', t3), Column('x', t4)), True)
+ is_(impl.compare_type(Column("x", t3), Column("x", t1)), False)
+ is_(impl.compare_type(Column("x", t3), Column("x", t2)), True)
+ is_(impl.compare_type(Column("x", t3), Column("x", t4)), True)
def test_numeric(self):
t1 = Numeric(10, 5)
@@ -637,16 +676,16 @@ class CompareTypeSpecificityTest(TestBase):
t4 = DateTime
impl = self._fixture()
- is_(impl.compare_type(Column('x', t3), Column('x', t1)), False)
- is_(impl.compare_type(Column('x', t3), Column('x', t2)), True)
- is_(impl.compare_type(Column('x', t3), Column('x', t4)), True)
+ is_(impl.compare_type(Column("x", t3), Column("x", t1)), False)
+ is_(impl.compare_type(Column("x", t3), Column("x", t2)), True)
+ is_(impl.compare_type(Column("x", t3), Column("x", t4)), True)
def test_numeric_noprecision(self):
t1 = Numeric()
t2 = Numeric(scale=5)
impl = self._fixture()
- is_(impl.compare_type(Column('x', t1), Column('x', t2)), False)
+ is_(impl.compare_type(Column("x", t1), Column("x", t2)), False)
def test_integer(self):
t1 = Integer()
@@ -657,12 +696,12 @@ class CompareTypeSpecificityTest(TestBase):
t6 = BigInteger()
impl = self._fixture()
- is_(impl.compare_type(Column('x', t5), Column('x', t1)), False)
- is_(impl.compare_type(Column('x', t3), Column('x', t1)), True)
- is_(impl.compare_type(Column('x', t3), Column('x', t6)), False)
- is_(impl.compare_type(Column('x', t3), Column('x', t2)), True)
- is_(impl.compare_type(Column('x', t5), Column('x', t2)), True)
- is_(impl.compare_type(Column('x', t1), Column('x', t4)), True)
+ is_(impl.compare_type(Column("x", t5), Column("x", t1)), False)
+ is_(impl.compare_type(Column("x", t3), Column("x", t1)), True)
+ is_(impl.compare_type(Column("x", t3), Column("x", t6)), False)
+ is_(impl.compare_type(Column("x", t3), Column("x", t2)), True)
+ is_(impl.compare_type(Column("x", t5), Column("x", t2)), True)
+ is_(impl.compare_type(Column("x", t1), Column("x", t4)), True)
def test_datetime(self):
t1 = DateTime()
@@ -670,22 +709,19 @@ class CompareTypeSpecificityTest(TestBase):
t3 = DateTime(timezone=True)
impl = self._fixture()
- is_(impl.compare_type(Column('x', t1), Column('x', t2)), False)
- is_(impl.compare_type(Column('x', t1), Column('x', t3)), True)
- is_(impl.compare_type(Column('x', t2), Column('x', t3)), True)
+ is_(impl.compare_type(Column("x", t1), Column("x", t2)), False)
+ is_(impl.compare_type(Column("x", t1), Column("x", t3)), True)
+ is_(impl.compare_type(Column("x", t2), Column("x", t3)), True)
class AutogenSystemColTest(AutogenTest, TestBase):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
@classmethod
def _get_db_schema(cls):
m = MetaData()
- Table(
- 'sometable', m,
- Column('id', Integer, primary_key=True),
- )
+ Table("sometable", m, Column("id", Integer, primary_key=True))
return m
@classmethod
@@ -695,9 +731,10 @@ class AutogenSystemColTest(AutogenTest, TestBase):
# 'xmin' is implicitly present, when added to a model should produce
# no change
Table(
- 'sometable', m,
- Column('id', Integer, primary_key=True),
- Column('xmin', Integer, system=True)
+ "sometable",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("xmin", Integer, system=True),
)
return m
@@ -715,30 +752,38 @@ class AutogenerateVariantCompareTest(AutogenTest, TestBase):
# 1.0.13 and lower fail on Postgresql due to variant / bigserial issue
# #3739
- __requires__ = ('sqlalchemy_1014', )
+ __requires__ = ("sqlalchemy_1014",)
@classmethod
def _get_db_schema(cls):
m = MetaData()
- Table('sometable', m,
- Column(
- 'id',
- BigInteger().with_variant(Integer, "sqlite"),
- primary_key=True),
- Column('value', String(50)))
+ Table(
+ "sometable",
+ m,
+ Column(
+ "id",
+ BigInteger().with_variant(Integer, "sqlite"),
+ primary_key=True,
+ ),
+ Column("value", String(50)),
+ )
return m
@classmethod
def _get_model_schema(cls):
m = MetaData()
- Table('sometable', m,
- Column(
- 'id',
- BigInteger().with_variant(Integer, "sqlite"),
- primary_key=True),
- Column('value', String(50)))
+ Table(
+ "sometable",
+ m,
+ Column(
+ "id",
+ BigInteger().with_variant(Integer, "sqlite"),
+ primary_key=True,
+ ),
+ Column("value", String(50)),
+ )
return m
def test_variant_no_issue(self):
@@ -750,24 +795,30 @@ class AutogenerateVariantCompareTest(AutogenTest, TestBase):
class AutogenerateCustomCompareTypeTest(AutogenTest, TestBase):
- __only_on__ = 'sqlite'
+ __only_on__ = "sqlite"
@classmethod
def _get_db_schema(cls):
m = MetaData()
- Table('sometable', m,
- Column('id', Integer, primary_key=True),
- Column('value', Integer))
+ Table(
+ "sometable",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("value", Integer),
+ )
return m
@classmethod
def _get_model_schema(cls):
m = MetaData()
- Table('sometable', m,
- Column('id', Integer, primary_key=True),
- Column('value', String))
+ Table(
+ "sometable",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("value", String),
+ )
return m
def test_uses_custom_compare_type_function(self):
@@ -779,15 +830,20 @@ class AutogenerateCustomCompareTypeTest(AutogenTest, TestBase):
ctx = self.autogen_context
autogenerate._produce_net_changes(ctx, uo)
- first_table = self.m2.tables['sometable']
- first_column = first_table.columns['id']
+ first_table = self.m2.tables["sometable"]
+ first_column = first_table.columns["id"]
eq_(len(my_compare_type.mock_calls), 2)
# We'll just test the first call
_, args, _ = my_compare_type.mock_calls[0]
- (context, inspected_column, metadata_column,
- inspected_type, metadata_type) = args
+ (
+ context,
+ inspected_column,
+ metadata_column,
+ inspected_type,
+ metadata_type,
+ ) = args
eq_(context, self.context)
eq_(metadata_column, first_column)
eq_(metadata_type, first_column.type)
@@ -816,8 +872,8 @@ class AutogenerateCustomCompareTypeTest(AutogenTest, TestBase):
autogenerate._produce_net_changes(ctx, uo)
diffs = uo.as_diffs()
- eq_(diffs[0][0][0], 'modify_type')
- eq_(diffs[1][0][0], 'modify_type')
+ eq_(diffs[0][0][0], "modify_type")
+ eq_(diffs[1][0][0], "modify_type")
class PKConstraintUpgradesIgnoresNullableTest(AutogenTest, TestBase):
@@ -829,10 +885,11 @@ class PKConstraintUpgradesIgnoresNullableTest(AutogenTest, TestBase):
m = MetaData()
Table(
- 'person_to_role', m,
- Column('person_id', Integer, autoincrement=False),
- Column('role_id', Integer, autoincrement=False),
- PrimaryKeyConstraint('person_id', 'role_id')
+ "person_to_role",
+ m,
+ Column("person_id", Integer, autoincrement=False),
+ Column("role_id", Integer, autoincrement=False),
+ PrimaryKeyConstraint("person_id", "role_id"),
)
return m
@@ -849,34 +906,40 @@ class PKConstraintUpgradesIgnoresNullableTest(AutogenTest, TestBase):
class AutogenKeyTest(AutogenTest, TestBase):
- __only_on__ = 'sqlite'
+ __only_on__ = "sqlite"
@classmethod
def _get_db_schema(cls):
m = MetaData()
- Table('someothertable', m,
- Column('id', Integer, primary_key=True),
- Column('value', Integer, key="somekey"),
- )
+ Table(
+ "someothertable",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("value", Integer, key="somekey"),
+ )
return m
@classmethod
def _get_model_schema(cls):
m = MetaData()
- Table('sometable', m,
- Column('id', Integer, primary_key=True),
- Column('value', Integer, key="someotherkey"),
- )
- Table('someothertable', m,
- Column('id', Integer, primary_key=True),
- Column('value', Integer, key="somekey"),
- Column("othervalue", Integer, key="otherkey")
- )
+ Table(
+ "sometable",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("value", Integer, key="someotherkey"),
+ )
+ Table(
+ "someothertable",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("value", Integer, key="somekey"),
+ Column("othervalue", Integer, key="otherkey"),
+ )
return m
- symbols = ['someothertable', 'sometable']
+ symbols = ["someothertable", "sometable"]
def test_autogen(self):
@@ -892,16 +955,19 @@ class AutogenKeyTest(AutogenTest, TestBase):
class AutogenVersionTableTest(AutogenTest, TestBase):
- __only_on__ = 'sqlite'
- version_table_name = 'alembic_version'
+ __only_on__ = "sqlite"
+ version_table_name = "alembic_version"
version_table_schema = None
@classmethod
def _get_db_schema(cls):
m = MetaData()
Table(
- cls.version_table_name, m,
- Column('x', Integer), schema=cls.version_table_schema)
+ cls.version_table_name,
+ m,
+ Column("x", Integer),
+ schema=cls.version_table_schema,
+ )
return m
@classmethod
@@ -919,7 +985,10 @@ class AutogenVersionTableTest(AutogenTest, TestBase):
def test_version_table_in_target(self):
Table(
self.version_table_name,
- self.m2, Column('x', Integer), schema=self.version_table_schema)
+ self.m2,
+ Column("x", Integer),
+ schema=self.version_table_schema,
+ )
ctx = self.autogen_context
uo = ops.UpgradeOps(ops=[])
@@ -928,29 +997,30 @@ class AutogenVersionTableTest(AutogenTest, TestBase):
class AutogenCustomVersionTableSchemaTest(AutogenVersionTableTest):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
__backend__ = True
- version_table_schema = 'test_schema'
- configure_opts = {'version_table_schema': 'test_schema'}
+ version_table_schema = "test_schema"
+ configure_opts = {"version_table_schema": "test_schema"}
class AutogenCustomVersionTableTest(AutogenVersionTableTest):
- version_table_name = 'my_version_table'
- configure_opts = {'version_table': 'my_version_table'}
+ version_table_name = "my_version_table"
+ configure_opts = {"version_table": "my_version_table"}
class AutogenCustomVersionTableAndSchemaTest(AutogenVersionTableTest):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
__backend__ = True
- version_table_name = 'my_version_table'
- version_table_schema = 'test_schema'
+ version_table_name = "my_version_table"
+ version_table_schema = "test_schema"
configure_opts = {
- 'version_table': 'my_version_table',
- 'version_table_schema': 'test_schema'}
+ "version_table": "my_version_table",
+ "version_table_schema": "test_schema",
+ }
class AutogenerateDiffOrderTest(AutogenTest, TestBase):
- __only_on__ = 'sqlite'
+ __only_on__ = "sqlite"
@classmethod
def _get_db_schema(cls):
@@ -959,13 +1029,11 @@ class AutogenerateDiffOrderTest(AutogenTest, TestBase):
@classmethod
def _get_model_schema(cls):
m = MetaData()
- Table('parent', m,
- Column('id', Integer, primary_key=True)
- )
+ Table("parent", m, Column("id", Integer, primary_key=True))
- Table('child', m,
- Column('parent_id', Integer, ForeignKey('parent.id')),
- )
+ Table(
+ "child", m, Column("parent_id", Integer, ForeignKey("parent.id"))
+ )
return m
@@ -980,32 +1048,29 @@ class AutogenerateDiffOrderTest(AutogenTest, TestBase):
autogenerate._produce_net_changes(ctx, uo)
diffs = uo.as_diffs()
- eq_(diffs[0][0], 'add_table')
+ eq_(diffs[0][0], "add_table")
eq_(diffs[0][1].name, "parent")
- eq_(diffs[1][0], 'add_table')
+ eq_(diffs[1][0], "add_table")
eq_(diffs[1][1].name, "child")
class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
- __only_on__ = 'sqlite'
+ __only_on__ = "sqlite"
def test_compare_metadata(self):
metadata = self.m2
diffs = autogenerate.compare_metadata(self.context, metadata)
- eq_(
- diffs[0],
- ('add_table', metadata.tables['item'])
- )
+ eq_(diffs[0], ("add_table", metadata.tables["item"]))
- eq_(diffs[1][0], 'remove_table')
+ eq_(diffs[1][0], "remove_table")
eq_(diffs[1][1].name, "extra")
eq_(diffs[2][0], "add_column")
eq_(diffs[2][1], None)
eq_(diffs[2][2], "address")
- eq_(diffs[2][3], metadata.tables['address'].c.street)
+ eq_(diffs[2][3], metadata.tables["address"].c.street)
eq_(diffs[3][0], "add_constraint")
eq_(diffs[3][1].name, "uq_email")
@@ -1013,7 +1078,7 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
eq_(diffs[4][0], "add_column")
eq_(diffs[4][1], None)
eq_(diffs[4][2], "order")
- eq_(diffs[4][3], metadata.tables['order'].c.user_id)
+ eq_(diffs[4][3], metadata.tables["order"].c.user_id)
eq_(diffs[5][0][0], "modify_type")
eq_(diffs[5][0][1], None)
@@ -1023,9 +1088,7 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
eq_(repr(diffs[5][0][6]), "Numeric(precision=10, scale=2)")
self._assert_fk_diff(
- diffs[6], "add_fk",
- "order", ["user_id"],
- "user", ["id"]
+ diffs[6], "add_fk", "order", ["user_id"], "user", ["id"]
)
eq_(diffs[7][0][0], "modify_default")
@@ -1034,15 +1097,15 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
eq_(diffs[7][0][3], "a1")
eq_(diffs[7][0][6].arg, "x")
- eq_(diffs[8][0][0], 'modify_nullable')
+ eq_(diffs[8][0][0], "modify_nullable")
eq_(diffs[8][0][5], True)
eq_(diffs[8][0][6], False)
- eq_(diffs[9][0], 'remove_index')
- eq_(diffs[9][1].name, 'pw_idx')
+ eq_(diffs[9][0], "remove_index")
+ eq_(diffs[9][1].name, "pw_idx")
- eq_(diffs[10][0], 'remove_column')
- eq_(diffs[10][3].name, 'pw')
+ eq_(diffs[10][0], "remove_column")
+ eq_(diffs[10][3].name, "pw")
def test_compare_metadata_include_object(self):
metadata = self.m2
@@ -1058,46 +1121,46 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
context = MigrationContext.configure(
connection=self.bind.connect(),
opts={
- 'compare_type': True,
- 'compare_server_default': True,
- 'include_object': include_object,
- }
+ "compare_type": True,
+ "compare_server_default": True,
+ "include_object": include_object,
+ },
)
diffs = autogenerate.compare_metadata(context, metadata)
- eq_(diffs[0][0], 'remove_table')
+ eq_(diffs[0][0], "remove_table")
eq_(diffs[0][1].name, "extra")
eq_(diffs[1][0], "add_column")
eq_(diffs[1][1], None)
eq_(diffs[1][2], "order")
- eq_(diffs[1][3], metadata.tables['order'].c.user_id)
+ eq_(diffs[1][3], metadata.tables["order"].c.user_id)
def test_compare_metadata_include_symbol(self):
metadata = self.m2
def include_symbol(table_name, schema_name):
- return table_name in ('extra', 'order')
+ return table_name in ("extra", "order")
context = MigrationContext.configure(
connection=self.bind.connect(),
opts={
- 'compare_type': True,
- 'compare_server_default': True,
- 'include_symbol': include_symbol,
- }
+ "compare_type": True,
+ "compare_server_default": True,
+ "include_symbol": include_symbol,
+ },
)
diffs = autogenerate.compare_metadata(context, metadata)
- eq_(diffs[0][0], 'remove_table')
+ eq_(diffs[0][0], "remove_table")
eq_(diffs[0][1].name, "extra")
eq_(diffs[1][0], "add_column")
eq_(diffs[1][1], None)
eq_(diffs[1][2], "order")
- eq_(diffs[1][3], metadata.tables['order'].c.user_id)
+ eq_(diffs[1][3], metadata.tables["order"].c.user_id)
eq_(diffs[2][0][0], "modify_type")
eq_(diffs[2][0][1], None)
@@ -1106,15 +1169,14 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
eq_(repr(diffs[2][0][5]), "NUMERIC(precision=8, scale=2)")
eq_(repr(diffs[2][0][6]), "Numeric(precision=10, scale=2)")
- eq_(diffs[2][1][0], 'modify_nullable')
- eq_(diffs[2][1][2], 'order')
+ eq_(diffs[2][1][0], "modify_nullable")
+ eq_(diffs[2][1][2], "order")
eq_(diffs[2][1][5], False)
eq_(diffs[2][1][6], True)
def test_compare_metadata_as_sql(self):
context = MigrationContext.configure(
- connection=self.bind.connect(),
- opts={'as_sql': True}
+ connection=self.bind.connect(), opts={"as_sql": True}
)
metadata = self.m2
@@ -1122,12 +1184,14 @@ class CompareMetadataTest(ModelOne, AutogenTest, TestBase):
CommandError,
"autogenerate can't use as_sql=True as it prevents "
"querying the database for schema information",
- autogenerate.compare_metadata, context, metadata
+ autogenerate.compare_metadata,
+ context,
+ metadata,
)
class PGCompareMetaData(ModelOne, AutogenTest, TestBase):
- __only_on__ = 'postgresql'
+ __only_on__ = "postgresql"
__backend__ = True
schema = "test_schema"
@@ -1135,26 +1199,20 @@ class PGCompareMetaData(ModelOne, AutogenTest, TestBase):
metadata = self.m2
context = MigrationContext.configure(
- connection=self.bind.connect(),
- opts={
- "include_schemas": True
- }
+ connection=self.bind.connect(), opts={"include_schemas": True}
)
diffs = autogenerate.compare_metadata(context, metadata)
- eq_(
- diffs[0],
- ('add_table', metadata.tables['test_schema.item'])
- )
+ eq_(diffs[0], ("add_table", metadata.tables["test_schema.item"]))
- eq_(diffs[1][0], 'remove_table')
+ eq_(diffs[1][0], "remove_table")
eq_(diffs[1][1].name, "extra")
eq_(diffs[2][0], "add_column")
eq_(diffs[2][1], "test_schema")
eq_(diffs[2][2], "address")
- eq_(diffs[2][3], metadata.tables['test_schema.address'].c.street)
+ eq_(diffs[2][3], metadata.tables["test_schema.address"].c.street)
eq_(diffs[3][0], "add_constraint")
eq_(diffs[3][1].name, "uq_email")
@@ -1162,27 +1220,25 @@ class PGCompareMetaData(ModelOne, AutogenTest, TestBase):
eq_(diffs[4][0], "add_column")
eq_(diffs[4][1], "test_schema")
eq_(diffs[4][2], "order")
- eq_(diffs[4][3], metadata.tables['test_schema.order'].c.user_id)
+ eq_(diffs[4][3], metadata.tables["test_schema.order"].c.user_id)
- eq_(diffs[5][0][0], 'modify_nullable')
+ eq_(diffs[5][0][0], "modify_nullable")
eq_(diffs[5][0][5], False)
eq_(diffs[5][0][6], True)
+
class OrigObjectTest(TestBase):
def setUp(self):
self.metadata = m = MetaData()
t = Table(
- 't', m,
- Column('id', Integer(), primary_key=True),
- Column('x', Integer())
- )
- self.ix = Index('ix1', t.c.id)
- fk = ForeignKeyConstraint(['t_id'], ['t.id'])
- q = Table(
- 'q', m,
- Column('t_id', Integer()),
- fk
+ "t",
+ m,
+ Column("id", Integer(), primary_key=True),
+ Column("x", Integer()),
)
+ self.ix = Index("ix1", t.c.id)
+ fk = ForeignKeyConstraint(["t_id"], ["t.id"])
+ q = Table("q", m, Column("t_id", Integer()), fk)
self.table = t
self.fk = fk
self.ck = CheckConstraint(t.c.x > 5)
@@ -1232,10 +1288,10 @@ class OrigObjectTest(TestBase):
is_not_(None, op.to_constraint().table)
def test_add_pk_no_orig(self):
- op = ops.CreatePrimaryKeyOp('pk1', 't', ['x', 'y'])
+ op = ops.CreatePrimaryKeyOp("pk1", "t", ["x", "y"])
pk = op.to_constraint()
- eq_(pk.name, 'pk1')
- eq_(pk.table.name, 't')
+ eq_(pk.name, "pk1")
+ eq_(pk.table.name, "t")
def test_add_pk(self):
pk = self.pk
@@ -1254,7 +1310,7 @@ class OrigObjectTest(TestBase):
def test_drop_column(self):
t = self.table
- op = ops.DropColumnOp.from_column_and_tablename(None, 't', t.c.x)
+ op = ops.DropColumnOp.from_column_and_tablename(None, "t", t.c.x)
is_(op.to_column(), t.c.x)
is_(op.reverse().to_column(), t.c.x)
is_not_(None, op.to_column().table)
@@ -1262,7 +1318,7 @@ class OrigObjectTest(TestBase):
def test_add_column(self):
t = self.table
- op = ops.AddColumnOp.from_column_and_tablename(None, 't', t.c.x)
+ op = ops.AddColumnOp.from_column_and_tablename(None, "t", t.c.x)
is_(op.to_column(), t.c.x)
is_(op.reverse().to_column(), t.c.x)
is_not_(None, op.to_column().table)
@@ -1304,25 +1360,33 @@ class MultipleMetaDataTest(AutogenFixtureTest, TestBase):
m2b = MetaData()
m2c = MetaData()
- Table('a', m1a, Column('id', Integer, primary_key=True))
- Table('b1', m1b, Column('id', Integer, primary_key=True))
- Table('b2', m1b, Column('id', Integer, primary_key=True))
- Table('c1', m1c, Column('id', Integer, primary_key=True),
- Column('x', Integer))
+ Table("a", m1a, Column("id", Integer, primary_key=True))
+ Table("b1", m1b, Column("id", Integer, primary_key=True))
+ Table("b2", m1b, Column("id", Integer, primary_key=True))
+ Table(
+ "c1",
+ m1c,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer),
+ )
- a = Table('a', m2a, Column('id', Integer, primary_key=True),
- Column('q', Integer))
- Table('b1', m2b, Column('id', Integer, primary_key=True))
- Table('c1', m2c, Column('id', Integer, primary_key=True))
- c2 = Table('c2', m2c, Column('id', Integer, primary_key=True))
+ a = Table(
+ "a",
+ m2a,
+ Column("id", Integer, primary_key=True),
+ Column("q", Integer),
+ )
+ Table("b1", m2b, Column("id", Integer, primary_key=True))
+ Table("c1", m2c, Column("id", Integer, primary_key=True))
+ c2 = Table("c2", m2c, Column("id", Integer, primary_key=True))
diffs = self._fixture([m1a, m1b, m1c], [m2a, m2b, m2c])
- eq_(diffs[0], ('add_table', c2))
- eq_(diffs[1][0], 'remove_table')
- eq_(diffs[1][1].name, 'b2')
- eq_(diffs[2], ('add_column', None, 'a', a.c.q))
- eq_(diffs[3][0:3], ('remove_column', None, 'c1'))
- eq_(diffs[3][3].name, 'x')
+ eq_(diffs[0], ("add_table", c2))
+ eq_(diffs[1][0], "remove_table")
+ eq_(diffs[1][1].name, "b2")
+ eq_(diffs[2], ("add_column", None, "a", a.c.q))
+ eq_(diffs[3][0:3], ("remove_column", None, "c1"))
+ eq_(diffs[3][3].name, "x")
def test_empty_list(self):
# because they're going to do it....
@@ -1339,18 +1403,19 @@ class MultipleMetaDataTest(AutogenFixtureTest, TestBase):
m2a = MetaData()
m2b = MetaData()
- Table('a', m1a, Column('id', Integer, primary_key=True))
- Table('b', m1b, Column('id', Integer, primary_key=True))
+ Table("a", m1a, Column("id", Integer, primary_key=True))
+ Table("b", m1b, Column("id", Integer, primary_key=True))
- Table('a', m2a, Column('id', Integer, primary_key=True))
- b = Table('b', m2b, Column('id', Integer, primary_key=True),
- Column('q', Integer))
+ Table("a", m2a, Column("id", Integer, primary_key=True))
+ b = Table(
+ "b",
+ m2b,
+ Column("id", Integer, primary_key=True),
+ Column("q", Integer),
+ )
diffs = self._fixture((m1a, m1b), (m2a, m2b))
- eq_(
- diffs,
- [('add_column', None, 'b', b.c.q)]
- )
+ eq_(diffs, [("add_column", None, "b", b.c.q)])
def test_raise_on_dupe(self):
m1a = MetaData()
@@ -1359,116 +1424,123 @@ class MultipleMetaDataTest(AutogenFixtureTest, TestBase):
m2a = MetaData()
m2b = MetaData()
- Table('a', m1a, Column('id', Integer, primary_key=True))
- Table('b1', m1b, Column('id', Integer, primary_key=True))
- Table('b2', m1b, Column('id', Integer, primary_key=True))
- Table('b3', m1b, Column('id', Integer, primary_key=True))
+ Table("a", m1a, Column("id", Integer, primary_key=True))
+ Table("b1", m1b, Column("id", Integer, primary_key=True))
+ Table("b2", m1b, Column("id", Integer, primary_key=True))
+ Table("b3", m1b, Column("id", Integer, primary_key=True))
- Table('a', m2a, Column('id', Integer, primary_key=True))
- Table('a', m2b, Column('id', Integer, primary_key=True))
- Table('b1', m2b, Column('id', Integer, primary_key=True))
- Table('b2', m2a, Column('id', Integer, primary_key=True))
- Table('b2', m2b, Column('id', Integer, primary_key=True))
+ Table("a", m2a, Column("id", Integer, primary_key=True))
+ Table("a", m2b, Column("id", Integer, primary_key=True))
+ Table("b1", m2b, Column("id", Integer, primary_key=True))
+ Table("b2", m2a, Column("id", Integer, primary_key=True))
+ Table("b2", m2b, Column("id", Integer, primary_key=True))
assert_raises_message(
ValueError,
'Duplicate table keys across multiple MetaData objects: "a", "b2"',
self._fixture,
- [m1a, m1b], [m2a, m2b]
+ [m1a, m1b],
+ [m2a, m2b],
)
class AutoincrementTest(AutogenFixtureTest, TestBase):
__backend__ = True
- __requires__ = 'integer_subtype_comparisons',
+ __requires__ = ("integer_subtype_comparisons",)
def test_alter_column_autoincrement_none(self):
m1 = MetaData()
m2 = MetaData()
- Table('a', m1, Column('x', Integer, nullable=False))
- Table('a', m2, Column('x', Integer, nullable=True))
+ Table("a", m1, Column("x", Integer, nullable=False))
+ Table("a", m2, Column("x", Integer, nullable=True))
ops = self._fixture(m1, m2, return_ops=True)
- assert 'autoincrement' not in ops.ops[0].ops[0].kw
+ assert "autoincrement" not in ops.ops[0].ops[0].kw
def test_alter_column_autoincrement_pk_false(self):
m1 = MetaData()
m2 = MetaData()
Table(
- 'a', m1,
- Column('x', Integer, primary_key=True, autoincrement=False))
+ "a",
+ m1,
+ Column("x", Integer, primary_key=True, autoincrement=False),
+ )
Table(
- 'a', m2,
- Column('x', BigInteger, primary_key=True, autoincrement=False))
+ "a",
+ m2,
+ Column("x", BigInteger, primary_key=True, autoincrement=False),
+ )
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], False)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], False)
def test_alter_column_autoincrement_pk_implicit_true(self):
m1 = MetaData()
m2 = MetaData()
- Table(
- 'a', m1,
- Column('x', Integer, primary_key=True))
- Table(
- 'a', m2,
- Column('x', BigInteger, primary_key=True))
+ Table("a", m1, Column("x", Integer, primary_key=True))
+ Table("a", m2, Column("x", BigInteger, primary_key=True))
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
def test_alter_column_autoincrement_pk_explicit_true(self):
m1 = MetaData()
m2 = MetaData()
Table(
- 'a', m1,
- Column('x', Integer, primary_key=True, autoincrement=True))
+ "a", m1, Column("x", Integer, primary_key=True, autoincrement=True)
+ )
Table(
- 'a', m2,
- Column('x', BigInteger, primary_key=True, autoincrement=True))
+ "a",
+ m2,
+ Column("x", BigInteger, primary_key=True, autoincrement=True),
+ )
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
def test_alter_column_autoincrement_nonpk_false(self):
m1 = MetaData()
m2 = MetaData()
Table(
- 'a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, autoincrement=False)
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, autoincrement=False),
)
Table(
- 'a', m2,
- Column('id', Integer, primary_key=True),
- Column('x', BigInteger, autoincrement=False)
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, autoincrement=False),
)
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], False)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], False)
def test_alter_column_autoincrement_nonpk_implicit_false(self):
m1 = MetaData()
m2 = MetaData()
Table(
- 'a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer)
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer),
)
Table(
- 'a', m2,
- Column('id', Integer, primary_key=True),
- Column('x', BigInteger)
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger),
)
ops = self._fixture(m1, m2, return_ops=True)
- assert 'autoincrement' not in ops.ops[0].ops[0].kw
+ assert "autoincrement" not in ops.ops[0].ops[0].kw
@config.requirements.fail_before_sqla_110
def test_alter_column_autoincrement_nonpk_explicit_true(self):
@@ -1476,54 +1548,60 @@ class AutoincrementTest(AutogenFixtureTest, TestBase):
m2 = MetaData()
Table(
- 'a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, autoincrement=True)
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, autoincrement=True),
)
Table(
- 'a', m2,
- Column('id', Integer, primary_key=True),
- Column('x', BigInteger, autoincrement=True)
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, autoincrement=True),
)
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
def test_alter_column_autoincrement_compositepk_false(self):
m1 = MetaData()
m2 = MetaData()
Table(
- 'a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, primary_key=True, autoincrement=False)
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, primary_key=True, autoincrement=False),
)
Table(
- 'a', m2,
- Column('id', Integer, primary_key=True),
- Column('x', BigInteger, primary_key=True, autoincrement=False)
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, primary_key=True, autoincrement=False),
)
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], False)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], False)
def test_alter_column_autoincrement_compositepk_implicit_false(self):
m1 = MetaData()
m2 = MetaData()
Table(
- 'a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, primary_key=True)
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, primary_key=True),
)
Table(
- 'a', m2,
- Column('id', Integer, primary_key=True),
- Column('x', BigInteger, primary_key=True)
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, primary_key=True),
)
ops = self._fixture(m1, m2, return_ops=True)
- assert 'autoincrement' not in ops.ops[0].ops[0].kw
+ assert "autoincrement" not in ops.ops[0].ops[0].kw
@config.requirements.autoincrement_on_composite_pk
def test_alter_column_autoincrement_compositepk_explicit_true(self):
@@ -1531,20 +1609,22 @@ class AutoincrementTest(AutogenFixtureTest, TestBase):
m2 = MetaData()
Table(
- 'a', m1,
- Column('id', Integer, primary_key=True, autoincrement=False),
- Column('x', Integer, primary_key=True, autoincrement=True),
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("x", Integer, primary_key=True, autoincrement=True),
# on SQLA 1.0 and earlier, this being present
# trips the "add KEY for the primary key" so that the
# AUTO_INCREMENT keyword is accepted by MySQL. SQLA 1.1 and
# greater the columns are just reorganized.
- mysql_engine='InnoDB'
+ mysql_engine="InnoDB",
)
Table(
- 'a', m2,
- Column('id', Integer, primary_key=True, autoincrement=False),
- Column('x', BigInteger, primary_key=True, autoincrement=True)
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("x", BigInteger, primary_key=True, autoincrement=True),
)
ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw['autoincrement'], True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)