fix usage of opportunistic test cases with enginefacade

This creates an in tree fixture that lets you rebuild the engine
facade for doing individual tests against MySQL. It is needed until
this is supported in upstream oslo.db, which might be some time given
we are just hitting library freeze. Without this we can't build
regression tests for mysql specific bugs.

Related-Bug: #1548960

Change-Id: I2c4148972e91dcf1d6df0960198dc842b633d9e7
This commit is contained in:
Roman Podoliaka 2016-02-23 19:33:54 +02:00 committed by Sean Dague
parent 243d543f0f
commit e59f95439b

View File

@ -23,6 +23,7 @@ import warnings
import fixtures
from oslo_config import cfg
from oslo_db.sqlalchemy import enginefacade
from oslo_messaging import conffixture as messaging_conffixture
import six
@ -476,3 +477,39 @@ class StableObjectJsonFixture(fixtures.Fixture):
self.useFixture(fixtures.MonkeyPatch(
'nova.objects.base.NovaObject.obj_to_primitive', _doit))
class EngineFacadeFixture(fixtures.Fixture):
"""Fixture to isolation EngineFacade during tests.
Because many elements of EngineFacade are based on globals, once
an engine facade has been initialized, all future code goes
through it. This means that the initialization of sqlite in
databases in our Database fixture will drive all connections to
sqlite. While that's fine in a production environment, during
testing this means we can't test againts multiple backends in the
same test run.
oslo.db does not yet support a reset mechanism here. This builds a
custom in tree engine facade fixture to handle this. Eventually
this will be added to oslo.db and this can be removed. Tracked by
https://bugs.launchpad.net/oslo.db/+bug/1548960
"""
def __init__(self, ctx_manager, engine, sessionmaker):
super(EngineFacadeFixture, self).__init__()
self._ctx_manager = ctx_manager
self._engine = engine
self._sessionmaker = sessionmaker
def setUp(self):
super(EngineFacadeFixture, self).setUp()
self._existing_factory = self._ctx_manager._root_factory
self._ctx_manager._root_factory = enginefacade._TestTransactionFactory(
self._engine, self._sessionmaker, apply_global=False,
synchronous_reader=True)
self.addCleanup(self.cleanup)
def cleanup(self):
self._ctx_manager._root_factory = self._existing_factory