text: # Copyright (c) 2016 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.
  
  import fixtures
  import os
  import testresources
  import testscenarios
  
  from oslo_db import exception
  from oslo_db.sqlalchemy import enginefacade
  from oslo_db.sqlalchemy import provision
  from oslo_db.sqlalchemy import utils
  
  
  class ReplaceEngineFacadeFixture(fixtures.Fixture):
      """A fixture that will plug the engine of one enginefacade into another.
  
      This fixture can be used by test suites that already have their own non-
      oslo_db database setup / teardown schemes, to plug any URL or test-oriented
      enginefacade as-is into an enginefacade-oriented API.
  
      For applications that use oslo.db's testing fixtures, the
      ReplaceEngineFacade fixture is used internally.
  
      E.g.::
  
          class MyDBTest(TestCase):
  
              def setUp(self):
                  from myapplication.api import main_enginefacade
  
                  my_test_enginefacade = enginefacade.transaction_context()
                  my_test_enginefacade.configure(connection=my_test_url)
  
                  self.useFixture(
                      ReplaceEngineFacadeFixture(
                          main_enginefacade, my_test_enginefacade))
  
      Above, the main_enginefacade object is the normal application level
      one, and my_test_enginefacade is a local one that we've created to
      refer to some testing database.   Throughout the fixture's setup,
      the application level enginefacade will use the engine factory and
      engines of the testing enginefacade, and at fixture teardown will be
      replaced back.
  
      """
      def __init__(self, enginefacade, replace_with_enginefacade):
          super(ReplaceEngineFacadeFixture, self).__init__()
          self.enginefacade = enginefacade
          self.replace_with_enginefacade = replace_with_enginefacade
  
      def _setUp(self):
          _reset_facade = self.enginefacade.patch_factory(
              self.replace_with_enginefacade._factory
          )
          self.addCleanup(_reset_facade)
  
  
  class BaseDbFixture(fixtures.Fixture):
      """Base database provisioning fixture.
  
      This serves as the base class for the other fixtures, but by itself
      does not implement _setUp(). It provides the basis for the flags
      implemented by the various capability mixins (GenerateSchema,
      DeletesFromSchema, etc.) as well as providing an abstraction over
      the provisioning objects, which are specific to testresources.
      Overall, consumers of this fixture just need to use the right classes
      and the testresources mechanics are taken care of.
  
      """
      DRIVER = "sqlite"
  
      _DROP_SCHEMA_PER_TEST = True
      _BUILD_SCHEMA = False
      _BUILD_WITH_MIGRATIONS = False
  
      _database_resources = {}
      _db_not_available = {}
      _schema_resources = {}
  
      def __init__(self, driver=None, ident=None):
          super(BaseDbFixture, self).__init__()
          self.driver = driver or self.DRIVER
          self.ident = ident or "default"
          self.resource_key = (self.driver, self.__class__, self.ident)
  
      def get_enginefacade(self):
          """Return an enginefacade._TransactionContextManager.
  
          This is typically a global variable like "context_manager" declared
          in the db/api.py module and is the object returned by
          enginefacade.transaction_context().
  
          If left not implemented, the global enginefacade manager is used.
  
          For the case where a project uses per-object or per-test enginefacades
          like Gnocchi, the get_per_test_enginefacade()
          method should also be implemented.
  
  
          """
          return enginefacade._context_manager
  
      def get_per_test_enginefacade(self):
          """Return an enginefacade._TransactionContextManager per test.
  
          This facade should be the one that the test expects the code to
          use.   Usually this is the same one returned by get_engineafacade()
          which is the default.  For special applications like Gnocchi,
          this can be overridden to provide an instance-level facade.
  
          """
          return self.get_enginefacade()
  
      def _get_db_resource_not_available_reason(self):
          return self._db_not_available.get(self.resource_key, None)
  
      def _has_db_resource(self):
          return self._database_resources.get(
              self.resource_key, None) is not None
  
      def _generate_schema_resource(self, database_resource):
          return provision.SchemaResource(
              database_resource,
              None if not self._BUILD_SCHEMA
              else self.generate_schema_create_all
              if not self._BUILD_WITH_MIGRATIONS
              else self.generate_schema_migrations,
              self._DROP_SCHEMA_PER_TEST
          )
  
      def _get_resources(self):
          key = self.resource_key
  
          # the DatabaseResource and SchemaResource provision objects
          # can be used by testresources as a marker outside of an individual
          # test to indicate that this database / schema can be used across
          # multiple tests.   To make this work, many instances of this
          # fixture have to return the *same* resource object given the same
          # inputs.  so we cache these in class-level dictionaries.
  
          if key not in self._database_resources:
              _enginefacade = self.get_enginefacade()
              try:
                  self._database_resources[key] = \
                      self._generate_database_resource(_enginefacade)
              except exception.BackendNotAvailable as bne:
                  self._database_resources[key] = None
                  self._db_not_available[key] = str(bne)
  
          database_resource = self._database_resources[key]
  
          if database_resource is None:
              return []
          else:
              if key in self._schema_resources:
                  schema_resource = self._schema_resources[key]
              else:
                  schema_resource = self._schema_resources[key] = \
                      self._generate_schema_resource(database_resource)
  
              return [
                  ('_schema_%s' % self.ident, schema_resource),
                  ('_db_%s' % self.ident, database_resource)
              ]
  
  
  class GeneratesSchema(object):
      """Mixin defining a fixture as generating a schema using create_all().
  
      This is a "capability" mixin that works in conjunction with classes
      that include BaseDbFixture as a base.
  
      """
  
      _BUILD_SCHEMA = True
      _BUILD_WITH_MIGRATIONS = False
  
      def generate_schema_create_all(self, engine):
          """A hook which should generate the model schema using create_all().
  
          This hook is called within the scope of creating the database
          assuming BUILD_WITH_MIGRATIONS is False.
  
          """
  
  
  class GeneratesSchemaFromMigrations(GeneratesSchema):
      """Mixin defining a fixture as generating a schema using migrations.
  
      This is a "capability" mixin that works in conjunction with classes
      that include BaseDbFixture as a base.
  
      """
  
      _BUILD_WITH_MIGRATIONS = True
  
      def generate_schema_migrations(self, engine):
          """A hook which should generate the model schema using migrations.
  
  
          This hook is called within the scope of creating the database
          assuming BUILD_WITH_MIGRATIONS is True.
  
          """
  
  
  class ResetsData(object):
      """Mixin defining a fixture that resets schema data without dropping."""
  
      _DROP_SCHEMA_PER_TEST = False
  
      def setup_for_reset(self, engine, enginefacade):
          """"Perform setup that may be needed before the test runs."""
  
      def reset_schema_data(self, engine, enginefacade):
          """Reset the data in the schema."""
  
  
  class DeletesFromSchema(ResetsData):
      """Mixin defining a fixture that can delete from all tables in place.
  
      When DeletesFromSchema is present in a fixture,
      _DROP_SCHEMA_PER_TEST is now False; this means that the
      "teardown" flag of provision.SchemaResource will be False, which
      prevents SchemaResource from dropping all objects within the schema
      after each test.
  
      This is a "capability" mixin that works in conjunction with classes
      that include BaseDbFixture as a base.
  
      """
  
      def reset_schema_data(self, engine, facade):
          self.delete_from_schema(engine)
  
      def delete_from_schema(self, engine):
          """A hook which should delete all data from an existing schema.
  
          Should *not* drop any objects, just remove data from tables
          that needs to be reset between tests.
          """
  
  
  class RollsBackTransaction(ResetsData):
      """Fixture class that maintains a database transaction per test.
  
      """
  
      def setup_for_reset(self, engine, facade):
          conn = engine.connect()
          engine = utils.NonCommittingEngine(conn)
          self._reset_engine = enginefacade._TestTransactionFactory.apply_engine(
              engine, facade)
  
      def reset_schema_data(self, engine, facade):
          self._reset_engine()
          engine._dispose()
  
  
  class SimpleDbFixture(BaseDbFixture):
      """Fixture which provides an engine from a fixed URL.
  
      The SimpleDbFixture is generally appropriate only for a SQLite memory
      database, as this database is naturally isolated from other processes and
      does not require management of schemas.   For tests that need to
      run specifically against MySQL or Postgresql, the OpportunisticDbFixture
      is more appropriate.
  
      The database connection information itself comes from the provisoning
      system, matching the desired driver (typically sqlite) to the default URL
      that provisioning provides for this driver (in the case of sqlite, it's
      the SQLite memory URL, e.g. sqlite://.  For MySQL and Postgresql, it's
      the familiar "openstack_citest" URL on localhost).
  
      There are a variety of create/drop schemes that can take place:
  
      * The default is to procure a database connection on setup,
        and at teardown, an instruction is issued to "drop" all
        objects in the schema (e.g. tables, indexes).  The SQLAlchemy
        engine itself remains referenced at the class level for subsequent
        re-use.
  
      * When the GeneratesSchema or GeneratesSchemaFromMigrations mixins
        are implemented, the appropriate generate_schema method is also
        called when the fixture is set up, by default this is per test.
  
      * When the DeletesFromSchema mixin is implemented, the generate_schema
        method is now only called **once**, and the "drop all objects"
        system is replaced with the delete_from_schema method.   This
        allows the same database to remain set up with all schema objects
        intact, so that expensive migrations need not be run on every test.
  
      * The fixture does **not** dispose the engine at the end of a test.
        It is assumed the same engine will be re-used many times across
        many tests.  The AdHocDbFixture extends this one to provide
        engine.dispose() at the end of a test.
  
      This fixture is intended to work without needing a reference to
      the test itself, and therefore cannot take advantage of the
      OptimisingTestSuite.
  
      """
  
      _dependency_resources = {}
  
      def _get_provisioned_db(self):
          return self._dependency_resources["_db_%s" % self.ident]
  
      def _generate_database_resource(self, _enginefacade):
          return provision.DatabaseResource(self.driver, _enginefacade,
                                            provision_new_database=False)
  
      def _setUp(self):
          super(SimpleDbFixture, self)._setUp()
  
          cls = self.__class__
  
          if "_db_%s" % self.ident not in cls._dependency_resources:
  
              resources = self._get_resources()
  
              # initialize resources the same way that testresources does.
              for name, resource in resources:
                  cls._dependency_resources[name] = resource.getResource()
  
          provisioned_db = self._get_provisioned_db()
  
          if not self._DROP_SCHEMA_PER_TEST:
              self.setup_for_reset(
                  provisioned_db.engine, provisioned_db.enginefacade)
  
          self.useFixture(ReplaceEngineFacadeFixture(
              self.get_per_test_enginefacade(),
              provisioned_db.enginefacade
              ))
  
          if not self._DROP_SCHEMA_PER_TEST:
              self.addCleanup(
                  self.reset_schema_data,
                  provisioned_db.engine, provisioned_db.enginefacade)
  
          self.addCleanup(self._cleanup)
  
      def _teardown_resources(self):
          for name, resource in self._get_resources():
              dep = self._dependency_resources.pop(name)
              resource.finishedWith(dep)
  
      def _cleanup(self):
          pass
  
  
  class AdHocDbFixture(SimpleDbFixture):
      """"Fixture which creates and disposes a database engine per test.
  
      Also allows a specific URL to be passed, meaning the fixture can
      be hardcoded to a specific SQLite file.
  
      For a SQLite, this fixture will create the named database upon setup
      and tear it down upon teardown.   For other databases, the
      database is assumed to exist already and will remain after teardown.
  
      """
      def __init__(self, url=None):
          if url:
              self.url = provision.sa_url.make_url(str(url))
              driver = self.url.get_backend_name()
          else:
              driver = None
              self.url = None
  
          BaseDbFixture.__init__(
              self, driver=driver,
              ident=provision._random_ident())
          self.url = url
  
      def _generate_database_resource(self, _enginefacade):
          return provision.DatabaseResource(
              self.driver, _enginefacade, ad_hoc_url=self.url,
              provision_new_database=False)
  
      def _cleanup(self):
          self._teardown_resources()
  
  
  class OpportunisticDbFixture(BaseDbFixture):
      """Fixture which uses testresources fully for optimised runs.
  
      This fixture relies upon the use of the OpportunisticDBTestMixin to supply
      a test.resources attribute, and also works much more effectively when
      combined the testresources.OptimisingTestSuite.   The
      optimize_db_test_loader() function should be used at the module and package
      levels to optimize database provisioning across many tests.
  
      """
      def __init__(self, test, driver=None, ident=None):
          super(OpportunisticDbFixture, self).__init__(
              driver=driver, ident=ident)
          self.test = test
  
      def _get_provisioned_db(self):
          return getattr(self.test, "_db_%s" % self.ident)
  
      def _generate_database_resource(self, _enginefacade):
          return provision.DatabaseResource(
              self.driver, _enginefacade, provision_new_database=True)
  
      def _setUp(self):
          super(OpportunisticDbFixture, self)._setUp()
  
          if not self._has_db_resource():
              return
  
          provisioned_db = self._get_provisioned_db()
  
          if not self._DROP_SCHEMA_PER_TEST:
              self.setup_for_reset(
                  provisioned_db.engine, provisioned_db.enginefacade)
  
          self.useFixture(ReplaceEngineFacadeFixture(
              self.get_per_test_enginefacade(),
              provisioned_db.enginefacade
              ))
  
          if not self._DROP_SCHEMA_PER_TEST:
              self.addCleanup(
                  self.reset_schema_data,
                  provisioned_db.engine, provisioned_db.enginefacade)
  
  
  class OpportunisticDBTestMixin(object):
      """Test mixin that integrates the test suite with testresources.
  
      There are three goals to this system:
  
      1. Allow creation of "stub" test suites that will run all the tests    in a
         parent suite against a specific    kind of database (e.g. Mysql,
         Postgresql), where the entire suite will be skipped if that    target
         kind of database is not available to the suite.
  
      2. provide a test with a process-local, anonymously named schema within a
         target database, so that the test can run concurrently with other tests
         without conflicting data
  
      3. provide compatibility with the testresources.OptimisingTestSuite, which
         organizes TestCase instances ahead of time into groups that all
         make use of the same type of database, setting up and tearing down
         a database schema once for the scope of any number of tests within.
         This technique is essential when testing against a non-SQLite database
         because building of a schema is expensive, and also is most ideally
         accomplished using the applications schema migration which are
         even more vastly slow than a straight create_all().
  
      This mixin provides the .resources attribute required by testresources when
      using the OptimisingTestSuite.The .resources attribute then provides a
      collection of testresources.TestResourceManager objects, which are defined
      here in oslo_db.sqlalchemy.provision.   These objects know how to find
      available database backends, build up temporary databases, and invoke
      schema generation and teardown instructions.   The actual "build the schema
      objects" part of the equation, and optionally a "delete from all the
      tables" step, is provided by the implementing application itself.
  
  
      """
      SKIP_ON_UNAVAILABLE_DB = True
  
      FIXTURE = OpportunisticDbFixture
  
      _collected_resources = None
      _instantiated_fixtures = None
  
      @property
      def resources(self):
          """Provide a collection of TestResourceManager objects.
  
          The collection here is memoized, both at the level of the test
          case itself, as well as in the fixture object(s) which provide
          those resources.
  
          """
  
          if self._collected_resources is not None:
              return self._collected_resources
  
          fixtures = self._instantiate_fixtures()
          self._collected_resources = []
          for fixture in fixtures:
              self._collected_resources.extend(fixture._get_resources())
          return self._collected_resources
  
      def setUp(self):
          self._setup_fixtures()
          super(OpportunisticDBTestMixin, self).setUp()
  
      def _get_default_provisioned_db(self):
          return self._db_default
  
      def _instantiate_fixtures(self):
          if self._instantiated_fixtures:
              return self._instantiated_fixtures
  
          self._instantiated_fixtures = utils.to_list(self.generate_fixtures())
          return self._instantiated_fixtures
  
      def generate_fixtures(self):
          return self.FIXTURE(test=self)
  
      def _setup_fixtures(self):
          testresources.setUpResources(
              self, self.resources, testresources._get_result())
          self.addCleanup(
              testresources.tearDownResources,
              self, self.resources, testresources._get_result()
          )
  
          fixtures = self._instantiate_fixtures()
          for fixture in fixtures:
              self.useFixture(fixture)
  
              if not fixture._has_db_resource():
                  msg = fixture._get_db_resource_not_available_reason()
                  if self.SKIP_ON_UNAVAILABLE_DB:
                      self.skip(msg)
                  else:
                      self.fail(msg)
  
  
  class MySQLOpportunisticFixture(OpportunisticDbFixture):
      DRIVER = 'mysql'
  
  
  class PostgresqlOpportunisticFixture(OpportunisticDbFixture):
      DRIVER = 'postgresql'
  
  
  def optimize_package_test_loader(file_):
      """Organize package-level tests into a testresources.OptimizingTestSuite.
  
      This function provides a unittest-compatible load_tests hook
      for a given package; for per-module, use the
      :func:`.optimize_module_test_loader` function.
  
      When a unitest or subunit style
      test runner is used, the function will be called in order to
      return a TestSuite containing the tests to run; this function
      ensures that this suite is an OptimisingTestSuite, which will organize
      the production of test resources across groups of tests at once.
  
      The function is invoked as::
  
          from oslo_db.sqlalchemy import test_base
  
          load_tests = test_base.optimize_package_test_loader(__file__)
  
      The loader *must* be present in the package level __init__.py.
  
      The function also applies testscenarios expansion to all  test collections.
      This so that an existing test suite that already needs to build
      TestScenarios from a load_tests call can still have this take place when
      replaced with this function.
  
      """
  
      this_dir = os.path.dirname(file_)
  
      def load_tests(loader, found_tests, pattern):
          result = testresources.OptimisingTestSuite()
          result.addTests(found_tests)
          pkg_tests = loader.discover(start_dir=this_dir, pattern=pattern)
          result.addTests(testscenarios.generate_scenarios(pkg_tests))
  
          return result
      return load_tests
  
  
  def optimize_module_test_loader():
      """Organize module-level tests into a testresources.OptimizingTestSuite.
  
      This function provides a unittest-compatible load_tests hook
      for a given module; for per-package, use the
      :func:`.optimize_package_test_loader` function.
  
      When a unitest or subunit style
      test runner is used, the function will be called in order to
      return a TestSuite containing the tests to run; this function
      ensures that this suite is an OptimisingTestSuite, which will organize
      the production of test resources across groups of tests at once.
  
      The function is invoked as::
  
          from oslo_db.sqlalchemy import test_base
  
          load_tests = test_base.optimize_module_test_loader()
  
      The loader *must* be present in an individual module, and *not* the
      package level __init__.py.
  
      The function also applies testscenarios expansion to all  test collections.
      This so that an existing test suite that already needs to build
      TestScenarios from a load_tests call can still have this take place when
      replaced with this function.
  
      """
  
      def load_tests(loader, found_tests, pattern):
          result = testresources.OptimisingTestSuite()
          result.addTests(testscenarios.generate_scenarios(found_tests))
          return result
      return load_tests
  