Skip to content

Commit

Permalink
- Fixed an issue with unique constraint autogenerate detection where
Browse files Browse the repository at this point in the history 
a named ``UniqueConstraint`` on both sides with column changes would
render with the "add" operation before the "drop", requiring the
user to reverse the order manually.
- reorganize the index/unique autogenerate test into individual test cases;
ideally the whole test suite would be broken out like this for those big
tests
  • Loading branch information
@zzzeek
zzzeek committed Dec 11, 2013
1 parent db44516 commit 74d1ccf
Show file tree
Hide file tree
Showing 3 changed files with 151 additions and 96 deletions.
20 changes: 17 additions & 3 deletions alembic/autogenerate/compare.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ def _compare_tables(conn_table_names, metadata_table_names,
name = '%s.%s' % (s, tname) if s else tname
metadata_table = metadata.tables[name]
conn_table = existing_metadata.tables[name]

if _run_filters(metadata_table, tname, "table", False, conn_table, object_filters):
_compare_columns(s, tname, object_filters,
conn_table,
Expand Down Expand Up @@ -187,6 +188,17 @@ def _compare_uniques(schema, tname, object_filters, conn_table,
)
c_keys = set(c_objs)

c_obj_by_name = dict((uq.name, uq) for uq in c_objs.values())
m_obj_by_name = dict((uq.name, uq) for uq in m_objs.values())

# for constraints that are named the same on both sides,
# keep these as a single "drop"/"add" so that the ordering
# comes out correctly
names_equal = set(c_obj_by_name).intersection(m_obj_by_name)
for name_equal in names_equal:
m_keys.remove(_uq_constraint_sig(m_obj_by_name[name_equal]))
c_keys.remove(_uq_constraint_sig(c_obj_by_name[name_equal]))

for key in m_keys.difference(c_keys):
meta_constraint = m_objs[key]
diffs.append(("add_constraint", meta_constraint))
Expand All @@ -202,9 +214,11 @@ def _compare_uniques(schema, tname, object_filters, conn_table,
key, tname
)

for key in m_keys.intersection(c_keys):
meta_constraint = m_objs[key]
conn_constraint = c_objs[key]
for meta_constraint, conn_constraint in [
(m_objs[key], c_objs[key]) for key in m_keys.intersection(c_keys)
] + [
(m_obj_by_name[key], c_obj_by_name[key]) for key in names_equal
]:
conn_cols = [col.name for col in conn_constraint.columns]
meta_cols = [col.name for col in meta_constraint.columns]

Expand Down
8 changes: 8 additions & 0 deletions docs/build/changelog.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,14 @@ Changelog
.. changelog::
:version: 0.6.2

.. change::
:tags: bug

Fixed an issue with unique constraint autogenerate detection where
a named ``UniqueConstraint`` on both sides with column changes would
render with the "add" operation before the "drop", requiring the
user to reverse the order manually.

.. change::
:tags: feature, mssql

Expand Down
219 changes: 126 additions & 93 deletions tests/test_autogenerate.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ def _default_include_object(obj, name, type_, reflected, compare_to):
if type_ == "table":
return name in ("parent", "child",
"user", "order", "item",
"address", "extra")
"address", "extra", "col_change")
else:
return True

Expand Down Expand Up @@ -750,100 +750,169 @@ def test_dont_barf_on_already_reflected(self):



class AutogenerateUniqueIndexTest(AutogenTest, TestCase):
class AutogenerateUniqueIndexTest(TestCase):

@classmethod
def _get_db_schema(cls):
m = MetaData()

Table('user', m,
def _fixture_one(self):
m1 = MetaData()
m2 = MetaData()

Table('user', m1,
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False, index=True),
Column('a1', Text, server_default="x")
)

Table('address', m,
Table('user', m2,
Column('id', Integer, primary_key=True),
Column('email_address', String(100), nullable=False),
Column('qpr', String(10), index=True),
)

Table('order', m,
Column('order_id', Integer, primary_key=True),
Column('amount', Numeric(10, 2), nullable=True),
Column('user_id', Integer, ForeignKey('user.id')),
CheckConstraint('amount > -1', name='ck_order_amount'),
UniqueConstraint('order_id', 'user_id',
name='order_order_id_user_id_unique'
),
Index('order_user_id_amount_idx', 'user_id', 'amount')
Column('name', String(50), nullable=False),
Column('a1', Text, server_default="x"),
UniqueConstraint("name", name="uq_user_name")
)
return m1, m2

Table('item', m,
Column('x', Integer),
UniqueConstraint('x', name="db_generated_name")
)
return m
def test_index_flag_becomes_named_unique_constraint(self):
diffs = self._fixture(self._fixture_one)

eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "uq_user_name")

@classmethod
def _get_model_schema(cls):
m = MetaData()
eq_(diffs[1][0], "remove_index")
eq_(diffs[1][1].name, "ix_user_name")

Table('user', m,
def _fixture_two(self):
m1 = MetaData()
m2 = MetaData()
Table('address', m1,
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False),
Column('a1', Text, server_default="x"),
UniqueConstraint("name", name="uq_user_name")
Column('email_address', String(100), nullable=False),
Column('qpr', String(10), index=True),
)

Table('address', m,
Table('address', m2,
Column('id', Integer, primary_key=True),
Column('email_address', String(100), nullable=False),
Column('qpr', String(10), index=True),
UniqueConstraint("email_address", name="uq_email_address")
)
return m1, m2

Table('order', m,
def test_add_unique_constraint(self):
diffs = self._fixture(self._fixture_two)
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "uq_email_address")

def _fixture_three(self):
m1 = MetaData()
m2 = MetaData()
Table('order', m1,
Column('order_id', Integer, primary_key=True),
Column('amount', Numeric(10, 2), nullable=True),
Column('user_id', Integer, ForeignKey('user.id')),
Column('user_id', Integer),
UniqueConstraint('order_id', 'user_id',
name='order_order_id_user_id_unique'
),
Index('order_user_id_amount_idx', 'user_id', 'amount')
)

Table('order', m2,
Column('order_id', Integer, primary_key=True),
Column('amount', Numeric(10, 2), nullable=True),
Column('user_id', Integer),
UniqueConstraint('order_id', 'user_id',
name='order_order_id_user_id_unique'
),
Index('order_user_id_amount_idx', 'user_id', 'amount', unique=True),
CheckConstraint('amount >= 0', name='ck_order_amount')
)
return m1, m2

def test_index_becomes_unique(self):
diffs = self._fixture(self._fixture_three)
eq_(diffs[0][0], "remove_index")
eq_(diffs[0][1].name, "order_user_id_amount_idx")
eq_(diffs[0][1].unique, False)

eq_(diffs[1][0], "add_index")
eq_(diffs[1][1].name, "order_user_id_amount_idx")
eq_(diffs[1][1].unique, True)

def _fixture_four(self):
m1 = MetaData()
m2 = MetaData()
Table('item', m1,
Column('x', Integer),
UniqueConstraint('x', name="db_generated_name")
)

# test mismatch between unique=True and
# named uq constraint
Table('item', m,
Table('item', m2,
Column('x', Integer, unique=True)
)
return m1, m2

def test_mismatch_db_named_col_flag(self):
diffs = self._fixture(self._fixture_four)

Table('extra', m,
eq_(diffs, [])

def _fixture_five(self):
m1 = MetaData()
m2 = MetaData()
Table('extra', m2,
Column('foo', Integer, index=True),
Column('bar', Integer),
Index('newtable_idx', 'bar')
)
return m
return m1, m2

@classmethod
@requires_07
def setup_class(cls):
def test_new_table_added(self):
diffs = self._fixture(self._fixture_five)

eq_(diffs[0][0], "add_table")

eq_(diffs[1][0], "add_index")
eq_(diffs[1][1].name, "ix_extra_foo")

eq_(diffs[2][0], "add_index")
eq_(diffs[2][1].name, "newtable_idx")

def _fixture_six(self):
m1 = MetaData()
m2 = MetaData()
Table('col_change', m1,
Column('x', Integer),
Column('y', Integer),
UniqueConstraint('x', name="nochange")
)
Table('col_change', m2,
Column('x', Integer),
Column('y', Integer),
UniqueConstraint('x', 'y', name="nochange")
)
return m1, m2

def test_named_cols_changed(self):
diffs = self._fixture(self._fixture_six)

eq_(diffs[0][0], "remove_constraint")
eq_(diffs[0][1].name, "nochange")

eq_(diffs[1][0], "add_constraint")
eq_(diffs[1][1].name, "nochange")

def _fixture(self, fn):
staging_env()
cls.bind = cls._get_bind()
cls.m2 = cls._get_db_schema()
cls.m2.create_all(cls.bind)
cls.m5 = cls._get_model_schema()
self.bind = sqlite_db()
self.metadata, model_metadata = fn()
self.metadata.create_all(self.bind)

conn = cls.bind.connect()
cls.context = context = MigrationContext.configure(
conn = self.bind.connect()
self.context = context = MigrationContext.configure(
connection=conn,
opts={
'compare_type': True,
'compare_server_default': True,
'target_metadata': cls.m5,
'target_metadata': model_metadata,
'upgrade_token': "upgrades",
'downgrade_token': "downgrades",
'alembic_module_prefix': 'op.',
Expand All @@ -852,61 +921,25 @@ def setup_class(cls):
)

connection = context.bind
cls.autogen_context = {
autogen_context = {
'imports': set(),
'connection': connection,
'dialect': connection.dialect,
'context': context
}

@classmethod
def teardown_class(cls):
cls.m2.drop_all(cls.bind)
clear_staging_env()

def test_diffs(self):
"""test generation of diff rules"""

metadata = self.m5
connection = self.context.bind
diffs = []
autogenerate._produce_net_changes(connection, metadata, diffs,
self.autogen_context,
autogenerate._produce_net_changes(connection, model_metadata, diffs,
autogen_context,
object_filters=_default_object_filters,
)
return diffs

eq_(diffs[0][0], "add_table")
eq_(diffs[0][1].name, "extra")

eq_(diffs[1][0], "add_index")
eq_(diffs[2][0], "add_index")

eq_(set([
diffs[1][1].name,
diffs[2][1].name
]),
set([
"ix_extra_foo", # sqlalchemy's naming scheme
"newtable_idx"
])
)

eq_(diffs[3][0], "add_constraint")
eq_(diffs[3][1].table.name, "address")

eq_(diffs[4][0], "remove_index")
eq_(diffs[4][1].name, "order_user_id_amount_idx")
eq_(diffs[4][1].unique, False)
def tearDown(self):
self.metadata.drop_all(self.bind)
clear_staging_env()

eq_(diffs[5][0], "add_index")
eq_(diffs[5][1].name, "order_user_id_amount_idx")
eq_(diffs[5][1].unique, True)

eq_(diffs[6][0], "add_constraint")
eq_(diffs[6][1].table.name, "user")

eq_(diffs[7][0], "remove_index")
eq_(diffs[7][1].name, "ix_user_name")

class PGUniqueIndexTest(AutogenerateUniqueIndexTest):
@classmethod
Expand Down

0 comments on commit 74d1ccf

Please sign in to comment.