# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import collections import contextlib import copy import fixtures import pickle import sys import warnings import mock from oslo_config import cfg from oslo_context import context as oslo_context from oslotest import base as oslo_test_base from sqlalchemy import Column from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy.orm import mapper from sqlalchemy.orm import Session from sqlalchemy import select from sqlalchemy import String from sqlalchemy import Table from oslo_db import exception from oslo_db import options from oslo_db.sqlalchemy import enginefacade from oslo_db.sqlalchemy import engines as oslo_engines from oslo_db.sqlalchemy import orm from oslo_db.tests.sqlalchemy import base as test_base from oslo_db import warning enginefacade.transaction_context_provider(oslo_context.RequestContext) class SingletonOnName(mock.MagicMock): def __init__(self, the_name, **kw): super(SingletonOnName, self).__init__( __eq__=lambda self, other: other._assert_name == self._assert_name, _assert_name=the_name, **kw ) def __deepcopy__(self, memo): return self class SingletonConnection(SingletonOnName): def __init__(self, **kw): super(SingletonConnection, self).__init__( "connection", **kw) self.info = {} class SingletonEngine(SingletonOnName): def __init__(self, connection, **kw): super(SingletonEngine, self).__init__( "engine", connect=mock.Mock(return_value=connection), pool=mock.Mock(), url=connection, _assert_connection=connection, **kw ) class NonDecoratedContext(object): """a Context object that's not run through transaction_context_provider.""" class AssertDataSource(collections.namedtuple( "AssertDataSource", ["writer", "reader", "async_reader"])): def element_for_writer(self, const): if const is enginefacade._WRITER: return self.writer elif const is enginefacade._READER: return self.reader elif const is enginefacade._ASYNC_READER: return self.async_reader else: assert False, "Unknown constant: %s" % const class MockFacadeTest(oslo_test_base.BaseTestCase): """test by applying mocks to internal call-points. This applies mocks to oslo.db.sqlalchemy.engines.create_engine() and oslo.db.sqlalchemy.orm.get_maker(), then mocking a _TransactionFactory into oslo.db.sqlalchemy.enginefacade._context_manager._root_factory. Various scenarios are run against the enginefacade functions, and the exact calls made against the mock create_engine(), get_maker(), and associated objects are tested exactly against expected calls. """ synchronous_reader = True engine_uri = 'some_connection' slave_uri = None def setUp(self): super(MockFacadeTest, self).setUp() writer_conn = SingletonConnection() writer_engine = SingletonEngine(writer_conn) writer_session = mock.Mock( connection=mock.Mock(return_value=writer_conn), info={}) writer_maker = mock.Mock(return_value=writer_session) if self.slave_uri: async_reader_conn = SingletonConnection() async_reader_engine = SingletonEngine(async_reader_conn) async_reader_session = mock.Mock( connection=mock.Mock(return_value=async_reader_conn), info={}) async_reader_maker = mock.Mock(return_value=async_reader_session) else: async_reader_conn = writer_conn async_reader_engine = writer_engine async_reader_session = writer_session async_reader_maker = writer_maker if self.synchronous_reader: reader_conn = writer_conn reader_engine = writer_engine reader_session = writer_session reader_maker = writer_maker else: reader_conn = async_reader_conn reader_engine = async_reader_engine reader_session = async_reader_session reader_maker = async_reader_maker self.connections = AssertDataSource( writer_conn, reader_conn, async_reader_conn ) self.engines = AssertDataSource( writer_engine, reader_engine, async_reader_engine ) self.sessions = AssertDataSource( writer_session, reader_session, async_reader_session ) self.makers = AssertDataSource( writer_maker, reader_maker, async_reader_maker ) def get_maker(engine, **kw): if engine is writer_engine: return self.makers.writer elif engine is reader_engine: return self.makers.reader elif engine is async_reader_engine: return self.makers.async_reader else: assert False session_patch = mock.patch.object( orm, "get_maker", side_effect=get_maker) self.get_maker = session_patch.start() self.addCleanup(session_patch.stop) def create_engine(sql_connection, **kw): if sql_connection == self.engine_uri: return self.engines.writer elif sql_connection == self.slave_uri: return self.engines.async_reader else: assert False engine_patch = mock.patch.object( oslo_engines, "create_engine", side_effect=create_engine) self.create_engine = engine_patch.start() self.addCleanup(engine_patch.stop) self.factory = enginefacade._TransactionFactory() self.factory.configure( synchronous_reader=self.synchronous_reader ) self.factory.configure( connection=self.engine_uri, slave_connection=self.slave_uri ) facade_patcher = mock.patch.object( enginefacade._context_manager, "_root_factory", self.factory) facade_patcher.start() self.addCleanup(facade_patcher.stop) def _assert_ctx_connection(self, context, connection): self.assertIs(context.connection, connection) def _assert_ctx_session(self, context, session): self.assertIs(context.session, session) def _assert_non_decorated_ctx_connection(self, context, connection): transaction_ctx = enginefacade._transaction_ctx_for_context(context) self.assertIs(transaction_ctx.connection, connection) def _assert_non_decorated_ctx_session(self, context, session): transaction_ctx = enginefacade._transaction_ctx_for_context(context) self.assertIs(transaction_ctx.session, session) @contextlib.contextmanager def _assert_engines(self): """produce a mock series of engine calls. These are expected to match engine-related calls established by the test subject. """ writer_conn = SingletonConnection() writer_engine = SingletonEngine(writer_conn) if self.slave_uri: async_reader_conn = SingletonConnection() async_reader_engine = SingletonEngine(async_reader_conn) else: async_reader_conn = writer_conn async_reader_engine = writer_engine if self.synchronous_reader: reader_engine = writer_engine else: reader_engine = async_reader_engine engines = AssertDataSource( writer_engine, reader_engine, async_reader_engine) def create_engine(sql_connection, **kw): if sql_connection == self.engine_uri: return engines.writer elif sql_connection == self.slave_uri: return engines.async_reader else: assert False engine_factory = mock.Mock(side_effect=create_engine) engine_factory( sql_connection=self.engine_uri, **dict((k, mock.ANY) for k in self.factory._engine_cfg.keys()) ) if self.slave_uri: engine_factory( sql_connection=self.slave_uri, **dict((k, mock.ANY) for k in self.factory._engine_cfg.keys()) ) yield AssertDataSource( writer_engine, reader_engine, async_reader_engine ) self.assertEqual( engine_factory.mock_calls, self.create_engine.mock_calls ) for sym in [ enginefacade._WRITER, enginefacade._READER, enginefacade._ASYNC_READER ]: self.assertEqual( engines.element_for_writer(sym).mock_calls, self.engines.element_for_writer(sym).mock_calls ) def _assert_async_reader_connection(self, engines, session=None): return self._assert_connection( engines, enginefacade._ASYNC_READER, session) def _assert_reader_connection(self, engines, session=None): return self._assert_connection(engines, enginefacade._READER, session) def _assert_writer_connection(self, engines, session=None): return self._assert_connection(engines, enginefacade._WRITER, session) @contextlib.contextmanager def _assert_connection(self, engines, writer, session=None): """produce a mock series of connection calls. These are expected to match connection-related calls established by the test subject. """ if session: connection = session.connection() yield connection else: connection = engines.element_for_writer(writer).connect() trans = connection.begin() yield connection if writer is enginefacade._WRITER: trans.commit() else: trans.rollback() connection.close() self.assertEqual( connection.mock_calls, self.connections.element_for_writer(writer).mock_calls) @contextlib.contextmanager def _assert_makers(self, engines): writer_session = mock.Mock(connection=mock.Mock( return_value=engines.writer._assert_connection) ) writer_maker = mock.Mock(return_value=writer_session) if self.slave_uri: async_reader_session = mock.Mock(connection=mock.Mock( return_value=engines.async_reader._assert_connection) ) async_reader_maker = mock.Mock(return_value=async_reader_session) else: async_reader_session = writer_session async_reader_maker = writer_maker if self.synchronous_reader: reader_maker = writer_maker else: reader_maker = async_reader_maker makers = AssertDataSource( writer_maker, reader_maker, async_reader_maker, ) def get_maker(engine, **kw): if engine is engines.writer: return makers.writer elif engine is engines.reader: return makers.reader elif engine is engines.async_reader: return makers.async_reader else: assert False maker_factories = mock.Mock(side_effect=get_maker) maker_factories( autocommit=True, engine=engines.writer, expire_on_commit=False) if self.slave_uri: maker_factories( autocommit=True, engine=engines.async_reader, expire_on_commit=False) yield makers self.assertEqual( maker_factories.mock_calls, self.get_maker.mock_calls) for sym in [ enginefacade._WRITER, enginefacade._READER, enginefacade._ASYNC_READER ]: self.assertEqual( makers.element_for_writer(sym).mock_calls, self.makers.element_for_writer(sym).mock_calls) def _assert_async_reader_session( self, makers, connection=None, assert_calls=True): return self._assert_session( makers, enginefacade._ASYNC_READER, connection, assert_calls) def _assert_reader_session( self, makers, connection=None, assert_calls=True): return self._assert_session( makers, enginefacade._READER, connection, assert_calls) def _assert_writer_session( self, makers, connection=None, assert_calls=True): return self._assert_session( makers, enginefacade._WRITER, connection, assert_calls) def _emit_sub_writer_session(self, session): return self._emit_sub_session(enginefacade._WRITER, session) def _emit_sub_reader_session(self, session): return self._emit_sub_session(enginefacade._READER, session) @contextlib.contextmanager def _assert_session( self, makers, writer, connection=None, assert_calls=True): """produce a mock series of session calls. These are expected to match session-related calls established by the test subject. """ if connection: session = makers.element_for_writer(writer)(bind=connection) else: session = makers.element_for_writer(writer)() session.begin() yield session if writer is enginefacade._WRITER: session.commit() elif enginefacade.\ _context_manager._factory._transaction_ctx_cfg[ 'rollback_reader_sessions']: session.rollback() session.close() if assert_calls: self.assertEqual( session.mock_calls, self.sessions.element_for_writer(writer).mock_calls) @contextlib.contextmanager def _emit_sub_session(self, writer, session): yield session if enginefacade._context_manager.\ _factory._transaction_ctx_cfg['flush_on_subtransaction']: session.flush() def test_dispose_pool(self): facade = enginefacade.transaction_context() facade.configure( connection=self.engine_uri, ) facade.dispose_pool() self.assertFalse(hasattr(facade._factory, '_writer_engine')) facade._factory._start() facade.dispose_pool() self.assertEqual( facade._factory._writer_engine.pool.mock_calls, [mock.call.dispose()] ) def test_dispose_pool_w_reader(self): facade = enginefacade.transaction_context() facade.configure( connection=self.engine_uri, slave_connection=self.slave_uri ) facade.dispose_pool() self.assertFalse(hasattr(facade._factory, '_writer_engine')) self.assertFalse(hasattr(facade._factory, '_reader_engine')) facade._factory._start() facade.dispose_pool() self.assertEqual( facade._factory._writer_engine.pool.mock_calls, [mock.call.dispose()] ) self.assertEqual( facade._factory._reader_engine.pool.mock_calls, [mock.call.dispose()] ) def test_session_reader_decorator(self): context = oslo_context.RequestContext() @enginefacade.reader def go(context): context.session.execute("test") go(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test") def test_connection_reader_decorator(self): context = oslo_context.RequestContext() @enginefacade.reader.connection def go(context): context.connection.execute("test") go(context) with self._assert_engines() as engines: with self._assert_reader_connection(engines) as connection: connection.execute("test") def test_session_reader_nested_in_connection_reader(self): context = oslo_context.RequestContext() @enginefacade.reader.connection def go1(context): context.connection.execute("test1") go2(context) @enginefacade.reader def go2(context): context.session.execute("test2") go1(context) with self._assert_engines() as engines: with self._assert_reader_connection(engines) as connection: connection.execute("test1") with self._assert_makers(engines) as makers: with self._assert_reader_session( makers, connection) as session: session.execute("test2") def test_connection_reader_nested_in_session_reader(self): context = oslo_context.RequestContext() @enginefacade.reader def go1(context): context.session.execute("test1") go2(context) @enginefacade.reader.connection def go2(context): context.connection.execute("test2") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test1") with self._assert_reader_connection( engines, session) as connection: connection.execute("test2") def test_session_reader_decorator_nested(self): context = oslo_context.RequestContext() @enginefacade.reader def go1(context): context.session.execute("test1") go2(context) @enginefacade.reader def go2(context): context.session.execute("test2") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test1") session.execute("test2") def test_reader_nested_in_writer_ok(self): context = oslo_context.RequestContext() @enginefacade.writer def go1(context): context.session.execute("test1") go2(context) @enginefacade.reader def go2(context): context.session.execute("test2") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_writer_session(makers) as session: session.execute("test1") session.execute("test2") def test_writer_nested_in_reader_raises(self): context = oslo_context.RequestContext() @enginefacade.reader def go1(context): context.session.execute("test1") go2(context) @enginefacade.writer def go2(context): context.session.execute("test2") exc = self.assertRaises( TypeError, go1, context ) self.assertEqual( "Can't upgrade a READER " "transaction to a WRITER mid-transaction", exc.args[0] ) def test_async_on_writer_raises(self): exc = self.assertRaises( TypeError, getattr, enginefacade.writer, "async_" ) self.assertEqual( "Setting async on a WRITER makes no sense", exc.args[0] ) def test_savepoint_and_independent_raises(self): exc = self.assertRaises( TypeError, getattr, enginefacade.writer.independent, "savepoint" ) self.assertEqual( "setting savepoint and independent makes no sense.", exc.args[0] ) def test_reader_nested_in_async_reader_raises(self): context = oslo_context.RequestContext() @enginefacade.reader.async_ def go1(context): context.session.execute("test1") go2(context) @enginefacade.reader def go2(context): context.session.execute("test2") exc = self.assertRaises( TypeError, go1, context ) self.assertEqual( "Can't upgrade an ASYNC_READER transaction " "to a READER mid-transaction", exc.args[0] ) def test_reader_allow_async_nested_in_async_reader(self): context = oslo_context.RequestContext() @enginefacade.reader.async_ def go1(context): context.session.execute("test1") go2(context) @enginefacade.reader.allow_async def go2(context): context.session.execute("test2") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_async_reader_session(makers) as session: session.execute("test1") session.execute("test2") def test_reader_allow_async_nested_in_reader(self): context = oslo_context.RequestContext() @enginefacade.reader.reader def go1(context): context.session.execute("test1") go2(context) @enginefacade.reader.allow_async def go2(context): context.session.execute("test2") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test1") session.execute("test2") def test_reader_allow_async_is_reader_by_default(self): context = oslo_context.RequestContext() @enginefacade.reader.allow_async def go1(context): context.session.execute("test1") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test1") def test_writer_nested_in_async_reader_raises(self): context = oslo_context.RequestContext() @enginefacade.reader.async_ def go1(context): context.session.execute("test1") go2(context) @enginefacade.writer def go2(context): context.session.execute("test2") exc = self.assertRaises( TypeError, go1, context ) self.assertEqual( "Can't upgrade an ASYNC_READER transaction to a " "WRITER mid-transaction", exc.args[0] ) def test_reader_then_writer_ok(self): context = oslo_context.RequestContext() @enginefacade.reader def go1(context): context.session.execute("test1") @enginefacade.writer def go2(context): context.session.execute("test2") go1(context) go2(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session( makers, assert_calls=False) as session: session.execute("test1") with self._assert_writer_session(makers) as session: session.execute("test2") def test_deprecated_async_reader_name(self): if sys.version_info >= (3, 7): self.skipTest("Test only runs on Python < 3.7") context = oslo_context.RequestContext() old = getattr(enginefacade.reader, "async") @old def go1(context): context.session.execute("test1") go1(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_async_reader_session( makers, assert_calls=False) as session: session.execute("test1") def test_async_reader_then_reader_ok(self): context = oslo_context.RequestContext() @enginefacade.reader.async_ def go1(context): context.session.execute("test1") @enginefacade.reader def go2(context): context.session.execute("test2") go1(context) go2(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_async_reader_session( makers, assert_calls=False) as session: session.execute("test1") with self._assert_reader_session(makers) as session: session.execute("test2") def test_using_reader(self): context = oslo_context.RequestContext() with enginefacade.reader.using(context) as session: self._assert_ctx_session(context, session) session.execute("test1") with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test1") def test_using_context_present_in_session_info(self): context = oslo_context.RequestContext() with enginefacade.reader.using(context) as session: self.assertEqual(context, session.info['using_context']) self.assertIsNone(session.info['using_context']) def test_using_context_present_in_connection_info(self): context = oslo_context.RequestContext() with enginefacade.writer.connection.using(context) as connection: self.assertEqual(context, connection.info['using_context']) self.assertIsNone(connection.info['using_context']) def test_using_reader_rollback_reader_session(self): enginefacade.configure(rollback_reader_sessions=True) context = oslo_context.RequestContext() with enginefacade.reader.using(context) as session: self._assert_ctx_session(context, session) session.execute("test1") with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test1") def test_using_flush_on_nested(self): enginefacade.configure(flush_on_nested=True) context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: with enginefacade.writer.using(context) as session: self._assert_ctx_session(context, session) session.execute("test1") with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_writer_session(makers) as session: with self._emit_sub_writer_session( session) as session: session.execute("test1") def test_using_writer(self): context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: self._assert_ctx_session(context, session) session.execute("test1") with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_writer_session(makers) as session: session.execute("test1") def test_using_writer_no_descriptors(self): context = NonDecoratedContext() with enginefacade.writer.using(context) as session: self._assert_non_decorated_ctx_session(context, session) session.execute("test1") with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_writer_session(makers) as session: session.execute("test1") def test_using_writer_connection_no_descriptors(self): context = NonDecoratedContext() with enginefacade.writer.connection.using(context) as connection: self._assert_non_decorated_ctx_connection(context, connection) connection.execute("test1") with self._assert_engines() as engines: with self._assert_writer_connection(engines) as conn: conn.execute("test1") def test_using_reader_connection(self): context = oslo_context.RequestContext() with enginefacade.reader.connection.using(context) as connection: self._assert_ctx_connection(context, connection) connection.execute("test1") with self._assert_engines() as engines: with self._assert_reader_connection(engines) as conn: conn.execute("test1") def test_using_writer_connection(self): context = oslo_context.RequestContext() with enginefacade.writer.connection.using(context) as connection: self._assert_ctx_connection(context, connection) connection.execute("test1") with self._assert_engines() as engines: with self._assert_writer_connection(engines) as conn: conn.execute("test1") def test_context_copied_using_existing_writer_connection(self): context = oslo_context.RequestContext() with enginefacade.writer.connection.using(context) as connection: self._assert_ctx_connection(context, connection) connection.execute("test1") ctx2 = copy.deepcopy(context) with enginefacade.reader.connection.using(ctx2) as conn2: self.assertIs(conn2, connection) self._assert_ctx_connection(ctx2, conn2) conn2.execute("test2") with self._assert_engines() as engines: with self._assert_writer_connection(engines) as conn: conn.execute("test1") conn.execute("test2") def test_context_nodesc_copied_using_existing_writer_connection(self): context = NonDecoratedContext() with enginefacade.writer.connection.using(context) as connection: self._assert_non_decorated_ctx_connection(context, connection) connection.execute("test1") ctx2 = copy.deepcopy(context) with enginefacade.reader.connection.using(ctx2) as conn2: self.assertIs(conn2, connection) self._assert_non_decorated_ctx_connection(ctx2, conn2) conn2.execute("test2") with self._assert_engines() as engines: with self._assert_writer_connection(engines) as conn: conn.execute("test1") conn.execute("test2") def test_session_context_notrequested_exception(self): context = oslo_context.RequestContext() with enginefacade.reader.connection.using(context): exc = self.assertRaises( exception.ContextNotRequestedError, getattr, context, 'session' ) self.assertRegex( exc.args[0], "The 'session' context attribute was requested but it has " "not been established for this context." ) def test_connection_context_notrequested_exception(self): context = oslo_context.RequestContext() with enginefacade.reader.using(context): exc = self.assertRaises( exception.ContextNotRequestedError, getattr, context, 'connection' ) self.assertRegex( exc.args[0], "The 'connection' context attribute was requested but it has " "not been established for this context." ) def test_session_context_exception(self): context = oslo_context.RequestContext() exc = self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'session' ) self.assertRegex( exc.args[0], "No TransactionContext is established for " "this .*RequestContext.* object within the current " "thread; the 'session' attribute is unavailable." ) def test_session_context_getattr(self): context = oslo_context.RequestContext() self.assertIsNone(getattr(context, 'session', None)) def test_connection_context_exception(self): context = oslo_context.RequestContext() exc = self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'connection' ) self.assertRegex( exc.args[0], "No TransactionContext is established for " "this .*RequestContext.* object within the current " "thread; the 'connection' attribute is unavailable." ) def test_connection_context_getattr(self): context = oslo_context.RequestContext() self.assertIsNone(getattr(context, 'connection', None)) def test_transaction_context_exception(self): context = oslo_context.RequestContext() exc = self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'transaction' ) self.assertRegex( exc.args[0], "No TransactionContext is established for " "this .*RequestContext.* object within the current " "thread; the 'transaction' attribute is unavailable." ) def test_transaction_context_getattr(self): context = oslo_context.RequestContext() self.assertIsNone(getattr(context, 'transaction', None)) def test_trans_ctx_context_exception(self): context = oslo_context.RequestContext() exc = self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'transaction_ctx' ) self.assertRegex( exc.args[0], "No TransactionContext is established for " "this .*RequestContext.* object within the current " "thread." ) def test_trans_ctx_context_getattr(self): context = oslo_context.RequestContext() self.assertIsNone(getattr(context, 'transaction_ctx', None)) def test_multiple_factories(self): """Test that the instrumentation applied to a context class is independent of a specific _TransactionContextManager / _TransactionFactory. """ mgr1 = enginefacade.transaction_context() mgr1.configure( connection=self.engine_uri, slave_connection=self.slave_uri ) mgr2 = enginefacade.transaction_context() mgr2.configure( connection=self.engine_uri, slave_connection=self.slave_uri ) context = oslo_context.RequestContext() self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'session' ) with mgr1.writer.using(context): self.assertIs(context.transaction_ctx.factory, mgr1._factory) self.assertIsNot(context.transaction_ctx.factory, mgr2._factory) self.assertIsNotNone(context.session) self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'session' ) with mgr2.writer.using(context): self.assertIsNot(context.transaction_ctx.factory, mgr1._factory) self.assertIs(context.transaction_ctx.factory, mgr2._factory) self.assertIsNotNone(context.session) def test_multiple_factories_nested(self): """Test that the instrumentation applied to a context class supports nested calls among multiple _TransactionContextManager objects. """ mgr1 = enginefacade.transaction_context() mgr1.configure( connection=self.engine_uri, slave_connection=self.slave_uri ) mgr2 = enginefacade.transaction_context() mgr2.configure( connection=self.engine_uri, slave_connection=self.slave_uri ) context = oslo_context.RequestContext() with mgr1.writer.using(context): self.assertIs(context.transaction_ctx.factory, mgr1._factory) self.assertIsNot(context.transaction_ctx.factory, mgr2._factory) with mgr2.writer.using(context): self.assertIsNot( context.transaction_ctx.factory, mgr1._factory) self.assertIs(context.transaction_ctx.factory, mgr2._factory) self.assertIsNotNone(context.session) # mgr1 is restored self.assertIs(context.transaction_ctx.factory, mgr1._factory) self.assertIsNot(context.transaction_ctx.factory, mgr2._factory) self.assertIsNotNone(context.session) self.assertRaises( exception.NoEngineContextEstablished, getattr, context, 'transaction_ctx' ) def test_context_found_for_bound_method(self): context = oslo_context.RequestContext() @enginefacade.reader def go(self, context): context.session.execute("test") go(self, context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test") def test_context_found_for_class_method(self): context = oslo_context.RequestContext() class Spam(object): @classmethod @enginefacade.reader def go(cls, context): context.session.execute("test") Spam.go(context) with self._assert_engines() as engines: with self._assert_makers(engines) as makers: with self._assert_reader_session(makers) as session: session.execute("test") class PatchFactoryTest(oslo_test_base.BaseTestCase): def test_patch_manager(self): normal_mgr = enginefacade.transaction_context() normal_mgr.configure(connection="sqlite:///foo.db") alt_mgr = enginefacade.transaction_context() alt_mgr.configure(connection="sqlite:///bar.db") @normal_mgr.writer def go1(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///foo.db") self.assertIs( s1.bind, normal_mgr._factory._writer_engine) @normal_mgr.writer def go2(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///bar.db") self.assertIs( normal_mgr._factory._writer_engine, alt_mgr._factory._writer_engine ) def create_engine(sql_connection, **kw): return mock.Mock(url=sql_connection) with mock.patch( "oslo_db.sqlalchemy.engines.create_engine", create_engine): context = oslo_context.RequestContext() go1(context) reset = normal_mgr.patch_factory(alt_mgr) go2(context) reset() go1(context) def test_patch_factory(self): normal_mgr = enginefacade.transaction_context() normal_mgr.configure(connection="sqlite:///foo.db") alt_mgr = enginefacade.transaction_context() alt_mgr.configure(connection="sqlite:///bar.db") @normal_mgr.writer def go1(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///foo.db") self.assertIs( s1.bind, normal_mgr._factory._writer_engine) @normal_mgr.writer def go2(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///bar.db") self.assertIs( normal_mgr._factory._writer_engine, alt_mgr._factory._writer_engine ) def create_engine(sql_connection, **kw): return mock.Mock(url=sql_connection) with mock.patch( "oslo_db.sqlalchemy.engines.create_engine", create_engine): context = oslo_context.RequestContext() go1(context) reset = normal_mgr.patch_factory(alt_mgr._factory) go2(context) reset() go1(context) def test_patch_engine(self): normal_mgr = enginefacade.transaction_context() normal_mgr.configure( connection="sqlite:///foo.db", rollback_reader_sessions=True ) @normal_mgr.writer def go1(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///foo.db") self.assertIs( s1.bind, normal_mgr._factory._writer_engine) @normal_mgr.writer def go2(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///bar.db") self.assertTrue( enginefacade._transaction_ctx_for_context( context).rollback_reader_sessions ) # ensure this defaults to True self.assertTrue( enginefacade._transaction_ctx_for_context( context).factory.synchronous_reader ) def create_engine(sql_connection, **kw): return mock.Mock(url=sql_connection) with mock.patch( "oslo_db.sqlalchemy.engines.create_engine", create_engine): mock_engine = create_engine("sqlite:///bar.db") context = oslo_context.RequestContext() go1(context) reset = normal_mgr.patch_engine(mock_engine) go2(context) self.assertIs( normal_mgr._factory._writer_engine, mock_engine) reset() go1(context) def test_patch_not_started(self): normal_mgr = enginefacade.transaction_context() normal_mgr.configure( connection="sqlite:///foo.db", rollback_reader_sessions=True ) @normal_mgr.writer def go1(context): s1 = context.session self.assertEqual( s1.bind.url, "sqlite:///bar.db") self.assertTrue( enginefacade._transaction_ctx_for_context( context).rollback_reader_sessions ) def create_engine(sql_connection, **kw): return mock.Mock(url=sql_connection) with mock.patch( "oslo_db.sqlalchemy.engines.create_engine", create_engine): mock_engine = create_engine("sqlite:///bar.db") context = oslo_context.RequestContext() reset = normal_mgr.patch_engine(mock_engine) go1(context) self.assertIs( normal_mgr._factory._writer_engine, mock_engine) reset() def test_new_manager_from_config(self): normal_mgr = enginefacade.transaction_context() normal_mgr.configure( connection="sqlite://", sqlite_fk=True, mysql_sql_mode="FOOBAR", max_overflow=38 ) normal_mgr._factory._start() copied_mgr = normal_mgr.make_new_manager() self.assertTrue(normal_mgr._factory._started) self.assertIsNotNone(normal_mgr._factory._writer_engine) self.assertIsNot(copied_mgr._factory, normal_mgr._factory) self.assertFalse(copied_mgr._factory._started) copied_mgr._factory._start() self.assertIsNot( normal_mgr._factory._writer_engine, copied_mgr._factory._writer_engine) engine_args = copied_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"]) class SynchronousReaderWSlaveMockFacadeTest(MockFacadeTest): synchronous_reader = True engine_uri = 'some_connection' slave_uri = 'some_slave_connection' class AsyncReaderWSlaveMockFacadeTest(MockFacadeTest): synchronous_reader = False engine_uri = 'some_connection' slave_uri = 'some_slave_connection' class LegacyIntegrationtest(test_base._DbTestCase): def test_legacy_integration(self): legacy_facade = enginefacade.get_legacy_facade() self.assertTrue( legacy_facade.get_engine() is enginefacade._context_manager._factory._writer_engine ) self.assertTrue( enginefacade.get_legacy_facade() is legacy_facade ) def test_get_sessionmaker(self): legacy_facade = enginefacade.get_legacy_facade() self.assertTrue( legacy_facade.get_sessionmaker() is enginefacade._context_manager._factory._writer_maker ) def test_legacy_facades_from_different_context_managers(self): transaction_context1 = enginefacade.transaction_context() transaction_context2 = enginefacade.transaction_context() transaction_context1.configure(connection='sqlite:///?conn1') transaction_context2.configure(connection='sqlite:///?conn2') legacy1 = transaction_context1.get_legacy_facade() legacy2 = transaction_context2.get_legacy_facade() self.assertNotEqual(legacy1, legacy2) def test_legacy_not_started(self): factory = enginefacade._TransactionFactory() self.assertRaises( exception.CantStartEngineError, factory.get_legacy_facade ) legacy_facade = factory.get_legacy_facade() self.assertRaises( exception.CantStartEngineError, legacy_facade.get_session ) self.assertRaises( exception.CantStartEngineError, legacy_facade.get_session ) self.assertRaises( exception.CantStartEngineError, legacy_facade.get_engine ) class ThreadingTest(test_base._DbTestCase): """Test copy/pickle on new threads using real connections and sessions.""" def _assert_ctx_connection(self, context, connection): self.assertIs(context.connection, connection) def _assert_ctx_session(self, context, session): self.assertIs(context.session, session) def _patch_thread_ident(self): self.ident = 1 test_instance = self class MockThreadingLocal(object): def __init__(self): self.__dict__['state'] = collections.defaultdict(dict) def __deepcopy__(self, memo): return self def __getattr__(self, key): ns = self.state[test_instance.ident] try: return ns[key] except KeyError: raise AttributeError(key) def __setattr__(self, key, value): ns = self.state[test_instance.ident] ns[key] = value def __delattr__(self, key): ns = self.state[test_instance.ident] try: del ns[key] except KeyError: raise AttributeError(key) return mock.patch.object( enginefacade, "_TransactionContextTLocal", MockThreadingLocal) def test_thread_ctxmanager_writer(self): context = oslo_context.RequestContext() with self._patch_thread_ident(): with enginefacade.writer.using(context) as session: self._assert_ctx_session(context, session) self.ident = 2 with enginefacade.reader.using(context) as sess2: # new session self.assertIsNot(sess2, session) # thread local shows the new session self._assert_ctx_session(context, sess2) self.ident = 1 with enginefacade.reader.using(context) as sess3: self.assertIs(sess3, session) self._assert_ctx_session(context, session) def test_thread_ctxmanager_writer_connection(self): context = oslo_context.RequestContext() with self._patch_thread_ident(): with enginefacade.writer.connection.using(context) as conn: self._assert_ctx_connection(context, conn) self.ident = 2 with enginefacade.reader.connection.using(context) as conn2: # new connection self.assertIsNot(conn2, conn) # thread local shows the new connection self._assert_ctx_connection(context, conn2) with enginefacade.reader.connection.using( context) as conn3: # we still get the right connection even though # this context is not the "copied" context self.assertIsNot(conn3, conn) self.assertIs(conn3, conn2) self.ident = 1 with enginefacade.reader.connection.using(context) as conn3: self.assertIs(conn3, conn) self._assert_ctx_connection(context, conn) def test_thread_ctxmanager_switch_styles(self): @enginefacade.writer.connection def go_one(context): self.assertRaises( exception.ContextNotRequestedError, getattr, context, "session" ) self.assertIsNotNone(context.connection) self.ident = 2 go_two(context) self.ident = 1 self.assertRaises( exception.ContextNotRequestedError, getattr, context, "session" ) self.assertIsNotNone(context.connection) @enginefacade.reader def go_two(context): self.assertRaises( exception.ContextNotRequestedError, getattr, context, "connection" ) self.assertIsNotNone(context.session) context = oslo_context.RequestContext() with self._patch_thread_ident(): go_one(context) def test_thread_decorator_writer(self): sessions = set() @enginefacade.writer def go_one(context): sessions.add(context.session) self.ident = 2 go_two(context) self.ident = 1 go_three(context) @enginefacade.reader def go_two(context): assert context.session not in sessions @enginefacade.reader def go_three(context): assert context.session in sessions context = oslo_context.RequestContext() with self._patch_thread_ident(): go_one(context) def test_thread_decorator_writer_connection(self): connections = set() @enginefacade.writer.connection def go_one(context): connections.add(context.connection) self.ident = 2 go_two(context) self.ident = 1 go_three(context) @enginefacade.reader.connection def go_two(context): assert context.connection not in connections @enginefacade.reader def go_three(context): assert context.connection in connections context = oslo_context.RequestContext() with self._patch_thread_ident(): go_one(context) def test_contexts_picklable(self): context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: self._assert_ctx_session(context, session) pickled = pickle.dumps(context) unpickled = pickle.loads(pickled) with enginefacade.writer.using(unpickled) as session2: self._assert_ctx_session(unpickled, session2) assert session is not session2 class LiveFacadeTest(test_base._DbTestCase): """test using live SQL with test-provisioned databases. Several of these tests require that multiple transactions run simultaenously; as the default SQLite :memory: connection can't achieve this, opportunistic test implementations against MySQL and PostgreSQL are supplied. """ def setUp(self): super(LiveFacadeTest, self).setUp() metadata = MetaData() user_table = Table( 'user', metadata, Column('id', Integer, primary_key=True), Column('name', String(30)), Column('favorite_color', String(10), default='yellow'), mysql_engine='InnoDB' ) self.user_table = user_table metadata.create_all(self.engine) self.addCleanup(metadata.drop_all, self.engine) class User(object): def __init__(self, name): self.name = name mapper(User, user_table) self.User = User def _assert_ctx_connection(self, context, connection): self.assertIs(context.connection, connection) def _assert_ctx_session(self, context, session): self.assertIs(context.session, session) def test_transaction_committed(self): context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: session.add(self.User(name="u1")) session = self.sessionmaker(autocommit=True) self.assertEqual( "u1", session.query(self.User.name).scalar() ) def test_transaction_rollback(self): context = oslo_context.RequestContext() class MyException(Exception): pass @enginefacade.writer def go(context): context.session.add(self.User(name="u1")) context.session.flush() raise MyException("a test") self.assertRaises(MyException, go, context) session = self.sessionmaker(autocommit=True) self.assertEqual( None, session.query(self.User.name).scalar() ) @mock.patch.object(Session, 'commit') @mock.patch.object(Session, 'rollback') def test_save_and_reraise_when_rollback_exception(self, rollback_patch, commit_patch): context = oslo_context.RequestContext() log = self.useFixture(fixtures.FakeLogger()) class RollbackException(Exception): pass class CommitException(Exception): pass commit_patch.side_effect = CommitException() rollback_patch.side_effect = RollbackException() @enginefacade.writer def go_session(context): context.session.add(self.User(name="u1")) self.assertRaises(RollbackException, go_session, context) self.assertIn('CommitException', log.output) def test_flush_on_subtransaction(self): facade = enginefacade.transaction_context() facade.configure( connection=self.engine.url, flush_on_subtransaction=True) facade.patch_engine(self.engine) context = oslo_context.RequestContext() with facade.writer.using(context): with facade.writer.using(context): u = self.User(name="u1") context.session.add(u) self.assertEqual('yellow', u.favorite_color) def test_flush_on_subtransaction_default_off(self): context = oslo_context.RequestContext() facade = enginefacade.transaction_context() facade.configure(connection=self.engine.url) facade.patch_engine(self.engine) with facade.writer.using(context): with facade.writer.using(context): u = self.User(name="u1") context.session.add(u) self.assertIsNone(u.favorite_color) self.assertEqual('yellow', u.favorite_color) def test_context_deepcopy_on_session(self): context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: ctx2 = copy.deepcopy(context) self._assert_ctx_session(ctx2, session) with enginefacade.writer.using(ctx2) as s2: self.assertIs(session, s2) self._assert_ctx_session(ctx2, s2) s2.add(self.User(name="u1")) s2.flush() session = self.sessionmaker(autocommit=True) self.assertEqual( "u1", session.query(self.User.name).scalar() ) def test_context_deepcopy_on_connection(self): context = oslo_context.RequestContext() with enginefacade.writer.connection.using(context) as conn: ctx2 = copy.deepcopy(context) self._assert_ctx_connection(ctx2, conn) with enginefacade.writer.connection.using(ctx2) as conn2: self.assertIs(conn, conn2) self._assert_ctx_connection(ctx2, conn2) conn2.execute(self.user_table.insert().values(name="u1")) self._assert_ctx_connection(ctx2, conn2) session = self.sessionmaker(autocommit=True) self.assertEqual( "u1", session.query(self.User.name).scalar() ) @test_base.backend_specific("postgresql", "mysql") def test_external_session_transaction(self): context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: session.add(self.User(name="u1")) session.flush() with enginefacade.writer.independent.using(context) as s2: # transaction() uses a new session self.assertIsNot(s2, session) self._assert_ctx_session(context, s2) # rows within a distinct transaction s2.add(self.User(name="u2")) # it also takes over the global enginefacade # within the context with enginefacade.writer.using(context) as s3: self.assertIs(s3, s2) s3.add(self.User(name="u3")) self._assert_ctx_session(context, session) # rollback the "outer" transaction session.rollback() # add more state on the "outer" transaction session.begin() session.add(self.User(name="u4")) session = self.sessionmaker(autocommit=True) # inner transaction + second part of "outer" transaction were committed self.assertEqual( [("u2",), ("u3",), ("u4", )], session.query( self.User.name).order_by(self.User.name).all() ) def test_savepoint_transaction_decorator(self): context = oslo_context.RequestContext() @enginefacade.writer def go1(context): session = context.session session.add(self.User(name="u1")) session.flush() try: go2(context) except Exception: pass go3(context) session.add(self.User(name="u4")) @enginefacade.writer.savepoint def go2(context): session = context.session session.add(self.User(name="u2")) raise Exception("nope") @enginefacade.writer.savepoint def go3(context): session = context.session session.add(self.User(name="u3")) go1(context) session = self.sessionmaker(autocommit=True) # inner transaction + second part of "outer" transaction were committed self.assertEqual( [("u1",), ("u3",), ("u4", )], session.query( self.User.name).order_by(self.User.name).all() ) def test_savepoint_transaction(self): context = oslo_context.RequestContext() with enginefacade.writer.using(context) as session: session.add(self.User(name="u1")) session.flush() try: with enginefacade.writer.savepoint.using(context) as session: session.add(self.User(name="u2")) raise Exception("nope") except Exception: pass with enginefacade.writer.savepoint.using(context) as session: session.add(self.User(name="u3")) session.add(self.User(name="u4")) session = self.sessionmaker(autocommit=True) # inner transaction + second part of "outer" transaction were committed self.assertEqual( [("u1",), ("u3",), ("u4", )], session.query( self.User.name).order_by(self.User.name).all() ) @test_base.backend_specific("postgresql", "mysql") def test_external_session_transaction_decorator(self): context = oslo_context.RequestContext() @enginefacade.writer def go1(context): session = context.session session.add(self.User(name="u1")) session.flush() go2(context, session) self._assert_ctx_session(context, session) # rollback the "outer" transaction session.rollback() # add more state on the "outer" transaction session.begin() session.add(self.User(name="u4")) @enginefacade.writer.independent def go2(context, session): s2 = context.session # uses a new session self.assertIsNot(s2, session) self._assert_ctx_session(context, s2) # rows within a distinct transaction s2.add(self.User(name="u2")) # it also takes over the global enginefacade # within the context with enginefacade.writer.using(context) as s3: self.assertIs(s3, s2) s3.add(self.User(name="u3")) go1(context) session = self.sessionmaker(autocommit=True) # inner transaction + second part of "outer" transaction were committed self.assertEqual( [("u2",), ("u3",), ("u4", )], session.query( self.User.name).order_by(self.User.name).all() ) @test_base.backend_specific("postgresql", "mysql") def test_external_connection_transaction(self): context = oslo_context.RequestContext() with enginefacade.writer.connection.using(context) as connection: connection.execute(self.user_table.insert().values(name="u1")) # transaction() uses a new Connection with enginefacade.writer.independent.connection.\ using(context) as c2: self.assertIsNot(c2, connection) self._assert_ctx_connection(context, c2) # rows within a distinct transaction c2.execute(self.user_table.insert().values(name="u2")) # it also takes over the global enginefacade # within the context with enginefacade.writer.connection.using(context) as c3: self.assertIs(c2, c3) c3.execute(self.user_table.insert().values(name="u3")) self._assert_ctx_connection(context, connection) # rollback the "outer" transaction transaction_ctx = context.transaction_ctx transaction_ctx.transaction.rollback() transaction_ctx.transaction = connection.begin() # add more state on the "outer" transaction connection.execute(self.user_table.insert().values(name="u4")) session = self.sessionmaker(autocommit=True) self.assertEqual( [("u2",), ("u3",), ("u4", )], session.query( self.User.name).order_by(self.User.name).all() ) @test_base.backend_specific("postgresql", "mysql") def test_external_writer_in_reader(self): context = oslo_context.RequestContext() with enginefacade.reader.using(context) as session: ping = session.scalar(select([1])) self.assertEqual(1, ping) # we're definitely a reader @enginefacade.writer def go(ctx): pass exc = self.assertRaises(TypeError, go, context) self.assertEqual( "Can't upgrade a READER transaction to a " "WRITER mid-transaction", exc.args[0]) # but we can do a writer on a new transaction with enginefacade.writer.independent.using(context) as sess2: self.assertIsNot(sess2, session) self._assert_ctx_session(context, sess2) session.add(self.User(name="u1_nocommit")) sess2.add(self.User(name="u1_commit")) user = session.query(self.User).first() self.assertEqual("u1_commit", user.name) session = self.sessionmaker(autocommit=True) self.assertEqual( [("u1_commit",)], session.query( self.User.name).order_by(self.User.name).all() ) def test_replace_scope(self): # "timeout" is an argument accepted by # the pysqlite dialect, which we set here to ensure # that even in an all-sqlite test, we test that the URL # is different in the context we are looking for alt_connection = "sqlite:///?timeout=90" alt_mgr1 = enginefacade.transaction_context() alt_mgr1.configure( connection=alt_connection, ) @enginefacade.writer def go1(context): s1 = context.session self.assertEqual( s1.bind.url, enginefacade._context_manager._factory._writer_engine.url) self.assertIs( s1.bind, enginefacade._context_manager._factory._writer_engine) self.assertEqual(s1.bind.url, self.engine.url) with alt_mgr1.replace.using(context): go2(context) go4(context) @enginefacade.writer def go2(context): s2 = context.session # factory is not replaced globally... self.assertIsNot( enginefacade._context_manager._factory._writer_engine, alt_mgr1._factory._writer_engine ) # but it is replaced for us self.assertIs(s2.bind, alt_mgr1._factory._writer_engine) self.assertEqual( str(s2.bind.url), alt_connection) go3(context) @enginefacade.reader def go3(context): s3 = context.session # in a call of a call, we still have the alt URL self.assertIs(s3.bind, alt_mgr1._factory._writer_engine) self.assertEqual( str(s3.bind.url), alt_connection) @enginefacade.writer def go4(context): s4 = context.session # outside the "replace" context, all is back to normal self.assertIs(s4.bind, self.engine) self.assertEqual( s4.bind.url, self.engine.url) context = oslo_context.RequestContext() go1(context) self.assertIsNot( enginefacade._context_manager._factory._writer_engine, alt_mgr1._factory._writer_engine ) def test_replace_scope_only_global_eng(self): # "timeout" is an argument accepted by # the pysqlite dialect, which we set here to ensure # that even in an all-sqlite test, we test that the URL # is different in the context we are looking for alt_connection1 = "sqlite:///?timeout=90" alt_mgr1 = enginefacade.transaction_context() alt_mgr1.configure( connection=alt_connection1, ) alt_connection2 = "sqlite:///?timeout=120" alt_mgr2 = enginefacade.transaction_context() alt_mgr2.configure( connection=alt_connection2, ) @enginefacade.writer def go1(context): s1 = context.session # global engine self.assertEqual(s1.bind.url, self.engine.url) # now replace global engine... with alt_mgr1.replace.using(context): go2(context) # and back go6(context) @enginefacade.writer def go2(context): s2 = context.session # we have the replace-the-global engine self.assertEqual(str(s2.bind.url), alt_connection1) self.assertIs(s2.bind, alt_mgr1._factory._writer_engine) go3(context) @alt_mgr2.writer def go3(context): s3 = context.session # we don't use the global engine in the first place. # make sure our own factory still used. self.assertEqual(str(s3.bind.url), alt_connection2) self.assertIs(s3.bind, alt_mgr2._factory._writer_engine) go4(context) @enginefacade.writer def go4(context): s4 = context.session # we *do* use the global, so we still want the replacement. self.assertEqual(str(s4.bind.url), alt_connection1) self.assertIs(s4.bind, alt_mgr1._factory._writer_engine) @enginefacade.writer def go5(context): s5 = context.session # ...and here also self.assertEqual(str(s5.bind.url), alt_connection1) self.assertIs(s5.bind, alt_mgr1._factory._writer_engine) @enginefacade.writer def go6(context): s6 = context.session # ...but not here! self.assertEqual(str(s6.bind.url), str(self.engine.url)) self.assertIs(s6.bind, self.engine) context = oslo_context.RequestContext() go1(context) class MySQLLiveFacadeTest( test_base._MySQLOpportunisticTestCase, LiveFacadeTest): pass class PGLiveFacadeTest( test_base._PostgreSQLOpportunisticTestCase, LiveFacadeTest): pass class ConfigOptionsTest(oslo_test_base.BaseTestCase): def test_all_options(self): """test that everything in CONF.database.iteritems() is accepted. There's a handful of options in oslo.db.options that seem to have no meaning, but need to be accepted. In particular, Cinder and maybe others are doing exactly this call. """ factory = enginefacade._TransactionFactory() cfg.CONF.register_opts(options.database_opts, 'database') factory.configure(**dict(cfg.CONF.database.items())) def test_options_not_supported(self): factory = enginefacade._TransactionFactory() with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") factory.configure( fake1='x', connection_recycle_time=200, wrong2='y') self.assertEqual(1, len(w)) self.assertTrue( issubclass(w[-1].category, warning.NotSupportedWarning)) self.assertEqual( "Configuration option(s) ['fake1', 'wrong2'] not supported", str(w[-1].message) ) def test_no_engine(self): factory = enginefacade._TransactionFactory() self.assertRaises( exception.CantStartEngineError, factory._create_session, enginefacade._WRITER ) self.assertRaises( exception.CantStartEngineError, factory._create_session, enginefacade._WRITER ) class TestTransactionFactoryCallback(oslo_test_base.BaseTestCase): def test_setup_for_connection_called_with_profiler(self): context_manager = enginefacade.transaction_context() context_manager.configure(connection='sqlite://') hook = mock.Mock() context_manager.append_on_engine_create(hook) self.assertEqual( [hook], context_manager._factory._facade_cfg['on_engine_create']) @context_manager.reader def go(context): hook.assert_called_once_with(context.session.bind) go(oslo_context.RequestContext()) # TODO(zzzeek): test configuration options, e.g. like # test_sqlalchemy->test_creation_from_config