check all directives in batch block until recreate selected

Fixed regression in batch mode due to :ticket:`883` where the "auto" mode
of batch would fail to accommodate any additional migration directives
beyond encountering an ``add_column()`` directive, due to a mis-application
of the conditional logic that was added as part of this change, leading to
"recreate" mode not being used in cases where it is required for SQLite
such as for unique constraints.

Change-Id: I6315569caff5f3b33d152974ebecc8b18d9cc523
Fixes: #896

diff --git a/alembic/ddl/sqlite.py b/alembic/ddl/sqlite.py
index 2f4ed77362c3cb152b71b18b53de3a2731d4f677..863accc6b50bef2e9c5e4a639111793fa49b0bf8 100644 (file)
--- a/alembic/ddl/sqlite.py
+++ b/alembic/ddl/sqlite.py
@@ -56,7 +56,6 @@ class SQLiteImpl(DefaultImpl):
                     and col.server_default.persisted
                 ):
                     return True
-                return False
             elif op[0] not in ("create_index", "drop_index"):
                 return True
         else:

diff --git a/docs/build/unreleased/896.rst b/docs/build/unreleased/896.rst
new file mode 100644 (file)
index 0000000..c6efce4
--- /dev/null
+++ b/docs/build/unreleased/896.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, regression, batch
+    :tickets: 896
+
+    Fixed regression in batch mode due to :ticket:`883` where the "auto" mode
+    of batch would fail to accommodate any additional migration directives
+    beyond encountering an ``add_column()`` directive, due to a mis-application
+    of the conditional logic that was added as part of this change, leading to
+    "recreate" mode not being used in cases where it is required for SQLite
+    such as for unique constraints.

diff --git a/tests/test_batch.py b/tests/test_batch.py
index ab5328339a818fda0f8c936d3d5b821d8c052f13..2753bdc3f9ff114726337211d4e1eb4533f670df 100644 (file)
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -1661,9 +1661,10 @@ class BatchRoundTripTest(TestBase):
         pk_const = inspect(self.conn).get_pk_constraint("foo")
         eq_(pk_const["constrained_columns"], ["id"])
 
-    def test_add_pk_constraint(self):
+    @testing.combinations(("always",), ("auto",), argnames="recreate")
+    def test_add_pk_constraint(self, recreate):
         self._no_pk_fixture()
-        with self.op.batch_alter_table("nopk", recreate="always") as batch_op:
+        with self.op.batch_alter_table("nopk", recreate=recreate) as batch_op:
             batch_op.create_primary_key("newpk", ["a", "b"])
 
         pk_const = inspect(self.conn).get_pk_constraint("nopk")
@@ -1671,9 +1672,10 @@ class BatchRoundTripTest(TestBase):
             eq_(pk_const["name"], "newpk")
         eq_(pk_const["constrained_columns"], ["a", "b"])
 
+    @testing.combinations(("always",), ("auto",), argnames="recreate")
     @config.requirements.check_constraint_reflection
-    def test_add_ck_constraint(self):
-        with self.op.batch_alter_table("foo", recreate="always") as batch_op:
+    def test_add_ck_constraint(self, recreate):
+        with self.op.batch_alter_table("foo", recreate=recreate) as batch_op:
             batch_op.create_check_constraint("newck", text("x > 0"))
 
         ck_consts = inspect(self.conn).get_check_constraints("foo")
@@ -1682,12 +1684,13 @@ class BatchRoundTripTest(TestBase):
         )
         eq_(ck_consts, [{"sqltext": "x > 0", "name": "newck"}])
 
+    @testing.combinations(("always",), ("auto",), argnames="recreate")
     @config.requirements.check_constraint_reflection
-    def test_drop_ck_constraint(self):
+    def test_drop_ck_constraint(self, recreate):
         self._ck_constraint_fixture()
 
         with self.op.batch_alter_table(
-            "ck_table", recreate="always"
+            "ck_table", recreate=recreate
         ) as batch_op:
             batch_op.drop_constraint("ck", "check")
 
@@ -1736,6 +1739,36 @@ class BatchRoundTripTest(TestBase):
         tcomment = insp.get_table_comment(tname)
         eq_(tcomment, {"text": comment})
 
+    @testing.combinations(("always",), ("auto",), argnames="recreate")
+    def test_add_uq(self, recreate):
+        with self.op.batch_alter_table("foo", recreate=recreate) as batch_op:
+            batch_op.create_unique_constraint("newuk", ["x"])
+
+        uq_consts = inspect(self.conn).get_unique_constraints("foo")
+        eq_(
+            [
+                {"name": uc["name"], "column_names": uc["column_names"]}
+                for uc in uq_consts
+            ],
+            [{"name": "newuk", "column_names": ["x"]}],
+        )
+
+    @testing.combinations(("always",), ("auto",), argnames="recreate")
+    def test_add_uq_plus_col(self, recreate):
+        with self.op.batch_alter_table("foo", recreate=recreate) as batch_op:
+            batch_op.add_column(Column("y", Integer))
+            batch_op.create_unique_constraint("newuk", ["x", "y"])
+
+        uq_consts = inspect(self.conn).get_unique_constraints("foo")
+
+        eq_(
+            [
+                {"name": uc["name"], "column_names": uc["column_names"]}
+                for uc in uq_consts
+            ],
+            [{"name": "newuk", "column_names": ["x", "y"]}],
+        )
+
     @config.requirements.comments
     def test_add_table_comment(self):
         with self.op.batch_alter_table("foo") as batch_op: