summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-03-16 10:55:46 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2022-03-16 10:58:54 -0400
commite533927189d7c6d9ce08d59ecb1f2bae0bd00783 (patch)
tree4df51753c293e1b6be785a2206559d6b9b0ad1b7 /tests
parenta9e6f9079a7d0157a842b3786ebd4f7031b08912 (diff)
downloadalembic-e533927189d7c6d9ce08d59ecb1f2bae0bd00783.tar.gz
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
Diffstat (limited to 'tests')
-rw-r--r--tests/requirements.py14
-rw-r--r--tests/test_autogen_indexes.py980
-rw-r--r--tests/test_postgresql.py206
3 files changed, 631 insertions, 569 deletions
diff --git a/tests/requirements.py b/tests/requirements.py
index 258c730..1060e25 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):
@@ -170,6 +178,10 @@ class DefaultRequirements(SuiteRequirements):
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):
if not exclusions.against(config, "postgresql"):
diff --git a/tests/test_autogen_indexes.py b/tests/test_autogen_indexes.py
index 8a4c26f..fb71099 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()
@@ -339,41 +325,6 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
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(
naming_convention={
@@ -433,32 +384,334 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
diffs = self._fixture(m1, m2)
eq_(diffs, [])
- def test_nothing_changed_ix_w_mixed_case_plain_name(self):
+ def test_nothing_changed_two(self):
m1 = MetaData()
m2 = MetaData()
Table(
"nothing_changed",
m1,
- Column("id", Integer, primary_key=True),
- Column("x", Integer),
- Index("SomeIndex", "x"),
+ Column("id1", Integer, primary_key=True),
+ Column("id2", Integer, primary_key=True),
+ Column("x", String(20), unique=True),
+ mysql_engine="InnoDB",
+ )
+ Table(
+ "nothing_changed_related",
+ m1,
+ Column("id1", Integer),
+ Column("id2", Integer),
+ ForeignKeyConstraint(
+ ["id1", "id2"], ["nothing_changed.id1", "nothing_changed.id2"]
+ ),
+ mysql_engine="InnoDB",
+ )
+
+ Table(
+ "nothing_changed",
+ m2,
+ Column("id1", Integer, primary_key=True),
+ Column("id2", Integer, primary_key=True),
+ Column("x", String(20), unique=True),
+ mysql_engine="InnoDB",
+ )
+ Table(
+ "nothing_changed_related",
+ m2,
+ Column("id1", Integer),
+ Column("id2", Integer),
+ ForeignKeyConstraint(
+ ["id1", "id2"], ["nothing_changed.id1", "nothing_changed.id2"]
+ ),
mysql_engine="InnoDB",
)
+ diffs = self._fixture(m1, m2)
+ eq_(diffs, [])
+
+ def test_nothing_changed_unique_w_colkeys(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "nothing_changed",
+ m1,
+ Column("x", String(20), key="nx"),
+ UniqueConstraint("nx"),
+ )
+
Table(
"nothing_changed",
m2,
+ Column("x", String(20), key="nx"),
+ UniqueConstraint("nx"),
+ )
+
+ diffs = self._fixture(m1, m2)
+ eq_(diffs, [])
+
+ @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(
+ "new_idx",
+ m1,
+ Column("id1", Integer, primary_key=True),
+ Column("x", String(20)),
+ uq1,
+ )
+
+ uq2 = UniqueConstraint("x", name="somecasingconvention")
+ Table(
+ "new_idx",
+ m2,
+ Column("id1", Integer, primary_key=True),
+ Column("x", String(20)),
+ uq2,
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ 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(
+ "some_table",
+ m1,
Column("id", Integer, primary_key=True),
+ Column("x", String(20)),
+ Column("y", String(20)),
+ UniqueConstraint("y", name="uq_y"),
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ 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(
+ "col_change",
+ m1,
Column("x", Integer),
- Index("SomeIndex", "x"),
- mysql_engine="InnoDB",
+ Column("y", Integer),
+ UniqueConstraint("x"),
+ )
+ Table(
+ "col_change",
+ m2,
+ Column("x", Integer),
+ Column("y", Integer),
+ UniqueConstraint("x", "y"),
+ )
+
+ 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)
+
+ 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(
+ "remove_idx",
+ m1,
+ 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__,
+ ),
+ [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(
+ "nothing_changed",
+ m1,
+ Column("x", String(20), index=True),
+ )
+
+ Table(
+ "nothing_changed",
+ m2,
+ Column("x", String(20), index=True),
+ )
+
diffs = self._fixture(m1, m2)
eq_(diffs, [])
@config.requirements.long_names
- def test_nothing_changed_ix_w_mixed_case_nconv_name(self):
+ def test_nothing_ix_changed_labels_were_truncated(self):
m1 = MetaData(
naming_convention={
"ix": "index_%(table_name)s_%(column_0_label)s",
@@ -473,89 +726,108 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
)
Table(
- "NothingChanged",
+ "nothing_changed",
m1,
- Column("id", Integer, primary_key=True),
- Column("XCol", Integer, index=True),
+ 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(
- "NothingChanged",
+ "nothing_changed",
m2,
- Column("id", Integer, primary_key=True),
- Column("XCol", Integer, index=True),
+ 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_nothing_changed_two(self):
+ def test_nothing_changed_ix_w_mixed_case_plain_name(self):
m1 = MetaData()
m2 = MetaData()
Table(
"nothing_changed",
m1,
- Column("id1", Integer, primary_key=True),
- Column("id2", Integer, primary_key=True),
- Column("x", String(20), unique=True),
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer),
+ Index("SomeIndex", "x"),
mysql_engine="InnoDB",
)
+
Table(
- "nothing_changed_related",
- m1,
- Column("id1", Integer),
- Column("id2", Integer),
- ForeignKeyConstraint(
- ["id1", "id2"], ["nothing_changed.id1", "nothing_changed.id2"]
- ),
+ "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(
- "nothing_changed",
- m2,
- Column("id1", Integer, primary_key=True),
- Column("id2", Integer, primary_key=True),
- Column("x", String(20), unique=True),
+ "NothingChanged",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("XCol", Integer, index=True),
mysql_engine="InnoDB",
)
+
Table(
- "nothing_changed_related",
+ "NothingChanged",
m2,
- Column("id1", Integer),
- Column("id2", Integer),
- ForeignKeyConstraint(
- ["id1", "id2"], ["nothing_changed.id1", "nothing_changed.id2"]
- ),
+ 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_unique_w_colkeys(self):
+ def test_new_table_added(self):
m1 = MetaData()
m2 = MetaData()
-
- Table(
- "nothing_changed",
- m1,
- Column("x", String(20), key="nx"),
- UniqueConstraint("nx"),
- )
-
Table(
- "nothing_changed",
+ "extra",
m2,
- Column("x", String(20), key="nx"),
- UniqueConstraint("nx"),
+ Column("foo", Integer, index=True),
+ Column("bar", Integer),
+ Index("newtable_idx", "bar"),
)
diffs = self._fixture(m1, m2)
- eq_(diffs, [])
+
+ 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()
@@ -729,47 +1001,6 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
],
)
- def test_uq_casing_convention_changed_so_put_drops_first(self):
- m1 = MetaData()
- m2 = MetaData()
-
- uq1 = UniqueConstraint("x", name="SomeCasingConvention")
- Table(
- "new_idx",
- m1,
- Column("id1", Integer, primary_key=True),
- Column("x", String(20)),
- uq1,
- )
-
- uq2 = UniqueConstraint("x", name="somecasingconvention")
- Table(
- "new_idx",
- m2,
- Column("id1", Integer, primary_key=True),
- Column("x", String(20)),
- uq2,
- )
-
- diffs = self._fixture(m1, m2)
-
- 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_idx_index_named_as_column(self):
m1 = MetaData()
m2 = MetaData()
@@ -795,6 +1026,7 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
diffs = self._fixture(m1, m2)
eq_(diffs, [("add_index", schemacompare.CompareIndex(idx))])
+ @exclusions.fails_on(["mysql", "mariadb"])
def test_removed_idx_index_named_as_column(self):
m1 = MetaData()
m2 = MetaData()
@@ -843,193 +1075,6 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
set([diffs[0][1].name, diffs[1][1].name]), set(["xy_idx", "y_idx"])
)
- def test_drop_table_w_uq_constraint(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "some_table",
- m1,
- Column("id", Integer, primary_key=True),
- Column("x", String(20)),
- Column("y", String(20)),
- UniqueConstraint("y", name="uq_y"),
- )
-
- diffs = self._fixture(m1, m2)
-
- 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):
- m1 = MetaData()
- m2 = MetaData()
- Table(
- "col_change",
- m1,
- Column("x", Integer),
- Column("y", Integer),
- UniqueConstraint("x"),
- )
- Table(
- "col_change",
- m2,
- Column("x", Integer),
- Column("y", Integer),
- UniqueConstraint("x", "y"),
- )
-
- 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)
-
- 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(
- "remove_idx",
- m1,
- 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__,
- ),
- [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
-
def test_add_ix_on_table_create(self):
m1 = MetaData()
m2 = 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):
+ @config.requirements.covering_indexes
+ @config.requirements.sqlalchemy_14
+ def test_nothing_changed_covering_index(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()
+ 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 dc0f544..6e76fd6 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, [])