Fix tests for sqlalchemy utils

This fix tests to be able to run it with postgresql and mysql.

Some tests should be run only on sqlite

Method create_shadow_tables raises different exception in different
backends (when we try to create duplicate table):
In postgresql it is ProgrammingError
In sqlite and mysql it is OperationalError
So create new Exception that will be raised in both cases.

bp db-common-migration-and-utils

Change-Id: I11be23106bc01bd3c72dad30edfb31a17177bb34

1 files changed, 74 insertions, 80 deletions

diff --git a/nova/tests/test_migration_utils.py b/nova/tests/test_migration_utils.py
index d6e1ee568..5155dba8f 100644
--- a/nova/tests/test_migration_utils.py
+++ b/nova/tests/test_migration_utils.py
@@ -19,7 +19,6 @@ from migrate.changeset import UniqueConstraint
 from sqlalchemy import Integer, DateTime, String
 from sqlalchemy import MetaData, Table, Column
 from sqlalchemy.exc import NoSuchTableError
-from sqlalchemy.exc import OperationalError
 from sqlalchemy.exc import SAWarning
 from sqlalchemy.sql import select
 from sqlalchemy.types import UserDefinedType
@@ -92,54 +91,51 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
             {'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', CustomType, default=0),
-                               UniqueConstraint('a', name='uniq_a'),
-                               UniqueConstraint('foo', name=uc_name))
-            test_table.create()
-
-            engine.execute(test_table.insert(), values)
-            if key == "sqlite":
-                warnings.simplefilter("ignore", SAWarning)
-                # NOTE(boris-42): Missing info about column `foo` that has
-                #                 unsupported type CustomType.
-                self.assertRaises(exception.NovaException,
-                                  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.NovaException,
-                                  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()
+        engine = self.engines['sqlite']
+        meta = MetaData(bind=engine)
 
-            for i in xrange(0, len(values)):
-                v = values[i]
-                self.assertEqual((v['id'], v['a'], v['foo']), rows[i])
+        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()
 
-            # 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()
+        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.NovaException,
+                          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.NovaException,
+                          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 _populate_db_for_drop_duplicate_entries(self, engine, meta,
                                                 table_name):
@@ -160,7 +156,7 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
                                   nullable=False),
                            Column('a', Integer),
                            Column('b', Integer),
-                           Column('c', String),
+                           Column('c', String(255)),
                            Column('deleted', Integer, default=0),
                            Column('deleted_at', DateTime),
                            Column('updated_at', DateTime))
@@ -296,22 +292,21 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
 
     def test_check_shadow_table_with_unsupported_type(self):
         table_name = 'abc'
-        for key, engine in self.engines.items():
-            meta = MetaData()
-            meta.bind = engine
-
-            table = Table(table_name, meta,
-                          Column('id', Integer, primary_key=True),
-                          Column('a', Integer),
-                          Column('c', CustomType))
-            table.create()
-
-            shadow_table = Table(db._SHADOW_TABLE_PREFIX + table_name, meta,
-                                 Column('id', Integer, primary_key=True),
-                                 Column('a', Integer),
-                                 Column('c', CustomType))
-            shadow_table.create()
-            self.assertTrue(utils.check_shadow_table(engine, table_name))
+        engine = self.engines['sqlite']
+        meta = MetaData(bind=engine)
+
+        table = Table(table_name, meta,
+                      Column('id', Integer, primary_key=True),
+                      Column('a', Integer),
+                      Column('c', CustomType))
+        table.create()
+
+        shadow_table = Table(db._SHADOW_TABLE_PREFIX + table_name, meta,
+                             Column('id', Integer, primary_key=True),
+                             Column('a', Integer),
+                             Column('c', CustomType))
+        shadow_table.create()
+        self.assertTrue(utils.check_shadow_table(engine, table_name))
 
     def test_create_shadow_table_by_table_instance(self):
         table_name = 'abc'
@@ -342,21 +337,20 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
 
     def test_create_shadow_table_not_supported_type(self):
         table_name = 'abc'
-        for key, engine in self.engines.items():
-            meta = MetaData()
-            meta.bind = engine
-
-            table = Table(table_name, meta,
-                          Column('id', Integer, primary_key=True),
-                          Column('a', CustomType))
-            table.create()
-            self.assertRaises(exception.NovaException,
-                              utils.create_shadow_table,
-                              engine, table_name=table_name)
-
-            utils.create_shadow_table(engine, table_name=table_name,
-                                      a=Column('a', CustomType()))
-            self.assertTrue(utils.check_shadow_table(engine, table_name))
+        engine = self.engines['sqlite']
+        meta = MetaData()
+        meta.bind = engine
+        table = Table(table_name, meta,
+                      Column('id', Integer, primary_key=True),
+                      Column('a', CustomType))
+        table.create()
+        self.assertRaises(exception.NovaException,
+                          utils.create_shadow_table,
+                          engine, table_name=table_name)
+
+        utils.create_shadow_table(engine, table_name=table_name,
+                                  a=Column('a', CustomType()))
+        self.assertTrue(utils.check_shadow_table(engine, table_name))
 
     def test_create_shadow_both_table_and_table_name_are_none(self):
         for key, engine in self.engines.items():
@@ -388,6 +382,6 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
                           Column('a', Integer))
             table.create()
             utils.create_shadow_table(engine, table_name=table_name)
-            self.assertRaises(OperationalError,
+            self.assertRaises(exception.ShadowTableExists,
                               utils.create_shadow_table,
                               engine, table_name=table_name)