diff --git a/neutron_lib/fixture.py b/neutron_lib/fixture.py index 8b1e0c110..50274de6c 100644 --- a/neutron_lib/fixture.py +++ b/neutron_lib/fixture.py @@ -16,6 +16,9 @@ import warnings import fixtures import mock from oslo_config import cfg +from oslo_db.sqlalchemy import enginefacade +from oslo_db.sqlalchemy import provision +from oslo_db.sqlalchemy import session from oslo_messaging import conffixture from neutron_lib.api import attributes @@ -83,25 +86,100 @@ class CallbackRegistryFixture(fixtures.Fixture): self.patcher.stop() -class SqlFixture(fixtures.Fixture): +class _EnableSQLiteFKsFixture(fixtures.Fixture): + """Turn SQLite PRAGMA foreign keys on and off for tests. + + FIXME(zzzeek): figure out some way to get oslo.db test_base to honor + oslo_db.engines.create_engine() arguments like sqlite_fks as well + as handling that it needs to be turned off during drops. + + """ + + def __init__(self, engine): + self.engine = engine + + def _setUp(self): + if self.engine.name == 'sqlite': + self.engine.execute("PRAGMA foreign_keys=ON") + + def disable_fks(): + with self.engine.connect() as conn: + conn.connection.rollback() + conn.execute("PRAGMA foreign_keys=OFF") + self.addCleanup(disable_fks) + + +class _SqlFixture(fixtures.Fixture): # flag to indicate that the models have been loaded _TABLES_ESTABLISHED = False - def _setUp(self): + def _init_resources(self): + pass + + @classmethod + def generate_schema(cls, engine): # Register all data models - engine = db_api.get_context_manager().writer.get_engine() - if not SqlFixture._TABLES_ESTABLISHED: + if not _SqlFixture._TABLES_ESTABLISHED: model_base.BASEV2.metadata.create_all(engine) - SqlFixture._TABLES_ESTABLISHED = True + _SqlFixture._TABLES_ESTABLISHED = True - def clear_tables(): - with engine.begin() as conn: - for table in reversed( - model_base.BASEV2.metadata.sorted_tables): - conn.execute(table.delete()) + def delete_from_schema(self, engine): + with engine.begin() as conn: + for table in reversed( + model_base.BASEV2.metadata.sorted_tables): + conn.execute(table.delete()) - self.addCleanup(clear_tables) + def _setUp(self): + self.engine = db_api.CONTEXT_WRITER.get_engine() + self.generate_schema(self.engine) + self._init_resources() + + self.sessionmaker = session.get_maker(self.engine) + + _restore_factory = db_api.get_context_manager()._root_factory + + self.enginefacade_factory = enginefacade._TestTransactionFactory( + self.engine, self.sessionmaker, from_factory=_restore_factory, + apply_global=False) + + db_api.get_context_manager()._root_factory = self.enginefacade_factory + + self.addCleanup(lambda: self.delete_from_schema(self.engine)) + self.addCleanup( + lambda: setattr( + db_api.get_context_manager(), + "_root_factory", _restore_factory)) + + self.useFixture(_EnableSQLiteFKsFixture(self.engine)) + + +class _StaticSqlFixture(_SqlFixture): + """Fixture which keeps a single sqlite memory database at global scope.""" + + _GLOBAL_RESOURCES = False + + @classmethod + def _init_resources(cls): + # this is a classlevel version of what testresources + # does w/ the resources attribute as well as the + # setUpResources() step (which requires a test instance, that + # SqlFixture does not have). Because this is a SQLite memory + # database, we don't actually tear it down, so we can keep + # it running throughout all tests. + if cls._GLOBAL_RESOURCES: + return + else: + cls._GLOBAL_RESOURCES = True + cls.schema_resource = provision.SchemaResource( + provision.DatabaseResource( + "sqlite", db_api.get_context_manager()), + cls.generate_schema, teardown=False) + dependency_resources = {} + for name, resource in cls.schema_resource.resources: + dependency_resources[name] = resource.getResource() + cls.schema_resource.make(dependency_resources) + cls.engine = dependency_resources['database'].engine class APIDefinitionFixture(fixtures.Fixture): diff --git a/neutron_lib/tests/unit/db/_base.py b/neutron_lib/tests/unit/db/_base.py index 058f9c7b4..c28b4c80b 100644 --- a/neutron_lib/tests/unit/db/_base.py +++ b/neutron_lib/tests/unit/db/_base.py @@ -23,4 +23,4 @@ class SqlTestCase(base.BaseTestCase): def setUp(self): super(SqlTestCase, self).setUp() - self.useFixture(fixture.SqlFixture()) + self.useFixture(fixture._SqlFixture()) diff --git a/neutron_lib/tests/unit/test_fixture.py b/neutron_lib/tests/unit/test_fixture.py index 5cbadcb19..ef659d8e4 100644 --- a/neutron_lib/tests/unit/test_fixture.py +++ b/neutron_lib/tests/unit/test_fixture.py @@ -61,10 +61,12 @@ class SqlFixtureTestCase(base.BaseTestCase): options.set_defaults( cfg.CONF, connection='sqlite://') - self.useFixture(fixture.SqlFixture()) + self.fixture = fixture._SqlFixture() + self.useFixture(self.fixture) def test_fixture(self): self.assertIsNotNone(model_base.BASEV2.metadata.sorted_tables) + self.assertIsNotNone(self.fixture.engine) class APIDefinitionFixtureTestCase(base.BaseTestCase):