womax/nova
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor Database fixture

Browse Source 
This patch apply the following changes to the fixture to make the
intention clearer and remove some unnecessary complexity:

* the fixture does a lot of dynamic thing in its __init__, these are
  moved to setUp() instead to facilitate proper reset functionality of
  the fixture

* the caching and applying of the DB schema is made a explicit and
  moved to setUp() too

* the explicit reset() function is removed as it is probably
  unintentionally overwrote the Fixture.reset(). Now the Fixture can be
  properly reset by calling the Fixture.reset() which is by default
  implemented by calling cleanUp() and setUp()

Change-Id: Ic58e93d6aafb88be4abeb6e52089f7ee43d8db01
This commit is contained in:
Balazs Gibizer
2021-10-27 14:33:16 +02:00
parent f318f822fc
commit c7ccbfe403
2 changed files with 39 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nova/tests/fixtures/nova.py
Vendored
+31 -30
View File
@@ -612,65 +612,66 @@ class Database(fixtures.Fixture):
super().__init__()
assert database in {'main', 'api'}, f'Unrecognized database {database}'
if database == 'api':
assert connection is None, 'Not supported for the API database'
self.database = database
self.version = version
self.connection = connection
if database == 'main':
# NOTE(gibi): this inject a new factory for each test and
# self.factory_reset is used to clean up the factory at the end
# of the test case. This way we can let each test configure the
# factory so we can avoid having a global flag guarding against
# factory re-configuration
self.factory_reset = main_db_api.context_manager.patch_factory(
enginefacade._TransactionFactory())
main_db_api.configure(CONF)
def setUp(self):
super().setUp()
if connection is not None:
if self.database == 'main':
if self.connection is not None:
ctxt_mgr = main_db_api.create_context_manager(
connection=connection)
connection=self.connection)
self.get_engine = ctxt_mgr.writer.get_engine
else:
self.get_engine = main_db_api.get_engine
elif database == 'api':
assert connection is None, 'Not supported for the API database'
# NOTE(gibi): this inject a new factory for each test and
# registers a cleanup that cleans the factory at the end
# of the test case. This way we can let each test configure the
# factory so we can avoid having a global flag guarding against
# factory re-configuration
self.addCleanup(main_db_api.context_manager.patch_factory(
enginefacade._TransactionFactory()))
main_db_api.configure(CONF)
self.get_engine = main_db_api.get_engine
elif self.database == 'api':
# NOTE(gibi): similar note applies here as for the main_db_api
# above
self.factory_reset = api_db_api.context_manager.patch_factory(
enginefacade._TransactionFactory())
self.addCleanup(api_db_api.context_manager.patch_factory(
enginefacade._TransactionFactory()))
api_db_api.configure(CONF)
self.get_engine = api_db_api.get_engine
def setUp(self):
super(Database, self).setUp()
self.reset()
self._apply_schema()
self.addCleanup(self.cleanup)
def _cache_schema(self):
def _apply_schema(self):
global DB_SCHEMA
if not DB_SCHEMA[(self.database, self.version)]:
# apply and cache schema
engine = self.get_engine()
conn = engine.connect()
migration.db_sync(database=self.database, version=self.version)
DB_SCHEMA[(self.database, self.version)] = "".join(
line for line in conn.connection.iterdump())
engine.dispose()
else:
# apply the cached schema
engine = self.get_engine()
conn = engine.connect()
conn.connection.executescript(
DB_SCHEMA[(self.database, self.version)])
def cleanup(self):
engine = self.get_engine()
engine.dispose()
def reset(self):
self.factory_reset()
engine = self.get_engine()
engine.dispose()
self._cache_schema()
conn = engine.connect()
conn.connection.executescript(
DB_SCHEMA[(self.database, self.version)])
class DefaultFlavorsFixture(fixtures.Fixture):
def setUp(self):
nova/tests/unit/test_fixtures.py
+8 -8
View File
@@ -121,7 +121,8 @@ class TestDatabaseFixture(testtools.TestCase):
def test_fixture_reset(self):
# because this sets up reasonable db connection strings
self.useFixture(fixtures.ConfFixture())
self.useFixture(fixtures.Database())
db_fixture = fixtures.Database()
self.useFixture(db_fixture)
engine = main_db_api.get_engine()
conn = engine.connect()
result = conn.execute("select * from instance_types")
@@ -138,12 +139,11 @@ class TestDatabaseFixture(testtools.TestCase):
rows = result.fetchall()
self.assertEqual(1, len(rows), "Rows %s" % rows)
# reset by invoking the fixture again
#
# NOTE(sdague): it's important to reestablish the db
# connection because otherwise we have a reference to the old
# in mem db.
self.useFixture(fixtures.Database())
db_fixture.reset()
engine = main_db_api.get_engine()
conn = engine.connect()
result = conn.execute("select * from instance_types")
rows = result.fetchall()
@@ -152,7 +152,8 @@ class TestDatabaseFixture(testtools.TestCase):
def test_api_fixture_reset(self):
# This sets up reasonable db connection strings
self.useFixture(fixtures.ConfFixture())
self.useFixture(fixtures.Database(database='api'))
db_fixture = fixtures.Database(database='api')
self.useFixture(db_fixture)
engine = api_db_api.get_engine()
conn = engine.connect()
result = conn.execute("select * from cell_mappings")
@@ -166,12 +167,11 @@ class TestDatabaseFixture(testtools.TestCase):
rows = result.fetchall()
self.assertEqual(1, len(rows), "Rows %s" % rows)
# reset by invoking the fixture again
#
# NOTE(sdague): it's important to reestablish the db
# connection because otherwise we have a reference to the old
# in mem db.
self.useFixture(fixtures.Database(database='api'))
db_fixture.reset()
engine = api_db_api.get_engine()
conn = engine.connect()
result = conn.execute("select * from cell_mappings")
rows = result.fetchall()