x/python-ganttclient
Code Issues Proposed changes

Merge "Fix tests for sqlalchemy utils"

Browse Source
This commit is contained in:
Jenkins
2013-05-24 19:38:59 +00:00
committed by Gerrit Code Review
parent 64271d9dd4 2d90f1ed0b
commit f2a8c8eac8
1 changed files with 67 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
nova/tests/test_migration_utils.py
View File

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