Merge "Organize provisioning to use testresources"

12 files changed, 491 insertions, 155 deletions

diff --git a/oslo_db/sqlalchemy/provision.py b/oslo_db/sqlalchemy/provision.py
index ffb0cca..0062782 100644
--- a/oslo_db/sqlalchemy/provision.py
+++ b/oslo_db/sqlalchemy/provision.py
@@ -16,7 +16,6 @@
 """Provision test environment for specific DB backends"""
 
 import abc
-import argparse
 import logging
 import os
 import random
@@ -28,6 +27,7 @@ from six import moves
 import sqlalchemy
 from sqlalchemy.engine import url as sa_url
 from sqlalchemy import schema
+import testresources
 
 from oslo_db._i18n import _LI
 from oslo_db import exception
@@ -39,31 +39,103 @@ LOG = logging.getLogger(__name__)
 
 
 class ProvisionedDatabase(object):
-    """Represent a single database node that can be used for testing in
+    pass
 
-    a serialized fashion.
 
-    ``ProvisionedDatabase`` includes features for full lifecycle management
-    of a node, in a way that is context-specific.   Depending on how the
-    test environment runs, ``ProvisionedDatabase`` should know if it needs
-    to create and drop databases or if it is making use of a database that
-    is maintained by an external process.
+class BackendResource(testresources.TestResourceManager):
+    def __init__(self, database_type):
+        super(BackendResource, self).__init__()
+        self.database_type = database_type
+        self.backend = Backend.backend_for_database_type(self.database_type)
 
-    """
+    def make(self, dependency_resources):
+        return self.backend
 
-    def __init__(self, database_type):
-        self.backend = Backend.backend_for_database_type(database_type)
-        self.db_token = _random_ident()
+    def isDirty(self):
+        return False
 
-        self.backend.create_named_database(self.db_token)
-        self.engine = self.backend.provisioned_engine(self.db_token)
 
-    def drop_all_objects(self):
-        self.backend.drop_all_objects(self.engine)
+class DatabaseResource(testresources.TestResourceManager):
 
-    def dispose(self):
-        self.engine.dispose()
-        self.backend.drop_named_database(self.db_token)
+    def __init__(self, database_type):
+        super(DatabaseResource, self).__init__()
+        self.database_type = database_type
+        self.resources = [
+            ('backend', BackendResource(database_type))
+        ]
+
+    def make(self, dependency_resources):
+        dependency_resources['db_token'] = db_token = _random_ident()
+        backend = dependency_resources['backend']
+        LOG.info(
+            "CREATE BACKEND %s TOKEN %s", backend.engine.url, db_token)
+        backend.create_named_database(db_token, conditional=True)
+        dependency_resources['engine'] = \
+            backend.provisioned_engine(db_token)
+        return ProvisionedDatabase()
+
+    def clean(self, resource):
+        resource.engine.dispose()
+        LOG.info(
+            "DROP BACKEND %s TOKEN %s",
+            resource.backend.engine, resource.db_token)
+        resource.backend.drop_named_database(resource.db_token)
+
+    def isDirty(self):
+        return False
+
+
+class TransactionResource(testresources.TestResourceManager):
+
+    def __init__(self, database_resource, schema_resource):
+        super(TransactionResource, self).__init__()
+        self.resources = [
+            ('database', database_resource),
+            ('schema', schema_resource)
+        ]
+
+    def clean(self, resource):
+        resource._dispose()
+
+    def make(self, dependency_resources):
+        conn = dependency_resources['database'].engine.connect()
+        return utils.NonCommittingEngine(conn)
+
+    def isDirty(self):
+        return True
+
+
+class Schema(object):
+    pass
+
+
+class SchemaResource(testresources.TestResourceManager):
+
+    def __init__(self, database_resource, generate_schema, teardown=False):
+        super(SchemaResource, self).__init__()
+        self.generate_schema = generate_schema
+        self.teardown = teardown
+        self.resources = [
+            ('database', database_resource)
+        ]
+
+    def clean(self, resource):
+        LOG.info(
+            "DROP ALL OBJECTS, BACKEND %s",
+            resource.database.engine.url)
+        resource.database.backend.drop_all_objects(
+            resource.database.engine)
+
+    def make(self, dependency_resources):
+        if self.generate_schema:
+            self.generate_schema(dependency_resources['database'].engine)
+        return Schema()
+
+    def isDirty(self):
+        if self.teardown:
+            return True
+        else:
+            return False
 
 
 class Backend(object):
