Merge "Fix changing the type of column deleted"

2 files changed, 26 insertions, 5 deletions

diff --git a/oslo/db/sqlalchemy/utils.py b/oslo/db/sqlalchemy/utils.py
index 04eb7ce..43a06c6 100644
--- a/oslo/db/sqlalchemy/utils.py
+++ b/oslo/db/sqlalchemy/utils.py
@@ -612,8 +612,15 @@ def _change_deleted_column_type_to_id_type_sqlite(migrate_engine, table_name,
         if not isinstance(constraint, CheckConstraint):
             return False
         sqltext = str(constraint.sqltext)
-        return (sqltext.endswith("deleted in (0, 1)") or
-                sqltext.endswith("deleted IN (:deleted_1, :deleted_2)"))
+        # NOTE(I159): in order to omit the CHECK constraint corresponding
+        # to `deleted` column we have to test these patterns which may
+        # vary depending on the SQLAlchemy version used.
+        constraint_markers = (
+            "deleted in (0, 1)",
+            "deleted IN (:deleted_1, :deleted_2)",
+            "deleted IN (:param_1, :param_2)"
+        )
+        return any(sqltext.endswith(marker) for marker in constraint_markers)
 
     constraints = []
     for constraint in table.constraints:
diff --git a/tests/sqlalchemy/test_utils.py b/tests/sqlalchemy/test_utils.py
index 36a2b14..ea464e2 100644
--- a/tests/sqlalchemy/test_utils.py
+++ b/tests/sqlalchemy/test_utils.py
@@ -183,9 +183,6 @@ class TestPaginateQuery(test_base.BaseTestCase):
 class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
     """Class for testing utils that are used in db migrations."""
 
-    def setUp(self):
-        super(TestMigrationUtils, self).setUp()
-
     def _populate_db_for_drop_duplicate_entries(self, engine, meta,
                                                 table_name):
         values = [
@@ -441,6 +438,23 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
             self.assertTrue(isinstance(table.c.foo.type, NullType))
         self.assertTrue(isinstance(table.c.deleted.type, Boolean))
 
+    def test_change_deleted_column_type_drops_check_constraint(self):
+        table_name = 'abc'
+        meta = MetaData()
+        engine = self.engines['sqlite']
+        meta.bind = engine
+        table = Table(table_name, meta,
+                      Column('id', Integer, primary_key=True),
+                      Column('deleted', Boolean))
+        table.create()
+
+        utils._change_deleted_column_type_to_id_type_sqlite(engine,
+                                                            table_name)
+        table = Table(table_name, meta, autoload=True)
+        # NOTE(I159): if the CHECK constraint has been dropped (expected
+        # behavior), any integer value can be inserted, otherwise only 1 or 0.
+        engine.execute(table.insert({'deleted': 10}))
+
     def test_utils_drop_unique_constraint(self):
         table_name = "__test_tmp_table__"
         uc_name = 'uniq_foo'