From e533927189d7c6d9ce08d59ecb1f2bae0bd00783 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Wed, 16 Mar 2022 10:55:46 -0400 Subject: [PATCH] add covering index tests; generalize autogen index tests In response to #1006, add tests for covering indexes for PostgreSQL and MSSQL. Refactor test_autogen_indexes to be generalized for all backends. Parts of this test module should eventually move to the suite tests. Fixes: #1006 Change-Id: I126b0bcbf4f065e5b148567929fece898e66821e --- alembic/testing/suite/_autogen_fixtures.py | 2 - tests/requirements.py | 14 +- tests/test_autogen_indexes.py | 1142 +++++++++----------- tests/test_postgresql.py | 206 ++++ 4 files changed, 712 insertions(+), 652 deletions(-) diff --git a/alembic/testing/suite/_autogen_fixtures.py b/alembic/testing/suite/_autogen_fixtures.py index ea1957ae..c47cc10e 100644 --- a/alembic/testing/suite/_autogen_fixtures.py +++ b/alembic/testing/suite/_autogen_fixtures.py @@ -323,8 +323,6 @@ class AutogenFixtureTest(_ComparesFKs): dialect._user_defined_max_identifier_length ) = existing_length - reports_unnamed_constraints = False - def setUp(self): staging_env() self.bind = config.db diff --git a/tests/requirements.py b/tests/requirements.py index 258c7300..1060e25e 100644 --- a/tests/requirements.py +++ b/tests/requirements.py @@ -124,7 +124,15 @@ class DefaultRequirements(SuiteRequirements): @property def reflects_unique_constraints_unambiguously(self): - return exclusions.fails_on(["mysql", "mariadb", "oracle"]) + return exclusions.fails_on(["mysql", "mariadb", "oracle", "mssql"]) + + @property + def reports_unique_constraints_as_indexes(self): + return exclusions.only_on(["mysql", "mariadb", "oracle"]) + + @property + def reports_unnamed_constraints(self): + return exclusions.skip_if(["sqlite"]) @property def reflects_indexes_w_sorting(self): @@ -169,6 +177,10 @@ class DefaultRequirements(SuiteRequirements): def mssql(self): return exclusions.only_on(["mssql"]) + @property + def covering_indexes(self): + return exclusions.only_on(["postgresql >= 11", "mssql"]) + @property def postgresql_uuid_ossp(self): def check_uuid_ossp(config): diff --git a/tests/test_autogen_indexes.py b/tests/test_autogen_indexes.py index 8a4c26f4..fb710991 100644 --- a/tests/test_autogen_indexes.py +++ b/tests/test_autogen_indexes.py @@ -1,7 +1,6 @@ from sqlalchemy import Column from sqlalchemy import ForeignKey from sqlalchemy import ForeignKeyConstraint -from sqlalchemy import func from sqlalchemy import Index from sqlalchemy import Integer from sqlalchemy import MetaData @@ -13,10 +12,10 @@ from sqlalchemy import UniqueConstraint from sqlalchemy.sql.expression import column from sqlalchemy.sql.expression import desc -from alembic.testing import assertions from alembic.testing import combinations from alembic.testing import config from alembic.testing import eq_ +from alembic.testing import exclusions from alembic.testing import schemacompare from alembic.testing import TestBase from alembic.testing import util @@ -24,15 +23,15 @@ from alembic.testing.env import staging_env from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest from alembic.util import sqla_compat -# TODO: create new suites that are taking tests from this suite, with a -# separate class for AutogenIndexes, AutogenUniqueConstraint, and a -# subset of the tests here. @zzzeek can work on this at a later point. -# (2021-06-10) - class NoUqReflection: + """mixin used to simulate dialects where unique constraints are + not reflected.""" + __requires__ = () + reports_unique_constraints = False + def setUp(self): staging_env() self.bind = eng = util.testing_engine() @@ -42,19 +41,32 @@ class NoUqReflection: eng.dialect.get_unique_constraints = unimpl - def test_add_ix_on_table_create(self): - return super(NoUqReflection, self).test_add_ix_on_table_create() - def test_add_idx_non_col(self): - return super(NoUqReflection, self).test_add_idx_non_col() +class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): + """tests that involve unique constraint reflection, or the lack of + this feature and the expected behaviors, and its interaction with index + reflection. + Tests that do not involve unique constraint reflection, but involve + indexes, should go into AutogenerateIndexTest. -class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): - reports_unique_constraints = True - reports_unique_constraints_as_indexes = False + """ - __requires__ = ("unique_constraint_reflection",) - __only_on__ = "sqlite" + __backend__ = True + + @property + def reports_unique_constraints(self): + return config.requirements.unique_constraint_reflection.enabled + + @property + def reports_unique_constraints_as_indexes(self): + return ( + config.requirements.reports_unique_constraints_as_indexes.enabled + ) + + @property + def reports_unnamed_constraints(self): + return config.requirements.reports_unnamed_constraints.enabled def test_index_flag_becomes_named_unique_constraint(self): m1 = MetaData() @@ -196,32 +208,6 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): eq_(diffs, []) - def test_new_table_added(self): - m1 = MetaData() - m2 = MetaData() - Table( - "extra", - m2, - Column("foo", Integer, index=True), - Column("bar", Integer), - Index("newtable_idx", "bar"), - ) - - diffs = self._fixture(m1, m2) - - eq_(diffs[0][0], "add_table") - - eq_(diffs[1][0], "add_index") - eq_( - sqla_compat._get_constraint_final_name( - diffs[1][1], config.db.dialect - ), - "ix_extra_foo", - ) - - eq_(diffs[2][0], "add_index") - eq_(diffs[2][1].name, "newtable_idx") - def test_named_cols_changed(self): m1 = MetaData() m2 = MetaData() @@ -338,41 +324,6 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): diffs = self._fixture(m1, m2, max_identifier_length=30) eq_(diffs, []) - @config.requirements.long_names - def test_nothing_ix_changed_labels_were_truncated(self): - m1 = MetaData( - naming_convention={ - "ix": "index_%(table_name)s_%(column_0_label)s", - "uq": "unique_%(table_name)s_%(column_0_label)s", - } - ) - m2 = MetaData( - naming_convention={ - "ix": "index_%(table_name)s_%(column_0_label)s", - "uq": "unique_%(table_name)s_%(column_0_label)s", - } - ) - - Table( - "nothing_changed", - m1, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), - Column("a_particularly_long_column_name", String(20), index=True), - mysql_engine="InnoDB", - ) - - Table( - "nothing_changed", - m2, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), - Column("a_particularly_long_column_name", String(20), index=True), - mysql_engine="InnoDB", - ) - diffs = self._fixture(m1, m2, max_identifier_length=30) - eq_(diffs, []) - @config.requirements.long_names def test_nothing_changed_uq_w_mixed_case_nconv_name(self): m1 = MetaData( @@ -433,64 +384,6 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): diffs = self._fixture(m1, m2) eq_(diffs, []) - def test_nothing_changed_ix_w_mixed_case_plain_name(self): - m1 = MetaData() - m2 = MetaData() - - Table( - "nothing_changed", - m1, - Column("id", Integer, primary_key=True), - Column("x", Integer), - Index("SomeIndex", "x"), - mysql_engine="InnoDB", - ) - - Table( - "nothing_changed", - m2, - Column("id", Integer, primary_key=True), - Column("x", Integer), - Index("SomeIndex", "x"), - mysql_engine="InnoDB", - ) - diffs = self._fixture(m1, m2) - eq_(diffs, []) - - @config.requirements.long_names - def test_nothing_changed_ix_w_mixed_case_nconv_name(self): - m1 = MetaData( - naming_convention={ - "ix": "index_%(table_name)s_%(column_0_label)s", - "uq": "unique_%(table_name)s_%(column_0_label)s", - } - ) - m2 = MetaData( - naming_convention={ - "ix": "index_%(table_name)s_%(column_0_label)s", - "uq": "unique_%(table_name)s_%(column_0_label)s", - } - ) - - Table( - "NothingChanged", - m1, - Column("id", Integer, primary_key=True), - Column("XCol", Integer, index=True), - mysql_engine="InnoDB", - ) - - Table( - "NothingChanged", - m2, - Column("id", Integer, primary_key=True), - Column("XCol", Integer, index=True), - mysql_engine="InnoDB", - ) - - diffs = self._fixture(m1, m2) - eq_(diffs, []) - def test_nothing_changed_two(self): m1 = MetaData() m2 = MetaData() @@ -557,478 +450,630 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): diffs = self._fixture(m1, m2) eq_(diffs, []) - def test_nothing_changed_index_w_colkeys(self): + @config.requirements.unique_constraint_reflection + def test_uq_casing_convention_changed_so_put_drops_first(self): m1 = MetaData() m2 = MetaData() + uq1 = UniqueConstraint("x", name="SomeCasingConvention") Table( - "nothing_changed", + "new_idx", m1, - Column("x", String(20), key="nx"), - Index("foobar", "nx"), + Column("id1", Integer, primary_key=True), + Column("x", String(20)), + uq1, ) + uq2 = UniqueConstraint("x", name="somecasingconvention") Table( - "nothing_changed", + "new_idx", m2, - Column("x", String(20), key="nx"), - Index("foobar", "nx"), + Column("id1", Integer, primary_key=True), + Column("x", String(20)), + uq2, ) diffs = self._fixture(m1, m2) - eq_(diffs, []) - def test_nothing_changed_index_named_as_column(self): + if self.reports_unique_constraints_as_indexes: + eq_( + [(d[0], d[1].name) for d in diffs], + [ + ("remove_index", "SomeCasingConvention"), + ("add_constraint", "somecasingconvention"), + ], + ) + else: + eq_( + [(d[0], d[1].name) for d in diffs], + [ + ("remove_constraint", "SomeCasingConvention"), + ("add_constraint", "somecasingconvention"), + ], + ) + + def test_drop_table_w_uq_constraint(self): m1 = MetaData() m2 = MetaData() Table( - "nothing_changed", + "some_table", m1, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), - Column("x", String(20)), - Index("x", "x"), - ) - - Table( - "nothing_changed", - m2, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), + Column("id", Integer, primary_key=True), Column("x", String(20)), - Index("x", "x"), + Column("y", String(20)), + UniqueConstraint("y", name="uq_y"), ) diffs = self._fixture(m1, m2) - eq_(diffs, []) - def test_nothing_changed_implicit_fk_index_named(self): + if self.reports_unique_constraints_as_indexes: + # for MySQL this UQ will look like an index, so + # make sure it at least sets it up correctly + eq_(diffs[0][0], "remove_index") + eq_(diffs[1][0], "remove_table") + eq_(len(diffs), 2) + + constraints = [ + c + for c in diffs[1][1].constraints + if isinstance(c, UniqueConstraint) + ] + eq_(len(constraints), 0) + else: + eq_(diffs[0][0], "remove_table") + eq_(len(diffs), 1) + constraints = [ + c + for c in diffs[0][1].constraints + if isinstance(c, UniqueConstraint) + ] + if self.reports_unique_constraints: + eq_(len(constraints), 1) + + @config.requirements.unique_constraint_reflection + def test_unnamed_cols_changed(self): m1 = MetaData() m2 = MetaData() - - Table( - "nothing_changed", - m1, - Column("id", Integer, primary_key=True), - Column( - "other_id", - ForeignKey("nc2.id", name="fk_my_table_other_table"), - nullable=False, - ), - Column("foo", Integer), - mysql_engine="InnoDB", - ) Table( - "nc2", + "col_change", m1, - Column("id", Integer, primary_key=True), - mysql_engine="InnoDB", + Column("x", Integer), + Column("y", Integer), + UniqueConstraint("x"), ) - Table( - "nothing_changed", + "col_change", m2, - Column("id", Integer, primary_key=True), - Column( - "other_id", - ForeignKey("nc2.id", name="fk_my_table_other_table"), - nullable=False, - ), - Column("foo", Integer), - mysql_engine="InnoDB", + Column("x", Integer), + Column("y", Integer), + UniqueConstraint("x", "y"), ) - Table( - "nc2", - m2, - Column("id", Integer, primary_key=True), - mysql_engine="InnoDB", + + diffs = self._fixture(m1, m2) + + diffs = set( + ( + cmd, + isinstance(obj, (UniqueConstraint, Index)) + if obj.name is not None + else False, + ) + for cmd, obj in diffs + ) + + if self.reports_unnamed_constraints: + if self.reports_unique_constraints_as_indexes: + eq_( + diffs, + set([("remove_index", True), ("add_constraint", False)]), + ) + else: + eq_( + diffs, + set( + [ + ("remove_constraint", True), + ("add_constraint", False), + ] + ), + ) + + def test_remove_named_unique_index(self): + m1 = MetaData() + m2 = MetaData() + + Table( + "remove_idx", + m1, + Column("x", Integer), + Index("xidx", "x", unique=True), ) + Table("remove_idx", m2, Column("x", Integer)) + diffs = self._fixture(m1, m2) - eq_(diffs, []) - def test_nothing_changed_implicit_composite_fk_index_named(self): + if self.reports_unique_constraints: + diffs = set((cmd, obj.name) for cmd, obj in diffs) + eq_(diffs, set([("remove_index", "xidx")])) + else: + eq_(diffs, []) + + def test_remove_named_unique_constraint(self): m1 = MetaData() m2 = MetaData() Table( - "nothing_changed", + "remove_idx", m1, - Column("id", Integer, primary_key=True), - Column("other_id_1", Integer), - Column("other_id_2", Integer), - Column("foo", Integer), - ForeignKeyConstraint( - ["other_id_1", "other_id_2"], - ["nc2.id1", "nc2.id2"], - name="fk_my_table_other_table", + Column("x", Integer), + UniqueConstraint("x", name="xidx"), + ) + Table("remove_idx", m2, Column("x", Integer)) + + diffs = self._fixture(m1, m2) + + if self.reports_unique_constraints: + diffs = set((cmd, obj.name) for cmd, obj in diffs) + if self.reports_unique_constraints_as_indexes: + eq_(diffs, set([("remove_index", "xidx")])) + else: + eq_(diffs, set([("remove_constraint", "xidx")])) + else: + eq_(diffs, []) + + def test_dont_add_uq_on_table_create(self): + m1 = MetaData() + m2 = MetaData() + Table("no_uq", m2, Column("x", String(50), unique=True)) + diffs = self._fixture(m1, m2) + + eq_(diffs[0][0], "add_table") + eq_(len(diffs), 1) + + # checking for dupes also + eq_( + sorted( + [type(cons) for cons in diffs[0][1].constraints], + key=lambda c: c.__name__, ), - mysql_engine="InnoDB", + [PrimaryKeyConstraint, UniqueConstraint], + ) + + @config.requirements.reflects_unique_constraints_unambiguously + def test_dont_add_uq_on_reverse_table_drop(self): + m1 = MetaData() + m2 = MetaData() + Table("no_uq", m1, Column("x", String(50), unique=True)) + diffs = self._fixture(m1, m2) + + eq_(diffs[0][0], "remove_table") + eq_(len(diffs), 1) + + # because the drop comes from reflection, the "unique=True" flag + # is lost in any case. + eq_( + sorted( + [type(cons) for cons in diffs[0][1].constraints], + key=lambda c: c.__name__, + ), + [PrimaryKeyConstraint, UniqueConstraint], + ) + + def test_add_uq_ix_on_table_create(self): + m1 = MetaData() + m2 = MetaData() + Table("add_ix", m2, Column("x", String(50), unique=True, index=True)) + diffs = self._fixture(m1, m2) + + eq_(diffs[0][0], "add_table") + eq_(len(diffs), 2) + assert UniqueConstraint not in set( + type(c) for c in diffs[0][1].constraints ) + + eq_(diffs[1][0], "add_index") + d_table = diffs[0][1] + d_idx = diffs[1][1] + eq_(d_idx.unique, True) + + # check for dupes + eq_(len(diffs), 2) + assert not d_table.indexes + + +class AutogenerateIndexTest(AutogenFixtureTest, TestBase): + """tests involving indexes but not unique constraints, as mssql + doesn't have these (?)...at least the dialect seems to not + reflect unique constraints which seems odd + + """ + + __backend__ = True + + def test_nothing_changed_one(self): + m1 = MetaData() + m2 = MetaData() + Table( - "nc2", + "nothing_changed", m1, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), - mysql_engine="InnoDB", + Column("x", String(20), index=True), ) Table( "nothing_changed", m2, - Column("id", Integer, primary_key=True), - Column("other_id_1", Integer), - Column("other_id_2", Integer), - Column("foo", Integer), - ForeignKeyConstraint( - ["other_id_1", "other_id_2"], - ["nc2.id1", "nc2.id2"], - name="fk_my_table_other_table", - ), + Column("x", String(20), index=True), + ) + + diffs = self._fixture(m1, m2) + eq_(diffs, []) + + @config.requirements.long_names + def test_nothing_ix_changed_labels_were_truncated(self): + m1 = MetaData( + naming_convention={ + "ix": "index_%(table_name)s_%(column_0_label)s", + "uq": "unique_%(table_name)s_%(column_0_label)s", + } + ) + m2 = MetaData( + naming_convention={ + "ix": "index_%(table_name)s_%(column_0_label)s", + "uq": "unique_%(table_name)s_%(column_0_label)s", + } + ) + + Table( + "nothing_changed", + m1, + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + Column("a_particularly_long_column_name", String(20), index=True), mysql_engine="InnoDB", ) + Table( - "nc2", + "nothing_changed", m2, Column("id1", Integer, primary_key=True), Column("id2", Integer, primary_key=True), + Column("a_particularly_long_column_name", String(20), index=True), mysql_engine="InnoDB", ) - diffs = self._fixture(m1, m2) + diffs = self._fixture(m1, m2, max_identifier_length=30) eq_(diffs, []) - def test_ix_casing_convention_changed_so_put_drops_first(self): + def test_nothing_changed_ix_w_mixed_case_plain_name(self): m1 = MetaData() m2 = MetaData() - ix1 = Index("SomeCasingConvention", "x") Table( - "new_idx", + "nothing_changed", m1, - Column("id1", Integer, primary_key=True), - Column("x", String(20)), - ix1, + Column("id", Integer, primary_key=True), + Column("x", Integer), + Index("SomeIndex", "x"), + mysql_engine="InnoDB", ) - ix2 = Index("somecasingconvention", "x") Table( - "new_idx", + "nothing_changed", m2, - Column("id1", Integer, primary_key=True), - Column("x", String(20)), - ix2, + Column("id", Integer, primary_key=True), + Column("x", Integer), + Index("SomeIndex", "x"), + mysql_engine="InnoDB", ) - diffs = self._fixture(m1, m2) + eq_(diffs, []) - eq_( - [(d[0], d[1].name) for d in diffs], - [ - ("remove_index", "SomeCasingConvention"), - ("add_index", "somecasingconvention"), - ], + @config.requirements.long_names + def test_nothing_changed_ix_w_mixed_case_nconv_name(self): + m1 = MetaData( + naming_convention={ + "ix": "index_%(table_name)s_%(column_0_label)s", + "uq": "unique_%(table_name)s_%(column_0_label)s", + } + ) + m2 = MetaData( + naming_convention={ + "ix": "index_%(table_name)s_%(column_0_label)s", + "uq": "unique_%(table_name)s_%(column_0_label)s", + } ) - def test_uq_casing_convention_changed_so_put_drops_first(self): - m1 = MetaData() - m2 = MetaData() - - uq1 = UniqueConstraint("x", name="SomeCasingConvention") Table( - "new_idx", + "NothingChanged", m1, - Column("id1", Integer, primary_key=True), - Column("x", String(20)), - uq1, + Column("id", Integer, primary_key=True), + Column("XCol", Integer, index=True), + mysql_engine="InnoDB", ) - uq2 = UniqueConstraint("x", name="somecasingconvention") Table( - "new_idx", + "NothingChanged", m2, - Column("id1", Integer, primary_key=True), - Column("x", String(20)), - uq2, + Column("id", Integer, primary_key=True), + Column("XCol", Integer, index=True), + mysql_engine="InnoDB", ) diffs = self._fixture(m1, m2) + eq_(diffs, []) - if self.reports_unique_constraints_as_indexes: - eq_( - [(d[0], d[1].name) for d in diffs], - [ - ("remove_index", "SomeCasingConvention"), - ("add_constraint", "somecasingconvention"), - ], - ) - else: - eq_( - [(d[0], d[1].name) for d in diffs], - [ - ("remove_constraint", "SomeCasingConvention"), - ("add_constraint", "somecasingconvention"), - ], - ) + def test_new_table_added(self): + m1 = MetaData() + m2 = MetaData() + Table( + "extra", + m2, + Column("foo", Integer, index=True), + Column("bar", Integer), + Index("newtable_idx", "bar"), + ) - def test_new_idx_index_named_as_column(self): + diffs = self._fixture(m1, m2) + + eq_(diffs[0][0], "add_table") + + eq_(diffs[1][0], "add_index") + eq_( + sqla_compat._get_constraint_final_name( + diffs[1][1], config.db.dialect + ), + "ix_extra_foo", + ) + + eq_(diffs[2][0], "add_index") + eq_(diffs[2][1].name, "newtable_idx") + + def test_nothing_changed_index_w_colkeys(self): m1 = MetaData() m2 = MetaData() Table( - "new_idx", + "nothing_changed", m1, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), - Column("x", String(20)), + Column("x", String(20), key="nx"), + Index("foobar", "nx"), ) - idx = Index("x", "x") Table( - "new_idx", + "nothing_changed", m2, - Column("id1", Integer, primary_key=True), - Column("id2", Integer, primary_key=True), - Column("x", String(20)), - idx, + Column("x", String(20), key="nx"), + Index("foobar", "nx"), ) diffs = self._fixture(m1, m2) - eq_(diffs, [("add_index", schemacompare.CompareIndex(idx))]) + eq_(diffs, []) - def test_removed_idx_index_named_as_column(self): + def test_nothing_changed_index_named_as_column(self): m1 = MetaData() m2 = MetaData() - idx = Index("x", "x") Table( - "new_idx", + "nothing_changed", m1, Column("id1", Integer, primary_key=True), Column("id2", Integer, primary_key=True), Column("x", String(20)), - idx, + Index("x", "x"), ) Table( - "new_idx", + "nothing_changed", m2, Column("id1", Integer, primary_key=True), Column("id2", Integer, primary_key=True), Column("x", String(20)), + Index("x", "x"), ) diffs = self._fixture(m1, m2) - eq_(diffs[0][0], "remove_index") + eq_(diffs, []) - def test_drop_table_w_indexes(self): + def test_nothing_changed_implicit_fk_index_named(self): m1 = MetaData() m2 = MetaData() - t = Table( - "some_table", + Table( + "nothing_changed", m1, Column("id", Integer, primary_key=True), - Column("x", String(20)), - Column("y", String(20)), - ) - Index("xy_idx", t.c.x, t.c.y) - Index("y_idx", t.c.y) - - diffs = self._fixture(m1, m2) - eq_(diffs[0][0], "remove_index") - eq_(diffs[1][0], "remove_index") - eq_(diffs[2][0], "remove_table") - - eq_( - set([diffs[0][1].name, diffs[1][1].name]), set(["xy_idx", "y_idx"]) + Column( + "other_id", + ForeignKey("nc2.id", name="fk_my_table_other_table"), + nullable=False, + ), + Column("foo", Integer), + mysql_engine="InnoDB", ) - - def test_drop_table_w_uq_constraint(self): - m1 = MetaData() - m2 = MetaData() - Table( - "some_table", + "nc2", m1, Column("id", Integer, primary_key=True), - Column("x", String(20)), - Column("y", String(20)), - UniqueConstraint("y", name="uq_y"), + mysql_engine="InnoDB", ) + Table( + "nothing_changed", + m2, + Column("id", Integer, primary_key=True), + Column( + "other_id", + ForeignKey("nc2.id", name="fk_my_table_other_table"), + nullable=False, + ), + Column("foo", Integer), + mysql_engine="InnoDB", + ) + Table( + "nc2", + m2, + Column("id", Integer, primary_key=True), + mysql_engine="InnoDB", + ) diffs = self._fixture(m1, m2) + eq_(diffs, []) - if self.reports_unique_constraints_as_indexes: - # for MySQL this UQ will look like an index, so - # make sure it at least sets it up correctly - eq_(diffs[0][0], "remove_index") - eq_(diffs[1][0], "remove_table") - eq_(len(diffs), 2) - - constraints = [ - c - for c in diffs[1][1].constraints - if isinstance(c, UniqueConstraint) - ] - eq_(len(constraints), 0) - else: - eq_(diffs[0][0], "remove_table") - eq_(len(diffs), 1) - constraints = [ - c - for c in diffs[0][1].constraints - if isinstance(c, UniqueConstraint) - ] - if self.reports_unique_constraints: - eq_(len(constraints), 1) - - def test_unnamed_cols_changed(self): + def test_nothing_changed_implicit_composite_fk_index_named(self): m1 = MetaData() m2 = MetaData() + Table( - "col_change", + "nothing_changed", m1, - Column("x", Integer), - Column("y", Integer), - UniqueConstraint("x"), + Column("id", Integer, primary_key=True), + Column("other_id_1", Integer), + Column("other_id_2", Integer), + Column("foo", Integer), + ForeignKeyConstraint( + ["other_id_1", "other_id_2"], + ["nc2.id1", "nc2.id2"], + name="fk_my_table_other_table", + ), + mysql_engine="InnoDB", ) Table( - "col_change", - m2, - Column("x", Integer), - Column("y", Integer), - UniqueConstraint("x", "y"), + "nc2", + m1, + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + mysql_engine="InnoDB", ) - diffs = self._fixture(m1, m2) - - diffs = set( - ( - cmd, - isinstance(obj, (UniqueConstraint, Index)) - if obj.name is not None - else False, - ) - for cmd, obj in diffs + Table( + "nothing_changed", + m2, + Column("id", Integer, primary_key=True), + Column("other_id_1", Integer), + Column("other_id_2", Integer), + Column("foo", Integer), + ForeignKeyConstraint( + ["other_id_1", "other_id_2"], + ["nc2.id1", "nc2.id2"], + name="fk_my_table_other_table", + ), + mysql_engine="InnoDB", ) - if self.reports_unnamed_constraints: - if self.reports_unique_constraints_as_indexes: - eq_( - diffs, - set([("remove_index", True), ("add_constraint", False)]), - ) - else: - eq_( - diffs, - set( - [ - ("remove_constraint", True), - ("add_constraint", False), - ] - ), - ) + Table( + "nc2", + m2, + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + mysql_engine="InnoDB", + ) + diffs = self._fixture(m1, m2) + eq_(diffs, []) - def test_remove_named_unique_index(self): + def test_ix_casing_convention_changed_so_put_drops_first(self): m1 = MetaData() m2 = MetaData() + ix1 = Index("SomeCasingConvention", "x") Table( - "remove_idx", + "new_idx", m1, - Column("x", Integer), - Index("xidx", "x", unique=True), + Column("id1", Integer, primary_key=True), + Column("x", String(20)), + ix1, + ) + + ix2 = Index("somecasingconvention", "x") + Table( + "new_idx", + m2, + Column("id1", Integer, primary_key=True), + Column("x", String(20)), + ix2, ) - Table("remove_idx", m2, Column("x", Integer)) diffs = self._fixture(m1, m2) - if self.reports_unique_constraints: - diffs = set((cmd, obj.name) for cmd, obj in diffs) - eq_(diffs, set([("remove_index", "xidx")])) - else: - eq_(diffs, []) + eq_( + [(d[0], d[1].name) for d in diffs], + [ + ("remove_index", "SomeCasingConvention"), + ("add_index", "somecasingconvention"), + ], + ) - def test_remove_named_unique_constraint(self): + def test_new_idx_index_named_as_column(self): m1 = MetaData() m2 = MetaData() Table( - "remove_idx", + "new_idx", m1, - Column("x", Integer), - UniqueConstraint("x", name="xidx"), + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + Column("x", String(20)), ) - Table("remove_idx", m2, Column("x", Integer)) - diffs = self._fixture(m1, m2) + idx = Index("x", "x") + Table( + "new_idx", + m2, + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + Column("x", String(20)), + idx, + ) - if self.reports_unique_constraints: - diffs = set((cmd, obj.name) for cmd, obj in diffs) - if self.reports_unique_constraints_as_indexes: - eq_(diffs, set([("remove_index", "xidx")])) - else: - eq_(diffs, set([("remove_constraint", "xidx")])) - else: - eq_(diffs, []) + diffs = self._fixture(m1, m2) + eq_(diffs, [("add_index", schemacompare.CompareIndex(idx))]) - def test_dont_add_uq_on_table_create(self): + @exclusions.fails_on(["mysql", "mariadb"]) + def test_removed_idx_index_named_as_column(self): m1 = MetaData() m2 = MetaData() - Table("no_uq", m2, Column("x", String(50), unique=True)) - diffs = self._fixture(m1, m2) - eq_(diffs[0][0], "add_table") - eq_(len(diffs), 1) + idx = Index("x", "x") + Table( + "new_idx", + m1, + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + Column("x", String(20)), + idx, + ) - # checking for dupes also - eq_( - sorted( - [type(cons) for cons in diffs[0][1].constraints], - key=lambda c: c.__name__, - ), - [PrimaryKeyConstraint, UniqueConstraint], + Table( + "new_idx", + m2, + Column("id1", Integer, primary_key=True), + Column("id2", Integer, primary_key=True), + Column("x", String(20)), ) - @config.requirements.reflects_unique_constraints_unambiguously - def test_dont_add_uq_on_reverse_table_drop(self): - m1 = MetaData() - m2 = MetaData() - Table("no_uq", m1, Column("x", String(50), unique=True)) diffs = self._fixture(m1, m2) + eq_(diffs[0][0], "remove_index") - eq_(diffs[0][0], "remove_table") - eq_(len(diffs), 1) - - # because the drop comes from reflection, the "unique=True" flag - # is lost in any case. - eq_( - sorted( - [type(cons) for cons in diffs[0][1].constraints], - key=lambda c: c.__name__, - ), - [PrimaryKeyConstraint, UniqueConstraint], - ) - - def test_add_uq_ix_on_table_create(self): + def test_drop_table_w_indexes(self): m1 = MetaData() m2 = MetaData() - Table("add_ix", m2, Column("x", String(50), unique=True, index=True)) - diffs = self._fixture(m1, m2) - eq_(diffs[0][0], "add_table") - eq_(len(diffs), 2) - assert UniqueConstraint not in set( - type(c) for c in diffs[0][1].constraints + t = Table( + "some_table", + m1, + Column("id", Integer, primary_key=True), + Column("x", String(20)), + Column("y", String(20)), ) + Index("xy_idx", t.c.x, t.c.y) + Index("y_idx", t.c.y) - eq_(diffs[1][0], "add_index") - d_table = diffs[0][1] - d_idx = diffs[1][1] - eq_(d_idx.unique, True) + diffs = self._fixture(m1, m2) + eq_(diffs[0][0], "remove_index") + eq_(diffs[1][0], "remove_index") + eq_(diffs[2][0], "remove_table") - # check for dupes - eq_(len(diffs), 2) - assert not d_table.indexes + eq_( + set([diffs[0][1].name, diffs[1][1].name]), set(["xy_idx", "y_idx"]) + ) def test_add_ix_on_table_create(self): m1 = MetaData() @@ -1106,237 +1151,36 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase): eq_(diffs, []) - -class PGUniqueIndexTest(AutogenerateUniqueIndexTest): - reports_unnamed_constraints = True - __only_on__ = "postgresql" - __backend__ = True - - def test_idx_added_schema(self): - m1 = MetaData() - m2 = MetaData() - Table("add_ix", m1, Column("x", String(50)), schema="test_schema") - Table( - "add_ix", - m2, - Column("x", String(50)), - Index("ix_1", "x"), - schema="test_schema", - ) - - diffs = self._fixture(m1, m2, include_schemas=True) - eq_(diffs[0][0], "add_index") - eq_(diffs[0][1].name, "ix_1") - - def test_idx_unchanged_schema(self): - m1 = MetaData() - m2 = MetaData() - Table( - "add_ix", - m1, - Column("x", String(50)), - Index("ix_1", "x"), - schema="test_schema", - ) - Table( - "add_ix", - m2, - Column("x", String(50)), - Index("ix_1", "x"), - schema="test_schema", - ) - - diffs = self._fixture(m1, m2, include_schemas=True) - eq_(diffs, []) - - def test_uq_added_schema(self): - m1 = MetaData() - m2 = MetaData() - Table("add_uq", m1, Column("x", String(50)), schema="test_schema") - Table( - "add_uq", - m2, - Column("x", String(50)), - UniqueConstraint("x", name="ix_1"), - schema="test_schema", - ) - - diffs = self._fixture(m1, m2, include_schemas=True) - eq_(diffs[0][0], "add_constraint") - eq_(diffs[0][1].name, "ix_1") - - def test_uq_unchanged_schema(self): + @config.requirements.covering_indexes + @config.requirements.sqlalchemy_14 + def test_nothing_changed_covering_index(self): m1 = MetaData() m2 = MetaData() - Table( - "add_uq", - m1, - Column("x", String(50)), - UniqueConstraint("x", name="ix_1"), - schema="test_schema", - ) - Table( - "add_uq", - m2, - Column("x", String(50)), - UniqueConstraint("x", name="ix_1"), - schema="test_schema", - ) - - diffs = self._fixture(m1, m2, include_schemas=True) - eq_(diffs, []) - - @config.requirements.btree_gist - def test_exclude_const_unchanged(self): - from sqlalchemy.dialects.postgresql import TSRANGE, ExcludeConstraint - m1 = MetaData() - m2 = MetaData() + cov_opt = {f"{config.db.name}_include": ["y"]} Table( - "add_excl", + "nothing_changed", m1, Column("id", Integer, primary_key=True), - Column("period", TSRANGE), - ExcludeConstraint(("period", "&&"), name="quarters_period_excl"), + Column("x", Integer), + Column("y", Integer), + Index("SomeIndex", "x", **cov_opt), ) Table( - "add_excl", + "nothing_changed", m2, Column("id", Integer, primary_key=True), - Column("period", TSRANGE), - ExcludeConstraint(("period", "&&"), name="quarters_period_excl"), + Column("x", Integer), + Column("y", Integer), + Index("SomeIndex", "x", **cov_opt), ) - diffs = self._fixture(m1, m2) eq_(diffs, []) - def test_same_tname_two_schemas(self): - m1 = MetaData() - m2 = MetaData() - - Table("add_ix", m1, Column("x", String(50)), Index("ix_1", "x")) - - Table("add_ix", m2, Column("x", String(50)), Index("ix_1", "x")) - Table("add_ix", m2, Column("x", String(50)), schema="test_schema") - - diffs = self._fixture(m1, m2, include_schemas=True) - eq_(diffs[0][0], "add_table") - eq_(len(diffs), 1) - - def test_uq_dropped(self): - m1 = MetaData() - m2 = MetaData() - Table( - "add_uq", - m1, - Column("id", Integer, primary_key=True), - Column("name", String), - UniqueConstraint("name", name="uq_name"), - ) - Table( - "add_uq", - m2, - Column("id", Integer, primary_key=True), - Column("name", String), - ) - diffs = self._fixture(m1, m2, include_schemas=True) - eq_(diffs[0][0], "remove_constraint") - eq_(diffs[0][1].name, "uq_name") - eq_(len(diffs), 1) - - def test_functional_ix_one(self): - m1 = MetaData() - m2 = MetaData() - - t1 = Table( - "foo", - m1, - Column("id", Integer, primary_key=True), - Column("email", String(50)), - ) - Index("email_idx", func.lower(t1.c.email), unique=True) - - t2 = Table( - "foo", - m2, - Column("id", Integer, primary_key=True), - Column("email", String(50)), - ) - Index("email_idx", func.lower(t2.c.email), unique=True) - - with assertions.expect_warnings( - "Skipped unsupported reflection", - "autogenerate skipping functional index", - ): - diffs = self._fixture(m1, m2) - eq_(diffs, []) - - def test_functional_ix_two(self): - m1 = MetaData() - m2 = MetaData() - - t1 = Table( - "foo", - m1, - Column("id", Integer, primary_key=True), - Column("email", String(50)), - Column("name", String(50)), - ) - Index( - "email_idx", - func.coalesce(t1.c.email, t1.c.name).desc(), - unique=True, - ) - - t2 = Table( - "foo", - m2, - Column("id", Integer, primary_key=True), - Column("email", String(50)), - Column("name", String(50)), - ) - Index( - "email_idx", - func.coalesce(t2.c.email, t2.c.name).desc(), - unique=True, - ) - - with assertions.expect_warnings( - "Skipped unsupported reflection", - "autogenerate skipping functional index", - ): - diffs = self._fixture(m1, m2) - eq_(diffs, []) - - -class MySQLUniqueIndexTest(AutogenerateUniqueIndexTest): - reports_unnamed_constraints = True - reports_unique_constraints_as_indexes = True - __only_on__ = "mysql", "mariadb" - __backend__ = True - - def test_removed_idx_index_named_as_column(self): - try: - super( - MySQLUniqueIndexTest, self - ).test_removed_idx_index_named_as_column() - except IndexError: - assert True - else: - assert False, "unexpected success" - - -class OracleUniqueIndexTest(AutogenerateUniqueIndexTest): - reports_unnamed_constraints = True - reports_unique_constraints_as_indexes = True - __only_on__ = "oracle" - __backend__ = True - class NoUqReflectionIndexTest(NoUqReflection, AutogenerateUniqueIndexTest): - reports_unique_constraints = False __only_on__ = "sqlite" def test_uq_casing_convention_changed_so_put_drops_first(self): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index dc0f5441..6e76fd61 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -16,6 +16,7 @@ from sqlalchemy import String from sqlalchemy import Table from sqlalchemy import text from sqlalchemy import types +from sqlalchemy import UniqueConstraint from sqlalchemy.dialects.postgresql import ARRAY from sqlalchemy.dialects.postgresql import BYTEA from sqlalchemy.dialects.postgresql import HSTORE @@ -38,6 +39,7 @@ from alembic.migration import MigrationContext from alembic.operations import ops from alembic.script import ScriptDirectory from alembic.testing import assert_raises_message +from alembic.testing import assertions from alembic.testing import combinations from alembic.testing import config from alembic.testing import eq_ @@ -52,6 +54,7 @@ from alembic.testing.fixtures import FutureEngineMixin from alembic.testing.fixtures import op_fixture from alembic.testing.fixtures import TablesTest from alembic.testing.fixtures import TestBase +from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest from alembic.util import sqla_compat @@ -1165,3 +1168,206 @@ class PostgresqlAutogenRenderTest(TestBase): autogenerate.render._repr_type(JSONB(), self.autogen_context), "postgresql.JSONB(astext_type=sa.Text())", ) + + +class PGUniqueIndexAutogenerateTest(AutogenFixtureTest, TestBase): + __only_on__ = "postgresql" + __backend__ = True + + def test_idx_added_schema(self): + m1 = MetaData() + m2 = MetaData() + Table("add_ix", m1, Column("x", String(50)), schema="test_schema") + Table( + "add_ix", + m2, + Column("x", String(50)), + Index("ix_1", "x"), + schema="test_schema", + ) + + diffs = self._fixture(m1, m2, include_schemas=True) + eq_(diffs[0][0], "add_index") + eq_(diffs[0][1].name, "ix_1") + + def test_idx_unchanged_schema(self): + m1 = MetaData() + m2 = MetaData() + Table( + "add_ix", + m1, + Column("x", String(50)), + Index("ix_1", "x"), + schema="test_schema", + ) + Table( + "add_ix", + m2, + Column("x", String(50)), + Index("ix_1", "x"), + schema="test_schema", + ) + + diffs = self._fixture(m1, m2, include_schemas=True) + eq_(diffs, []) + + def test_uq_added_schema(self): + m1 = MetaData() + m2 = MetaData() + Table("add_uq", m1, Column("x", String(50)), schema="test_schema") + Table( + "add_uq", + m2, + Column("x", String(50)), + UniqueConstraint("x", name="ix_1"), + schema="test_schema", + ) + + diffs = self._fixture(m1, m2, include_schemas=True) + eq_(diffs[0][0], "add_constraint") + eq_(diffs[0][1].name, "ix_1") + + def test_uq_unchanged_schema(self): + m1 = MetaData() + m2 = MetaData() + Table( + "add_uq", + m1, + Column("x", String(50)), + UniqueConstraint("x", name="ix_1"), + schema="test_schema", + ) + Table( + "add_uq", + m2, + Column("x", String(50)), + UniqueConstraint("x", name="ix_1"), + schema="test_schema", + ) + + diffs = self._fixture(m1, m2, include_schemas=True) + eq_(diffs, []) + + @config.requirements.btree_gist + def test_exclude_const_unchanged(self): + from sqlalchemy.dialects.postgresql import TSRANGE, ExcludeConstraint + + m1 = MetaData() + m2 = MetaData() + + Table( + "add_excl", + m1, + Column("id", Integer, primary_key=True), + Column("period", TSRANGE), + ExcludeConstraint(("period", "&&"), name="quarters_period_excl"), + ) + + Table( + "add_excl", + m2, + Column("id", Integer, primary_key=True), + Column("period", TSRANGE), + ExcludeConstraint(("period", "&&"), name="quarters_period_excl"), + ) + + diffs = self._fixture(m1, m2) + eq_(diffs, []) + + def test_same_tname_two_schemas(self): + m1 = MetaData() + m2 = MetaData() + + Table("add_ix", m1, Column("x", String(50)), Index("ix_1", "x")) + + Table("add_ix", m2, Column("x", String(50)), Index("ix_1", "x")) + Table("add_ix", m2, Column("x", String(50)), schema="test_schema") + + diffs = self._fixture(m1, m2, include_schemas=True) + eq_(diffs[0][0], "add_table") + eq_(len(diffs), 1) + + def test_uq_dropped(self): + m1 = MetaData() + m2 = MetaData() + Table( + "add_uq", + m1, + Column("id", Integer, primary_key=True), + Column("name", String), + UniqueConstraint("name", name="uq_name"), + ) + Table( + "add_uq", + m2, + Column("id", Integer, primary_key=True), + Column("name", String), + ) + diffs = self._fixture(m1, m2, include_schemas=True) + eq_(diffs[0][0], "remove_constraint") + eq_(diffs[0][1].name, "uq_name") + eq_(len(diffs), 1) + + def test_functional_ix_one(self): + m1 = MetaData() + m2 = MetaData() + + t1 = Table( + "foo", + m1, + Column("id", Integer, primary_key=True), + Column("email", String(50)), + ) + Index("email_idx", func.lower(t1.c.email), unique=True) + + t2 = Table( + "foo", + m2, + Column("id", Integer, primary_key=True), + Column("email", String(50)), + ) + Index("email_idx", func.lower(t2.c.email), unique=True) + + with assertions.expect_warnings( + "Skipped unsupported reflection", + "autogenerate skipping functional index", + ): + diffs = self._fixture(m1, m2) + eq_(diffs, []) + + def test_functional_ix_two(self): + m1 = MetaData() + m2 = MetaData() + + t1 = Table( + "foo", + m1, + Column("id", Integer, primary_key=True), + Column("email", String(50)), + Column("name", String(50)), + ) + Index( + "email_idx", + func.coalesce(t1.c.email, t1.c.name).desc(), + unique=True, + ) + + t2 = Table( + "foo", + m2, + Column("id", Integer, primary_key=True), + Column("email", String(50)), + Column("name", String(50)), + ) + Index( + "email_idx", + func.coalesce(t2.c.email, t2.c.name).desc(), + unique=True, + ) + + with assertions.expect_warnings( + "Skipped unsupported reflection", + "autogenerate skipping functional index", + ): + diffs = self._fixture(m1, m2) + eq_(diffs, []) -- 2.47.2