From 74d1ccfabc40853b8018e7847f4c3c2b799424c8 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Wed, 11 Dec 2013 17:00:06 -0500 Subject: [PATCH] - 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. - 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 --- alembic/autogenerate/compare.py | 20 ++- docs/build/changelog.rst | 8 ++ tests/test_autogenerate.py | 219 ++++++++++++++++++-------------- 3 files changed, 151 insertions(+), 96 deletions(-) diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py index 3b2e1d7..4fb61ab 100644 --- a/alembic/autogenerate/compare.py +++ b/alembic/autogenerate/compare.py @@ -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, @@ -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)) @@ -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] diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst index f960c59..c51842b 100644 --- a/docs/build/changelog.rst +++ b/docs/build/changelog.rst @@ -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 diff --git a/tests/test_autogenerate.py b/tests/test_autogenerate.py index 47c207d..5e2a488 100644 --- a/tests/test_autogenerate.py +++ b/tests/test_autogenerate.py @@ -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 @@ -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.', @@ -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