summaryrefslogtreecommitdiff
path: root/oslo_db/tests
diff options
context:
space:
mode:
Diffstat (limited to 'oslo_db/tests')
-rw-r--r--oslo_db/tests/sqlalchemy/base.py31
-rw-r--r--oslo_db/tests/sqlalchemy/test_exc_filters.py11
-rw-r--r--oslo_db/tests/sqlalchemy/test_fixtures.py97
-rw-r--r--oslo_db/tests/sqlalchemy/test_sqlalchemy.py2
-rw-r--r--oslo_db/tests/sqlalchemy/test_utils.py18
-rw-r--r--oslo_db/tests/test_api.py6
6 files changed, 6 insertions, 159 deletions
diff --git a/oslo_db/tests/sqlalchemy/base.py b/oslo_db/tests/sqlalchemy/base.py
index 80d9812..c66618a 100644
--- a/oslo_db/tests/sqlalchemy/base.py
+++ b/oslo_db/tests/sqlalchemy/base.py
@@ -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.
diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py
index 796ba6c..816ef7e 100644
--- a/oslo_db/tests/sqlalchemy/test_exc_filters.py
+++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py
@@ -416,16 +416,7 @@ class TestNonExistentDatabase(
super(TestNonExistentDatabase, self).setUp()
url = utils.make_url(self.engine.url)
-
- # TODO(zzzeek): remove hasattr() conditional in favor of "url.set()"
- # when SQLAlchemy 1.4 is the minimum version in requirements
- if hasattr(url, "set"):
- self.url = url.set(database="non_existent_database")
- else:
- # TODO(zzzeek): remove when SQLAlchemy 1.4
- # is the minimum version in requirements
- url.database = 'non_existent_database'
- self.url = url
+ self.url = url.set(database="non_existent_database")
def test_raise(self):
matched = self.assertRaises(
diff --git a/oslo_db/tests/sqlalchemy/test_fixtures.py b/oslo_db/tests/sqlalchemy/test_fixtures.py
index 72f4f93..eff5c7e 100644
--- a/oslo_db/tests/sqlalchemy/test_fixtures.py
+++ b/oslo_db/tests/sqlalchemy/test_fixtures.py
@@ -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.
diff --git a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py
index 5f7b27f..aa66ae5 100644
--- a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py
+++ b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py
@@ -206,7 +206,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
diff --git a/oslo_db/tests/sqlalchemy/test_utils.py b/oslo_db/tests/sqlalchemy/test_utils.py
index 059d015..94a52eb 100644
--- a/oslo_db/tests/sqlalchemy/test_utils.py
+++ b/oslo_db/tests/sqlalchemy/test_utils.py
@@ -19,7 +19,6 @@ from urllib import parse
import fixtures
import sqlalchemy
from sqlalchemy import Boolean, Index, Integer, DateTime, String
-from sqlalchemy import CheckConstraint
from sqlalchemy import MetaData, Table, Column
from sqlalchemy import ForeignKey, ForeignKeyConstraint
from sqlalchemy.dialects.postgresql import psycopg2
@@ -801,23 +800,6 @@ class TestMigrationUtils(db_test_base._DbTestCase):
for value in soft_deleted_values:
self.assertIn(value['id'], deleted_rows_ids)
- def test_detect_boolean_deleted_constraint_detection(self):
- table_name = 'abc'
- table = Table(table_name, self.meta,
- Column('id', Integer, primary_key=True),
- Column('deleted', Boolean(create_constraint=True)))
- ck = [
- const for const in table.constraints if
- isinstance(const, CheckConstraint)][0]
-
- self.assertTrue(utils._is_deleted_column_constraint(ck))
-
- self.assertFalse(
- utils._is_deleted_column_constraint(
- CheckConstraint("deleted > 5")
- )
- )
-
def test_get_foreign_key_constraint_name(self):
table_1 = Table('table_name_1', self.meta,
Column('id', Integer, primary_key=True),
diff --git a/oslo_db/tests/test_api.py b/oslo_db/tests/test_api.py
index 2540780..d45dce6 100644
--- a/oslo_db/tests/test_api.py
+++ b/oslo_db/tests/test_api.py
@@ -210,7 +210,8 @@ class DBRetryRequestCase(DBAPITestCase):
some_method()
- def test_retry_wrapper_reaches_limit(self):
+ @mock.patch('oslo_db.api.time.sleep', return_value=None)
+ def test_retry_wrapper_reaches_limit(self, mock_sleep):
max_retries = 2
@api.wrap_db_retry(max_retries=max_retries)
@@ -222,7 +223,8 @@ class DBRetryRequestCase(DBAPITestCase):
self.assertRaises(ValueError, some_method, res)
self.assertEqual(max_retries + 1, res['result'])
- def test_retry_wrapper_exception_checker(self):
+ @mock.patch('oslo_db.api.time.sleep', return_value=None)
+ def test_retry_wrapper_exception_checker(self, mock_sleep):
def exception_checker(exc):
return isinstance(exc, ValueError) and exc.args[0] < 5