openstack
/
oslo.db
Code Issues Proposed changes

Remove legacy base test classes 

The fixtures have been available for a very long time now. We can drop
these.

Change-Id: I1e71150ba6daeba464b6ed8d46163f1f34959db3
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2023-02-17 12:54:24 +00:00 committed by Takashi Kajinami
parent 94d6e24ca1
commit a609333c54
5 changed files with 1 additions and 344 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
oslo_db/exception.py
View File

@ -43,11 +43,9 @@ with `try/except` statement. This is required for consistent handling of
database errors.
"""
from debtcollector import moves
from oslo_utils.excutils import CausedByException
from oslo_db._i18n import _
from oslo_db import warning
class DBError(CausedByException):
@ -287,12 +285,3 @@ class ContextNotRequestedError(AttributeError):
class CantStartEngineError(Exception):
"""Error raised when the enginefacade cannot start up correctly."""
moves.moved_class(warning.NotSupportedWarning,
'NotSupportedWarning',
__name__, version='Stein')
moves.moved_class(warning.OsloDBDeprecationWarning,
'OsloDBDeprecationWarning',
__name__, version='Stein')

204
oslo_db/sqlalchemy/test_base.py
View File

@ -15,192 +15,8 @@
import functools
import debtcollector
import debtcollector.moves
import fixtures
import testresources
try:
from oslotest import base as test_base
except ImportError:
raise NameError('Oslotest is not installed. Please add oslotest in your'
' test-requirements')
from oslo_utils import reflection
from oslo_db import exception
from oslo_db.sqlalchemy import enginefacade
from oslo_db.sqlalchemy import provision
from oslo_db.sqlalchemy import session
@debtcollector.removals.removed_class(
"DbFixture",
message="Please use oslo_db.sqlalchemy.test_fixtures directly")
class DbFixture(fixtures.Fixture):
"""Basic database fixture.
Allows to run tests on various db backends, such as SQLite, MySQL and
PostgreSQL. By default use sqlite backend. To override default backend
uri set env variable OS_TEST_DBAPI_ADMIN_CONNECTION with database admin
credentials for specific backend.
"""
DRIVER = "sqlite"
# these names are deprecated, and are not used by DbFixture.
# they are here for backwards compatibility with test suites that
# are referring to them directly.
DBNAME = PASSWORD = USERNAME = 'openstack_citest'
def __init__(self, test, skip_on_unavailable_db=True):
super(DbFixture, self).__init__()
self.test = test
self.skip_on_unavailable_db = skip_on_unavailable_db
def setUp(self):
super(DbFixture, self).setUp()
testresources.setUpResources(
self.test, self.test.resources, testresources._get_result())
self.addCleanup(
testresources.tearDownResources,
self.test, self.test.resources, testresources._get_result()
)
if not self.test._has_db_resource():
msg = self.test._get_db_resource_not_available_reason()
if self.test.SKIP_ON_UNAVAILABLE_DB:
self.test.skipTest(msg)
else:
self.test.fail(msg)
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.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, 'engine', None)
self.test.enginefacade = enginefacade._TestTransactionFactory(
self.test.engine, self.test.sessionmaker, apply_global=True)
self.addCleanup(self.test.enginefacade.dispose_global)
@debtcollector.removals.removed_class(
"DbTestCase",
message="Please use oslo_db.sqlalchemy.test_fixtures directly")
class DbTestCase(test_base.BaseTestCase):
"""Base class for testing of DB code.
"""
FIXTURE = DbFixture
SCHEMA_SCOPE = None
SKIP_ON_UNAVAILABLE_DB = True
_db_not_available = {}
_schema_resources = {}
_database_resources = {}
def _get_db_resource_not_available_reason(self):
return self._db_not_available.get(self.FIXTURE.DRIVER, None)
def _has_db_resource(self):
return self._database_resources.get(
self.FIXTURE.DRIVER, None) is not None
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,
provision_new_database=True)
except exception.BackendNotAvailable as bne:
self._database_resources[driver] = None
self._db_not_available[driver] = str(bne)
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, skip_on_unavailable_db=self.SKIP_ON_UNAVAILABLE_DB))
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().")
@debtcollector.removals.removed_class("OpportunisticTestCase")
class OpportunisticTestCase(DbTestCase):
"""Placeholder for backwards compatibility."""
ALLOWED_DIALECTS = ['sqlite', 'mysql', 'postgresql']
@ -226,23 +42,3 @@ def backend_specific(*dialects):
return f(self)
return ins_wrap
return wrap
@debtcollector.removals.removed_class("MySQLOpportunisticFixture")
class MySQLOpportunisticFixture(DbFixture):
DRIVER = 'mysql'
@debtcollector.removals.removed_class("PostgreSQLOpportunisticFixture")
class PostgreSQLOpportunisticFixture(DbFixture):
DRIVER = 'postgresql'
@debtcollector.removals.removed_class("MySQLOpportunisticTestCase")
class MySQLOpportunisticTestCase(OpportunisticTestCase):
FIXTURE = MySQLOpportunisticFixture
@debtcollector.removals.removed_class("PostgreSQLOpportunisticTestCase")
class PostgreSQLOpportunisticTestCase(OpportunisticTestCase):
FIXTURE = PostgreSQLOpportunisticFixture

31
oslo_db/tests/sqlalchemy/base.py
View File

@ -13,8 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import debtcollector
from oslo_db.sqlalchemy import enginefacade
from oslo_db.sqlalchemy.test_base import backend_specific # noqa
from oslo_db.sqlalchemy import test_fixtures as db_fixtures
@ -29,35 +27,6 @@ class Context(object):
context = Context()
@debtcollector.removals.removed_class(
"DbTestCase",
message="Do not import from oslo_db.tests! "
"Please use oslo_db.sqlalchemy.test_fixtures directly")
class DbTestCase(db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase):
def setUp(self):
super(DbTestCase, self).setUp()
self.engine = enginefacade.writer.get_engine()
self.sessionmaker = enginefacade.writer.get_sessionmaker()
@debtcollector.removals.removed_class(
"MySQLOpportunisticTestCase",
message="Do not import from oslo_db.tests! "
"Please use oslo_db.sqlalchemy.test_fixtures directly")
class MySQLOpportunisticTestCase(DbTestCase):
FIXTURE = db_fixtures.MySQLOpportunisticFixture
@debtcollector.removals.removed_class(
"PostgreSQLOpportunisticTestCase",
message="Do not import from oslo_db.tests! "
"Please use oslo_db.sqlalchemy.test_fixtures directly")
class PostgreSQLOpportunisticTestCase(DbTestCase):
FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
# NOTE (zzzeek) These test classes are **private to oslo.db**. Please
# make use of oslo_db.sqlalchemy.test_fixtures directly.

97
oslo_db/tests/sqlalchemy/test_fixtures.py
View File

@ -16,10 +16,8 @@ import testscenarios
import unittest
from unittest import mock
from oslo_db import exception
from oslo_db.sqlalchemy import enginefacade
from oslo_db.sqlalchemy import provision
from oslo_db.sqlalchemy import test_base as legacy_test_base
from oslo_db.sqlalchemy import test_fixtures
from oslo_db.tests import base as test_base
@ -96,74 +94,6 @@ class BackendSkipTest(test_base.BaseTestCase):
str(ex)
)
def test_skip_no_dbapi_legacy(self):
class FakeDatabaseOpportunisticFixture(
legacy_test_base.DbFixture,
):
DRIVER = 'postgresql'
class SomeTest(legacy_test_base.DbTestCase):
FIXTURE = FakeDatabaseOpportunisticFixture
def runTest(self):
pass
st = SomeTest()
# patch in replacement lookup dictionaries to avoid
# leaking from/to other tests
with mock.patch(
"oslo_db.sqlalchemy.provision."
"Backend.backends_by_database_type", {
"postgresql":
provision.Backend("postgresql", "postgresql://")}):
st._database_resources = {}
st._db_not_available = {}
st._schema_resources = {}
with mock.patch(
"sqlalchemy.create_engine",
mock.Mock(side_effect=ImportError())):
self.assertEqual([], st.resources)
ex = self.assertRaises(
self.skipException,
st.setUp
)
self.assertEqual(
"Backend 'postgresql' is unavailable: No DBAPI installed",
str(ex)
)
def test_skip_no_such_backend_legacy(self):
class FakeDatabaseOpportunisticFixture(
legacy_test_base.DbFixture,
):
DRIVER = 'postgresql+nosuchdbapi'
class SomeTest(legacy_test_base.DbTestCase):
FIXTURE = FakeDatabaseOpportunisticFixture
def runTest(self):
pass
st = SomeTest()
ex = self.assertRaises(
self.skipException,
st.setUp
)
self.assertEqual(
"Backend 'postgresql+nosuchdbapi' is unavailable: No such backend",
str(ex)
)
class EnginefacadeIntegrationTest(test_base.BaseTestCase):
def test_db_fixture(self):
@ -206,33 +136,6 @@ class EnginefacadeIntegrationTest(test_base.BaseTestCase):
self.assertFalse(normal_mgr._factory._started)
class LegacyBaseClassTest(test_base.BaseTestCase):
def test_new_db_is_provisioned_by_default_pg(self):
self._test_new_db_is_provisioned_by_default(
legacy_test_base.PostgreSQLOpportunisticTestCase
)
def test_new_db_is_provisioned_by_default_mysql(self):
self._test_new_db_is_provisioned_by_default(
legacy_test_base.MySQLOpportunisticTestCase
)
def _test_new_db_is_provisioned_by_default(self, base_cls):
try:
provision.DatabaseResource(base_cls.FIXTURE.DRIVER)
except exception.BackendNotAvailable:
self.skipTest("Backend %s is not available" %
base_cls.FIXTURE.DRIVER)
class SomeTest(base_cls):
def runTest(self):
pass
st = SomeTest()
db_resource = dict(st.resources)['db']
self.assertTrue(db_resource.provision_new_database)
class TestLoadHook(unittest.TestCase):
"""Test the 'load_tests' hook supplied by test_base.

2
oslo_db/tests/sqlalchemy/test_sqlalchemy.py
View File

@ -205,7 +205,7 @@ class ProgrammingError(Exception):
pass
class QueryParamTest(db_test_base.DbTestCase):
class QueryParamTest(db_test_base._DbTestCase):
def _fixture(self):
from sqlalchemy import create_engine