Repair boolean CHECK constraint detection

The test here always passed because SQLAlchemy has never
reflected CHECK constraints before, so the test never exercised
anything.   An upcoming feature in SQLAlchemy 1.1 will allow
the Postgresql and SQLite dialects to reflect CHECK constraints,
and the existing test for the "boolean" constraint fails.
This patch repairs the test with a more generic regular expression
match and passes against all SQLAlchemy versions.

Change-Id: Idc17c39325ccfee3bbbe171ed2833a4f0991d261

oslo_db/sqlalchemy/utils.py

@@ -589,6 +589,19 @@ def change_deleted_column_type_to_id_type(migrate_engine, table_name,
     _restore_indexes_on_deleted_columns(migrate_engine, table_name, indexes)
 
 
+def _is_deleted_column_constraint(constraint):
+    # NOTE(boris-42): There is no other way to check is CheckConstraint
+    #                 associated with deleted column.
+    if not isinstance(constraint, CheckConstraint):
+        return False
+    sqltext = str(constraint.sqltext)
+    # NOTE(zzzeek): SQLite never reflected CHECK contraints here
+    # in any case until version 1.1.   Safe to assume that any CHECK
+    # that's talking about the value of "deleted in (something)" is
+    # the boolean constraint we're looking to get rid of.
+    return bool(re.match(r".*deleted in \(.*\)", sqltext, re.I))
+
+
 def _change_deleted_column_type_to_id_type_sqlite(migrate_engine, table_name,
                                                   **col_name_col_instance):
     # NOTE(boris-42): sqlaclhemy-migrate can't drop column with check
@@ -619,25 +632,9 @@ def _change_deleted_column_type_to_id_type_sqlite(migrate_engine, table_name,
                                  default=default_deleted_value)
         columns.append(column_copy)
 
-    def is_deleted_column_constraint(constraint):
-        # NOTE(boris-42): There is no other way to check is CheckConstraint
-        #                 associated with deleted column.
-        if not isinstance(constraint, CheckConstraint):
-            return False
-        sqltext = str(constraint.sqltext)
-        # 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:
-        if not is_deleted_column_constraint(constraint):
+        if not _is_deleted_column_constraint(constraint):
             constraints.append(constraint.copy())
 
     new_table = Table(table_name + "__tmp__", meta,

oslo_db/tests/sqlalchemy/test_utils.py

@@ -24,6 +24,7 @@ from six.moves.urllib import parse
 import sqlalchemy
 from sqlalchemy.dialects import mysql
 from sqlalchemy import Boolean, Index, Integer, DateTime, String, SmallInteger
+from sqlalchemy import CheckConstraint
 from sqlalchemy import MetaData, Table, Column, ForeignKey
 from sqlalchemy.engine import reflection
 from sqlalchemy.engine import url as sa_url
@@ -546,6 +547,23 @@ class TestMigrationUtils(db_test_base.DbTestCase):
             self.assertTrue(isinstance(table.c.foo.type, NullType))
         self.assertTrue(isinstance(table.c.deleted.type, Boolean))
 
+    def test_detect_boolean_deleted_constraint_detection(self):
+        table_name = 'abc'
+        table = Table(table_name, self.meta,
+                      Column('id', Integer, primary_key=True),
+                      Column('deleted', Boolean))
+        ck = [
+            const for const in table.constraints if
+            isinstance(const, CheckConstraint)][0]
+
+        self.assertTrue(utils._is_deleted_column_constraint(ck))
+
+        self.assertFalse(
+            utils._is_deleted_column_constraint(
+                CheckConstraint("deleted > 5")
+            )
+        )
+
     @db_test_base.backend_specific('sqlite')
     def test_change_deleted_column_type_sqlite_drops_check_constraint(self):
         table_name = 'abc'