@@ -85,19 +157,12 @@ class Backend(object):
         self.verified = False
         self.engine = None
         self.impl = BackendImpl.impl(database_type)
+        self.current_dbs = set()
         Backend.backends_by_database_type[database_type] = self
 
     @classmethod
     def backend_for_database_type(cls, database_type):
-        """Return and verify the ``Backend`` for the given database type.
-
-        Creates the engine if it does not already exist and raises
-        ``BackendNotAvailable`` if it cannot be produced.
-
-        :return: a base ``Engine`` that allows provisioning of databases.
-
-        :raises: ``BackendNotAvailable``, if an engine for this backend
-         cannot be produced.
+        """Return the ``Backend`` for the given database type.
 
         """
         try:
@@ -172,10 +237,13 @@ class Backend(object):
                 conn.close()
                 return eng
 
-    def create_named_database(self, ident):
+    def create_named_database(self, ident, conditional=False):
         """Create a database with the given name."""
 
-        self.impl.create_named_database(self.engine, ident)
+        if not conditional or ident not in self.current_dbs:
+            self.current_dbs.add(ident)
+            self.impl.create_named_database(
+                self.engine, ident, conditional=conditional)
 
     def drop_named_database(self, ident, conditional=False):
         """Drop a database with the given name."""
@@ -183,6 +251,7 @@ class Backend(object):
         self.impl.drop_named_database(
             self.engine, ident,
             conditional=conditional)
+        self.current_dbs.discard(ident)
 
     def drop_all_objects(self, engine):
         """Drop all database objects.
