openstack/deb-glance
Code Issues Proposed changes

add postgresql support to test_migrations

Browse Source 
fixes bug 908180

Change-Id: Ic8afdfb3bce4c40c8cb47664dae053ec5e40d03c
This commit is contained in:
J. Daniel Schmidt 2012-04-11 16:11:04 +02:00
parent f8766a8fb5
commit 936b6a6e24
2 changed files with 26 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
glance/tests/unit/test_migrations.conf
View File

@ -3,3 +3,4 @@
# The "name" used in the test is the config variable key.
sqlite=sqlite:///test_migrations.db
#mysql=mysql://root:@localhost/test_migrations
#postgresql=postgresql://root:@localhost/test_migrations

31
glance/tests/unit/test_migrations.py
View File

@ -89,10 +89,19 @@ class TestMigrations(unittest.TestCase):
self._reset_databases()
def _reset_databases(self):
def _is_sqlite(conn_string):
return conn_string.startswith('sqlite')
def _is_mysql(conn_string):
return conn_string.startswith('mysql')
def _is_postgresql(conn_string):
return conn_string.startswith('postgresql')
for key, engine in self.engines.items():
conn_string = TestMigrations.TEST_DATABASES[key]
conn_pieces = urlparse.urlparse(conn_string)
if conn_string.startswith('sqlite'):
if _is_sqlite(conn_string):
# We can just delete the SQLite database, which is
# the easiest and cleanest solution
db_path = conn_pieces.path.strip('/')
@ -100,7 +109,7 @@ class TestMigrations(unittest.TestCase):
os.unlink(db_path)
# No need to recreate the SQLite DB. SQLite will
# create it for us if it's not there...
elif conn_string.startswith('mysql'):
elif _is_mysql(conn_string) or _is_postgresql(conn_string):
# We can execute the MySQL client to destroy and re-create
# the MYSQL database, which is easier and less error-prone
# than using SQLAlchemy to do this via MetaData...trust me.
@ -113,10 +122,20 @@ class TestMigrations(unittest.TestCase):
if len(auth_pieces) > 1:
if auth_pieces[1].strip():
password = "-p%s" % auth_pieces[1]
sql = ("drop database if exists %(database)s; "
"create database %(database)s;") % locals()
cmd = ("mysql -u%(user)s %(password)s -h%(host)s "
"-e\"%(sql)s\"") % locals()
if _is_mysql(conn_string):
sql = ("drop database if exists %(database)s; "
"create database %(database)s;") % locals()
cmd = ("mysql -u%(user)s %(password)s -h%(host)s "
"-e\"%(sql)s\"") % locals()
if _is_postgresql(conn_string):
cmd = ("dropdb %(database)s ; "
"createdb %(database)s -O %(user)s ; "
"echo \"select 'drop table if exists ' "
"|| tablename || ' cascade;' "
"from pg_tables where schemaname = 'public';\" | "
"psql -d %(database)s | grep '^\s*drop' | "
"psql -d %(database)s"
) % locals()
exitcode, out, err = utils.execute(cmd)
self.assertEqual(0, exitcode)