<s> import threading <EOL> import time <EOL> from sqlalchemy import pool , select , event <EOL> import sqlalchemy as tsa <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing . util import gc_collect , lazy_gc <EOL> from sqlalchemy . testing import eq_ , assert_raises , is_not_ , is_ <EOL> from sqlalchemy . testing . engines import testing_engine <EOL> from sqlalchemy . testing import fixtures <EOL> import random <EOL> from sqlalchemy . testing . mock import Mock , call , patch , ANY <EOL> import weakref <EOL> import collections <EOL> join_timeout = <NUM_LIT:10> <EOL> def MockDBAPI ( ) : <EOL> def cursor ( ) : <EOL> return Mock ( ) <EOL> def connect ( * arg , ** kw ) : <EOL> return Mock ( cursor = Mock ( side_effect = cursor ) ) <EOL> def shutdown ( value ) : <EOL> if value : <EOL> db . connect = Mock ( side_effect = Exception ( "<STR_LIT>" ) ) <EOL> else : <EOL> db . connect = Mock ( side_effect = connect ) <EOL> db . is_shutdown = value <EOL> db = Mock ( <EOL> connect = Mock ( side_effect = connect ) , <EOL> shutdown = shutdown , <EOL> is_shutdown = False ) <EOL> return db <EOL> class PoolTestBase ( fixtures . TestBase ) : <EOL> def setup ( self ) : <EOL> pool . clear_managers ( ) <EOL> self . _teardown_conns = [ ] <EOL> def teardown ( self ) : <EOL> for ref in self . _teardown_conns : <EOL> conn = ref ( ) <EOL> if conn : <EOL> conn . close ( ) <EOL> @ classmethod <EOL> def teardown_class ( cls ) : <EOL> pool . clear_managers ( ) <EOL> def _with_teardown ( self , connection ) : <EOL> self . _teardown_conns . append ( weakref . ref ( connection ) ) <EOL> return connection <EOL> def _queuepool_fixture ( self , ** kw ) : <EOL> dbapi , pool = self . _queuepool_dbapi_fixture ( ** kw ) <EOL> return pool <EOL> def _queuepool_dbapi_fixture ( self , ** kw ) : <EOL> dbapi = MockDBAPI ( ) <EOL> return dbapi , pool . QueuePool ( creator = lambda : dbapi . connect ( '<STR_LIT>' ) , <EOL> ** kw ) <EOL> class PoolTest ( PoolTestBase ) : <EOL> def test_manager ( self ) : <EOL> manager = pool . manage ( MockDBAPI ( ) , use_threadlocal = True ) <EOL> c1 = manager . connect ( '<STR_LIT>' ) <EOL> c2 = manager . connect ( '<STR_LIT>' ) <EOL> c3 = manager . connect ( '<STR_LIT>' ) <EOL> c4 = manager . connect ( "<STR_LIT>" , bar = "<STR_LIT>" ) <EOL> c5 = manager . connect ( "<STR_LIT>" , bar = "<STR_LIT>" ) <EOL> c6 = manager . connect ( "<STR_LIT>" , bar = "<STR_LIT>" ) <EOL> assert c1 . cursor ( ) is not None <EOL> assert c1 is c2 <EOL> assert c1 is not c3 <EOL> assert c4 is c6 <EOL> assert c4 is not c5 <EOL> def test_manager_with_key ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> manager = pool . manage ( dbapi , use_threadlocal = True ) <EOL> c1 = manager . connect ( '<STR_LIT>' , sa_pool_key = "<STR_LIT:a>" ) <EOL> c2 = manager . connect ( '<STR_LIT>' , sa_pool_key = "<STR_LIT:b>" ) <EOL> c3 = manager . connect ( '<STR_LIT>' , sa_pool_key = "<STR_LIT:a>" ) <EOL> assert c1 . cursor ( ) is not None <EOL> assert c1 is not c2 <EOL> assert c1 is c3 <EOL> eq_ ( dbapi . connect . mock_calls , <EOL> [ <EOL> call ( "<STR_LIT>" ) , <EOL> call ( "<STR_LIT>" ) , <EOL> ] <EOL> ) <EOL> def test_bad_args ( self ) : <EOL> manager = pool . manage ( MockDBAPI ( ) ) <EOL> manager . connect ( None ) <EOL> def test_non_thread_local_manager ( self ) : <EOL> manager = pool . manage ( MockDBAPI ( ) , use_threadlocal = False ) <EOL> connection = manager . connect ( '<STR_LIT>' ) <EOL> connection2 = manager . connect ( '<STR_LIT>' ) <EOL> self . assert_ ( connection . cursor ( ) is not None ) <EOL> self . assert_ ( connection is not connection2 ) <EOL> @ testing . fails_on ( '<STR_LIT>' , <EOL> "<STR_LIT>" ) <EOL> @ testing . fails_on ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_cursor_iterable ( self ) : <EOL> conn = testing . db . raw_connection ( ) <EOL> cursor = conn . cursor ( ) <EOL> cursor . execute ( str ( select ( [ <NUM_LIT:1> ] , bind = testing . db ) ) ) <EOL> expected = [ ( <NUM_LIT:1> , ) ] <EOL> for row in cursor : <EOL> eq_ ( row , expected . pop ( <NUM_LIT:0> ) ) <EOL> def test_no_connect_on_recreate ( self ) : <EOL> def creator ( ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> for cls in ( pool . SingletonThreadPool , pool . StaticPool , <EOL> pool . QueuePool , pool . NullPool , pool . AssertionPool ) : <EOL> p = cls ( creator = creator ) <EOL> p . dispose ( ) <EOL> p2 = p . recreate ( ) <EOL> assert p2 . __class__ is cls <EOL> mock_dbapi = MockDBAPI ( ) <EOL> p = cls ( creator = mock_dbapi . connect ) <EOL> conn = p . connect ( ) <EOL> conn . close ( ) <EOL> mock_dbapi . connect . side_effect = Exception ( "<STR_LIT>" ) <EOL> p . dispose ( ) <EOL> p . recreate ( ) <EOL> def test_threadlocal_del ( self ) : <EOL> self . _do_testthreadlocal ( useclose = False ) <EOL> def test_threadlocal_close ( self ) : <EOL> self . _do_testthreadlocal ( useclose = True ) <EOL> def _do_testthreadlocal ( self , useclose = False ) : <EOL> dbapi = MockDBAPI ( ) <EOL> for p in pool . QueuePool ( creator = dbapi . connect , <EOL> pool_size = <NUM_LIT:3> , max_overflow = - <NUM_LIT:1> , <EOL> use_threadlocal = True ) , pool . SingletonThreadPool ( creator = dbapi . connect , <EOL> use_threadlocal = True ) : <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> self . assert_ ( c1 is c2 ) <EOL> c3 = p . unique_connection ( ) <EOL> self . assert_ ( c3 is not c1 ) <EOL> if useclose : <EOL> c2 . close ( ) <EOL> else : <EOL> c2 = None <EOL> c2 = p . connect ( ) <EOL> self . assert_ ( c1 is c2 ) <EOL> self . assert_ ( c3 is not c1 ) <EOL> if useclose : <EOL> c2 . close ( ) <EOL> else : <EOL> c2 = None <EOL> lazy_gc ( ) <EOL> if useclose : <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> c3 = p . connect ( ) <EOL> c3 . close ( ) <EOL> c2 . close ( ) <EOL> self . assert_ ( c1 . connection is not None ) <EOL> c1 . close ( ) <EOL> c1 = c2 = c3 = None <EOL> if isinstance ( p , pool . QueuePool ) : <EOL> lazy_gc ( ) <EOL> self . assert_ ( p . checkedout ( ) == <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> if useclose : <EOL> c2 . close ( ) <EOL> c1 . close ( ) <EOL> else : <EOL> c2 = None <EOL> c1 = None <EOL> lazy_gc ( ) <EOL> self . assert_ ( p . checkedout ( ) == <NUM_LIT:0> ) <EOL> def test_info ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:0> ) <EOL> c = p . connect ( ) <EOL> self . assert_ ( not c . info ) <EOL> self . assert_ ( c . info is c . _connection_record . info ) <EOL> c . info [ '<STR_LIT:foo>' ] = '<STR_LIT:bar>' <EOL> c . close ( ) <EOL> del c <EOL> c = p . connect ( ) <EOL> self . assert_ ( '<STR_LIT:foo>' in c . info ) <EOL> c . invalidate ( ) <EOL> c = p . connect ( ) <EOL> self . assert_ ( '<STR_LIT:foo>' not in c . info ) <EOL> c . info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> c . detach ( ) <EOL> self . assert_ ( '<STR_LIT>' in c . info ) <EOL> c2 = p . connect ( ) <EOL> is_not_ ( c . connection , c2 . connection ) <EOL> assert not c2 . info <EOL> assert '<STR_LIT>' in c . info <EOL> class PoolDialectTest ( PoolTestBase ) : <EOL> def _dialect ( self ) : <EOL> canary = [ ] <EOL> class PoolDialect ( object ) : <EOL> def do_rollback ( self , dbapi_connection ) : <EOL> canary . append ( '<STR_LIT:R>' ) <EOL> dbapi_connection . rollback ( ) <EOL> def do_commit ( self , dbapi_connection ) : <EOL> canary . append ( '<STR_LIT:C>' ) <EOL> dbapi_connection . commit ( ) <EOL> def do_close ( self , dbapi_connection ) : <EOL> canary . append ( '<STR_LIT>' ) <EOL> dbapi_connection . close ( ) <EOL> return PoolDialect ( ) , canary <EOL> def _do_test ( self , pool_cls , assertion ) : <EOL> mock_dbapi = MockDBAPI ( ) <EOL> dialect , canary = self . _dialect ( ) <EOL> p = pool_cls ( creator = mock_dbapi . connect ) <EOL> p . _dialect = dialect <EOL> conn = p . connect ( ) <EOL> conn . close ( ) <EOL> p . dispose ( ) <EOL> p . recreate ( ) <EOL> conn = p . connect ( ) <EOL> conn . close ( ) <EOL> eq_ ( canary , assertion ) <EOL> def test_queue_pool ( self ) : <EOL> self . _do_test ( pool . QueuePool , [ '<STR_LIT:R>' , '<STR_LIT>' , '<STR_LIT:R>' ] ) <EOL> def test_assertion_pool ( self ) : <EOL> self . _do_test ( pool . AssertionPool , [ '<STR_LIT:R>' , '<STR_LIT>' , '<STR_LIT:R>' ] ) <EOL> def test_singleton_pool ( self ) : <EOL> self . _do_test ( pool . SingletonThreadPool , [ '<STR_LIT:R>' , '<STR_LIT>' , '<STR_LIT:R>' ] ) <EOL> def test_null_pool ( self ) : <EOL> self . _do_test ( pool . NullPool , [ '<STR_LIT:R>' , '<STR_LIT>' , '<STR_LIT:R>' , '<STR_LIT>' ] ) <EOL> def test_static_pool ( self ) : <EOL> self . _do_test ( pool . StaticPool , [ '<STR_LIT:R>' , '<STR_LIT:R>' ] ) <EOL> class PoolEventsTest ( PoolTestBase ) : <EOL> def _first_connect_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = [ ] <EOL> def first_connect ( * arg , ** kw ) : <EOL> canary . append ( '<STR_LIT>' ) <EOL> event . listen ( p , '<STR_LIT>' , first_connect ) <EOL> return p , canary <EOL> def _connect_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = [ ] <EOL> def connect ( * arg , ** kw ) : <EOL> canary . append ( '<STR_LIT>' ) <EOL> event . listen ( p , '<STR_LIT>' , connect ) <EOL> return p , canary <EOL> def _checkout_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = [ ] <EOL> def checkout ( * arg , ** kw ) : <EOL> canary . append ( '<STR_LIT>' ) <EOL> event . listen ( p , '<STR_LIT>' , checkout ) <EOL> return p , canary <EOL> def _checkin_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = [ ] <EOL> def checkin ( * arg , ** kw ) : <EOL> canary . append ( '<STR_LIT>' ) <EOL> event . listen ( p , '<STR_LIT>' , checkin ) <EOL> return p , canary <EOL> def _reset_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = [ ] <EOL> def reset ( * arg , ** kw ) : <EOL> canary . append ( '<STR_LIT>' ) <EOL> event . listen ( p , '<STR_LIT>' , reset ) <EOL> return p , canary <EOL> def _invalidate_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = Mock ( ) <EOL> event . listen ( p , '<STR_LIT>' , canary ) <EOL> return p , canary <EOL> def _soft_invalidate_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = Mock ( ) <EOL> event . listen ( p , '<STR_LIT>' , canary ) <EOL> return p , canary <EOL> def _close_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = Mock ( ) <EOL> event . listen ( p , '<STR_LIT>' , canary ) <EOL> return p , canary <EOL> def _detach_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = Mock ( ) <EOL> event . listen ( p , '<STR_LIT>' , canary ) <EOL> return p , canary <EOL> def _close_detached_event_fixture ( self ) : <EOL> p = self . _queuepool_fixture ( ) <EOL> canary = Mock ( ) <EOL> event . listen ( p , '<STR_LIT>' , canary ) <EOL> return p , canary <EOL> def test_close ( self ) : <EOL> p , canary = self . _close_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> connection = c1 . connection <EOL> rec = c1 . _connection_record <EOL> c1 . close ( ) <EOL> eq_ ( canary . mock_calls , [ ] ) <EOL> p . dispose ( ) <EOL> eq_ ( canary . mock_calls , [ call ( connection , rec ) ] ) <EOL> def test_detach ( self ) : <EOL> p , canary = self . _detach_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> connection = c1 . connection <EOL> rec = c1 . _connection_record <EOL> c1 . detach ( ) <EOL> eq_ ( canary . mock_calls , [ call ( connection , rec ) ] ) <EOL> def test_detach_close ( self ) : <EOL> p , canary = self . _close_detached_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> connection = c1 . connection <EOL> c1 . detach ( ) <EOL> c1 . close ( ) <EOL> eq_ ( canary . mock_calls , [ call ( connection ) ] ) <EOL> def test_first_connect_event ( self ) : <EOL> p , canary = self . _first_connect_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_first_connect_event_fires_once ( self ) : <EOL> p , canary = self . _first_connect_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_first_connect_on_previously_recreated ( self ) : <EOL> p , canary = self . _first_connect_event_fixture ( ) <EOL> p2 = p . recreate ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p2 . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_first_connect_on_subsequently_recreated ( self ) : <EOL> p , canary = self . _first_connect_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> p2 = p . recreate ( ) <EOL> c2 = p2 . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_connect_event ( self ) : <EOL> p , canary = self . _connect_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_connect_event_fires_subsequent ( self ) : <EOL> p , canary = self . _connect_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_connect_on_previously_recreated ( self ) : <EOL> p , canary = self . _connect_event_fixture ( ) <EOL> p2 = p . recreate ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p2 . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_connect_on_subsequently_recreated ( self ) : <EOL> p , canary = self . _connect_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> p2 = p . recreate ( ) <EOL> c2 = p2 . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_checkout_event ( self ) : <EOL> p , canary = self . _checkout_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_checkout_event_fires_subsequent ( self ) : <EOL> p , canary = self . _checkout_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_checkout_event_on_subsequently_recreated ( self ) : <EOL> p , canary = self . _checkout_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> p2 = p . recreate ( ) <EOL> c2 = p2 . connect ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_checkin_event ( self ) : <EOL> p , canary = self . _checkin_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> eq_ ( canary , [ ] ) <EOL> c1 . close ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_reset_event ( self ) : <EOL> p , canary = self . _reset_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> eq_ ( canary , [ ] ) <EOL> c1 . close ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_soft_invalidate_event_no_exception ( self ) : <EOL> p , canary = self . _soft_invalidate_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert not canary . called <EOL> c1 = p . connect ( ) <EOL> dbapi_con = c1 . connection <EOL> c1 . invalidate ( soft = True ) <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] is dbapi_con <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] is None <EOL> def test_soft_invalidate_event_exception ( self ) : <EOL> p , canary = self . _soft_invalidate_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert not canary . called <EOL> c1 = p . connect ( ) <EOL> dbapi_con = c1 . connection <EOL> exc = Exception ( "<STR_LIT>" ) <EOL> c1 . invalidate ( exc , soft = True ) <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] is dbapi_con <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] is exc <EOL> def test_invalidate_event_no_exception ( self ) : <EOL> p , canary = self . _invalidate_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert not canary . called <EOL> c1 = p . connect ( ) <EOL> dbapi_con = c1 . connection <EOL> c1 . invalidate ( ) <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] is dbapi_con <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] is None <EOL> def test_invalidate_event_exception ( self ) : <EOL> p , canary = self . _invalidate_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert not canary . called <EOL> c1 = p . connect ( ) <EOL> dbapi_con = c1 . connection <EOL> exc = Exception ( "<STR_LIT>" ) <EOL> c1 . invalidate ( exc ) <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] is dbapi_con <EOL> assert canary . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] is exc <EOL> def test_checkin_event_gc ( self ) : <EOL> p , canary = self . _checkin_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> eq_ ( canary , [ ] ) <EOL> del c1 <EOL> lazy_gc ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> def test_checkin_event_on_subsequently_recreated ( self ) : <EOL> p , canary = self . _checkin_event_fixture ( ) <EOL> c1 = p . connect ( ) <EOL> p2 = p . recreate ( ) <EOL> c2 = p2 . connect ( ) <EOL> eq_ ( canary , [ ] ) <EOL> c1 . close ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' ] ) <EOL> c2 . close ( ) <EOL> eq_ ( canary , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_listen_targets_scope ( self ) : <EOL> canary = [ ] <EOL> def listen_one ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> def listen_two ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> def listen_three ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> def listen_four ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> engine = testing_engine ( testing . db . url ) <EOL> event . listen ( pool . Pool , '<STR_LIT>' , listen_one ) <EOL> event . listen ( engine . pool , '<STR_LIT>' , listen_two ) <EOL> event . listen ( engine , '<STR_LIT>' , listen_three ) <EOL> event . listen ( engine . __class__ , '<STR_LIT>' , listen_four ) <EOL> engine . execute ( select ( [ <NUM_LIT:1> ] ) ) . close ( ) <EOL> eq_ ( <EOL> canary , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> ) <EOL> def test_listen_targets_per_subclass ( self ) : <EOL> """<STR_LIT>""" <EOL> canary = [ ] <EOL> def listen_one ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> def listen_two ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> def listen_three ( * args ) : <EOL> canary . append ( "<STR_LIT>" ) <EOL> event . listen ( pool . Pool , '<STR_LIT>' , listen_one ) <EOL> event . listen ( pool . QueuePool , '<STR_LIT>' , listen_two ) <EOL> event . listen ( pool . SingletonThreadPool , '<STR_LIT>' , listen_three ) <EOL> p1 = pool . QueuePool ( creator = MockDBAPI ( ) . connect ) <EOL> p2 = pool . SingletonThreadPool ( creator = MockDBAPI ( ) . connect ) <EOL> assert listen_one in p1 . dispatch . connect <EOL> assert listen_two in p1 . dispatch . connect <EOL> assert listen_three not in p1 . dispatch . connect <EOL> assert listen_one in p2 . dispatch . connect <EOL> assert listen_two not in p2 . dispatch . connect <EOL> assert listen_three in p2 . dispatch . connect <EOL> p1 . connect ( ) <EOL> eq_ ( canary , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> p2 . connect ( ) <EOL> eq_ ( canary , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def teardown ( self ) : <EOL> pool . Pool . dispatch . _clear ( ) <EOL> class PoolFirstConnectSyncTest ( PoolTestBase ) : <EOL> @ testing . requires . timing_intensive <EOL> def test_sync ( self ) : <EOL> pool = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , max_overflow = <NUM_LIT:0> ) <EOL> evt = Mock ( ) <EOL> @ event . listens_for ( pool , '<STR_LIT>' ) <EOL> def slow_first_connect ( dbapi_con , rec ) : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> evt . first_connect ( ) <EOL> @ event . listens_for ( pool , '<STR_LIT>' ) <EOL> def on_connect ( dbapi_con , rec ) : <EOL> evt . connect ( ) <EOL> def checkout ( ) : <EOL> for j in range ( <NUM_LIT:2> ) : <EOL> c1 = pool . connect ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> c1 . close ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:5> ) : <EOL> th = threading . Thread ( target = checkout ) <EOL> th . start ( ) <EOL> threads . append ( th ) <EOL> for th in threads : <EOL> th . join ( join_timeout ) <EOL> eq_ ( evt . mock_calls , <EOL> [ <EOL> call . first_connect ( ) , <EOL> call . connect ( ) , <EOL> call . connect ( ) , <EOL> call . connect ( ) ] <EOL> ) <EOL> class DeprecatedPoolListenerTest ( PoolTestBase ) : <EOL> @ testing . requires . predictable_gc <EOL> @ testing . uses_deprecated ( r"<STR_LIT>" ) <EOL> def test_listeners ( self ) : <EOL> class InstrumentingListener ( object ) : <EOL> def __init__ ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . connect = self . inst_connect <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . first_connect = self . inst_first_connect <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . checkout = self . inst_checkout <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . checkin = self . inst_checkin <EOL> self . clear ( ) <EOL> def clear ( self ) : <EOL> self . connected = [ ] <EOL> self . first_connected = [ ] <EOL> self . checked_out = [ ] <EOL> self . checked_in = [ ] <EOL> def assert_total ( innerself , conn , fconn , cout , cin ) : <EOL> eq_ ( len ( innerself . connected ) , conn ) <EOL> eq_ ( len ( innerself . first_connected ) , fconn ) <EOL> eq_ ( len ( innerself . checked_out ) , cout ) <EOL> eq_ ( len ( innerself . checked_in ) , cin ) <EOL> def assert_in ( innerself , item , in_conn , in_fconn , <EOL> in_cout , in_cin ) : <EOL> self . assert_ ( ( item in innerself . connected ) == in_conn ) <EOL> self . assert_ ( ( item in innerself . first_connected ) == in_fconn ) <EOL> self . assert_ ( ( item in innerself . checked_out ) == in_cout ) <EOL> self . assert_ ( ( item in innerself . checked_in ) == in_cin ) <EOL> def inst_connect ( self , con , record ) : <EOL> print ( "<STR_LIT>" % ( con , record ) ) <EOL> assert con is not None <EOL> assert record is not None <EOL> self . connected . append ( con ) <EOL> def inst_first_connect ( self , con , record ) : <EOL> print ( "<STR_LIT>" % ( con , record ) ) <EOL> assert con is not None <EOL> assert record is not None <EOL> self . first_connected . append ( con ) <EOL> def inst_checkout ( self , con , record , proxy ) : <EOL> print ( "<STR_LIT>" % ( con , record , proxy ) ) <EOL> assert con is not None <EOL> assert record is not None <EOL> assert proxy is not None <EOL> self . checked_out . append ( con ) <EOL> def inst_checkin ( self , con , record ) : <EOL> print ( "<STR_LIT>" % ( con , record ) ) <EOL> assert record is not None <EOL> self . checked_in . append ( con ) <EOL> class ListenAll ( tsa . interfaces . PoolListener , InstrumentingListener ) : <EOL> pass <EOL> class ListenConnect ( InstrumentingListener ) : <EOL> def connect ( self , con , record ) : <EOL> pass <EOL> class ListenFirstConnect ( InstrumentingListener ) : <EOL> def first_connect ( self , con , record ) : <EOL> pass <EOL> class ListenCheckOut ( InstrumentingListener ) : <EOL> def checkout ( self , con , record , proxy , num ) : <EOL> pass <EOL> class ListenCheckIn ( InstrumentingListener ) : <EOL> def checkin ( self , con , record ) : <EOL> pass <EOL> def assert_listeners ( p , total , conn , fconn , cout , cin ) : <EOL> for instance in ( p , p . recreate ( ) ) : <EOL> self . assert_ ( len ( instance . dispatch . connect ) == conn ) <EOL> self . assert_ ( len ( instance . dispatch . first_connect ) == fconn ) <EOL> self . assert_ ( len ( instance . dispatch . checkout ) == cout ) <EOL> self . assert_ ( len ( instance . dispatch . checkin ) == cin ) <EOL> p = self . _queuepool_fixture ( ) <EOL> assert_listeners ( p , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> p . add_listener ( ListenAll ( ) ) <EOL> assert_listeners ( p , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . add_listener ( ListenConnect ( ) ) <EOL> assert_listeners ( p , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . add_listener ( ListenFirstConnect ( ) ) <EOL> assert_listeners ( p , <NUM_LIT:3> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . add_listener ( ListenCheckOut ( ) ) <EOL> assert_listeners ( p , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> p . add_listener ( ListenCheckIn ( ) ) <EOL> assert_listeners ( p , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> ) <EOL> del p <EOL> snoop = ListenAll ( ) <EOL> p = self . _queuepool_fixture ( listeners = [ snoop ] ) <EOL> assert_listeners ( p , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> cc = c . connection <EOL> snoop . assert_in ( cc , True , True , True , False ) <EOL> c . close ( ) <EOL> snoop . assert_in ( cc , True , True , True , True ) <EOL> del c , cc <EOL> snoop . clear ( ) <EOL> c = p . connect ( ) <EOL> cc = c . connection <EOL> snoop . assert_in ( cc , False , False , True , False ) <EOL> snoop . assert_total ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> del c , cc <EOL> lazy_gc ( ) <EOL> snoop . assert_total ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . dispose ( ) <EOL> snoop . clear ( ) <EOL> c = p . connect ( ) <EOL> c . close ( ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> c . close ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:2> ) <EOL> p . dispose ( ) <EOL> snoop . clear ( ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> c . invalidate ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> c . close ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> del c <EOL> lazy_gc ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> c . close ( ) <EOL> del c <EOL> lazy_gc ( ) <EOL> snoop . assert_total ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:2> ) <EOL> p . dispose ( ) <EOL> snoop . clear ( ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> c . detach ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> c . close ( ) <EOL> del c <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> c . close ( ) <EOL> del c <EOL> snoop . assert_total ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> p = p . recreate ( ) <EOL> snoop . clear ( ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> c . close ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> c = p . connect ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> c . close ( ) <EOL> snoop . assert_total ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:2> ) <EOL> @ testing . uses_deprecated ( r"<STR_LIT>" ) <EOL> def test_listeners_callables ( self ) : <EOL> def connect ( dbapi_con , con_record ) : <EOL> counts [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> def checkout ( dbapi_con , con_record , con_proxy ) : <EOL> counts [ <NUM_LIT:1> ] += <NUM_LIT:1> <EOL> def checkin ( dbapi_con , con_record ) : <EOL> counts [ <NUM_LIT:2> ] += <NUM_LIT:1> <EOL> i_all = dict ( connect = connect , checkout = checkout , checkin = checkin ) <EOL> i_connect = dict ( connect = connect ) <EOL> i_checkout = dict ( checkout = checkout ) <EOL> i_checkin = dict ( checkin = checkin ) <EOL> for cls in ( pool . QueuePool , pool . StaticPool ) : <EOL> counts = [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] <EOL> def assert_listeners ( p , total , conn , cout , cin ) : <EOL> for instance in ( p , p . recreate ( ) ) : <EOL> eq_ ( len ( instance . dispatch . connect ) , conn ) <EOL> eq_ ( len ( instance . dispatch . checkout ) , cout ) <EOL> eq_ ( len ( instance . dispatch . checkin ) , cin ) <EOL> p = self . _queuepool_fixture ( ) <EOL> assert_listeners ( p , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> p . add_listener ( i_all ) <EOL> assert_listeners ( p , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . add_listener ( i_connect ) <EOL> assert_listeners ( p , <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . add_listener ( i_checkout ) <EOL> assert_listeners ( p , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> p . add_listener ( i_checkin ) <EOL> assert_listeners ( p , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> del p <EOL> p = self . _queuepool_fixture ( listeners = [ i_all ] ) <EOL> assert_listeners ( p , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> c = p . connect ( ) <EOL> assert counts == [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> ] <EOL> c . close ( ) <EOL> assert counts == [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] <EOL> c = p . connect ( ) <EOL> assert counts == [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:1> ] <EOL> p . add_listener ( i_checkin ) <EOL> c . close ( ) <EOL> assert counts == [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:2> ] <EOL> class QueuePoolTest ( PoolTestBase ) : <EOL> def test_queuepool_del ( self ) : <EOL> self . _do_testqueuepool ( useclose = False ) <EOL> def test_queuepool_close ( self ) : <EOL> self . _do_testqueuepool ( useclose = True ) <EOL> def _do_testqueuepool ( self , useclose = False ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , <EOL> max_overflow = - <NUM_LIT:1> ) <EOL> def status ( pool ) : <EOL> return pool . size ( ) , pool . checkedin ( ) , pool . overflow ( ) , pool . checkedout ( ) <EOL> c1 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:0> , - <NUM_LIT:2> , <NUM_LIT:1> ) ) <EOL> c2 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:0> , - <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> c3 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:3> ) ) <EOL> c4 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:4> ) ) <EOL> c5 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:5> ) ) <EOL> c6 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT:6> ) ) <EOL> if useclose : <EOL> c4 . close ( ) <EOL> c3 . close ( ) <EOL> c2 . close ( ) <EOL> else : <EOL> c4 = c3 = c2 = None <EOL> lazy_gc ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:3> , <NUM_LIT:3> , <NUM_LIT:3> ) ) <EOL> if useclose : <EOL> c1 . close ( ) <EOL> c5 . close ( ) <EOL> c6 . close ( ) <EOL> else : <EOL> c1 = c5 = c6 = None <EOL> lazy_gc ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:2> ) , status ( p ) ) <EOL> if useclose : <EOL> c2 . close ( ) <EOL> else : <EOL> c2 = None <EOL> lazy_gc ( ) <EOL> self . assert_ ( status ( p ) == ( <NUM_LIT:3> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:1> ) ) <EOL> c1 . close ( ) <EOL> lazy_gc ( ) <EOL> assert not pool . _refs <EOL> @ testing . requires . timing_intensive <EOL> def test_timeout ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , <EOL> max_overflow = <NUM_LIT:0> , <EOL> timeout = <NUM_LIT:2> ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> c3 = p . connect ( ) <EOL> now = time . time ( ) <EOL> try : <EOL> c4 = p . connect ( ) <EOL> assert False <EOL> except tsa . exc . TimeoutError : <EOL> assert int ( time . time ( ) - now ) == <NUM_LIT:2> <EOL> @ testing . requires . threading_with_mock <EOL> @ testing . requires . timing_intensive <EOL> def test_timeout_race ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> p = pool . QueuePool ( <EOL> creator = lambda : dbapi . connect ( delay = <NUM_LIT> ) , <EOL> pool_size = <NUM_LIT:2> , <EOL> max_overflow = <NUM_LIT:1> , use_threadlocal = False , timeout = <NUM_LIT:3> ) <EOL> timeouts = [ ] <EOL> def checkout ( ) : <EOL> for x in range ( <NUM_LIT:1> ) : <EOL> now = time . time ( ) <EOL> try : <EOL> c1 = p . connect ( ) <EOL> except tsa . exc . TimeoutError : <EOL> timeouts . append ( time . time ( ) - now ) <EOL> continue <EOL> time . sleep ( <NUM_LIT:4> ) <EOL> c1 . close ( ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> th = threading . Thread ( target = checkout ) <EOL> th . start ( ) <EOL> threads . append ( th ) <EOL> for th in threads : <EOL> th . join ( join_timeout ) <EOL> assert len ( timeouts ) > <NUM_LIT:0> <EOL> for t in timeouts : <EOL> assert t >= <NUM_LIT:3> , "<STR_LIT>" % timeouts <EOL> assert t < <NUM_LIT> , "<STR_LIT>" % timeouts <EOL> def _test_overflow ( self , thread_count , max_overflow ) : <EOL> gc_collect ( ) <EOL> dbapi = MockDBAPI ( ) <EOL> mutex = threading . Lock ( ) <EOL> def creator ( ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> with mutex : <EOL> return dbapi . connect ( ) <EOL> p = pool . QueuePool ( creator = creator , <EOL> pool_size = <NUM_LIT:3> , timeout = <NUM_LIT:2> , <EOL> max_overflow = max_overflow ) <EOL> peaks = [ ] <EOL> def whammy ( ) : <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> try : <EOL> con = p . connect ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> peaks . append ( p . overflow ( ) ) <EOL> con . close ( ) <EOL> del con <EOL> except tsa . exc . TimeoutError : <EOL> pass <EOL> threads = [ ] <EOL> for i in range ( thread_count ) : <EOL> th = threading . Thread ( target = whammy ) <EOL> th . start ( ) <EOL> threads . append ( th ) <EOL> for th in threads : <EOL> th . join ( join_timeout ) <EOL> self . assert_ ( max ( peaks ) <= max_overflow ) <EOL> lazy_gc ( ) <EOL> assert not pool . _refs <EOL> def test_overflow_reset_on_failed_connect ( self ) : <EOL> dbapi = Mock ( ) <EOL> def failing_dbapi ( ) : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> creator = dbapi . connect <EOL> def create ( ) : <EOL> return creator ( ) <EOL> p = pool . QueuePool ( creator = create , pool_size = <NUM_LIT:2> , max_overflow = <NUM_LIT:3> ) <EOL> c1 = self . _with_teardown ( p . connect ( ) ) <EOL> c2 = self . _with_teardown ( p . connect ( ) ) <EOL> c3 = self . _with_teardown ( p . connect ( ) ) <EOL> eq_ ( p . _overflow , <NUM_LIT:1> ) <EOL> creator = failing_dbapi <EOL> assert_raises ( Exception , p . connect ) <EOL> eq_ ( p . _overflow , <NUM_LIT:1> ) <EOL> @ testing . requires . threading_with_mock <EOL> @ testing . requires . timing_intensive <EOL> def test_hanging_connect_within_overflow ( self ) : <EOL> """<STR_LIT>""" <EOL> dbapi = Mock ( ) <EOL> mutex = threading . Lock ( ) <EOL> def hanging_dbapi ( ) : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> with mutex : <EOL> return dbapi . connect ( ) <EOL> def fast_dbapi ( ) : <EOL> with mutex : <EOL> return dbapi . connect ( ) <EOL> creator = threading . local ( ) <EOL> def create ( ) : <EOL> return creator . mock_connector ( ) <EOL> def run_test ( name , pool , should_hang ) : <EOL> if should_hang : <EOL> creator . mock_connector = hanging_dbapi <EOL> else : <EOL> creator . mock_connector = fast_dbapi <EOL> conn = pool . connect ( ) <EOL> conn . operation ( name ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> conn . close ( ) <EOL> p = pool . QueuePool ( creator = create , pool_size = <NUM_LIT:2> , max_overflow = <NUM_LIT:3> ) <EOL> threads = [ <EOL> threading . Thread ( <EOL> target = run_test , args = ( "<STR_LIT>" , p , False ) ) , <EOL> threading . Thread ( <EOL> target = run_test , args = ( "<STR_LIT>" , p , False ) ) , <EOL> threading . Thread ( <EOL> target = run_test , args = ( "<STR_LIT>" , p , True ) ) , <EOL> threading . Thread ( <EOL> target = run_test , args = ( "<STR_LIT>" , p , False ) ) , <EOL> threading . Thread ( <EOL> target = run_test , args = ( "<STR_LIT>" , p , False ) ) <EOL> ] <EOL> for t in threads : <EOL> t . start ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> for t in threads : <EOL> t . join ( timeout = join_timeout ) <EOL> eq_ ( <EOL> dbapi . connect ( ) . operation . mock_calls , <EOL> [ call ( "<STR_LIT>" ) , call ( "<STR_LIT>" ) , <EOL> call ( "<STR_LIT>" ) , call ( "<STR_LIT>" ) , <EOL> call ( "<STR_LIT>" ) ] <EOL> ) <EOL> @ testing . requires . threading_with_mock <EOL> @ testing . requires . timing_intensive <EOL> def test_waiters_handled ( self ) : <EOL> """<STR_LIT>""" <EOL> mutex = threading . Lock ( ) <EOL> dbapi = MockDBAPI ( ) <EOL> def creator ( ) : <EOL> mutex . acquire ( ) <EOL> try : <EOL> return dbapi . connect ( ) <EOL> finally : <EOL> mutex . release ( ) <EOL> success = [ ] <EOL> for timeout in ( None , <NUM_LIT:30> ) : <EOL> for max_overflow in ( <NUM_LIT:0> , - <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> p = pool . QueuePool ( creator = creator , <EOL> pool_size = <NUM_LIT:2> , timeout = timeout , <EOL> max_overflow = max_overflow ) <EOL> def waiter ( p , timeout , max_overflow ) : <EOL> success_key = ( timeout , max_overflow ) <EOL> conn = p . connect ( ) <EOL> success . append ( success_key ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> conn . close ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> t = threading . Thread ( target = waiter , <EOL> args = ( p , timeout , max_overflow ) ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> threads . append ( t ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> p . _invalidate ( c2 ) <EOL> for t in threads : <EOL> t . join ( join_timeout ) <EOL> eq_ ( len ( success ) , <NUM_LIT:12> , "<STR_LIT>" % success ) <EOL> def test_connrec_invalidated_within_checkout_no_race ( self ) : <EOL> """<STR_LIT>""" <EOL> dbapi = MockDBAPI ( ) <EOL> def creator ( ) : <EOL> return dbapi . connect ( ) <EOL> p = pool . QueuePool ( creator = creator , pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:0> ) <EOL> conn = p . connect ( ) <EOL> conn . close ( ) <EOL> _existing_checkout = pool . _ConnectionRecord . checkout <EOL> @ classmethod <EOL> def _decorate_existing_checkout ( cls , * arg , ** kw ) : <EOL> fairy = _existing_checkout ( * arg , ** kw ) <EOL> connrec = fairy . _connection_record <EOL> connrec . invalidate ( ) <EOL> return fairy <EOL> with patch ( <EOL> "<STR_LIT>" , <EOL> _decorate_existing_checkout ) : <EOL> conn = p . connect ( ) <EOL> is_ ( conn . _connection_record . connection , None ) <EOL> conn . close ( ) <EOL> @ testing . requires . threading_with_mock <EOL> @ testing . requires . timing_intensive <EOL> def test_notify_waiters ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> canary = [ ] <EOL> def creator ( ) : <EOL> canary . append ( <NUM_LIT:1> ) <EOL> return dbapi . connect ( ) <EOL> p1 = pool . QueuePool ( creator = creator , <EOL> pool_size = <NUM_LIT:1> , timeout = None , <EOL> max_overflow = <NUM_LIT:0> ) <EOL> def waiter ( p ) : <EOL> conn = p . connect ( ) <EOL> canary . append ( <NUM_LIT:2> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> conn . close ( ) <EOL> c1 = p1 . connect ( ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:5> ) : <EOL> t = threading . Thread ( target = waiter , args = ( p1 , ) ) <EOL> t . start ( ) <EOL> threads . append ( t ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> eq_ ( canary , [ <NUM_LIT:1> ] ) <EOL> p1 . _invalidate ( c1 ) <EOL> for t in threads : <EOL> t . join ( join_timeout ) <EOL> eq_ ( canary , [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> def test_dispose_closes_pooled ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> p = pool . QueuePool ( creator = dbapi . connect , <EOL> pool_size = <NUM_LIT:2> , timeout = None , <EOL> max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> c1_con = c1 . connection <EOL> c2_con = c2 . connection <EOL> c1 . close ( ) <EOL> eq_ ( c1_con . close . call_count , <NUM_LIT:0> ) <EOL> eq_ ( c2_con . close . call_count , <NUM_LIT:0> ) <EOL> p . dispose ( ) <EOL> eq_ ( c1_con . close . call_count , <NUM_LIT:1> ) <EOL> eq_ ( c2_con . close . call_count , <NUM_LIT:0> ) <EOL> c2 . close ( ) <EOL> eq_ ( c1_con . close . call_count , <NUM_LIT:1> ) <EOL> eq_ ( c2_con . close . call_count , <NUM_LIT:0> ) <EOL> c3 = p . connect ( ) <EOL> assert c3 . connection is c2_con <EOL> @ testing . requires . threading_with_mock <EOL> @ testing . requires . timing_intensive <EOL> def test_no_overflow ( self ) : <EOL> self . _test_overflow ( <NUM_LIT> , <NUM_LIT:0> ) <EOL> @ testing . requires . threading_with_mock <EOL> @ testing . requires . timing_intensive <EOL> def test_max_overflow ( self ) : <EOL> self . _test_overflow ( <NUM_LIT> , <NUM_LIT:5> ) <EOL> def test_mixed_close ( self ) : <EOL> pool . _refs . clear ( ) <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , max_overflow = - <NUM_LIT:1> , <EOL> use_threadlocal = True ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> assert c1 is c2 <EOL> c1 . close ( ) <EOL> c2 = None <EOL> assert p . checkedout ( ) == <NUM_LIT:1> <EOL> c1 = None <EOL> lazy_gc ( ) <EOL> assert p . checkedout ( ) == <NUM_LIT:0> <EOL> lazy_gc ( ) <EOL> assert not pool . _refs <EOL> def test_overflow_no_gc_tlocal ( self ) : <EOL> self . _test_overflow_no_gc ( True ) <EOL> def test_overflow_no_gc ( self ) : <EOL> self . _test_overflow_no_gc ( False ) <EOL> def _test_overflow_no_gc ( self , threadlocal ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:2> , <EOL> max_overflow = <NUM_LIT:2> ) <EOL> strong_refs = set ( ) <EOL> def _conn ( ) : <EOL> c = p . connect ( ) <EOL> strong_refs . add ( c . connection ) <EOL> return c <EOL> for j in range ( <NUM_LIT:5> ) : <EOL> conns = [ _conn ( ) for i in range ( <NUM_LIT:4> ) ] <EOL> for c in conns : <EOL> c . close ( ) <EOL> eq_ ( <EOL> set ( [ c . close . call_count for c in strong_refs ] ) , <EOL> set ( [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> ] ) <EOL> ) <EOL> @ testing . requires . predictable_gc <EOL> def test_weakref_kaboom ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , <EOL> max_overflow = - <NUM_LIT:1> , use_threadlocal = True ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> c1 . close ( ) <EOL> c2 = None <EOL> del c1 <EOL> del c2 <EOL> gc_collect ( ) <EOL> assert p . checkedout ( ) == <NUM_LIT:0> <EOL> c3 = p . connect ( ) <EOL> assert c3 is not None <EOL> def test_trick_the_counter ( self ) : <EOL> """<STR_LIT>""" <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , <EOL> max_overflow = - <NUM_LIT:1> , use_threadlocal = True ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> assert c1 is c2 <EOL> c1 . close ( ) <EOL> c2 = p . connect ( ) <EOL> c2 . close ( ) <EOL> self . assert_ ( p . checkedout ( ) != <NUM_LIT:0> ) <EOL> c2 . close ( ) <EOL> self . assert_ ( p . checkedout ( ) == <NUM_LIT:0> ) <EOL> def test_recycle ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:1> , <EOL> max_overflow = <NUM_LIT:0> , <EOL> recycle = <NUM_LIT:3> ) <EOL> c1 = p . connect ( ) <EOL> c_id = id ( c1 . connection ) <EOL> c1 . close ( ) <EOL> c2 = p . connect ( ) <EOL> assert id ( c2 . connection ) == c_id <EOL> c2 . close ( ) <EOL> time . sleep ( <NUM_LIT:4> ) <EOL> c3 = p . connect ( ) <EOL> assert id ( c3 . connection ) != c_id <EOL> @ testing . requires . timing_intensive <EOL> def test_recycle_on_invalidate ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:1> , <EOL> max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c_id = id ( c1 . connection ) <EOL> c1 . close ( ) <EOL> c2 = p . connect ( ) <EOL> assert id ( c2 . connection ) == c_id <EOL> c2_rec = c2 . _connection_record <EOL> p . _invalidate ( c2 ) <EOL> assert c2_rec . connection is None <EOL> c2 . close ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> c3 = p . connect ( ) <EOL> assert id ( c3 . connection ) != c_id <EOL> @ testing . requires . timing_intensive <EOL> def test_recycle_on_soft_invalidate ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:1> , <EOL> max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c_id = id ( c1 . connection ) <EOL> c1 . close ( ) <EOL> c2 = p . connect ( ) <EOL> assert id ( c2 . connection ) == c_id <EOL> c2_rec = c2 . _connection_record <EOL> c2 . invalidate ( soft = True ) <EOL> assert c2_rec . connection is c2 . connection <EOL> c2 . close ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> c3 = p . connect ( ) <EOL> assert id ( c3 . connection ) != c_id <EOL> assert c3 . _connection_record is c2_rec <EOL> assert c2_rec . connection is c3 . connection <EOL> def _no_wr_finalize ( self ) : <EOL> finalize_fairy = pool . _finalize_fairy <EOL> def assert_no_wr_callback ( <EOL> connection , connection_record , <EOL> pool , ref , echo , fairy = None ) : <EOL> if fairy is None : <EOL> raise AssertionError ( <EOL> "<STR_LIT>" ) <EOL> return finalize_fairy ( <EOL> connection , connection_record , pool , ref , echo , fairy ) <EOL> return patch . object ( <EOL> pool , '<STR_LIT>' , assert_no_wr_callback ) <EOL> def _assert_cleanup_on_pooled_reconnect ( self , dbapi , p ) : <EOL> with self . _no_wr_finalize ( ) : <EOL> eq_ ( p . checkedout ( ) , <NUM_LIT:0> ) <EOL> eq_ ( p . _overflow , <NUM_LIT:0> ) <EOL> dbapi . shutdown ( True ) <EOL> assert_raises ( <EOL> Exception , <EOL> p . connect <EOL> ) <EOL> eq_ ( p . _overflow , <NUM_LIT:0> ) <EOL> eq_ ( p . checkedout ( ) , <NUM_LIT:0> ) <EOL> dbapi . shutdown ( False ) <EOL> c1 = self . _with_teardown ( p . connect ( ) ) <EOL> assert p . _pool . empty ( ) <EOL> c2 = self . _with_teardown ( p . connect ( ) ) <EOL> eq_ ( p . _overflow , <NUM_LIT:1> ) <EOL> c3 = self . _with_teardown ( p . connect ( ) ) <EOL> def test_error_on_pooled_reconnect_cleanup_invalidate ( self ) : <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:2> ) <EOL> c1 = p . connect ( ) <EOL> c1 . invalidate ( ) <EOL> c1 . close ( ) <EOL> self . _assert_cleanup_on_pooled_reconnect ( dbapi , p ) <EOL> @ testing . requires . timing_intensive <EOL> def test_error_on_pooled_reconnect_cleanup_recycle ( self ) : <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( pool_size = <NUM_LIT:1> , <EOL> max_overflow = <NUM_LIT:2> , recycle = <NUM_LIT:1> ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . _assert_cleanup_on_pooled_reconnect ( dbapi , p ) <EOL> def test_connect_handler_not_called_for_recycled ( self ) : <EOL> """<STR_LIT>""" <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( <EOL> pool_size = <NUM_LIT:2> , max_overflow = <NUM_LIT:2> ) <EOL> canary = Mock ( ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> c1 . close ( ) <EOL> c2 . close ( ) <EOL> dbapi . shutdown ( True ) <EOL> bad = p . connect ( ) <EOL> p . _invalidate ( bad ) <EOL> bad . close ( ) <EOL> assert p . _invalidate_time <EOL> event . listen ( p , "<STR_LIT>" , canary . connect ) <EOL> event . listen ( p , "<STR_LIT>" , canary . checkout ) <EOL> assert_raises ( <EOL> Exception , <EOL> p . connect <EOL> ) <EOL> p . _pool . queue = collections . deque ( <EOL> [ <EOL> c for c in p . _pool . queue <EOL> if c . connection is not None <EOL> ] <EOL> ) <EOL> dbapi . shutdown ( False ) <EOL> c = p . connect ( ) <EOL> c . close ( ) <EOL> eq_ ( <EOL> canary . mock_calls , <EOL> [ <EOL> call . connect ( ANY , ANY ) , <EOL> call . checkout ( ANY , ANY , ANY ) <EOL> ] <EOL> ) <EOL> def test_connect_checkout_handler_always_gets_info ( self ) : <EOL> """<STR_LIT>""" <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( <EOL> pool_size = <NUM_LIT:2> , max_overflow = <NUM_LIT:2> ) <EOL> c1 = p . connect ( ) <EOL> c2 = p . connect ( ) <EOL> c1 . close ( ) <EOL> c2 . close ( ) <EOL> dbapi . shutdown ( True ) <EOL> bad = p . connect ( ) <EOL> p . _invalidate ( bad ) <EOL> bad . close ( ) <EOL> assert p . _invalidate_time <EOL> @ event . listens_for ( p , "<STR_LIT>" ) <EOL> def connect ( conn , conn_rec ) : <EOL> conn_rec . info [ '<STR_LIT:x>' ] = True <EOL> @ event . listens_for ( p , "<STR_LIT>" ) <EOL> def checkout ( conn , conn_rec , conn_f ) : <EOL> assert '<STR_LIT:x>' in conn_rec . info <EOL> assert_raises ( <EOL> Exception , <EOL> p . connect <EOL> ) <EOL> p . _pool . queue = collections . deque ( <EOL> [ <EOL> c for c in p . _pool . queue <EOL> if c . connection is not None <EOL> ] <EOL> ) <EOL> dbapi . shutdown ( False ) <EOL> c = p . connect ( ) <EOL> c . close ( ) <EOL> def test_error_on_pooled_reconnect_cleanup_wcheckout_event ( self ) : <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( pool_size = <NUM_LIT:1> , <EOL> max_overflow = <NUM_LIT:2> ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> @ event . listens_for ( p , "<STR_LIT>" ) <EOL> def handle_checkout_event ( dbapi_con , con_record , con_proxy ) : <EOL> if dbapi . is_shutdown : <EOL> raise tsa . exc . DisconnectionError ( ) <EOL> self . _assert_cleanup_on_pooled_reconnect ( dbapi , p ) <EOL> @ testing . requires . timing_intensive <EOL> def test_recycle_pool_no_race ( self ) : <EOL> def slow_close ( ) : <EOL> slow_closing_connection . _slow_close ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> slow_closing_connection = Mock ( ) <EOL> slow_closing_connection . connect . return_value . close = slow_close <EOL> class Error ( Exception ) : <EOL> pass <EOL> dialect = Mock ( ) <EOL> dialect . is_disconnect = lambda * arg , ** kw : True <EOL> dialect . dbapi . Error = Error <EOL> pools = [ ] <EOL> class TrackQueuePool ( pool . QueuePool ) : <EOL> def __init__ ( self , * arg , ** kw ) : <EOL> pools . append ( self ) <EOL> super ( TrackQueuePool , self ) . __init__ ( * arg , ** kw ) <EOL> def creator ( ) : <EOL> return slow_closing_connection . connect ( ) <EOL> p1 = TrackQueuePool ( creator = creator , pool_size = <NUM_LIT:20> ) <EOL> from sqlalchemy import create_engine <EOL> eng = create_engine ( testing . db . url , pool = p1 , _initialize = False ) <EOL> eng . dialect = dialect <EOL> conns = [ eng . connect ( ) for i in range ( <NUM_LIT:15> ) ] <EOL> for conn in conns [ <NUM_LIT:3> : <NUM_LIT:10> ] : <EOL> conn . close ( ) <EOL> def attempt ( conn ) : <EOL> time . sleep ( random . random ( ) ) <EOL> try : <EOL> conn . _handle_dbapi_exception ( Error ( ) , "<STR_LIT>" , { } , <EOL> Mock ( ) , Mock ( ) ) <EOL> except tsa . exc . DBAPIError : <EOL> pass <EOL> threads = [ ] <EOL> for conn in conns : <EOL> t = threading . Thread ( target = attempt , args = ( conn , ) ) <EOL> t . start ( ) <EOL> threads . append ( t ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> for conn in conns : <EOL> conn . close ( ) <EOL> conns = [ eng . connect ( ) for i in range ( <NUM_LIT:15> ) ] <EOL> assert slow_closing_connection . _slow_close . call_count == <NUM_LIT:15> <EOL> assert slow_closing_connection . connect . call_count == <NUM_LIT:30> <EOL> assert len ( pools ) <= <NUM_LIT:2> , len ( pools ) <EOL> def test_invalidate ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c_id = c1 . connection . id <EOL> c1 . close ( ) <EOL> c1 = None <EOL> c1 = p . connect ( ) <EOL> assert c1 . connection . id == c_id <EOL> c1 . invalidate ( ) <EOL> c1 = None <EOL> c1 = p . connect ( ) <EOL> assert c1 . connection . id != c_id <EOL> def test_recreate ( self ) : <EOL> p = self . _queuepool_fixture ( reset_on_return = None , pool_size = <NUM_LIT:1> , <EOL> max_overflow = <NUM_LIT:0> ) <EOL> p2 = p . recreate ( ) <EOL> assert p2 . size ( ) == <NUM_LIT:1> <EOL> assert p2 . _reset_on_return is pool . reset_none <EOL> assert p2 . _use_threadlocal is False <EOL> assert p2 . _max_overflow == <NUM_LIT:0> <EOL> def test_reconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c_id = c1 . connection . id <EOL> c1 . close ( ) <EOL> c1 = None <EOL> c1 = p . connect ( ) <EOL> assert c1 . connection . id == c_id <EOL> dbapi . raise_error = True <EOL> c1 . invalidate ( ) <EOL> c1 = None <EOL> c1 = p . connect ( ) <EOL> assert c1 . connection . id != c_id <EOL> def test_detach ( self ) : <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c1 . detach ( ) <EOL> c2 = p . connect ( ) <EOL> eq_ ( dbapi . connect . mock_calls , [ call ( "<STR_LIT>" ) , call ( "<STR_LIT>" ) ] ) <EOL> c1_con = c1 . connection <EOL> assert c1_con is not None <EOL> eq_ ( c1_con . close . call_count , <NUM_LIT:0> ) <EOL> c1 . close ( ) <EOL> eq_ ( c1_con . close . call_count , <NUM_LIT:1> ) <EOL> def test_detach_via_invalidate ( self ) : <EOL> dbapi , p = self . _queuepool_dbapi_fixture ( pool_size = <NUM_LIT:1> , max_overflow = <NUM_LIT:0> ) <EOL> c1 = p . connect ( ) <EOL> c1_con = c1 . connection <EOL> c1 . invalidate ( ) <EOL> assert c1 . connection is None <EOL> eq_ ( c1_con . close . call_count , <NUM_LIT:1> ) <EOL> c2 = p . connect ( ) <EOL> assert c2 . connection is not c1_con <EOL> c2_con = c2 . connection <EOL> c2 . close ( ) <EOL> eq_ ( c2_con . close . call_count , <NUM_LIT:0> ) <EOL> def test_threadfairy ( self ) : <EOL> p = self . _queuepool_fixture ( pool_size = <NUM_LIT:3> , max_overflow = - <NUM_LIT:1> , <EOL> use_threadlocal = True ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> c2 = p . connect ( ) <EOL> assert c2 . connection is not None <EOL> class ResetOnReturnTest ( PoolTestBase ) : <EOL> def _fixture ( self , ** kw ) : <EOL> dbapi = Mock ( ) <EOL> return dbapi , pool . QueuePool ( creator = lambda : dbapi . connect ( '<STR_LIT>' ) , <EOL> ** kw ) <EOL> def test_plain_rollback ( self ) : <EOL> dbapi , p = self . _fixture ( reset_on_return = '<STR_LIT>' ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert dbapi . connect ( ) . rollback . called <EOL> assert not dbapi . connect ( ) . commit . called <EOL> def test_plain_commit ( self ) : <EOL> dbapi , p = self . _fixture ( reset_on_return = '<STR_LIT>' ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert not dbapi . connect ( ) . rollback . called <EOL> assert dbapi . connect ( ) . commit . called <EOL> def test_plain_none ( self ) : <EOL> dbapi , p = self . _fixture ( reset_on_return = None ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> assert not dbapi . connect ( ) . rollback . called <EOL> assert not dbapi . connect ( ) . commit . called <EOL> def test_agent_rollback ( self ) : <EOL> dbapi , p = self . _fixture ( reset_on_return = '<STR_LIT>' ) <EOL> class Agent ( object ) : <EOL> def __init__ ( self , conn ) : <EOL> self . conn = conn <EOL> def rollback ( self ) : <EOL> self . conn . special_rollback ( ) <EOL> def commit ( self ) : <EOL> self . conn . special_commit ( ) <EOL> c1 = p . connect ( ) <EOL> c1 . _reset_agent = Agent ( c1 ) <EOL> c1 . close ( ) <EOL> assert dbapi . connect ( ) . special_rollback . called <EOL> assert not dbapi . connect ( ) . special_commit . called <EOL> assert not dbapi . connect ( ) . rollback . called <EOL> assert not dbapi . connect ( ) . commit . called <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> eq_ ( dbapi . connect ( ) . special_rollback . call_count , <NUM_LIT:1> ) <EOL> eq_ ( dbapi . connect ( ) . special_commit . call_count , <NUM_LIT:0> ) <EOL> assert dbapi . connect ( ) . rollback . called <EOL> assert not dbapi . connect ( ) . commit . called <EOL> def test_agent_commit ( self ) : <EOL> dbapi , p = self . _fixture ( reset_on_return = '<STR_LIT>' ) <EOL> class Agent ( object ) : <EOL> def __init__ ( self , conn ) : <EOL> self . conn = conn <EOL> def rollback ( self ) : <EOL> self . conn . special_rollback ( ) <EOL> def commit ( self ) : <EOL> self . conn . special_commit ( ) <EOL> c1 = p . connect ( ) <EOL> c1 . _reset_agent = Agent ( c1 ) <EOL> c1 . close ( ) <EOL> assert not dbapi . connect ( ) . special_rollback . called <EOL> assert dbapi . connect ( ) . special_commit . called <EOL> assert not dbapi . connect ( ) . rollback . called <EOL> assert not dbapi . connect ( ) . commit . called <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> eq_ ( dbapi . connect ( ) . special_rollback . call_count , <NUM_LIT:0> ) <EOL> eq_ ( dbapi . connect ( ) . special_commit . call_count , <NUM_LIT:1> ) <EOL> assert not dbapi . connect ( ) . rollback . called <EOL> assert dbapi . connect ( ) . commit . called <EOL> class SingletonThreadPoolTest ( PoolTestBase ) : <EOL> @ testing . requires . threading_with_mock <EOL> def test_cleanup ( self ) : <EOL> self . _test_cleanup ( False ) <EOL> @ testing . requires . threading_with_mock <EOL> def test_cleanup_no_gc ( self ) : <EOL> self . _test_cleanup ( True ) <EOL> def _test_cleanup ( self , strong_refs ) : <EOL> """<STR_LIT>""" <EOL> dbapi = MockDBAPI ( ) <EOL> lock = threading . Lock ( ) <EOL> def creator ( ) : <EOL> with lock : <EOL> return dbapi . connect ( ) <EOL> p = pool . SingletonThreadPool ( creator = creator , pool_size = <NUM_LIT:3> ) <EOL> if strong_refs : <EOL> sr = set ( ) <EOL> def _conn ( ) : <EOL> c = p . connect ( ) <EOL> sr . add ( c . connection ) <EOL> return c <EOL> else : <EOL> def _conn ( ) : <EOL> return p . connect ( ) <EOL> def checkout ( ) : <EOL> for x in range ( <NUM_LIT:10> ) : <EOL> c = _conn ( ) <EOL> assert c <EOL> c . cursor ( ) <EOL> c . close ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> th = threading . Thread ( target = checkout ) <EOL> th . start ( ) <EOL> threads . append ( th ) <EOL> for th in threads : <EOL> th . join ( join_timeout ) <EOL> assert len ( p . _all_conns ) == <NUM_LIT:3> <EOL> if strong_refs : <EOL> still_opened = len ( [ c for c in sr if not c . close . call_count ] ) <EOL> eq_ ( still_opened , <NUM_LIT:3> ) <EOL> class AssertionPoolTest ( PoolTestBase ) : <EOL> def test_connect_error ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> p = pool . AssertionPool ( creator = lambda : dbapi . connect ( '<STR_LIT>' ) ) <EOL> c1 = p . connect ( ) <EOL> assert_raises ( AssertionError , p . connect ) <EOL> def test_connect_multiple ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> p = pool . AssertionPool ( creator = lambda : dbapi . connect ( '<STR_LIT>' ) ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> c2 = p . connect ( ) <EOL> c2 . close ( ) <EOL> c3 = p . connect ( ) <EOL> assert_raises ( AssertionError , p . connect ) <EOL> class NullPoolTest ( PoolTestBase ) : <EOL> def test_reconnect ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> p = pool . NullPool ( creator = lambda : dbapi . connect ( '<STR_LIT>' ) ) <EOL> c1 = p . connect ( ) <EOL> c1 . close ( ) <EOL> c1 = None <EOL> c1 = p . connect ( ) <EOL> c1 . invalidate ( ) <EOL> c1 = None <EOL> c1 = p . connect ( ) <EOL> dbapi . connect . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' ) , <EOL> call ( '<STR_LIT>' ) ] , <EOL> any_order = True ) <EOL> class StaticPoolTest ( PoolTestBase ) : <EOL> def test_recreate ( self ) : <EOL> dbapi = MockDBAPI ( ) <EOL> creator = lambda : dbapi . connect ( '<STR_LIT>' ) <EOL> p = pool . StaticPool ( creator ) <EOL> p2 = p . recreate ( ) <EOL> assert p . _creator is p2 . _creator <EOL> class CreatorCompatibilityTest ( PoolTestBase ) : <EOL> def test_creator_callable_outside_noarg ( self ) : <EOL> e = testing_engine ( ) <EOL> creator = e . pool . _creator <EOL> try : <EOL> conn = creator ( ) <EOL> finally : <EOL> conn . close ( ) <EOL> def test_creator_callable_outside_witharg ( self ) : <EOL> e = testing_engine ( ) <EOL> creator = e . pool . _creator <EOL> try : <EOL> conn = creator ( Mock ( ) ) <EOL> finally : <EOL> conn . close ( ) <EOL> def test_creator_patching_arg_to_noarg ( self ) : <EOL> e = testing_engine ( ) <EOL> creator = e . pool . _creator <EOL> try : <EOL> conn = creator ( Mock ( ) ) <EOL> finally : <EOL> conn . close ( ) <EOL> def mock_create ( ) : <EOL> return creator ( ) <EOL> conn = e . connect ( ) <EOL> conn . invalidate ( ) <EOL> conn . close ( ) <EOL> with patch . object ( e . pool , "<STR_LIT>" , mock_create ) : <EOL> conn = e . connect ( ) <EOL> conn . invalidate ( ) <EOL> conn . close ( ) <EOL> conn = e . connect ( ) <EOL> conn . close ( ) </s>
