<s> '''<STR_LIT>''' <EOL> from __future__ import print_function <EOL> import sys <EOL> import os <EOL> import traceback <EOL> from unittest import TestCase , main <EOL> from tests import mock , skip_unless , skip_with_pyevent , get_database_auth <EOL> from eventlet import event <EOL> from eventlet import db_pool <EOL> from eventlet . support import six <EOL> import eventlet <EOL> class DBTester ( object ) : <EOL> __test__ = False <EOL> def setUp ( self ) : <EOL> self . create_db ( ) <EOL> self . connection = None <EOL> connection = self . _dbmodule . connect ( ** self . _auth ) <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( """<STR_LIT>""" ) <EOL> connection . commit ( ) <EOL> cursor . close ( ) <EOL> connection . close ( ) <EOL> def tearDown ( self ) : <EOL> if self . connection : <EOL> self . connection . close ( ) <EOL> self . drop_db ( ) <EOL> def set_up_dummy_table ( self , connection = None ) : <EOL> close_connection = False <EOL> if connection is None : <EOL> close_connection = True <EOL> if self . connection is None : <EOL> connection = self . _dbmodule . connect ( ** self . _auth ) <EOL> else : <EOL> connection = self . connection <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( self . dummy_table_sql ) <EOL> connection . commit ( ) <EOL> cursor . close ( ) <EOL> if close_connection : <EOL> connection . close ( ) <EOL> class Mock ( object ) : <EOL> pass <EOL> class DBConnectionPool ( DBTester ) : <EOL> __test__ = False <EOL> def setUp ( self ) : <EOL> super ( DBConnectionPool , self ) . setUp ( ) <EOL> self . pool = self . create_pool ( ) <EOL> self . connection = self . pool . get ( ) <EOL> def tearDown ( self ) : <EOL> if self . connection : <EOL> self . pool . put ( self . connection ) <EOL> self . pool . clear ( ) <EOL> super ( DBConnectionPool , self ) . tearDown ( ) <EOL> def assert_cursor_works ( self , cursor ) : <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> rows = cursor . fetchall ( ) <EOL> assert rows <EOL> def test_connecting ( self ) : <EOL> assert self . connection is not None <EOL> def test_create_cursor ( self ) : <EOL> cursor = self . connection . cursor ( ) <EOL> cursor . close ( ) <EOL> def test_run_query ( self ) : <EOL> cursor = self . connection . cursor ( ) <EOL> self . assert_cursor_works ( cursor ) <EOL> cursor . close ( ) <EOL> def test_run_bad_query ( self ) : <EOL> cursor = self . connection . cursor ( ) <EOL> try : <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> assert False <EOL> except AssertionError : <EOL> raise <EOL> except Exception : <EOL> pass <EOL> cursor . close ( ) <EOL> def test_put_none ( self ) : <EOL> assert self . pool . free ( ) == <NUM_LIT:0> <EOL> self . pool . put ( None ) <EOL> assert self . pool . free ( ) == <NUM_LIT:1> <EOL> conn2 = self . pool . get ( ) <EOL> assert conn2 is not None <EOL> assert conn2 . cursor <EOL> self . pool . put ( conn2 ) <EOL> def test_close_does_a_put ( self ) : <EOL> assert self . pool . free ( ) == <NUM_LIT:0> <EOL> self . connection . close ( ) <EOL> assert self . pool . free ( ) == <NUM_LIT:1> <EOL> self . assertRaises ( AttributeError , self . connection . cursor ) <EOL> def test_put_doesnt_double_wrap ( self ) : <EOL> self . pool . put ( self . connection ) <EOL> conn = self . pool . get ( ) <EOL> assert not isinstance ( conn . _base , db_pool . PooledConnectionWrapper ) <EOL> self . pool . put ( conn ) <EOL> def test_bool ( self ) : <EOL> assert self . connection <EOL> self . connection . close ( ) <EOL> assert not self . connection <EOL> def fill_up_table ( self , conn ) : <EOL> curs = conn . cursor ( ) <EOL> for i in six . moves . range ( <NUM_LIT:1000> ) : <EOL> curs . execute ( '<STR_LIT>' % i ) <EOL> conn . commit ( ) <EOL> def test_returns_immediately ( self ) : <EOL> self . pool = self . create_pool ( ) <EOL> conn = self . pool . get ( ) <EOL> self . set_up_dummy_table ( conn ) <EOL> self . fill_up_table ( conn ) <EOL> curs = conn . cursor ( ) <EOL> results = [ ] <EOL> SHORT_QUERY = "<STR_LIT>" <EOL> evt = event . Event ( ) <EOL> def a_query ( ) : <EOL> self . assert_cursor_works ( curs ) <EOL> curs . execute ( SHORT_QUERY ) <EOL> results . append ( <NUM_LIT:2> ) <EOL> evt . send ( ) <EOL> eventlet . spawn ( a_query ) <EOL> results . append ( <NUM_LIT:1> ) <EOL> self . assertEqual ( [ <NUM_LIT:1> ] , results ) <EOL> evt . wait ( ) <EOL> self . assertEqual ( [ <NUM_LIT:1> , <NUM_LIT:2> ] , results ) <EOL> self . pool . put ( conn ) <EOL> def test_connection_is_clean_after_put ( self ) : <EOL> self . pool = self . create_pool ( ) <EOL> conn = self . pool . get ( ) <EOL> self . set_up_dummy_table ( conn ) <EOL> curs = conn . cursor ( ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> curs . execute ( '<STR_LIT>' % i ) <EOL> self . pool . put ( conn ) <EOL> del conn <EOL> conn2 = self . pool . get ( ) <EOL> curs2 = conn2 . cursor ( ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> curs2 . execute ( '<STR_LIT>' % i ) <EOL> conn2 . commit ( ) <EOL> curs2 . execute ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:10> , curs2 . rowcount ) <EOL> self . pool . put ( conn2 ) <EOL> def test_visibility_from_other_connections ( self ) : <EOL> self . pool = self . create_pool ( max_size = <NUM_LIT:3> ) <EOL> conn = self . pool . get ( ) <EOL> conn2 = self . pool . get ( ) <EOL> curs = conn . cursor ( ) <EOL> try : <EOL> curs2 = conn2 . cursor ( ) <EOL> curs2 . execute ( "<STR_LIT>" % ( <NUM_LIT> ) ) <EOL> self . assertEqual ( curs2 . rowcount , <NUM_LIT:1> ) <EOL> conn2 . commit ( ) <EOL> selection_query = "<STR_LIT>" <EOL> curs2 . execute ( selection_query ) <EOL> self . assertEqual ( curs2 . rowcount , <NUM_LIT:1> ) <EOL> del curs2 <EOL> self . pool . put ( conn2 ) <EOL> conn3 = self . pool . get ( ) <EOL> curs3 = conn3 . cursor ( ) <EOL> curs3 . execute ( selection_query ) <EOL> self . assertEqual ( curs3 . rowcount , <NUM_LIT:1> ) <EOL> curs . execute ( selection_query ) <EOL> self . assertEqual ( curs . rowcount , <NUM_LIT:1> ) <EOL> self . pool . put ( conn3 ) <EOL> finally : <EOL> curs . execute ( "<STR_LIT>" ) <EOL> conn . commit ( ) <EOL> self . pool . put ( conn ) <EOL> def test_clear ( self ) : <EOL> self . pool = self . create_pool ( ) <EOL> self . pool . put ( self . connection ) <EOL> self . pool . clear ( ) <EOL> self . assertEqual ( len ( self . pool . free_items ) , <NUM_LIT:0> ) <EOL> def test_clear_warmup ( self ) : <EOL> """<STR_LIT>""" <EOL> self . pool = self . create_pool ( min_size = <NUM_LIT:1> ) <EOL> self . pool . clear ( ) <EOL> self . assertEqual ( len ( self . pool . free_items ) , <NUM_LIT:0> ) <EOL> def test_unwrap_connection ( self ) : <EOL> self . assert_ ( isinstance ( self . connection , <EOL> db_pool . GenericConnectionWrapper ) ) <EOL> conn = self . pool . _unwrap_connection ( self . connection ) <EOL> assert not isinstance ( conn , db_pool . GenericConnectionWrapper ) <EOL> self . assertEqual ( None , self . pool . _unwrap_connection ( None ) ) <EOL> self . assertEqual ( None , self . pool . _unwrap_connection ( <NUM_LIT:1> ) ) <EOL> x = Mock ( ) <EOL> x . _base = '<STR_LIT>' <EOL> self . assertEqual ( '<STR_LIT>' , self . pool . _unwrap_connection ( x ) ) <EOL> conn . close ( ) <EOL> def test_safe_close ( self ) : <EOL> self . pool . _safe_close ( self . connection , quiet = True ) <EOL> self . assertEqual ( len ( self . pool . free_items ) , <NUM_LIT:1> ) <EOL> self . pool . _safe_close ( None ) <EOL> self . pool . _safe_close ( <NUM_LIT:1> ) <EOL> x = Mock ( ) <EOL> def fail ( ) : <EOL> raise KeyboardInterrupt ( ) <EOL> x . close = fail <EOL> self . assertRaises ( KeyboardInterrupt , self . pool . _safe_close , x ) <EOL> x = Mock ( ) <EOL> def fail2 ( ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> x . close = fail2 <EOL> self . pool . _safe_close ( x , quiet = False ) <EOL> def test_zero_max_idle ( self ) : <EOL> self . pool . put ( self . connection ) <EOL> self . pool . clear ( ) <EOL> self . pool = self . create_pool ( max_size = <NUM_LIT:2> , max_idle = <NUM_LIT:0> ) <EOL> self . connection = self . pool . get ( ) <EOL> self . connection . close ( ) <EOL> self . assertEqual ( len ( self . pool . free_items ) , <NUM_LIT:0> ) <EOL> def test_zero_max_age ( self ) : <EOL> self . pool . put ( self . connection ) <EOL> self . pool . clear ( ) <EOL> self . pool = self . create_pool ( max_size = <NUM_LIT:2> , max_age = <NUM_LIT:0> ) <EOL> self . connection = self . pool . get ( ) <EOL> self . connection . close ( ) <EOL> self . assertEqual ( len ( self . pool . free_items ) , <NUM_LIT:0> ) <EOL> def test_waiters_get_woken ( self ) : <EOL> self . pool . put ( self . connection ) <EOL> self . pool . clear ( ) <EOL> self . pool = self . create_pool ( max_size = <NUM_LIT:1> , max_age = <NUM_LIT:0> ) <EOL> self . connection = self . pool . get ( ) <EOL> self . assertEqual ( self . pool . free ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( self . pool . waiting ( ) , <NUM_LIT:0> ) <EOL> e = event . Event ( ) <EOL> def retrieve ( pool , ev ) : <EOL> c = pool . get ( ) <EOL> ev . send ( c ) <EOL> eventlet . spawn ( retrieve , self . pool , e ) <EOL> eventlet . sleep ( <NUM_LIT:0> ) <EOL> eventlet . sleep ( <NUM_LIT:0> ) <EOL> self . assertEqual ( self . pool . free ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( self . pool . waiting ( ) , <NUM_LIT:1> ) <EOL> self . pool . put ( self . connection ) <EOL> timer = eventlet . Timeout ( <NUM_LIT:1> ) <EOL> conn = e . wait ( ) <EOL> timer . cancel ( ) <EOL> self . assertEqual ( self . pool . free ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( self . pool . waiting ( ) , <NUM_LIT:0> ) <EOL> self . pool . put ( conn ) <EOL> def test_raising_create ( self ) : <EOL> self . pool = self . create_pool ( max_size = <NUM_LIT:1> , module = RaisingDBModule ( ) ) <EOL> self . assertRaises ( RuntimeError , self . pool . get ) <EOL> self . assertEqual ( self . pool . free ( ) , <NUM_LIT:1> ) <EOL> class DummyConnection ( object ) : <EOL> def rollback ( self ) : <EOL> pass <EOL> class DummyDBModule ( object ) : <EOL> def connect ( self , * args , ** kwargs ) : <EOL> return DummyConnection ( ) <EOL> class RaisingDBModule ( object ) : <EOL> def connect ( self , * args , ** kw ) : <EOL> raise RuntimeError ( ) <EOL> class TpoolConnectionPool ( DBConnectionPool ) : <EOL> __test__ = False <EOL> def create_pool ( self , min_size = <NUM_LIT:0> , max_size = <NUM_LIT:1> , max_idle = <NUM_LIT:10> , max_age = <NUM_LIT:10> , <EOL> connect_timeout = <NUM_LIT:0.5> , module = None ) : <EOL> if module is None : <EOL> module = self . _dbmodule <EOL> return db_pool . TpooledConnectionPool ( <EOL> module , <EOL> min_size = min_size , max_size = max_size , <EOL> max_idle = max_idle , max_age = max_age , <EOL> connect_timeout = connect_timeout , <EOL> ** self . _auth ) <EOL> @ skip_with_pyevent <EOL> def setUp ( self ) : <EOL> super ( TpoolConnectionPool , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> super ( TpoolConnectionPool , self ) . tearDown ( ) <EOL> from eventlet import tpool <EOL> tpool . killall ( ) <EOL> class RawConnectionPool ( DBConnectionPool ) : <EOL> __test__ = False <EOL> def create_pool ( self , min_size = <NUM_LIT:0> , max_size = <NUM_LIT:1> , max_idle = <NUM_LIT:10> , max_age = <NUM_LIT:10> , <EOL> connect_timeout = <NUM_LIT:0.5> , module = None ) : <EOL> if module is None : <EOL> module = self . _dbmodule <EOL> return db_pool . RawConnectionPool ( <EOL> module , <EOL> min_size = min_size , max_size = max_size , <EOL> max_idle = max_idle , max_age = max_age , <EOL> connect_timeout = connect_timeout , <EOL> ** self . _auth ) <EOL> def test_raw_pool_issue_125 ( ) : <EOL> pool = db_pool . RawConnectionPool ( <EOL> DummyDBModule ( ) , <EOL> dsn = "<STR_LIT>" , <EOL> min_size = <NUM_LIT:3> , max_size = <NUM_LIT:5> ) <EOL> conn = pool . get ( ) <EOL> pool . put ( conn ) <EOL> def test_raw_pool_custom_cleanup_ok ( ) : <EOL> cleanup_mock = mock . Mock ( ) <EOL> pool = db_pool . RawConnectionPool ( DummyDBModule ( ) , cleanup = cleanup_mock ) <EOL> conn = pool . get ( ) <EOL> pool . put ( conn ) <EOL> assert cleanup_mock . call_count == <NUM_LIT:1> <EOL> with pool . item ( ) as conn : <EOL> pass <EOL> assert cleanup_mock . call_count == <NUM_LIT:2> <EOL> def test_raw_pool_custom_cleanup_arg_error ( ) : <EOL> cleanup_mock = mock . Mock ( side_effect = NotImplementedError ) <EOL> pool = db_pool . RawConnectionPool ( DummyDBModule ( ) ) <EOL> conn = pool . get ( ) <EOL> pool . put ( conn , cleanup = cleanup_mock ) <EOL> assert cleanup_mock . call_count == <NUM_LIT:1> <EOL> with pool . item ( cleanup = cleanup_mock ) : <EOL> pass <EOL> assert cleanup_mock . call_count == <NUM_LIT:2> <EOL> def test_raw_pool_custom_cleanup_fatal ( ) : <EOL> state = [ <NUM_LIT:0> ] <EOL> def cleanup ( conn ) : <EOL> state [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> raise KeyboardInterrupt <EOL> pool = db_pool . RawConnectionPool ( DummyDBModule ( ) , cleanup = cleanup ) <EOL> conn = pool . get ( ) <EOL> try : <EOL> pool . put ( conn ) <EOL> except KeyboardInterrupt : <EOL> pass <EOL> else : <EOL> assert False , '<STR_LIT>' <EOL> assert state [ <NUM_LIT:0> ] == <NUM_LIT:1> <EOL> def test_raw_pool_clear_update_current_size ( ) : <EOL> pool = db_pool . RawConnectionPool ( DummyDBModule ( ) ) <EOL> pool . get ( ) . close ( ) <EOL> assert pool . current_size == <NUM_LIT:1> <EOL> assert len ( pool . free_items ) == <NUM_LIT:1> <EOL> pool . clear ( ) <EOL> assert pool . current_size == <NUM_LIT:0> <EOL> assert len ( pool . free_items ) == <NUM_LIT:0> <EOL> get_auth = get_database_auth <EOL> def mysql_requirement ( _f ) : <EOL> verbose = os . environ . get ( '<STR_LIT>' ) <EOL> try : <EOL> import MySQLdb <EOL> try : <EOL> auth = get_auth ( ) [ '<STR_LIT>' ] . copy ( ) <EOL> MySQLdb . connect ( ** auth ) <EOL> return True <EOL> except MySQLdb . OperationalError : <EOL> if verbose : <EOL> print ( "<STR_LIT>" , file = sys . stderr ) <EOL> traceback . print_exc ( ) <EOL> return False <EOL> except ImportError : <EOL> if verbose : <EOL> print ( "<STR_LIT>" , file = sys . stderr ) <EOL> return False <EOL> class MysqlConnectionPool ( object ) : <EOL> dummy_table_sql = """<STR_LIT>""" <EOL> @ skip_unless ( mysql_requirement ) <EOL> def setUp ( self ) : <EOL> import MySQLdb <EOL> self . _dbmodule = MySQLdb <EOL> self . _auth = get_auth ( ) [ '<STR_LIT>' ] <EOL> super ( MysqlConnectionPool , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> super ( MysqlConnectionPool , self ) . tearDown ( ) <EOL> def create_db ( self ) : <EOL> auth = self . _auth . copy ( ) <EOL> try : <EOL> self . drop_db ( ) <EOL> except Exception : <EOL> pass <EOL> dbname = '<STR_LIT>' % os . getpid ( ) <EOL> db = self . _dbmodule . connect ( ** auth ) . cursor ( ) <EOL> db . execute ( "<STR_LIT>" + dbname ) <EOL> db . close ( ) <EOL> self . _auth [ '<STR_LIT>' ] = dbname <EOL> del db <EOL> def drop_db ( self ) : <EOL> db = self . _dbmodule . connect ( ** self . _auth ) . cursor ( ) <EOL> db . execute ( "<STR_LIT>" + self . _auth [ '<STR_LIT>' ] ) <EOL> db . close ( ) <EOL> del db <EOL> class Test01MysqlTpool ( MysqlConnectionPool , TpoolConnectionPool , TestCase ) : <EOL> __test__ = True <EOL> class Test02MysqlRaw ( MysqlConnectionPool , RawConnectionPool , TestCase ) : <EOL> __test__ = True <EOL> def postgres_requirement ( _f ) : <EOL> try : <EOL> import psycopg2 <EOL> try : <EOL> auth = get_auth ( ) [ '<STR_LIT>' ] . copy ( ) <EOL> psycopg2 . connect ( ** auth ) <EOL> return True <EOL> except psycopg2 . OperationalError : <EOL> print ( "<STR_LIT>" ) <EOL> return False <EOL> except ImportError : <EOL> print ( "<STR_LIT>" ) <EOL> return False <EOL> class Psycopg2ConnectionPool ( object ) : <EOL> dummy_table_sql = """<STR_LIT>""" <EOL> @ skip_unless ( postgres_requirement ) <EOL> def setUp ( self ) : <EOL> import psycopg2 <EOL> self . _dbmodule = psycopg2 <EOL> self . _auth = get_auth ( ) [ '<STR_LIT>' ] <EOL> super ( Psycopg2ConnectionPool , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> super ( Psycopg2ConnectionPool , self ) . tearDown ( ) <EOL> def create_db ( self ) : <EOL> dbname = '<STR_LIT>' % os . getpid ( ) <EOL> self . _auth [ '<STR_LIT>' ] = dbname <EOL> try : <EOL> self . drop_db ( ) <EOL> except Exception : <EOL> pass <EOL> auth = self . _auth . copy ( ) <EOL> auth . pop ( '<STR_LIT>' ) <EOL> conn = self . _dbmodule . connect ( ** auth ) <EOL> conn . set_isolation_level ( <NUM_LIT:0> ) <EOL> db = conn . cursor ( ) <EOL> db . execute ( "<STR_LIT>" + dbname ) <EOL> db . close ( ) <EOL> conn . close ( ) <EOL> def drop_db ( self ) : <EOL> auth = self . _auth . copy ( ) <EOL> auth . pop ( '<STR_LIT>' ) <EOL> conn = self . _dbmodule . connect ( ** auth ) <EOL> conn . set_isolation_level ( <NUM_LIT:0> ) <EOL> db = conn . cursor ( ) <EOL> db . execute ( "<STR_LIT>" + self . _auth [ '<STR_LIT>' ] ) <EOL> db . close ( ) <EOL> conn . close ( ) <EOL> class TestPsycopg2Base ( TestCase ) : <EOL> __test__ = False <EOL> def test_cursor_works_as_context_manager ( self ) : <EOL> with self . connection . cursor ( ) as c : <EOL> c . execute ( '<STR_LIT>' ) <EOL> row = c . fetchone ( ) <EOL> assert row == ( <NUM_LIT:1> , ) <EOL> class Test01Psycopg2Tpool ( Psycopg2ConnectionPool , TpoolConnectionPool , TestPsycopg2Base ) : <EOL> __test__ = True <EOL> class Test02Psycopg2Raw ( Psycopg2ConnectionPool , RawConnectionPool , TestPsycopg2Base ) : <EOL> __test__ = True <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
