summaryrefslogtreecommitdiff
path: root/oslo_db/tests/sqlalchemy/test_fixtures.py
diff options
context:
space:
mode:
Diffstat (limited to 'oslo_db/tests/sqlalchemy/test_fixtures.py')
-rw-r--r--oslo_db/tests/sqlalchemy/test_fixtures.py125
1 files changed, 120 insertions, 5 deletions
diff --git a/oslo_db/tests/sqlalchemy/test_fixtures.py b/oslo_db/tests/sqlalchemy/test_fixtures.py
index 5d63d69..08723b4 100644
--- a/oslo_db/tests/sqlalchemy/test_fixtures.py
+++ b/oslo_db/tests/sqlalchemy/test_fixtures.py
@@ -11,9 +11,12 @@
# under the License.
import mock
+import testresources
+from oslo_db.sqlalchemy import enginefacade
from oslo_db.sqlalchemy import provision
-from oslo_db.sqlalchemy import test_base
+from oslo_db.sqlalchemy import test_base as legacy_test_base
+from oslo_db.sqlalchemy import test_fixtures
from oslotest import base as oslo_test_base
@@ -21,10 +24,12 @@ class BackendSkipTest(oslo_test_base.BaseTestCase):
def test_skip_no_dbapi(self):
- class FakeDatabaseOpportunisticFixture(test_base.DbFixture):
+ class FakeDatabaseOpportunisticFixture(
+ test_fixtures.OpportunisticDbFixture):
DRIVER = 'postgresql'
- class SomeTest(test_base.DbTestCase):
+ class SomeTest(test_fixtures.OpportunisticDBTestMixin,
+ oslo_test_base.BaseTestCase):
FIXTURE = FakeDatabaseOpportunisticFixture
def runTest(self):
@@ -61,10 +66,79 @@ class BackendSkipTest(oslo_test_base.BaseTestCase):
def test_skip_no_such_backend(self):
- class FakeDatabaseOpportunisticFixture(test_base.DbFixture):
+ class FakeDatabaseOpportunisticFixture(
+ test_fixtures.OpportunisticDbFixture):
DRIVER = 'postgresql+nosuchdbapi'
- class SomeTest(test_base.DbTestCase):
+ class SomeTest(test_fixtures.OpportunisticDBTestMixin,
+ oslo_test_base.BaseTestCase):
+
+ 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)
+ )
+
+ 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):
@@ -81,3 +155,44 @@ class BackendSkipTest(oslo_test_base.BaseTestCase):
"Backend 'postgresql+nosuchdbapi' is unavailable: No such backend",
str(ex)
)
+
+
+class EnginefacadeIntegrationTest(oslo_test_base.BaseTestCase):
+ def test_db_fixture(self):
+ normal_mgr = enginefacade.transaction_context()
+ normal_mgr.configure(
+ connection="sqlite://",
+ sqlite_fk=True,
+ mysql_sql_mode="FOOBAR",
+ max_overflow=38
+ )
+
+ class MyFixture(test_fixtures.OpportunisticDbFixture):
+ def get_enginefacade(self):
+ return normal_mgr
+
+ test = mock.Mock(SCHEMA_SCOPE=None)
+ fixture = MyFixture(test=test)
+ resources = fixture._get_resources()
+
+ testresources.setUpResources(test, resources, None)
+ self.addCleanup(
+ testresources.tearDownResources,
+ test, resources, None
+ )
+ fixture.setUp()
+ self.addCleanup(fixture.cleanUp)
+
+ self.assertTrue(normal_mgr._factory._started)
+
+ test.engine = normal_mgr.writer.get_engine()
+ self.assertEqual("sqlite://", str(test.engine.url))
+ self.assertIs(test.engine, normal_mgr._factory._writer_engine)
+ engine_args = normal_mgr._factory._engine_args_for_conf(None)
+ self.assertTrue(engine_args['sqlite_fk'])
+ self.assertEqual("FOOBAR", engine_args["mysql_sql_mode"])
+ self.assertEqual(38, engine_args["max_overflow"])
+
+ fixture.cleanUp()
+ fixture._clear_cleanups() # so the real cleanUp works
+ self.assertFalse(normal_mgr._factory._started)