Add function drop_unique_constraint()

We forgot to copy this function from nova, so we are not able to
switch to oslo sqlalchemy.utils until add this function.

Sqlalchemy doesn't supports some sqlite column types so we have no
possibility to drop unique constraint in general way on sqlite (we
loose these column types after migration).
Added unified drop_unique_constraint() function to drop unique
constraint for sql backends with tests.

blueprint: oslo-sqlalchemy-utils

Change-Id: I34000ad2277a97c31a29539d047faaf19876c9d5

2 files changed, 171 insertions, 1 deletions

diff --git a/openstack/common/db/sqlalchemy/utils.py b/openstack/common/db/sqlalchemy/utils.py
index 3ff7bdb..caf5569 100755
--- a/openstack/common/db/sqlalchemy/utils.py
+++ b/openstack/common/db/sqlalchemy/utils.py
@@ -18,6 +18,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from migrate.changeset import UniqueConstraint
 import sqlalchemy
 from sqlalchemy import Boolean
 from sqlalchemy import CheckConstraint
@@ -191,6 +192,43 @@ def _get_not_supported_column(col_name_col_instance, column_name):
     return column
 
 
+def drop_unique_constraint(migrate_engine, table_name, uc_name, *columns,
+                           **col_name_col_instance):
+    """Drop unique constraint from table.
+
+    This method drops UC from table and works for mysql, postgresql and sqlite.
+    In mysql and postgresql we are able to use "alter table" construction.
+    Sqlalchemy doesn't support some sqlite column types and replaces their
+    type with NullType in metadata. We process these columns and replace
+    NullType with the correct column type.
+
+    :param migrate_engine: sqlalchemy engine
+    :param table_name:     name of table that contains uniq constraint.
+    :param uc_name:        name of uniq constraint that will be dropped.
+    :param columns:        columns that are in uniq constraint.
+    :param col_name_col_instance:   contains pair column_name=column_instance.
+                            column_instance is instance of Column. These params
+                            are required only for columns that have unsupported
+                            types by sqlite. For example BigInteger.
+    """
+
+    meta = MetaData()
+    meta.bind = migrate_engine
+    t = Table(table_name, meta, autoload=True)
+
+    if migrate_engine.name == "sqlite":
+        override_cols = [
+            _get_not_supported_column(col_name_col_instance, col.name)
+            for col in t.columns
+            if isinstance(col.type, NullType)
+        ]
+        for col in override_cols:
+            t.columns.replace(col)
+
+    uc = UniqueConstraint(*columns, table=t, name=uc_name)
+    uc.drop()
+
+
 def drop_old_duplicate_entries_from_table(migrate_engine, table_name,
                                           use_soft_delete, *uc_column_names):
     """Drop all old rows having the same values for columns in uc_columns.
diff --git a/tests/unit/db/sqlalchemy/test_utils.py b/tests/unit/db/sqlalchemy/test_utils.py
index 15a1e25..78e99da 100644
--- a/tests/unit/db/sqlalchemy/test_utils.py
+++ b/tests/unit/db/sqlalchemy/test_utils.py
@@ -15,11 +15,15 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import warnings
+
+from migrate.changeset import UniqueConstraint
 import sqlalchemy
 from sqlalchemy.dialects import mysql
 from sqlalchemy import Boolean, Index, Integer, DateTime, String
-from sqlalchemy import MetaData, Table, Column
+from sqlalchemy import MetaData, Table, Column, ForeignKey
 from sqlalchemy.engine import reflection
+from sqlalchemy.exc import SAWarning
 from sqlalchemy.sql import select
 from sqlalchemy.types import UserDefinedType, NullType
 
@@ -371,3 +375,131 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
         #                 but sqlalchemy will set it to NullType.
         self.assertTrue(isinstance(table.c.foo.type, NullType))
         self.assertTrue(isinstance(table.c.deleted.type, Boolean))
+
+    def test_utils_drop_unique_constraint(self):
+        table_name = "__test_tmp_table__"
+        uc_name = 'uniq_foo'
+        values = [
+            {'id': 1, 'a': 3, 'foo': 10},
+            {'id': 2, 'a': 2, 'foo': 20},
+            {'id': 3, 'a': 1, 'foo': 30},
+        ]
+        for key, engine in self.engines.items():
+            meta = MetaData()
+            meta.bind = engine
+            test_table = Table(
+                table_name, meta,
+                Column('id', Integer, primary_key=True, nullable=False),
+                Column('a', Integer),
+                Column('foo', Integer),
+                UniqueConstraint('a', name='uniq_a'),
+                UniqueConstraint('foo', name=uc_name),
+            )
+            test_table.create()
+
+            engine.execute(test_table.insert(), values)
+            # NOTE(boris-42): This method is generic UC dropper.
+            utils.drop_unique_constraint(engine, table_name, uc_name, 'foo')
+
+            s = test_table.select().order_by(test_table.c.id)
+            rows = engine.execute(s).fetchall()
+
+            for i in xrange(0, len(values)):
+                v = values[i]
+                self.assertEqual((v['id'], v['a'], v['foo']), rows[i])
+
+            # NOTE(boris-42): Update data about Table from DB.
+            meta = MetaData()
+            meta.bind = engine
+            test_table = Table(table_name, meta, autoload=True)
+            constraints = filter(
+                lambda c: c.name == uc_name, test_table.constraints)
+            self.assertEqual(len(constraints), 0)
+            self.assertEqual(len(test_table.constraints), 1)
+
+            test_table.drop()
+
+    def test_util_drop_unique_constraint_with_not_supported_sqlite_type(self):
+        table_name = "__test_tmp_table__"
+        uc_name = 'uniq_foo'
+        values = [
+            {'id': 1, 'a': 3, 'foo': 10},
+            {'id': 2, 'a': 2, 'foo': 20},
+            {'id': 3, 'a': 1, 'foo': 30}
+        ]
+
+        engine = self.engines['sqlite']
+        meta = MetaData(bind=engine)
+
+        test_table = Table(
+            table_name, meta,
+            Column('id', Integer, primary_key=True, nullable=False),
+            Column('a', Integer),
+            Column('foo', CustomType, default=0),
+            UniqueConstraint('a', name='uniq_a'),
+            UniqueConstraint('foo', name=uc_name),
+        )
+        test_table.create()
+
+        engine.execute(test_table.insert(), values)
+        warnings.simplefilter("ignore", SAWarning)
+        # NOTE(boris-42): Missing info about column `foo` that has
+        #                 unsupported type CustomType.
+        self.assertRaises(exception.OpenstackException,
+                          utils.drop_unique_constraint,
+                          engine, table_name, uc_name, 'foo')
+
+        # NOTE(boris-42): Wrong type of foo instance. it should be
+        #                 instance of sqlalchemy.Column.
+        self.assertRaises(exception.OpenstackException,
+                          utils.drop_unique_constraint,
+                          engine, table_name, uc_name, 'foo', foo=Integer())
+
+        foo = Column('foo', CustomType, default=0)
+        utils.drop_unique_constraint(
+            engine, table_name, uc_name, 'foo', foo=foo)
+
+        s = test_table.select().order_by(test_table.c.id)
+        rows = engine.execute(s).fetchall()
+
+        for i in xrange(0, len(values)):
+            v = values[i]
+            self.assertEqual((v['id'], v['a'], v['foo']), rows[i])
+
+        # NOTE(boris-42): Update data about Table from DB.
+        meta = MetaData(bind=engine)
+        test_table = Table(table_name, meta, autoload=True)
+        constraints = filter(
+            lambda c: c.name == uc_name, test_table.constraints)
+        self.assertEqual(len(constraints), 0)
+        self.assertEqual(len(test_table.constraints), 1)
+        test_table.drop()
+
+    def test_drop_unique_constraint_in_sqlite_fk_recreate(self):
+        engine = self.engines['sqlite']
+        meta = MetaData()
+        meta.bind = engine
+        parent_table = Table(
+            'table0', meta,
+            Column('id', Integer, primary_key=True),
+            Column('foo', Integer),
+        )
+        parent_table.create()
+        table_name = 'table1'
+        table = Table(
+            table_name, meta,
+            Column('id', Integer, primary_key=True),
+            Column('baz', Integer),
+            Column('bar', Integer, ForeignKey("table0.id")),
+            UniqueConstraint('baz', name='constr1')
+        )
+        table.create()
+        utils.drop_unique_constraint(engine, table_name, 'constr1', 'baz')
+
+        insp = reflection.Inspector.from_engine(engine)
+        f_keys = insp.get_foreign_keys(table_name)
+        self.assertEqual(len(f_keys), 1)
+        f_key = f_keys[0]
+        self.assertEqual(f_key['referred_table'], 'table0')
+        self.assertEqual(f_key['referred_columns'], ['id'])
+        self.assertEqual(f_key['constrained_columns'], ['bar'])