<s> import collections <EOL> import contextlib <EOL> import copy <EOL> import pickle <EOL> import warnings <EOL> import mock <EOL> from oslo_config import cfg <EOL> from oslo_context import context as oslo_context <EOL> from oslotest import base as oslo_test_base <EOL> from sqlalchemy import Column <EOL> from sqlalchemy import Integer <EOL> from sqlalchemy import MetaData <EOL> from sqlalchemy . orm import mapper <EOL> from sqlalchemy import select <EOL> from sqlalchemy import String <EOL> from sqlalchemy import Table <EOL> from oslo_db import exception <EOL> from oslo_db import options <EOL> from oslo_db . sqlalchemy import enginefacade <EOL> from oslo_db . sqlalchemy import engines as oslo_engines <EOL> from oslo_db . sqlalchemy import orm <EOL> from oslo_db . sqlalchemy import test_base <EOL> enginefacade . transaction_context_provider ( oslo_context . RequestContext ) <EOL> class SingletonOnName ( mock . MagicMock ) : <EOL> def __init__ ( self , the_name , ** kw ) : <EOL> super ( SingletonOnName , self ) . __init__ ( <EOL> __eq__ = lambda self , other : other . _assert_name == self . _assert_name , <EOL> _assert_name = the_name , <EOL> ** kw <EOL> ) <EOL> def __deepcopy__ ( self , memo ) : <EOL> return self <EOL> class SingletonConnection ( SingletonOnName ) : <EOL> def __init__ ( self , ** kw ) : <EOL> super ( SingletonConnection , self ) . __init__ ( <EOL> "<STR_LIT>" , ** kw ) <EOL> class SingletonEngine ( SingletonOnName ) : <EOL> def __init__ ( self , connection , ** kw ) : <EOL> super ( SingletonEngine , self ) . __init__ ( <EOL> "<STR_LIT>" , <EOL> connect = mock . Mock ( return_value = connection ) , <EOL> url = connection , <EOL> _assert_connection = connection , <EOL> ** kw <EOL> ) <EOL> class NonDecoratedContext ( object ) : <EOL> """<STR_LIT>""" <EOL> class AssertDataSource ( collections . namedtuple ( <EOL> "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) ) : <EOL> def element_for_writer ( self , const ) : <EOL> if const is enginefacade . _WRITER : <EOL> return self . writer <EOL> elif const is enginefacade . _READER : <EOL> return self . reader <EOL> elif const is enginefacade . _ASYNC_READER : <EOL> return self . async_reader <EOL> else : <EOL> assert False , "<STR_LIT>" % const <EOL> class MockFacadeTest ( oslo_test_base . BaseTestCase ) : <EOL> """<STR_LIT>""" <EOL> synchronous_reader = True <EOL> engine_uri = '<STR_LIT>' <EOL> slave_uri = None <EOL> def setUp ( self ) : <EOL> super ( MockFacadeTest , self ) . setUp ( ) <EOL> writer_conn = SingletonConnection ( ) <EOL> writer_engine = SingletonEngine ( writer_conn ) <EOL> writer_session = mock . Mock ( <EOL> connection = mock . Mock ( return_value = writer_conn ) ) <EOL> writer_maker = mock . Mock ( return_value = writer_session ) <EOL> if self . slave_uri : <EOL> async_reader_conn = SingletonConnection ( ) <EOL> async_reader_engine = SingletonEngine ( async_reader_conn ) <EOL> async_reader_session = mock . Mock ( <EOL> connection = mock . Mock ( return_value = async_reader_conn ) ) <EOL> async_reader_maker = mock . Mock ( return_value = async_reader_session ) <EOL> else : <EOL> async_reader_conn = writer_conn <EOL> async_reader_engine = writer_engine <EOL> async_reader_session = writer_session <EOL> async_reader_maker = writer_maker <EOL> if self . synchronous_reader : <EOL> reader_conn = async_reader_conn <EOL> reader_engine = async_reader_engine <EOL> reader_session = async_reader_session <EOL> reader_maker = async_reader_maker <EOL> else : <EOL> reader_conn = writer_conn <EOL> reader_engine = writer_engine <EOL> reader_session = writer_session <EOL> reader_maker = writer_maker <EOL> self . connections = AssertDataSource ( <EOL> writer_conn , reader_conn , async_reader_conn <EOL> ) <EOL> self . engines = AssertDataSource ( <EOL> writer_engine , reader_engine , async_reader_engine <EOL> ) <EOL> self . sessions = AssertDataSource ( <EOL> writer_session , reader_session , async_reader_session <EOL> ) <EOL> self . makers = AssertDataSource ( <EOL> writer_maker , reader_maker , async_reader_maker <EOL> ) <EOL> def get_maker ( engine , ** kw ) : <EOL> if engine is writer_engine : <EOL> return self . makers . writer <EOL> elif engine is reader_engine : <EOL> return self . makers . reader <EOL> elif engine is async_reader_engine : <EOL> return self . makers . async_reader <EOL> else : <EOL> assert False <EOL> session_patch = mock . patch . object ( <EOL> orm , "<STR_LIT>" , <EOL> side_effect = get_maker ) <EOL> self . get_maker = session_patch . start ( ) <EOL> self . addCleanup ( session_patch . stop ) <EOL> def create_engine ( sql_connection , ** kw ) : <EOL> if sql_connection == self . engine_uri : <EOL> return self . engines . writer <EOL> elif sql_connection == self . slave_uri : <EOL> return self . engines . async_reader <EOL> else : <EOL> assert False <EOL> engine_patch = mock . patch . object ( <EOL> oslo_engines , "<STR_LIT>" , side_effect = create_engine ) <EOL> self . create_engine = engine_patch . start ( ) <EOL> self . addCleanup ( engine_patch . stop ) <EOL> self . factory = enginefacade . _TransactionFactory ( ) <EOL> self . factory . configure ( <EOL> synchronous_reader = self . synchronous_reader <EOL> ) <EOL> self . factory . configure ( <EOL> connection = self . engine_uri , <EOL> slave_connection = self . slave_uri <EOL> ) <EOL> facade_patcher = mock . patch . object ( <EOL> enginefacade . _context_manager , "<STR_LIT>" , self . factory ) <EOL> facade_patcher . start ( ) <EOL> self . addCleanup ( facade_patcher . stop ) <EOL> def _assert_ctx_connection ( self , context , connection ) : <EOL> self . assertIs ( context . connection , connection ) <EOL> def _assert_ctx_session ( self , context , session ) : <EOL> self . assertIs ( context . session , session ) <EOL> def _assert_non_decorated_ctx_connection ( self , context , connection ) : <EOL> transaction_ctx = enginefacade . _transaction_ctx_for_context ( context ) <EOL> self . assertIs ( transaction_ctx . connection , connection ) <EOL> def _assert_non_decorated_ctx_session ( self , context , session ) : <EOL> transaction_ctx = enginefacade . _transaction_ctx_for_context ( context ) <EOL> self . assertIs ( transaction_ctx . session , session ) <EOL> @ contextlib . contextmanager <EOL> def _assert_engines ( self ) : <EOL> """<STR_LIT>""" <EOL> writer_conn = SingletonConnection ( ) <EOL> writer_engine = SingletonEngine ( writer_conn ) <EOL> if self . slave_uri : <EOL> async_reader_conn = SingletonConnection ( ) <EOL> async_reader_engine = SingletonEngine ( async_reader_conn ) <EOL> else : <EOL> async_reader_conn = writer_conn <EOL> async_reader_engine = writer_engine <EOL> if self . synchronous_reader : <EOL> reader_engine = async_reader_engine <EOL> else : <EOL> reader_engine = writer_engine <EOL> engines = AssertDataSource ( <EOL> writer_engine , reader_engine , async_reader_engine ) <EOL> def create_engine ( sql_connection , ** kw ) : <EOL> if sql_connection == self . engine_uri : <EOL> return engines . writer <EOL> elif sql_connection == self . slave_uri : <EOL> return engines . async_reader <EOL> else : <EOL> assert False <EOL> engine_factory = mock . Mock ( side_effect = create_engine ) <EOL> engine_factory ( <EOL> sql_connection = self . engine_uri , <EOL> ** dict ( ( k , mock . ANY ) for k in self . factory . _engine_cfg . keys ( ) ) <EOL> ) <EOL> if self . slave_uri : <EOL> engine_factory ( <EOL> sql_connection = self . slave_uri , <EOL> ** dict ( ( k , mock . ANY ) for k in self . factory . _engine_cfg . keys ( ) ) <EOL> ) <EOL> yield AssertDataSource ( <EOL> writer_engine , reader_engine , async_reader_engine <EOL> ) <EOL> self . assertEqual ( <EOL> engine_factory . mock_calls , <EOL> self . create_engine . mock_calls <EOL> ) <EOL> for sym in [ <EOL> enginefacade . _WRITER , enginefacade . _READER , <EOL> enginefacade . _ASYNC_READER <EOL> ] : <EOL> self . assertEqual ( <EOL> engines . element_for_writer ( sym ) . mock_calls , <EOL> self . engines . element_for_writer ( sym ) . mock_calls <EOL> ) <EOL> def _assert_async_reader_connection ( self , engines , session = None ) : <EOL> return self . _assert_connection ( <EOL> engines , enginefacade . _ASYNC_READER , session ) <EOL> def _assert_reader_connection ( self , engines , session = None ) : <EOL> return self . _assert_connection ( engines , enginefacade . _READER , session ) <EOL> def _assert_writer_connection ( self , engines , session = None ) : <EOL> return self . _assert_connection ( engines , enginefacade . _WRITER , session ) <EOL> @ contextlib . contextmanager <EOL> def _assert_connection ( self , engines , writer , session = None ) : <EOL> """<STR_LIT>""" <EOL> if session : <EOL> connection = session . connection ( ) <EOL> yield connection <EOL> else : <EOL> connection = engines . element_for_writer ( writer ) . connect ( ) <EOL> trans = connection . begin ( ) <EOL> yield connection <EOL> if writer is enginefacade . _WRITER : <EOL> trans . commit ( ) <EOL> else : <EOL> trans . rollback ( ) <EOL> connection . close ( ) <EOL> self . assertEqual ( <EOL> connection . mock_calls , <EOL> self . connections . element_for_writer ( writer ) . mock_calls ) <EOL> @ contextlib . contextmanager <EOL> def _assert_makers ( self , engines ) : <EOL> writer_session = mock . Mock ( connection = mock . Mock ( <EOL> return_value = engines . writer . _assert_connection ) <EOL> ) <EOL> writer_maker = mock . Mock ( return_value = writer_session ) <EOL> if self . slave_uri : <EOL> async_reader_session = mock . Mock ( connection = mock . Mock ( <EOL> return_value = engines . async_reader . _assert_connection ) <EOL> ) <EOL> async_reader_maker = mock . Mock ( return_value = async_reader_session ) <EOL> else : <EOL> async_reader_session = writer_session <EOL> async_reader_maker = writer_maker <EOL> if self . synchronous_reader : <EOL> reader_maker = async_reader_maker <EOL> else : <EOL> reader_maker = writer_maker <EOL> makers = AssertDataSource ( <EOL> writer_maker , <EOL> reader_maker , <EOL> async_reader_maker , <EOL> ) <EOL> def get_maker ( engine , ** kw ) : <EOL> if engine is engines . writer : <EOL> return makers . writer <EOL> elif engine is engines . reader : <EOL> return makers . reader <EOL> elif engine is engines . async_reader : <EOL> return makers . async_reader <EOL> else : <EOL> assert False <EOL> maker_factories = mock . Mock ( side_effect = get_maker ) <EOL> maker_factories ( <EOL> autocommit = True , engine = engines . writer , <EOL> expire_on_commit = False ) <EOL> if self . slave_uri : <EOL> maker_factories ( <EOL> autocommit = True , engine = engines . async_reader , <EOL> expire_on_commit = False ) <EOL> yield makers <EOL> self . assertEqual ( <EOL> maker_factories . mock_calls , <EOL> self . get_maker . mock_calls ) <EOL> for sym in [ <EOL> enginefacade . _WRITER , enginefacade . _READER , <EOL> enginefacade . _ASYNC_READER <EOL> ] : <EOL> self . assertEqual ( <EOL> makers . element_for_writer ( sym ) . mock_calls , <EOL> self . makers . element_for_writer ( sym ) . mock_calls ) <EOL> def _assert_async_reader_session ( <EOL> self , makers , connection = None , assert_calls = True ) : <EOL> return self . _assert_session ( <EOL> makers , enginefacade . _ASYNC_READER , connection , assert_calls ) <EOL> def _assert_reader_session ( <EOL> self , makers , connection = None , assert_calls = True ) : <EOL> return self . _assert_session ( <EOL> makers , enginefacade . _READER , <EOL> connection , assert_calls ) <EOL> def _assert_writer_session ( <EOL> self , makers , connection = None , assert_calls = True ) : <EOL> return self . _assert_session ( <EOL> makers , enginefacade . _WRITER , <EOL> connection , assert_calls ) <EOL> @ contextlib . contextmanager <EOL> def _assert_session ( <EOL> self , makers , writer , connection = None , assert_calls = True ) : <EOL> """<STR_LIT>""" <EOL> if connection : <EOL> session = makers . element_for_writer ( writer ) ( bind = connection ) <EOL> else : <EOL> session = makers . element_for_writer ( writer ) ( ) <EOL> session . begin ( ) <EOL> yield session <EOL> if writer is enginefacade . _WRITER : <EOL> session . commit ( ) <EOL> elif enginefacade . _context_manager . _factory . _transaction_ctx_cfg [ <EOL> '<STR_LIT>' ] : <EOL> session . rollback ( ) <EOL> session . close ( ) <EOL> if assert_calls : <EOL> self . assertEqual ( <EOL> session . mock_calls , <EOL> self . sessions . element_for_writer ( writer ) . mock_calls ) <EOL> def test_session_reader_decorator ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader <EOL> def go ( context ) : <EOL> context . session . execute ( "<STR_LIT:test>" ) <EOL> go ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT:test>" ) <EOL> def test_connection_reader_decorator ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . connection <EOL> def go ( context ) : <EOL> context . connection . execute ( "<STR_LIT:test>" ) <EOL> go ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_reader_connection ( engines ) as connection : <EOL> connection . execute ( "<STR_LIT:test>" ) <EOL> def test_session_reader_nested_in_connection_reader ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . connection <EOL> def go1 ( context ) : <EOL> context . connection . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_reader_connection ( engines ) as connection : <EOL> connection . execute ( "<STR_LIT>" ) <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( <EOL> makers , connection ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_connection_reader_nested_in_session_reader ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader . connection <EOL> def go2 ( context ) : <EOL> context . connection . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_reader_connection ( <EOL> engines , session ) as connection : <EOL> connection . execute ( "<STR_LIT>" ) <EOL> def test_session_reader_decorator_nested ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_reader_nested_in_writer_ok ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . writer <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_writer_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_writer_nested_in_reader_raises ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . writer <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> exc = self . assertRaises ( <EOL> TypeError , go1 , context <EOL> ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> exc . args [ <NUM_LIT:0> ] <EOL> ) <EOL> def test_async_on_writer_raises ( self ) : <EOL> exc = self . assertRaises ( <EOL> TypeError , getattr , enginefacade . writer , "<STR_LIT>" <EOL> ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , <EOL> exc . args [ <NUM_LIT:0> ] <EOL> ) <EOL> def test_savepoint_and_independent_raises ( self ) : <EOL> exc = self . assertRaises ( <EOL> TypeError , getattr , enginefacade . writer . independent , "<STR_LIT>" <EOL> ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , <EOL> exc . args [ <NUM_LIT:0> ] <EOL> ) <EOL> def test_reader_nested_in_async_reader_raises ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . async <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> exc = self . assertRaises ( <EOL> TypeError , go1 , context <EOL> ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> exc . args [ <NUM_LIT:0> ] <EOL> ) <EOL> def test_reader_allow_async_nested_in_async_reader ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . async <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader . allow_async <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_async_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_reader_allow_async_nested_in_reader ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . reader <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . reader . allow_async <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_reader_allow_async_is_reader_by_default ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . allow_async <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_writer_nested_in_async_reader_raises ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . async <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go2 ( context ) <EOL> @ enginefacade . writer <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> exc = self . assertRaises ( <EOL> TypeError , go1 , context <EOL> ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> exc . args [ <NUM_LIT:0> ] <EOL> ) <EOL> def test_reader_then_writer_ok ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> @ enginefacade . writer <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> go2 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( <EOL> makers , assert_calls = False ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_writer_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_async_reader_then_reader_ok ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader . async <EOL> def go1 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> @ enginefacade . reader <EOL> def go2 ( context ) : <EOL> context . session . execute ( "<STR_LIT>" ) <EOL> go1 ( context ) <EOL> go2 ( context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_async_reader_session ( <EOL> makers , assert_calls = False ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_using_reader ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . reader . using ( context ) as session : <EOL> self . _assert_ctx_session ( context , session ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_using_reader_rollback_reader_session ( self ) : <EOL> enginefacade . configure ( rollback_reader_sessions = True ) <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . reader . using ( context ) as session : <EOL> self . _assert_ctx_session ( context , session ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_using_writer ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> self . _assert_ctx_session ( context , session ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_writer_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_using_writer_no_descriptors ( self ) : <EOL> context = NonDecoratedContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> self . _assert_non_decorated_ctx_session ( context , session ) <EOL> session . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_writer_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT>" ) <EOL> def test_using_writer_connection_no_descriptors ( self ) : <EOL> context = NonDecoratedContext ( ) <EOL> with enginefacade . writer . connection . using ( context ) as connection : <EOL> self . _assert_non_decorated_ctx_connection ( context , connection ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_writer_connection ( engines ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> def test_using_reader_connection ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . reader . connection . using ( context ) as connection : <EOL> self . _assert_ctx_connection ( context , connection ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_reader_connection ( engines ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> def test_using_writer_connection ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . connection . using ( context ) as connection : <EOL> self . _assert_ctx_connection ( context , connection ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_writer_connection ( engines ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> def test_context_copied_using_existing_writer_connection ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . connection . using ( context ) as connection : <EOL> self . _assert_ctx_connection ( context , connection ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> ctx2 = copy . deepcopy ( context ) <EOL> with enginefacade . reader . connection . using ( ctx2 ) as conn2 : <EOL> self . assertIs ( conn2 , connection ) <EOL> self . _assert_ctx_connection ( ctx2 , conn2 ) <EOL> conn2 . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_writer_connection ( engines ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> conn . execute ( "<STR_LIT>" ) <EOL> def test_context_nodesc_copied_using_existing_writer_connection ( self ) : <EOL> context = NonDecoratedContext ( ) <EOL> with enginefacade . writer . connection . using ( context ) as connection : <EOL> self . _assert_non_decorated_ctx_connection ( context , connection ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> ctx2 = copy . deepcopy ( context ) <EOL> with enginefacade . reader . connection . using ( ctx2 ) as conn2 : <EOL> self . assertIs ( conn2 , connection ) <EOL> self . _assert_non_decorated_ctx_connection ( ctx2 , conn2 ) <EOL> conn2 . execute ( "<STR_LIT>" ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_writer_connection ( engines ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> conn . execute ( "<STR_LIT>" ) <EOL> def test_session_context_notrequested_exception ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . reader . connection . using ( context ) : <EOL> exc = self . assertRaises ( <EOL> exception . ContextNotRequestedError , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> self . assertRegexpMatches ( <EOL> exc . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_connection_context_notrequested_exception ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . reader . using ( context ) : <EOL> exc = self . assertRaises ( <EOL> exception . ContextNotRequestedError , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> self . assertRegexpMatches ( <EOL> exc . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_session_context_exception ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> exc = self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> self . assertRegexpMatches ( <EOL> exc . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_session_context_getattr ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> self . assertIsNone ( getattr ( context , '<STR_LIT>' , None ) ) <EOL> def test_connection_context_exception ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> exc = self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> self . assertRegexpMatches ( <EOL> exc . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_connection_context_getattr ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> self . assertIsNone ( getattr ( context , '<STR_LIT>' , None ) ) <EOL> def test_transaction_context_exception ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> exc = self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> self . assertRegexpMatches ( <EOL> exc . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_transaction_context_getattr ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> self . assertIsNone ( getattr ( context , '<STR_LIT>' , None ) ) <EOL> def test_trans_ctx_context_exception ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> exc = self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> self . assertRegexpMatches ( <EOL> exc . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_trans_ctx_context_getattr ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> self . assertIsNone ( getattr ( context , '<STR_LIT>' , None ) ) <EOL> def test_multiple_factories ( self ) : <EOL> """<STR_LIT>""" <EOL> mgr1 = enginefacade . transaction_context ( ) <EOL> mgr1 . configure ( <EOL> connection = self . engine_uri , <EOL> slave_connection = self . slave_uri <EOL> ) <EOL> mgr2 = enginefacade . transaction_context ( ) <EOL> mgr2 . configure ( <EOL> connection = self . engine_uri , <EOL> slave_connection = self . slave_uri <EOL> ) <EOL> context = oslo_context . RequestContext ( ) <EOL> self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> with mgr1 . writer . using ( context ) : <EOL> self . assertIs ( context . transaction_ctx . factory , mgr1 . _factory ) <EOL> self . assertIsNot ( context . transaction_ctx . factory , mgr2 . _factory ) <EOL> self . assertIsNotNone ( context . session ) <EOL> self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> with mgr2 . writer . using ( context ) : <EOL> self . assertIsNot ( context . transaction_ctx . factory , mgr1 . _factory ) <EOL> self . assertIs ( context . transaction_ctx . factory , mgr2 . _factory ) <EOL> self . assertIsNotNone ( context . session ) <EOL> def test_multiple_factories_nested ( self ) : <EOL> """<STR_LIT>""" <EOL> mgr1 = enginefacade . transaction_context ( ) <EOL> mgr1 . configure ( <EOL> connection = self . engine_uri , <EOL> slave_connection = self . slave_uri <EOL> ) <EOL> mgr2 = enginefacade . transaction_context ( ) <EOL> mgr2 . configure ( <EOL> connection = self . engine_uri , <EOL> slave_connection = self . slave_uri <EOL> ) <EOL> context = oslo_context . RequestContext ( ) <EOL> with mgr1 . writer . using ( context ) : <EOL> self . assertIs ( context . transaction_ctx . factory , mgr1 . _factory ) <EOL> self . assertIsNot ( context . transaction_ctx . factory , mgr2 . _factory ) <EOL> with mgr2 . writer . using ( context ) : <EOL> self . assertIsNot ( <EOL> context . transaction_ctx . factory , mgr1 . _factory ) <EOL> self . assertIs ( context . transaction_ctx . factory , mgr2 . _factory ) <EOL> self . assertIsNotNone ( context . session ) <EOL> self . assertIs ( context . transaction_ctx . factory , mgr1 . _factory ) <EOL> self . assertIsNot ( context . transaction_ctx . factory , mgr2 . _factory ) <EOL> self . assertIsNotNone ( context . session ) <EOL> self . assertRaises ( <EOL> exception . NoEngineContextEstablished , <EOL> getattr , context , '<STR_LIT>' <EOL> ) <EOL> def test_context_found_for_bound_method ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . reader <EOL> def go ( self , context ) : <EOL> context . session . execute ( "<STR_LIT:test>" ) <EOL> go ( self , context ) <EOL> with self . _assert_engines ( ) as engines : <EOL> with self . _assert_makers ( engines ) as makers : <EOL> with self . _assert_reader_session ( makers ) as session : <EOL> session . execute ( "<STR_LIT:test>" ) <EOL> class SynchronousReaderWSlaveMockFacadeTest ( MockFacadeTest ) : <EOL> synchronous_reader = True <EOL> engine_uri = '<STR_LIT>' <EOL> slave_uri = '<STR_LIT>' <EOL> class AsyncReaderWSlaveMockFacadeTest ( MockFacadeTest ) : <EOL> synchronous_reader = False <EOL> engine_uri = '<STR_LIT>' <EOL> slave_uri = '<STR_LIT>' <EOL> class LegacyIntegrationtest ( test_base . DbTestCase ) : <EOL> def test_legacy_integration ( self ) : <EOL> legacy_facade = enginefacade . get_legacy_facade ( ) <EOL> self . assertTrue ( <EOL> legacy_facade . get_engine ( ) is <EOL> enginefacade . _context_manager . _factory . _writer_engine <EOL> ) <EOL> self . assertTrue ( <EOL> enginefacade . get_legacy_facade ( ) is legacy_facade <EOL> ) <EOL> def test_get_sessionmaker ( self ) : <EOL> legacy_facade = enginefacade . get_legacy_facade ( ) <EOL> self . assertTrue ( <EOL> legacy_facade . get_sessionmaker ( ) is <EOL> enginefacade . _context_manager . _factory . _writer_maker <EOL> ) <EOL> def test_legacy_facades_from_different_context_managers ( self ) : <EOL> transaction_context1 = enginefacade . transaction_context ( ) <EOL> transaction_context2 = enginefacade . transaction_context ( ) <EOL> transaction_context1 . configure ( connection = '<STR_LIT>' ) <EOL> transaction_context2 . configure ( connection = '<STR_LIT>' ) <EOL> legacy1 = transaction_context1 . get_legacy_facade ( ) <EOL> legacy2 = transaction_context2 . get_legacy_facade ( ) <EOL> self . assertNotEqual ( legacy1 , legacy2 ) <EOL> def test_legacy_not_started ( self ) : <EOL> factory = enginefacade . _TransactionFactory ( ) <EOL> self . assertRaises ( <EOL> exception . CantStartEngineError , <EOL> factory . get_legacy_facade <EOL> ) <EOL> legacy_facade = factory . get_legacy_facade ( ) <EOL> self . assertRaises ( <EOL> exception . CantStartEngineError , <EOL> legacy_facade . get_session <EOL> ) <EOL> self . assertRaises ( <EOL> exception . CantStartEngineError , <EOL> legacy_facade . get_session <EOL> ) <EOL> self . assertRaises ( <EOL> exception . CantStartEngineError , <EOL> legacy_facade . get_engine <EOL> ) <EOL> class ThreadingTest ( test_base . DbTestCase ) : <EOL> """<STR_LIT>""" <EOL> def _assert_ctx_connection ( self , context , connection ) : <EOL> self . assertIs ( context . connection , connection ) <EOL> def _assert_ctx_session ( self , context , session ) : <EOL> self . assertIs ( context . session , session ) <EOL> def _patch_thread_ident ( self ) : <EOL> self . ident = <NUM_LIT:1> <EOL> test_instance = self <EOL> class MockThreadingLocal ( object ) : <EOL> def __init__ ( self ) : <EOL> self . __dict__ [ '<STR_LIT:state>' ] = collections . defaultdict ( dict ) <EOL> def __deepcopy__ ( self , memo ) : <EOL> return self <EOL> def __getattr__ ( self , key ) : <EOL> ns = self . state [ test_instance . ident ] <EOL> try : <EOL> return ns [ key ] <EOL> except KeyError : <EOL> raise AttributeError ( key ) <EOL> def __setattr__ ( self , key , value ) : <EOL> ns = self . state [ test_instance . ident ] <EOL> ns [ key ] = value <EOL> def __delattr__ ( self , key ) : <EOL> ns = self . state [ test_instance . ident ] <EOL> try : <EOL> del ns [ key ] <EOL> except KeyError : <EOL> raise AttributeError ( key ) <EOL> return mock . patch . object ( <EOL> enginefacade , "<STR_LIT>" , MockThreadingLocal ) <EOL> def test_thread_ctxmanager_writer ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with self . _patch_thread_ident ( ) : <EOL> with enginefacade . writer . using ( context ) as session : <EOL> self . _assert_ctx_session ( context , session ) <EOL> self . ident = <NUM_LIT:2> <EOL> with enginefacade . reader . using ( context ) as sess2 : <EOL> self . assertIsNot ( sess2 , session ) <EOL> self . _assert_ctx_session ( context , sess2 ) <EOL> self . ident = <NUM_LIT:1> <EOL> with enginefacade . reader . using ( context ) as sess3 : <EOL> self . assertIs ( sess3 , session ) <EOL> self . _assert_ctx_session ( context , session ) <EOL> def test_thread_ctxmanager_writer_connection ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with self . _patch_thread_ident ( ) : <EOL> with enginefacade . writer . connection . using ( context ) as conn : <EOL> self . _assert_ctx_connection ( context , conn ) <EOL> self . ident = <NUM_LIT:2> <EOL> with enginefacade . reader . connection . using ( context ) as conn2 : <EOL> self . assertIsNot ( conn2 , conn ) <EOL> self . _assert_ctx_connection ( context , conn2 ) <EOL> with enginefacade . reader . connection . using ( <EOL> context ) as conn3 : <EOL> self . assertIsNot ( conn3 , conn ) <EOL> self . assertIs ( conn3 , conn2 ) <EOL> self . ident = <NUM_LIT:1> <EOL> with enginefacade . reader . connection . using ( context ) as conn3 : <EOL> self . assertIs ( conn3 , conn ) <EOL> self . _assert_ctx_connection ( context , conn ) <EOL> def test_thread_ctxmanager_switch_styles ( self ) : <EOL> @ enginefacade . writer . connection <EOL> def go_one ( context ) : <EOL> self . assertRaises ( <EOL> exception . ContextNotRequestedError , <EOL> getattr , context , "<STR_LIT>" <EOL> ) <EOL> self . assertIsNotNone ( context . connection ) <EOL> self . ident = <NUM_LIT:2> <EOL> go_two ( context ) <EOL> self . ident = <NUM_LIT:1> <EOL> self . assertRaises ( <EOL> exception . ContextNotRequestedError , <EOL> getattr , context , "<STR_LIT>" <EOL> ) <EOL> self . assertIsNotNone ( context . connection ) <EOL> @ enginefacade . reader <EOL> def go_two ( context ) : <EOL> self . assertRaises ( <EOL> exception . ContextNotRequestedError , <EOL> getattr , context , "<STR_LIT>" <EOL> ) <EOL> self . assertIsNotNone ( context . session ) <EOL> context = oslo_context . RequestContext ( ) <EOL> with self . _patch_thread_ident ( ) : <EOL> go_one ( context ) <EOL> def test_thread_decorator_writer ( self ) : <EOL> sessions = set ( ) <EOL> @ enginefacade . writer <EOL> def go_one ( context ) : <EOL> sessions . add ( context . session ) <EOL> self . ident = <NUM_LIT:2> <EOL> go_two ( context ) <EOL> self . ident = <NUM_LIT:1> <EOL> go_three ( context ) <EOL> @ enginefacade . reader <EOL> def go_two ( context ) : <EOL> assert context . session not in sessions <EOL> @ enginefacade . reader <EOL> def go_three ( context ) : <EOL> assert context . session in sessions <EOL> context = oslo_context . RequestContext ( ) <EOL> with self . _patch_thread_ident ( ) : <EOL> go_one ( context ) <EOL> def test_thread_decorator_writer_connection ( self ) : <EOL> connections = set ( ) <EOL> @ enginefacade . writer . connection <EOL> def go_one ( context ) : <EOL> connections . add ( context . connection ) <EOL> self . ident = <NUM_LIT:2> <EOL> go_two ( context ) <EOL> self . ident = <NUM_LIT:1> <EOL> go_three ( context ) <EOL> @ enginefacade . reader . connection <EOL> def go_two ( context ) : <EOL> assert context . connection not in connections <EOL> @ enginefacade . reader <EOL> def go_three ( context ) : <EOL> assert context . connection in connections <EOL> context = oslo_context . RequestContext ( ) <EOL> with self . _patch_thread_ident ( ) : <EOL> go_one ( context ) <EOL> def test_contexts_picklable ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> self . _assert_ctx_session ( context , session ) <EOL> pickled = pickle . dumps ( context ) <EOL> unpickled = pickle . loads ( pickled ) <EOL> with enginefacade . writer . using ( unpickled ) as session2 : <EOL> self . _assert_ctx_session ( unpickled , session2 ) <EOL> assert session is not session2 <EOL> class LiveFacadeTest ( test_base . DbTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( LiveFacadeTest , self ) . setUp ( ) <EOL> metadata = MetaData ( ) <EOL> user_table = Table ( <EOL> '<STR_LIT:user>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( <NUM_LIT:30> ) ) , <EOL> mysql_engine = '<STR_LIT>' <EOL> ) <EOL> self . user_table = user_table <EOL> metadata . create_all ( self . engine ) <EOL> self . addCleanup ( metadata . drop_all , self . engine ) <EOL> class User ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> mapper ( User , user_table ) <EOL> self . User = User <EOL> def _assert_ctx_connection ( self , context , connection ) : <EOL> self . assertIs ( context . connection , connection ) <EOL> def _assert_ctx_session ( self , context , session ) : <EOL> self . assertIs ( context . session , session ) <EOL> def test_transaction_committed ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , <EOL> session . query ( self . User . name ) . scalar ( ) <EOL> ) <EOL> def test_transaction_rollback ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> class MyException ( Exception ) : <EOL> pass <EOL> @ enginefacade . writer <EOL> def go ( context ) : <EOL> context . session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> context . session . flush ( ) <EOL> raise MyException ( "<STR_LIT>" ) <EOL> self . assertRaises ( MyException , go , context ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> None , <EOL> session . query ( self . User . name ) . scalar ( ) <EOL> ) <EOL> def test_context_deepcopy_on_session ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> ctx2 = copy . deepcopy ( context ) <EOL> self . _assert_ctx_session ( ctx2 , session ) <EOL> with enginefacade . writer . using ( ctx2 ) as s2 : <EOL> self . assertIs ( session , s2 ) <EOL> self . _assert_ctx_session ( ctx2 , s2 ) <EOL> s2 . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> s2 . flush ( ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , <EOL> session . query ( self . User . name ) . scalar ( ) <EOL> ) <EOL> def test_context_deepcopy_on_connection ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . connection . using ( context ) as conn : <EOL> ctx2 = copy . deepcopy ( context ) <EOL> self . _assert_ctx_connection ( ctx2 , conn ) <EOL> with enginefacade . writer . connection . using ( ctx2 ) as conn2 : <EOL> self . assertIs ( conn , conn2 ) <EOL> self . _assert_ctx_connection ( ctx2 , conn2 ) <EOL> conn2 . execute ( self . user_table . insert ( ) . values ( name = "<STR_LIT>" ) ) <EOL> self . _assert_ctx_connection ( ctx2 , conn2 ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , <EOL> session . query ( self . User . name ) . scalar ( ) <EOL> ) <EOL> @ test_base . backend_specific ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_external_session_transaction ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session . flush ( ) <EOL> with enginefacade . writer . independent . using ( context ) as s2 : <EOL> self . assertIsNot ( s2 , session ) <EOL> self . _assert_ctx_session ( context , s2 ) <EOL> s2 . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> with enginefacade . writer . using ( context ) as s3 : <EOL> self . assertIs ( s3 , s2 ) <EOL> s3 . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> self . _assert_ctx_session ( context , session ) <EOL> session . rollback ( ) <EOL> session . begin ( ) <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> [ ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) ] , <EOL> session . query ( <EOL> self . User . name ) . order_by ( self . User . name ) . all ( ) <EOL> ) <EOL> def test_savepoint_transaction_decorator ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . writer <EOL> def go1 ( context ) : <EOL> session = context . session <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session . flush ( ) <EOL> try : <EOL> go2 ( context ) <EOL> except Exception : <EOL> pass <EOL> go3 ( context ) <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> @ enginefacade . writer . savepoint <EOL> def go2 ( context ) : <EOL> session = context . session <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> @ enginefacade . writer . savepoint <EOL> def go3 ( context ) : <EOL> session = context . session <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> go1 ( context ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> [ ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) ] , <EOL> session . query ( <EOL> self . User . name ) . order_by ( self . User . name ) . all ( ) <EOL> ) <EOL> def test_savepoint_transaction ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . using ( context ) as session : <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session . flush ( ) <EOL> try : <EOL> with enginefacade . writer . savepoint . using ( context ) as session : <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> except Exception : <EOL> pass <EOL> with enginefacade . writer . savepoint . using ( context ) as session : <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> [ ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) ] , <EOL> session . query ( <EOL> self . User . name ) . order_by ( self . User . name ) . all ( ) <EOL> ) <EOL> @ test_base . backend_specific ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_external_session_transaction_decorator ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> @ enginefacade . writer <EOL> def go1 ( context ) : <EOL> session = context . session <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> session . flush ( ) <EOL> go2 ( context , session ) <EOL> self . _assert_ctx_session ( context , session ) <EOL> session . rollback ( ) <EOL> session . begin ( ) <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> @ enginefacade . writer . independent <EOL> def go2 ( context , session ) : <EOL> s2 = context . session <EOL> self . assertIsNot ( s2 , session ) <EOL> self . _assert_ctx_session ( context , s2 ) <EOL> s2 . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> with enginefacade . writer . using ( context ) as s3 : <EOL> self . assertIs ( s3 , s2 ) <EOL> s3 . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> go1 ( context ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> [ ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) ] , <EOL> session . query ( <EOL> self . User . name ) . order_by ( self . User . name ) . all ( ) <EOL> ) <EOL> @ test_base . backend_specific ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_external_connection_transaction ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . writer . connection . using ( context ) as connection : <EOL> connection . execute ( self . user_table . insert ( ) . values ( name = "<STR_LIT>" ) ) <EOL> with enginefacade . writer . independent . connection . using ( context ) as c2 : <EOL> self . assertIsNot ( c2 , connection ) <EOL> self . _assert_ctx_connection ( context , c2 ) <EOL> c2 . execute ( self . user_table . insert ( ) . values ( name = "<STR_LIT>" ) ) <EOL> with enginefacade . writer . connection . using ( context ) as c3 : <EOL> self . assertIs ( c2 , c3 ) <EOL> c3 . execute ( self . user_table . insert ( ) . values ( name = "<STR_LIT>" ) ) <EOL> self . _assert_ctx_connection ( context , connection ) <EOL> transaction_ctx = context . transaction_ctx <EOL> transaction_ctx . transaction . rollback ( ) <EOL> transaction_ctx . transaction = connection . begin ( ) <EOL> connection . execute ( self . user_table . insert ( ) . values ( name = "<STR_LIT>" ) ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> [ ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) , ( "<STR_LIT>" , ) ] , <EOL> session . query ( <EOL> self . User . name ) . order_by ( self . User . name ) . all ( ) <EOL> ) <EOL> @ test_base . backend_specific ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_external_writer_in_reader ( self ) : <EOL> context = oslo_context . RequestContext ( ) <EOL> with enginefacade . reader . using ( context ) as session : <EOL> ping = session . scalar ( select ( [ <NUM_LIT:1> ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ping ) <EOL> @ enginefacade . writer <EOL> def go ( ctx ) : <EOL> pass <EOL> exc = self . assertRaises ( TypeError , go , context ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> exc . args [ <NUM_LIT:0> ] ) <EOL> with enginefacade . writer . independent . using ( context ) as sess2 : <EOL> self . assertIsNot ( sess2 , session ) <EOL> self . _assert_ctx_session ( context , sess2 ) <EOL> session . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> sess2 . add ( self . User ( name = "<STR_LIT>" ) ) <EOL> user = session . query ( self . User ) . first ( ) <EOL> self . assertEqual ( "<STR_LIT>" , user . name ) <EOL> session = self . sessionmaker ( autocommit = True ) <EOL> self . assertEqual ( <EOL> session . query ( <EOL> self . User . name ) . order_by ( self . User . name ) . all ( ) , <EOL> [ ( "<STR_LIT>" , ) ] <EOL> ) <EOL> def test_replace_scope ( self ) : <EOL> alt_connection = "<STR_LIT>" <EOL> alt_mgr1 = enginefacade . transaction_context ( ) <EOL> alt_mgr1 . configure ( <EOL> connection = alt_connection , <EOL> ) <EOL> @ enginefacade . writer <EOL> def go1 ( context ) : <EOL> s1 = context . session <EOL> self . assertEqual ( <EOL> s1 . bind . url , <EOL> enginefacade . _context_manager . _factory . _writer_engine . url ) <EOL> self . assertIs ( <EOL> s1 . bind , <EOL> enginefacade . _context_manager . _factory . _writer_engine ) <EOL> self . assertEqual ( s1 . bind . url , self . engine . url ) <EOL> with alt_mgr1 . replace . using ( context ) : <EOL> go2 ( context ) <EOL> go4 ( context ) <EOL> @ enginefacade . writer <EOL> def go2 ( context ) : <EOL> s2 = context . session <EOL> self . assertIsNot ( <EOL> enginefacade . _context_manager . _factory . _writer_engine , <EOL> alt_mgr1 . _factory . _writer_engine <EOL> ) <EOL> self . assertIs ( s2 . bind , alt_mgr1 . _factory . _writer_engine ) <EOL> self . assertEqual ( <EOL> str ( s2 . bind . url ) , alt_connection ) <EOL> go3 ( context ) <EOL> @ enginefacade . reader <EOL> def go3 ( context ) : <EOL> s3 = context . session <EOL> self . assertIs ( s3 . bind , alt_mgr1 . _factory . _writer_engine ) <EOL> self . assertEqual ( <EOL> str ( s3 . bind . url ) , alt_connection ) <EOL> @ enginefacade . writer <EOL> def go4 ( context ) : <EOL> s4 = context . session <EOL> self . assertIs ( s4 . bind , self . engine ) <EOL> self . assertEqual ( <EOL> s4 . bind . url , self . engine . url ) <EOL> context = oslo_context . RequestContext ( ) <EOL> go1 ( context ) <EOL> self . assertIsNot ( <EOL> enginefacade . _context_manager . _factory . _writer_engine , <EOL> alt_mgr1 . _factory . _writer_engine <EOL> ) <EOL> def test_replace_scope_only_global_eng ( self ) : <EOL> alt_connection1 = "<STR_LIT>" <EOL> alt_mgr1 = enginefacade . transaction_context ( ) <EOL> alt_mgr1 . configure ( <EOL> connection = alt_connection1 , <EOL> ) <EOL> alt_connection2 = "<STR_LIT>" <EOL> alt_mgr2 = enginefacade . transaction_context ( ) <EOL> alt_mgr2 . configure ( <EOL> connection = alt_connection2 , <EOL> ) <EOL> @ enginefacade . writer <EOL> def go1 ( context ) : <EOL> s1 = context . session <EOL> self . assertEqual ( s1 . bind . url , self . engine . url ) <EOL> with alt_mgr1 . replace . using ( context ) : <EOL> go2 ( context ) <EOL> go6 ( context ) <EOL> @ enginefacade . writer <EOL> def go2 ( context ) : <EOL> s2 = context . session <EOL> self . assertEqual ( str ( s2 . bind . url ) , alt_connection1 ) <EOL> self . assertIs ( s2 . bind , alt_mgr1 . _factory . _writer_engine ) <EOL> go3 ( context ) <EOL> @ alt_mgr2 . writer <EOL> def go3 ( context ) : <EOL> s3 = context . session <EOL> self . assertEqual ( str ( s3 . bind . url ) , alt_connection2 ) <EOL> self . assertIs ( s3 . bind , alt_mgr2 . _factory . _writer_engine ) <EOL> go4 ( context ) <EOL> @ enginefacade . writer <EOL> def go4 ( context ) : <EOL> s4 = context . session <EOL> self . assertEqual ( str ( s4 . bind . url ) , alt_connection1 ) <EOL> self . assertIs ( s4 . bind , alt_mgr1 . _factory . _writer_engine ) <EOL> @ enginefacade . writer <EOL> def go5 ( context ) : <EOL> s5 = context . session <EOL> self . assertEqual ( str ( s5 . bind . url ) , alt_connection1 ) <EOL> self . assertIs ( s5 . bind , alt_mgr1 . _factory . _writer_engine ) <EOL> @ enginefacade . writer <EOL> def go6 ( context ) : <EOL> s6 = context . session <EOL> self . assertEqual ( str ( s6 . bind . url ) , str ( self . engine . url ) ) <EOL> self . assertIs ( s6 . bind , self . engine ) <EOL> context = oslo_context . RequestContext ( ) <EOL> go1 ( context ) <EOL> class MySQLLiveFacadeTest ( <EOL> test_base . MySQLOpportunisticTestCase , LiveFacadeTest ) : <EOL> pass <EOL> class PGLiveFacadeTest ( <EOL> test_base . PostgreSQLOpportunisticTestCase , LiveFacadeTest ) : <EOL> pass <EOL> class ConfigOptionsTest ( oslo_test_base . BaseTestCase ) : <EOL> def test_all_options ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = enginefacade . _TransactionFactory ( ) <EOL> cfg . CONF . register_opts ( options . database_opts , '<STR_LIT>' ) <EOL> factory . configure ( ** dict ( cfg . CONF . database . items ( ) ) ) <EOL> def test_options_not_supported ( self ) : <EOL> factory = enginefacade . _TransactionFactory ( ) <EOL> with warnings . catch_warnings ( record = True ) as w : <EOL> warnings . simplefilter ( "<STR_LIT>" ) <EOL> factory . configure ( fake1 = '<STR_LIT:x>' , idle_timeout = <NUM_LIT:200> , wrong2 = '<STR_LIT:y>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( w ) ) <EOL> self . assertTrue ( <EOL> issubclass ( w [ - <NUM_LIT:1> ] . category , exception . NotSupportedWarning ) ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , <EOL> str ( w [ - <NUM_LIT:1> ] . message ) <EOL> ) <EOL> def test_no_engine ( self ) : <EOL> factory = enginefacade . _TransactionFactory ( ) <EOL> self . assertRaises ( <EOL> exception . CantStartEngineError , <EOL> factory . _create_session , enginefacade . _WRITER <EOL> ) <EOL> self . assertRaises ( <EOL> exception . CantStartEngineError , <EOL> factory . _create_session , enginefacade . _WRITER <EOL> ) </s>
