diff options
author | Angus Lees <gus@inodes.org> | 2014-07-02 18:23:19 +1000 |
---|---|---|
committer | Angus Lees <gus@inodes.org> | 2014-07-22 11:10:08 +1000 |
commit | 62729fb352ff23007174b2b6ea6f355f9baef63a (patch) | |
tree | 9176817209f4007c345ef8c7b61332972ebbdbe8 | |
parent | bf4eff9ffd577c519846c98251942a3d998311e5 (diff) | |
download | oslo-db-62729fb352ff23007174b2b6ea6f355f9baef63a.tar.gz |
Allow tox tests with complex OS_TEST_DBAPI_CONNECTION URLs
In particular, this allows OS_TEST_DBAPI_ADMIN_CONNECTION URLs with
alternate drivers like mysql+mysqlconnector://... to survive through
into OS_TEST_DBAPI_CONNECTION for the actual unittest.
Change-Id: I712d9366e616a551f4b2bd225de42469a35b363f
-rw-r--r-- | oslo/db/sqlalchemy/provision.py | 44 |
1 files changed, 23 insertions, 21 deletions
diff --git a/oslo/db/sqlalchemy/provision.py b/oslo/db/sqlalchemy/provision.py index 315d599..f1aa2cd 100644 --- a/oslo/db/sqlalchemy/provision.py +++ b/oslo/db/sqlalchemy/provision.py @@ -16,6 +16,7 @@ """Provision test environment for specific DB backends""" import argparse +import copy import logging import os import random @@ -34,9 +35,9 @@ def get_engine(uri): """Engine creation Call the function without arguments to get admin connection. Admin - connection required to create temporary user and database for each - particular test. Otherwise use existing connection to recreate connection - to the temporary database. + connection required to create temporary database for each + particular test. Otherwise use existing connection to recreate + connection to the temporary database. """ return sqlalchemy.create_engine(uri, poolclass=sqlalchemy.pool.NullPool) @@ -57,31 +58,33 @@ def _execute_sql(engine, sql, driver): def create_database(engine): - """Provide temporary user and database for each particular test.""" + """Provide temporary database for each particular test.""" driver = engine.name - auth = { - 'database': ''.join(random.choice(string.ascii_lowercase) - for i in moves.range(10)), - 'user': engine.url.username, - 'passwd': engine.url.password, - } + database = ''.join(random.choice(string.ascii_lowercase) + for i in moves.range(10)) if driver == 'sqlite': - return 'sqlite:////tmp/%s' % auth['database'] + database = '/tmp/%s' % database elif driver in ['mysql', 'postgresql']: - sql = 'create database %s;' % auth['database'] + sql = 'create database %s;' % database _execute_sql(engine, [sql], driver) else: raise ValueError('Unsupported RDBMS %s' % driver) - params = auth.copy() - params['backend'] = driver - return "%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s" % params + # Both shallow and deep copies may lead to surprising behaviour + # without knowing the implementation of sqlalchemy.engine.url. + # Use a shallow copy here, since we're only overriding a single + # property, invoking __str__ and then discarding our copy. This + # is currently safe and _should_ remain safe into the future. + new_url = copy.copy(engine.url) + + new_url.database = database + return str(new_url) def drop_database(admin_engine, current_uri): - """Drop temporary database and user after each particular test.""" + """Drop temporary database after each particular test.""" engine = get_engine(current_uri) driver = engine.name @@ -101,8 +104,8 @@ def drop_database(admin_engine, current_uri): def main(): """Controller to handle commands - ::create: Create test user and database with random names. - ::drop: Drop user and database created by previous command. + ::create: Create test database with random names. + ::drop: Drop database created by previous command. """ parser = argparse.ArgumentParser( description='Controller to handle database creation and dropping' @@ -115,8 +118,7 @@ def main(): create = subparsers.add_parser( 'create', - help='Create temporary test ' - 'databases and users.') + help='Create temporary test databases.') create.set_defaults(which='create') create.add_argument( 'instances_count', @@ -125,7 +127,7 @@ def main(): drop = subparsers.add_parser( 'drop', - help='Drop temporary test databases and users.') + help='Drop temporary test databases.') drop.set_defaults(which='drop') drop.add_argument( 'instances', |