@@ -303,7 +372,7 @@ class BackendImpl(object):
         """
 
     @abc.abstractmethod
-    def create_named_database(self, engine, ident):
+    def create_named_database(self, engine, ident, conditional=False):
         """Create a database with the given name."""
 
     @abc.abstractmethod
@@ -388,9 +457,10 @@ class MySQLBackendImpl(BackendImpl):
     def create_opportunistic_driver_url(self):
         return "mysql://openstack_citest:openstack_citest@localhost/"
 
-    def create_named_database(self, engine, ident):
+    def create_named_database(self, engine, ident, conditional=False):
         with engine.connect() as conn:
-            conn.execute("CREATE DATABASE %s" % ident)
+            if not conditional or not self.database_exists(conn, ident):
+                conn.execute("CREATE DATABASE %s" % ident)
 
     def drop_named_database(self, engine, ident, conditional=False):
         with engine.connect() as conn:
@@ -409,10 +479,12 @@ class SQLiteBackendImpl(BackendImpl):
     def create_opportunistic_driver_url(self):
         return "sqlite://"
 
-    def create_named_database(self, engine, ident):
+    def create_named_database(self, engine, ident, conditional=False):
         url = self._provisioned_database_url(engine.url, ident)
-        eng = sqlalchemy.create_engine(url)
-        eng.connect().close()
+        filename = url.database
+        if filename and (not conditional or not os.access(filename, os.F_OK)):
+            eng = sqlalchemy.create_engine(url)
+            eng.connect().close()
 
     def provisioned_engine(self, base_url, ident):
         return session.create_engine(
@@ -442,10 +514,11 @@ class PostgresqlBackendImpl(BackendImpl):
         return "postgresql://openstack_citest:openstack_citest"\
             "@localhost/postgres"
 
-    def create_named_database(self, engine, ident):
+    def create_named_database(self, engine, ident, conditional=False):
         with engine.connect().execution_options(
                 isolation_level="AUTOCOMMIT") as conn:
-            conn.execute("CREATE DATABASE %s" % ident)
+            if not conditional or not self.database_exists(conn, ident):
+                conn.execute("CREATE DATABASE %s" % ident)
 
     def drop_named_database(self, engine, ident, conditional=False):
         with engine.connect().execution_options(
@@ -501,82 +574,4 @@ def _random_ident():
         for i in moves.range(10))
 
 
-def _echo_cmd(args):
-    idents = [_random_ident() for i in moves.range(args.instances_count)]
-    print("\n".join(idents))
-
-
-def _create_cmd(args):
-    idents = [_random_ident() for i in moves.range(args.instances_count)]
-
-    for backend in Backend.all_viable_backends():
-        for ident in idents:
-            backend.create_named_database(ident)
-
-    print("\n".join(idents))
-
-
-def _drop_cmd(args):
-    for backend in Backend.all_viable_backends():
-        for ident in args.instances:
-            backend.drop_named_database(ident, args.conditional)
-
 Backend._setup()
-
-
-def main(argv=None):
-    """Command line interface to create/drop databases.
-
-    ::create: Create test database with random names.
-    ::drop: Drop database created by previous command.
-    ::echo: create random names and display them; don't create.
-    """
-    parser = argparse.ArgumentParser(
-        description='Controller to handle database creation and dropping'
-        ' commands.',
-        epilog='Typically called by the test runner, e.g. shell script, '
-        'testr runner via .testr.conf, or other system.')
-    subparsers = parser.add_subparsers(
-        help='Subcommands to manipulate temporary test databases.')
-
-    create = subparsers.add_parser(
-        'create',
-        help='Create temporary test databases.')
-    create.set_defaults(which=_create_cmd)
-    create.add_argument(
-        'instances_count',
-        type=int,
-        help='Number of databases to create.')
-
-    drop = subparsers.add_parser(
-        'drop',
-        help='Drop temporary test databases.')
-    drop.set_defaults(which=_drop_cmd)
-    drop.add_argument(
-        'instances',
-        nargs='+',
-        help='List of databases uri to be dropped.')
-    drop.add_argument(
-        '--conditional',
-        action="store_true",
-        help="Check if database exists first before dropping"
-    )
-
-    echo = subparsers.add_parser(
-        'echo',
-        help="Create random database names and display only."
-    )
-    echo.set_defaults(which=_echo_cmd)
-    echo.add_argument(
-        'instances_count',
-        type=int,
-        help='Number of identifiers to create.')
-
-    args = parser.parse_args(argv)
-
-    cmd = args.which
-    cmd(args)
-
-
-if __name__ == "__main__":
-    main()
diff --git a/oslo_db/sqlalchemy/test_base.py b/oslo_db/sqlalchemy/test_base.py
index 601a2c8..96171ea 100644
--- a/oslo_db/sqlalchemy/test_base.py
+++ b/oslo_db/sqlalchemy/test_base.py
@@ -14,6 +14,8 @@
 # under the License.
 
 import fixtures
+import testresources
+import testscenarios
 
 try:
     from oslotest import base as test_base
@@ -22,6 +24,7 @@ except ImportError:
                     ' test-requirements')
 
 
+import os
 import six
 
 from oslo_db import exception
@@ -48,40 +51,118 @@ class DbFixture(fixtures.Fixture):
 
     def __init__(self, test):
         super(DbFixture, self).__init__()
-
         self.test = test
 
     def setUp(self):
         super(DbFixture, self).setUp()
 
-        try:
-            self.provision = provision.ProvisionedDatabase(self.DRIVER)
-            self.addCleanup(self.provision.dispose)
-        except exception.BackendNotAvailable:
-            msg = '%s backend is not available.' % self.DRIVER
-            return self.test.skip(msg)
+        testresources.setUpResources(
+            self.test, self.test.resources, testresources._get_result())
+        self.addCleanup(
+            testresources.tearDownResources,
+            self.test, self.test.resources, testresources._get_result()
+        )
+        if not hasattr(self.test, 'db'):
+            self.test.skip("database '%s' unavailable" % self.DRIVER)
+
+        if self.test.SCHEMA_SCOPE:
+            self.test.engine = self.test.transaction_engine
+            self.test.sessionmaker = session.get_maker(
+                self.test.transaction_engine)
         else:
-            self.test.provision = self.provision
-            self.test.engine = self.provision.engine
-            self.addCleanup(setattr, self.test, 'engine', None)
+            self.test.engine = self.test.db.engine
             self.test.sessionmaker = session.get_maker(self.test.engine)
-            self.addCleanup(setattr, self.test, 'sessionmaker', None)
+        self.addCleanup(setattr, self.test, 'sessionmaker', None)
+        self.addCleanup(setattr, self.test, 'engine', None)
 
 
 class DbTestCase(test_base.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
+    SCHEMA_SCOPE = None
+
+    _schema_resources = {}
+    _database_resources = {}
+
+    def _resources_for_driver(self, driver, schema_scope, generate_schema):
+        # testresources relies on the identity and state of the
+        # TestResourceManager objects in play to correctly manage
+        # resources, and it also hardcodes to looking at the
+        # ".resources" attribute on the test object, even though the
+        # setUpResources() function passes the list of resources in,
+        # so we have to code the TestResourceManager logic into the
+        # .resources attribute and ensure that the same set of test
+        # variables always produces the same TestResourceManager objects.
+        if driver not in self._database_resources:
+            try:
+                self._database_resources[driver] = \
+                    provision.DatabaseResource(driver)
+            except exception.BackendNotAvailable:
+                self._database_resources[driver] = None
+
+        database_resource = self._database_resources[driver]
+        if database_resource is None:
+            return []
+
+        if schema_scope:
+            key = (driver, schema_scope)
+            if key not in self._schema_resources:
+                schema_resource = provision.SchemaResource(
+                    database_resource, generate_schema)
+
+                transaction_resource = provision.TransactionResource(
+                    database_resource, schema_resource)
+
+                self._schema_resources[key] = \
+                    transaction_resource
+
+            transaction_resource = self._schema_resources[key]
+
+            return [
+                ('transaction_engine', transaction_resource),
+                ('db', database_resource),
+            ]
+        else:
+            key = (driver, None)
+            if key not in self._schema_resources:
+                self._schema_resources[key] = provision.SchemaResource(
+                    database_resource, generate_schema, teardown=True)
+
+            schema_resource = self._schema_resources[key]
+            return [
+                ('schema', schema_resource),
+                ('db', database_resource)
+            ]
+
+    @property
+    def resources(self):
+        return self._resources_for_driver(
+            self.FIXTURE.DRIVER, self.SCHEMA_SCOPE, self.generate_schema)
 
     def setUp(self):
         super(DbTestCase, self).setUp()
         self.useFixture(self.FIXTURE(self))
 
+    def generate_schema(self, engine):
+        """Generate schema objects to be used within a test.
+
+        The function is separate from the setUp() case as the scope
+        of this method is controlled by the provisioning system.  A
+        test that specifies SCHEMA_SCOPE may not call this method
+        for each test, as the schema may be maintained from a previous run.
+
+        """
+        if self.SCHEMA_SCOPE:
+            # if SCHEMA_SCOPE is set, then this method definitely
+            # has to be implemented.  This is a guard against a test
+            # that inadvertently does schema setup within setUp().
+            raise NotImplementedError(
+                "This test requires schema-level setup to be "
+                "implemented within generate_schema().")
+
 
 class OpportunisticTestCase(DbTestCase):
     """Placeholder for backwards compatibility."""
@@ -126,3 +207,41 @@ class MySQLOpportunisticTestCase(OpportunisticTestCase):
 
 class PostgreSQLOpportunisticTestCase(OpportunisticTestCase):
     FIXTURE = PostgreSQLOpportunisticFixture
+
+
+def optimize_db_test_loader(file_):
+    """Package level load_tests() function.
+
+    Will apply an optimizing test suite to all sub-tests, which groups DB
+    tests and other resources appropriately.
+
+    Place this in an __init__.py package file within the root of the test
+    suite, at the level where testresources loads it as a package::
+
+        from oslo.db.sqlalchemy import test_base
+
+        load_tests = test_base.optimize_db_test_loader(__file__)
+
+    Alternatively, the directive can be placed into a test module directly.
+
+    """
+
+    this_dir = os.path.dirname(file_)
+
+    def load_tests(loader, found_tests, pattern):
+        # pattern is None if the directive is placed within
+        # a test module directly, as well as within certain test
+        # discovery patterns
+
+        if pattern is not None:
+            pkg_tests = loader.discover(start_dir=this_dir, pattern=pattern)
+
+        result = testresources.OptimisingTestSuite()
+        found_tests = testscenarios.load_tests_apply_scenarios(
+            loader, found_tests, pattern)
+        result.addTest(found_tests)
+
+        if pattern is not None:
+            result.addTest(pkg_tests)
+        return result
+    return load_tests
diff --git a/oslo_db/sqlalchemy/test_migrations.py b/oslo_db/sqlalchemy/test_migrations.py
index 7627d21..2f9372e 100644
--- a/oslo_db/sqlalchemy/test_migrations.py
+++ b/oslo_db/sqlalchemy/test_migrations.py
@@ -16,6 +16,7 @@
 
 import abc
 import collections
+import functools
 import logging
 import pprint
 
@@ -483,9 +484,6 @@ class ModelsMigrationsSync(object):
                 return meta_def != insp_def
             return insp_def != "'%s'::character varying" % meta_def.arg
 
-    def _cleanup(self):
-        self.provision.drop_all_objects()
-
     FKInfo = collections.namedtuple('fk_info', ['constrained_columns',
                                                 'referred_table',
                                                 'referred_columns'])
@@ -567,7 +565,8 @@ class ModelsMigrationsSync(object):
                           ' for running of this test: %s' % e)
 
         # drop all tables after a test run
-        self.addCleanup(self._cleanup)
+        self.addCleanup(functools.partial(self.db.backend.drop_all_objects,
+                                          self.get_engine()))
 
         # run migration scripts
         self.db_sync(self.get_engine())
diff --git a/oslo_db/sqlalchemy/utils.py b/oslo_db/sqlalchemy/utils.py
index 5505b79..932d409 100644
--- a/oslo_db/sqlalchemy/utils.py
+++ b/oslo_db/sqlalchemy/utils.py
@@ -17,6 +17,7 @@
 #    under the License.
 
 import collections
+import contextlib
 import logging
 import re
 
@@ -1012,3 +1013,126 @@ def get_non_innodb_tables(connectable, skip_tables=('migrate_version',
     query = text(query_str)
     noninnodb = connectable.execute(query, **params)
     return [i[0] for i in noninnodb]
+
+
+class NonCommittingConnectable(object):
+    """A ``Connectable`` substitute which rolls all operations back.
+
+    ``NonCommittingConnectable`` forms the basis of mock
+    ``Engine`` and ``Connection`` objects within a test.   It provides
+    only that part of the API that should reasonably be used within
+    a single-connection test environment (e.g. no engine.dispose(),
+    connection.invalidate(), etc. ).   The connection runs both within
+    a transaction as well as a savepoint.   The transaction is there
+    so that any operations upon the connection can be rolled back.
+    If the test calls begin(), a "pseduo" transaction is returned that
+    won't actually commit anything.   The subtransaction is there to allow
+    a test to successfully call rollback(), however, where all operations
+    to that point will be rolled back and the operations can continue,
+    simulating a real rollback while still remaining within a transaction
+    external to the test.
+
+    """
+
+    def __init__(self, connection):
+        self.connection = connection
+        self._trans = connection.begin()
+        self._restart_nested()
+
+    def _restart_nested(self):
+        self._nested_trans = self.connection.begin_nested()
+
+    def _dispose(self):
+        if not self.connection.closed:
+            self._nested_trans.rollback()
+            self._trans.rollback()
+            self.connection.close()
+
+    def execute(self, obj, *multiparams, **params):
+        """Executes the given construct and returns a :class:`.ResultProxy`."""
+
+        return self.connection.execute(obj, *multiparams, **params)
+
+    def scalar(self, obj, *multiparams, **params):
+        """Executes and returns the first column of the first row."""
+
+        return self.connection.scalar(obj, *multiparams, **params)
+
+
+class NonCommittingEngine(NonCommittingConnectable):
+    """``Engine`` -specific non committing connectbale."""
+
+    @property
+    def url(self):
+        return self.connection.engine.url
+
+    @property
+    def engine(self):
+        return self
+
+    def connect(self):
+        return NonCommittingConnection(self.connection)
+
+    @contextlib.contextmanager
+    def begin(self):
+        conn = self.connect()
+        trans = conn.begin()
+        try:
+            yield conn
+        except Exception:
+            trans.rollback()
+        else:
+            trans.commit()
+
+
+class NonCommittingConnection(NonCommittingConnectable):
+    """``Connection`` -specific non committing connectbale."""
+
+    def close(self):
+        """Close the 'Connection'.
+
+        In this context, close() is a no-op.
+
+        """
+        pass
+
+    def begin(self):
+        return NonCommittingTransaction(self, self.connection.begin())
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *arg):
+        pass
+
+
+class NonCommittingTransaction(object):
+    """A wrapper for ``Transaction``.
+
+    This is to accommodate being able to guaranteed start a new
+    SAVEPOINT when a transaction is rolled back.
+
+    """
+    def __init__(self, provisioned, transaction):
+        self.provisioned = provisioned
+        self.transaction = transaction
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        if type is None:
+            try:
+                self.commit()
+            except Exception:
+                self.rollback()
+                raise
+        else:
+            self.rollback()
+
+    def commit(self):
+        self.transaction.commit()
+
+    def rollback(self):
+        self.transaction.rollback()
+        self.provisioned._restart_nested()
diff --git a/oslo_db/tests/sqlalchemy/__init__.py b/oslo_db/tests/sqlalchemy/__init__.py
index e69de29..e4bfd87 100644
--- a/oslo_db/tests/sqlalchemy/__init__.py
+++ b/oslo_db/tests/sqlalchemy/__init__.py
@@ -0,0 +1,18 @@
+# Copyright (c) 2014 OpenStack Foundation
+# 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.
+
+from oslo.db.sqlalchemy import test_base
+
+load_tests = test_base.optimize_db_test_loader(__file__)
diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py
index edab9d6..aafdcfb 100644
--- a/oslo_db/tests/sqlalchemy/test_exc_filters.py
+++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py
@@ -326,12 +326,17 @@ class TestReferenceErrorMySQL(TestReferenceErrorSQLite,
         self.assertEqual("resource_foo", matched.key_table)
 
     def test_raise_ansi_quotes(self):
-        self.engine.execute("SET SESSION sql_mode = 'ANSI';")
-        matched = self.assertRaises(
-            exception.DBReferenceError,
-            self.engine.execute,
-            self.table_2.insert({'id': 1, 'foo_id': 2})
-        )
+        with self.engine.connect() as conn:
+            conn.detach()  # will not be returned to the pool when closed
+
+            # this is incompatible with some internals of the engine
+            conn.execute("SET SESSION sql_mode = 'ANSI';")
+
+            matched = self.assertRaises(
+                exception.DBReferenceError,
+                conn.execute,
+                self.table_2.insert({'id': 1, 'foo_id': 2})
+            )
 
         self.assertInnerException(
             matched,
diff --git a/oslo_db/tests/sqlalchemy/test_migrations.py b/oslo_db/tests/sqlalchemy/test_migrations.py
index 1bae701..96e7e31 100644
--- a/oslo_db/tests/sqlalchemy/test_migrations.py
+++ b/oslo_db/tests/sqlalchemy/test_migrations.py
@@ -171,7 +171,7 @@ class TestWalkVersions(test.BaseTestCase, migrate.WalkVersionsMixin):
         self.assertEqual(upgraded, self.migrate_up.call_args_list)
 
 
-class ModelsMigrationSyncMixin(test.BaseTestCase):
+class ModelsMigrationSyncMixin(test_base.DbTestCase):
 
     def setUp(self):
         super(ModelsMigrationSyncMixin, self).setUp()
diff --git a/oslo_db/tests/sqlalchemy/test_provision.py b/oslo_db/tests/sqlalchemy/test_provision.py
index 7c57de3..25f5bc4 100644
--- a/oslo_db/tests/sqlalchemy/test_provision.py
+++ b/oslo_db/tests/sqlalchemy/test_provision.py
@@ -10,11 +10,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
+from oslotest import base as oslo_test_base
 from sqlalchemy import inspect
 from sqlalchemy import schema
 from sqlalchemy import types
 
+from oslo.db import exception
+from oslo.db.sqlalchemy import provision
 from oslo_db.sqlalchemy import test_base
 
 
@@ -62,7 +64,7 @@ class DropAllObjectsTest(test_base.DbTestCase):
             set(insp.get_table_names())
         )
 
-        self.provision.drop_all_objects()
+        self.db.backend.drop_all_objects(self.engine)
 
         insp = inspect(self.engine)
         self.assertEqual(
@@ -71,11 +73,83 @@ class DropAllObjectsTest(test_base.DbTestCase):
         )
 
 
-class MySQLRetainSchemaTest(
+class MySQLDropAllObjectsTest(
         DropAllObjectsTest, test_base.MySQLOpportunisticTestCase):
     pass
 
 
-class PostgresqlRetainSchemaTest(
+class PostgreSQLDropAllObjectsTest(
         DropAllObjectsTest, test_base.PostgreSQLOpportunisticTestCase):
     pass
+
+
+class RetainSchemaTest(oslo_test_base.BaseTestCase):
+    DRIVER = "sqlite"
+
+    def setUp(self):
+        super(RetainSchemaTest, self).setUp()
+
+        metadata = schema.MetaData()
+        self.test_table = schema.Table(
+            'test_table', metadata,
+            schema.Column('x', types.Integer),
+            schema.Column('y', types.Integer),
+            mysql_engine='InnoDB'
+        )
+
+        def gen_schema(engine):
+            metadata.create_all(engine, checkfirst=False)
+        self._gen_schema = gen_schema
+
+    def test_once(self):
+        self._run_test()
+
+    def test_twice(self):
+        self._run_test()
+
+    def _run_test(self):
+        try:
+            database_resource = provision.DatabaseResource(self.DRIVER)
+        except exception.BackendNotAvailable:
+            self.skip("database not available")
+
+        schema_resource = provision.SchemaResource(
+            database_resource, self._gen_schema)
+        transaction_resource = provision.TransactionResource(
+            database_resource, schema_resource)
+
+        engine = transaction_resource.getResource()
+
+        with engine.connect() as conn:
+            rows = conn.execute(self.test_table.select())
+            self.assertEqual(rows.fetchall(), [])
+
+            trans = conn.begin()
+            conn.execute(
+                self.test_table.insert(),
+                {"x": 1, "y": 2}
+            )
+            trans.rollback()
+
+            rows = conn.execute(self.test_table.select())
+            self.assertEqual(rows.fetchall(), [])
+
+            trans = conn.begin()
+            conn.execute(
+                self.test_table.insert(),
+                {"x": 2, "y": 3}
+            )
+            trans.commit()
+
+            rows = conn.execute(self.test_table.select())
+            self.assertEqual(rows.fetchall(), [(2, 3)])
+
+        transaction_resource.finishedWith(engine)
+
+
+class MySQLRetainSchemaTest(RetainSchemaTest):
+    DRIVER = "mysql"
+
+
+class PostgresqlRetainSchemaTest(RetainSchemaTest):
+    DRIVER = "postgresql"
diff --git a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py
index bcc4a1c..24aeb22 100644
--- a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py
+++ b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py
@@ -229,20 +229,22 @@ class MySQLModeTestCase(test_base.MySQLOpportunisticTestCase):
 
     def setUp(self):
         super(MySQLModeTestCase, self).setUp()
-
-        self.engine = session.create_engine(self.engine.url,
-                                            mysql_sql_mode=self.mysql_mode)
-        self.connection = self.engine.connect()
+        mode_engine = session.create_engine(
+            self.engine.url,
+            mysql_sql_mode=self.mysql_mode)
+        self.connection = mode_engine.connect()
 
         meta = MetaData()
-        meta.bind = self.engine
         self.test_table = Table(_TABLE_NAME + "mode", meta,
                                 Column('id', Integer, primary_key=True),
                                 Column('bar', String(255)))
-        self.test_table.create()
+        self.test_table.create(self.connection)
 
-        self.addCleanup(self.test_table.drop)
-        self.addCleanup(self.connection.close)
+        def cleanup():
+            self.test_table.drop(self.connection)
+            self.connection.close()
+            mode_engine.dispose()
+        self.addCleanup(cleanup)
 
     def _test_string_too_long(self, value):
         with self.connection.begin():
diff --git a/requirements.txt b/requirements.txt
index 1673a51..545d892 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -13,3 +13,5 @@ SQLAlchemy>=0.9.7,<=0.9.99
 sqlalchemy-migrate>=0.9.1,!=0.9.2
 stevedore>=1.1.0  # Apache-2.0
 six>=1.7.0
+testresources>=0.2.4
+testscenarios>=0.4
diff --git a/test-requirements-py2.txt b/test-requirements-py2.txt
index 72d7a8c..34e18db 100644
--- a/test-requirements-py2.txt
+++ b/test-requirements-py2.txt
@@ -15,6 +15,5 @@ sphinx>=1.1.2,!=1.2.0,!=1.3b1,<1.3
 oslosphinx>=2.2.0  # Apache-2.0
 oslotest>=1.2.0  # Apache-2.0
 testrepository>=0.0.18
-testscenarios>=0.4
 testtools>=0.9.36,!=1.2.0
 tempest-lib>=0.2.0
diff --git a/test-requirements-py3.txt b/test-requirements-py3.txt
index 8501d75..b9c1c33 100644
--- a/test-requirements-py3.txt
+++ b/test-requirements-py3.txt
@@ -14,7 +14,6 @@ sphinx>=1.1.2,!=1.2.0,!=1.3b1,<1.3
 oslosphinx>=2.2.0  # Apache-2.0
 oslotest>=1.2.0  # Apache-2.0
 testrepository>=0.0.18
-testscenarios>=0.4
 testtools>=0.9.36,!=1.2.0
 tempest-lib>=0.2.0