The ability to run tests at various backend
Added testing functionality for using different db backends. Сurrently exists ability to define backend-specific tests along with regular tests. Assuming that the tests will be run under a variety of backend there exists opportunity to skip backend-specific tests at inappropriate backends. For the correct functioning of the tests required to export environment variable with a root access database uri. Blueprint: tests-given-db-backend Change-Id: Ic5d3cfe8764b3c04affe79088473255c17535e54
This commit is contained in:
Pekelny Ilya
Ilya Pekelny
parent
232942278f
commit
8a01dd8e68
|
|
@ -6,3 +6,12 @@ test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
|
|||
|
||||
test_id_option=--load-list $IDFILE
|
||||
test_list_option=--list
|
||||
|
||||
# NOTE(ipekelny): create an isolated DB instance for each test running process
|
||||
# to prevent race conditions. Please see
|
||||
# https://testrepository.readthedocs.org/en/latest/MANUAL.html#remote-or-isolated-test-environments
|
||||
# for details.
|
||||
|
||||
instance_provision=${PYTHON:-python} ./openstack/common/db/sqlalchemy/provision.py create $INSTANCE_COUNT
|
||||
instance_dispose=${PYTHON:-python} ./openstack/common/db/sqlalchemy/provision.py drop $INSTANCE_IDS
|
||||
instance_execute=OS_TEST_DBAPI_CONNECTION=$INSTANCE_ID $COMMAND
|
||||
|
|
|
|||
|
|
@ -29,3 +29,9 @@ To run tests in the current environment:
|
|||
|
||||
sudo pip install -r requirements.txt
|
||||
nosetests
|
||||
|
||||
To run tests using MySQL or PostgreSQL as a DB backend do:
|
||||
|
||||
OS_TEST_DBAPI_ADMIN_CONNECTION=mysql://user:password@host/database tox -e py27
|
||||
|
||||
Note, that your DB user must have permissions to create and drop databases.
|
||||
|
|
|
|||
|
|
@ -47,3 +47,8 @@ class DbMigrationError(DBError):
|
|||
"""Wraps migration specific exception."""
|
||||
def __init__(self, message=None):
|
||||
super(DbMigrationError, self).__init__(str(message))
|
||||
|
||||
|
||||
class DBConnectionError(DBError):
|
||||
"""Wraps connection specific exception."""
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -0,0 +1,187 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2013 Mirantis.inc
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
"""Provision test environment for specific DB backends"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import random
|
||||
import string
|
||||
|
||||
import sqlalchemy
|
||||
|
||||
from openstack.common.db import exception as exc
|
||||
|
||||
|
||||
SQL_CONNECTION = os.getenv('OS_TEST_DBAPI_ADMIN_CONNECTION', 'sqlite://')
|
||||
|
||||
|
||||
def _gen_credentials(*names):
|
||||
"""Generate credentials."""
|
||||
auth_dict = {}
|
||||
for name in names:
|
||||
val = ''.join(random.choice(string.lowercase) for i in xrange(10))
|
||||
auth_dict[name] = val
|
||||
return auth_dict
|
||||
|
||||
|
||||
def _get_engine(uri=SQL_CONNECTION):
|
||||
"""Engine creation
|
||||
|
||||
By default the uri is SQL_CONNECTION which is admin credentials.
|
||||
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.
|
||||
"""
|
||||
return sqlalchemy.create_engine(uri, poolclass=sqlalchemy.pool.NullPool)
|
||||
|
||||
|
||||
def _execute_sql(engine, sql, driver):
|
||||
"""Initialize connection, execute sql query and close it."""
|
||||
try:
|
||||
with engine.connect() as conn:
|
||||
if driver == 'postgresql':
|
||||
conn.connection.set_isolation_level(0)
|
||||
for s in sql:
|
||||
conn.execute(s)
|
||||
except sqlalchemy.exc.OperationalError:
|
||||
msg = ('%s does not match database admin '
|
||||
'credentials or database does not exist.')
|
||||
raise exc.DBConnectionError(msg % SQL_CONNECTION)
|
||||
|
||||
|
||||
def create_database(engine):
|
||||
"""Provide temporary user and database for each particular test."""
|
||||
driver = engine.name
|
||||
|
||||
auth = _gen_credentials('database', 'user', 'passwd')
|
||||
|
||||
sqls = {
|
||||
'mysql': [
|
||||
"drop database if exists %(database)s;",
|
||||
"grant all on %(database)s.* to '%(user)s'@'localhost'"
|
||||
" identified by '%(passwd)s';",
|
||||
"create database %(database)s;",
|
||||
],
|
||||
'postgresql': [
|
||||
"drop database if exists %(database)s;",
|
||||
"drop user if exists %(user)s;",
|
||||
"create user %(user)s with password '%(passwd)s';",
|
||||
"create database %(database)s owner %(user)s;",
|
||||
]
|
||||
}
|
||||
|
||||
if driver == 'sqlite':
|
||||
return 'sqlite:////tmp/%s' % auth['database']
|
||||
|
||||
try:
|
||||
sql_rows = sqls[driver]
|
||||
except KeyError:
|
||||
raise ValueError('Unsupported RDBMS %s' % driver)
|
||||
sql_query = map(lambda x: x % auth, sql_rows)
|
||||
|
||||
_execute_sql(engine, sql_query, driver)
|
||||
|
||||
params = auth.copy()
|
||||
params['backend'] = driver
|
||||
return "%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s" % params
|
||||
|
||||
|
||||
def drop_database(engine, current_uri):
|
||||
"""Drop temporary database and user after each particular test."""
|
||||
engine = _get_engine(current_uri)
|
||||
admin_engine = _get_engine()
|
||||
driver = engine.name
|
||||
auth = {'database': engine.url.database, 'user': engine.url.username}
|
||||
|
||||
if driver == 'sqlite':
|
||||
try:
|
||||
os.remove(auth['database'])
|
||||
except OSError:
|
||||
pass
|
||||
return
|
||||
|
||||
sqls = {
|
||||
'mysql': [
|
||||
"drop database if exists %(database)s;",
|
||||
"drop user '%(user)s'@'localhost';",
|
||||
],
|
||||
'postgresql': [
|
||||
"drop database if exists %(database)s;",
|
||||
"drop user if exists %(user)s;",
|
||||
]
|
||||
}
|
||||
|
||||
try:
|
||||
sql_rows = sqls[driver]
|
||||
except KeyError:
|
||||
raise ValueError('Unsupported RDBMS %s' % driver)
|
||||
sql_query = map(lambda x: x % auth, sql_rows)
|
||||
|
||||
_execute_sql(admin_engine, sql_query, driver)
|
||||
|
||||
|
||||
def main():
|
||||
"""Controller to handle commands
|
||||
|
||||
::create: Create test user and database with random names.
|
||||
::drop: Drop user and database created by previous command.
|
||||
"""
|
||||
parser = argparse.ArgumentParser(
|
||||
description='Controller to handle database creation and dropping'
|
||||
' commands.',
|
||||
epilog='Under normal circumstances is not used directly.'
|
||||
' Used in .testr.conf to automate test database creation'
|
||||
' and dropping processes.')
|
||||
subparsers = parser.add_subparsers(
|
||||
help='Subcommands to manipulate temporary test databases.')
|
||||
|
||||
create = subparsers.add_parser(
|
||||
'create',
|
||||
help='Create temporary test '
|
||||
'databases and users.')
|
||||
create.set_defaults(which='create')
|
||||
create.add_argument(
|
||||
'instances_count',
|
||||
type=int,
|
||||
help='Number of databases to create.')
|
||||
|
||||
drop = subparsers.add_parser(
|
||||
'drop',
|
||||
help='Drop temporary test databases and users.')
|
||||
drop.set_defaults(which='drop')
|
||||
drop.add_argument(
|
||||
'instances',
|
||||
nargs='+',
|
||||
help='List of databases uri to be dropped.')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
engine = _get_engine()
|
||||
which = args.which
|
||||
|
||||
if which == "create":
|
||||
for i in range(int(args.instances_count)):
|
||||
print(create_database(engine))
|
||||
elif which == "drop":
|
||||
for db in args.instances:
|
||||
drop_database(engine, db)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -13,33 +13,80 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from functools import wraps
|
||||
import os
|
||||
|
||||
import fixtures
|
||||
from oslo.config import cfg
|
||||
|
||||
from openstack.common.db.sqlalchemy import session
|
||||
from openstack.common.fixture import config
|
||||
from openstack.common import test
|
||||
from tests import utils as test_utils
|
||||
|
||||
|
||||
class SqliteInMemoryFixture(fixtures.Fixture):
|
||||
"""SQLite in-memory DB recreated for each test case."""
|
||||
class DbFixture(fixtures.Fixture):
|
||||
"""Basic database fixture.
|
||||
|
||||
def setUp(self):
|
||||
super(SqliteInMemoryFixture, self).setUp()
|
||||
config_fixture = self.useFixture(config.Config())
|
||||
self.conf = config_fixture.conf
|
||||
Allows to run tests on various db backends, such as SQLite, MySQL and
|
||||
PostgreSQL. By default use sqlite backend. To override default backend
|
||||
uri set env variable OS_TEST_DBAPI_CONNECTION with database admin
|
||||
credentials for specific backend.
|
||||
"""
|
||||
|
||||
def _get_uri(self):
|
||||
return os.getenv('OS_TEST_DBAPI_CONNECTION', 'sqlite://')
|
||||
|
||||
def __init__(self):
|
||||
super(DbFixture, self).__init__()
|
||||
self.conf = cfg.CONF
|
||||
self.conf.import_opt('connection',
|
||||
'openstack.common.db.sqlalchemy.session',
|
||||
group='database')
|
||||
|
||||
self.conf.set_default('connection', "sqlite://", group='database')
|
||||
self.addCleanup(session.cleanup)
|
||||
def setUp(self):
|
||||
super(DbFixture, self).setUp()
|
||||
|
||||
self.conf.set_default('connection', self._get_uri(), group='database')
|
||||
self.addCleanup(self.conf.reset)
|
||||
|
||||
|
||||
class DbTestCase(test.BaseTestCase):
|
||||
"""Base class for testing of DB code (uses in-memory SQLite DB fixture)."""
|
||||
class DbTestCase(test_utils.BaseTestCase):
|
||||
"""Base class for testing of DB code.
|
||||
|
||||
Using `DbFixture`. Intended to be the main database test case to use all
|
||||
the tests on a given backend with user defined uri. Backend specific
|
||||
tests should be decorated with `backend_specific` decorator.
|
||||
"""
|
||||
|
||||
FIXTURE = DbFixture
|
||||
|
||||
def setUp(self):
|
||||
super(DbTestCase, self).setUp()
|
||||
self.useFixture(self.FIXTURE())
|
||||
|
||||
self.useFixture(SqliteInMemoryFixture())
|
||||
self.addCleanup(session.cleanup)
|
||||
|
||||
|
||||
ALLOWED_DIALECTS = ['sqlite', 'mysql', 'postgresql']
|
||||
|
||||
|
||||
def backend_specific(*dialects):
|
||||
"""Decorator to skip backend specific tests on inappropriate engines.
|
||||
|
||||
::dialects: list of dialects names under which the test will be launched.
|
||||
"""
|
||||
def wrap(f):
|
||||
@wraps(f)
|
||||
def ins_wrap(self):
|
||||
if not set(dialects).issubset(ALLOWED_DIALECTS):
|
||||
raise ValueError(
|
||||
"Please use allowed dialects: %s" % ALLOWED_DIALECTS)
|
||||
engine = session.get_engine()
|
||||
if engine.name not in dialects:
|
||||
msg = ('The test "%s" can be run '
|
||||
'only on %s. Current engine is %s.')
|
||||
args = (f.__name__, ' '.join(dialects), engine.name)
|
||||
self.skip(msg % args)
|
||||
else:
|
||||
return f(self)
|
||||
return ins_wrap
|
||||
return wrap
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@ class TestSqliteUniqueConstraints(test_base.DbTestCase):
|
|||
autoload=True
|
||||
)
|
||||
|
||||
@test_base.backend_specific('sqlite')
|
||||
def test_get_unique_constraints(self):
|
||||
table = self.reflected_table
|
||||
|
||||
|
|
@ -73,6 +74,7 @@ class TestSqliteUniqueConstraints(test_base.DbTestCase):
|
|||
)
|
||||
self.assertEqual(should_be, existing)
|
||||
|
||||
@test_base.backend_specific('sqlite')
|
||||
def test_add_unique_constraint(self):
|
||||
table = self.reflected_table
|
||||
UniqueConstraint(table.c.a, table.c.c, name='unique_a_c').create()
|
||||
|
|
@ -85,6 +87,7 @@ class TestSqliteUniqueConstraints(test_base.DbTestCase):
|
|||
)
|
||||
self.assertEqual(should_be, existing)
|
||||
|
||||
@test_base.backend_specific('sqlite')
|
||||
def test_drop_unique_constraint(self):
|
||||
table = self.reflected_table
|
||||
UniqueConstraint(table.c.a, table.c.b, name='unique_a_b').drop()
|
||||
|
|
|
|||
Loading…
Reference in New Issue