Merge "Fix sqlalchemy utils."

2 files changed, 39 insertions, 4 deletions

diff --git a/nova/db/sqlalchemy/utils.py b/nova/db/sqlalchemy/utils.py
index 6c0a53846..918f50bfd 100644
--- a/nova/db/sqlalchemy/utils.py
+++ b/nova/db/sqlalchemy/utils.py
@@ -17,7 +17,7 @@
 
 import re
 
-from migrate.changeset import UniqueConstraint
+from migrate.changeset import UniqueConstraint, ForeignKeyConstraint
 from sqlalchemy import Boolean
 from sqlalchemy import CheckConstraint
 from sqlalchemy import Column
@@ -101,7 +101,8 @@ def _get_unique_constraints_in_sqlite(migrate_engine, table_name):
 
     uniques = set([
         schema.UniqueConstraint(
-            *[getattr(table.c, c.strip()) for c in cols.split(",")], name=name
+            *[getattr(table.c, c.strip(' "'))
+              for c in cols.split(",")], name=name
         )
         for name, cols in re.findall(regexp, sql_data)
     ])
@@ -128,7 +129,8 @@ def _drop_unique_constraint_in_sqlite(migrate_engine, table_name, uc_name,
     table.constraints.update(uniques)
 
     constraints = [constraint for constraint in table.constraints
-                    if not constraint.name == uc_name]
+                    if not constraint.name == uc_name and
+                    not isinstance(constraint, schema.ForeignKeyConstraint)]
 
     new_table = Table(table_name + "__tmp__", meta, *(columns + constraints))
     new_table.create()
@@ -139,12 +141,20 @@ def _drop_unique_constraint_in_sqlite(migrate_engine, table_name, uc_name,
         indexes.append(Index(index["name"],
                              *column_names,
                              unique=index["unique"]))
+    f_keys = []
+    for fk in insp.get_foreign_keys(table_name):
+        refcolumns = [fk['referred_table'] + '.' + col
+                      for col in fk['referred_columns']]
+        f_keys.append(ForeignKeyConstraint(fk['constrained_columns'],
+                      refcolumns, table=new_table, name=fk['name']))
 
     ins = InsertFromSelect(new_table, table.select())
     migrate_engine.execute(ins)
     table.drop()
 
     [index.create(migrate_engine) for index in indexes]
+    for fkey in f_keys:
+        fkey.create()
     new_table.rename(table_name)
 
 
diff --git a/nova/tests/db/test_migration_utils.py b/nova/tests/db/test_migration_utils.py
index 3674a7c45..af206632d 100644
--- a/nova/tests/db/test_migration_utils.py
+++ b/nova/tests/db/test_migration_utils.py
@@ -19,7 +19,7 @@ import warnings
 from migrate.changeset import UniqueConstraint
 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 NoSuchTableError
 from sqlalchemy.exc import SAWarning
@@ -514,3 +514,28 @@ 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_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'])