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
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
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()
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()
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()
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(
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()
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()
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):
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
