<s> import sys <EOL> from time import time , sleep <EOL> import random <EOL> import weakref <EOL> import greentest <EOL> import gevent . threadpool <EOL> from gevent . threadpool import ThreadPool <EOL> import gevent <EOL> from greentest import ExpectedException <EOL> import six <EOL> import gc <EOL> PYPY = hasattr ( sys , '<STR_LIT>' ) <EOL> class TestCase ( greentest . TestCase ) : <EOL> pool = None <EOL> def cleanup ( self ) : <EOL> pool = getattr ( self , '<STR_LIT>' , None ) <EOL> if pool is not None : <EOL> kill = getattr ( pool , '<STR_LIT>' , None ) or getattr ( pool , '<STR_LIT>' , None ) <EOL> kill ( ) <EOL> del kill <EOL> del self . pool <EOL> class PoolBasicTests ( TestCase ) : <EOL> def test_execute_async ( self ) : <EOL> self . pool = pool = ThreadPool ( <NUM_LIT:2> ) <EOL> r = [ ] <EOL> first = pool . spawn ( r . append , <NUM_LIT:1> ) <EOL> first . get ( ) <EOL> self . assertEqual ( r , [ <NUM_LIT:1> ] ) <EOL> gevent . sleep ( <NUM_LIT:0> ) <EOL> pool . apply_async ( r . append , ( <NUM_LIT:2> , ) ) <EOL> self . assertEqual ( r , [ <NUM_LIT:1> ] ) <EOL> pool . apply_async ( r . append , ( <NUM_LIT:3> , ) ) <EOL> self . assertEqual ( r , [ <NUM_LIT:1> ] ) <EOL> pool . apply_async ( r . append , ( <NUM_LIT:4> , ) ) <EOL> self . assertEqual ( r , [ <NUM_LIT:1> ] ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( sorted ( r ) , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] ) <EOL> def test_apply ( self ) : <EOL> self . pool = pool = ThreadPool ( <NUM_LIT:1> ) <EOL> result = pool . apply ( lambda a : ( '<STR_LIT:foo>' , a ) , ( <NUM_LIT:1> , ) ) <EOL> self . assertEqual ( result , ( '<STR_LIT:foo>' , <NUM_LIT:1> ) ) <EOL> def test_apply_raises ( self ) : <EOL> self . pool = pool = ThreadPool ( <NUM_LIT:1> ) <EOL> def raiser ( ) : <EOL> raise ExpectedException ( ) <EOL> try : <EOL> pool . apply ( raiser ) <EOL> except ExpectedException : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> test_apply_raises . error_fatal = False <EOL> def test_init_valueerror ( self ) : <EOL> self . switch_expected = False <EOL> self . assertRaises ( ValueError , ThreadPool , - <NUM_LIT:1> ) <EOL> self . pool = None <EOL> class TimingWrapper ( object ) : <EOL> def __init__ ( self , func ) : <EOL> self . func = func <EOL> self . elapsed = None <EOL> def __call__ ( self , * args , ** kwds ) : <EOL> t = time ( ) <EOL> try : <EOL> return self . func ( * args , ** kwds ) <EOL> finally : <EOL> self . elapsed = time ( ) - t <EOL> def sqr ( x , wait = <NUM_LIT:0.0> ) : <EOL> sleep ( wait ) <EOL> return x * x <EOL> def sqr_random_sleep ( x ) : <EOL> sleep ( random . random ( ) * <NUM_LIT:0.1> ) <EOL> return x * x <EOL> TIMEOUT1 , TIMEOUT2 , TIMEOUT3 = <NUM_LIT> , <NUM_LIT> , <NUM_LIT> <EOL> class _AbstractPoolTest ( TestCase ) : <EOL> __timeout__ = <NUM_LIT:5> <EOL> size = <NUM_LIT:1> <EOL> ClassUnderTest = ThreadPool <EOL> MAP_IS_GEN = False <EOL> def setUp ( self ) : <EOL> greentest . TestCase . setUp ( self ) <EOL> self . pool = self . ClassUnderTest ( self . size ) <EOL> @ greentest . ignores_leakcheck <EOL> def test_map ( self ) : <EOL> pmap = self . pool . map <EOL> if self . MAP_IS_GEN : <EOL> pmap = lambda * args : list ( self . pool . map ( * args ) ) <EOL> self . assertEqual ( pmap ( sqr , range ( <NUM_LIT:10> ) ) , list ( map ( sqr , range ( <NUM_LIT:10> ) ) ) ) <EOL> self . assertEqual ( pmap ( sqr , range ( <NUM_LIT:100> ) ) , list ( map ( sqr , range ( <NUM_LIT:100> ) ) ) ) <EOL> self . pool . kill ( ) <EOL> del self . pool <EOL> del pmap <EOL> class TestPool ( _AbstractPoolTest ) : <EOL> def test_apply ( self ) : <EOL> papply = self . pool . apply <EOL> self . assertEqual ( papply ( sqr , ( <NUM_LIT:5> , ) ) , sqr ( <NUM_LIT:5> ) ) <EOL> self . assertEqual ( papply ( sqr , ( ) , { '<STR_LIT:x>' : <NUM_LIT:3> } ) , sqr ( x = <NUM_LIT:3> ) ) <EOL> def test_async ( self ) : <EOL> res = self . pool . apply_async ( sqr , ( <NUM_LIT:7> , TIMEOUT1 , ) ) <EOL> get = TimingWrapper ( res . get ) <EOL> self . assertEqual ( get ( ) , <NUM_LIT> ) <EOL> self . assertAlmostEqual ( get . elapsed , TIMEOUT1 , <NUM_LIT:1> ) <EOL> def test_async_callback ( self ) : <EOL> result = [ ] <EOL> res = self . pool . apply_async ( sqr , ( <NUM_LIT:7> , TIMEOUT1 , ) , callback = lambda x : result . append ( x ) ) <EOL> get = TimingWrapper ( res . get ) <EOL> self . assertEqual ( get ( ) , <NUM_LIT> ) <EOL> self . assertAlmostEqual ( get . elapsed , TIMEOUT1 , <NUM_LIT:1> ) <EOL> gevent . sleep ( <NUM_LIT:0> ) <EOL> assert result == [ <NUM_LIT> ] , result <EOL> def test_async_timeout ( self ) : <EOL> res = self . pool . apply_async ( sqr , ( <NUM_LIT:6> , TIMEOUT2 + <NUM_LIT> ) ) <EOL> get = TimingWrapper ( res . get ) <EOL> self . assertRaises ( gevent . Timeout , get , timeout = TIMEOUT2 ) <EOL> self . assertAlmostEqual ( get . elapsed , TIMEOUT2 , <NUM_LIT:1> ) <EOL> self . pool . join ( ) <EOL> def test_imap ( self ) : <EOL> it = self . pool . imap ( sqr , range ( <NUM_LIT:10> ) ) <EOL> self . assertEqual ( list ( it ) , list ( map ( sqr , range ( <NUM_LIT:10> ) ) ) ) <EOL> it = self . pool . imap ( sqr , range ( <NUM_LIT:10> ) ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> self . assertEqual ( six . advance_iterator ( it ) , i * i ) <EOL> self . assertRaises ( StopIteration , lambda : six . advance_iterator ( it ) ) <EOL> it = self . pool . imap ( sqr , range ( <NUM_LIT:1000> ) ) <EOL> for i in range ( <NUM_LIT:1000> ) : <EOL> self . assertEqual ( six . advance_iterator ( it ) , i * i ) <EOL> self . assertRaises ( StopIteration , lambda : six . advance_iterator ( it ) ) <EOL> def test_imap_gc ( self ) : <EOL> it = self . pool . imap ( sqr , range ( <NUM_LIT:10> ) ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> self . assertEqual ( six . advance_iterator ( it ) , i * i ) <EOL> gc . collect ( ) <EOL> self . assertRaises ( StopIteration , lambda : six . advance_iterator ( it ) ) <EOL> def test_imap_unordered_gc ( self ) : <EOL> it = self . pool . imap_unordered ( sqr , range ( <NUM_LIT:10> ) ) <EOL> result = [ ] <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> result . append ( six . advance_iterator ( it ) ) <EOL> gc . collect ( ) <EOL> self . assertRaises ( StopIteration , lambda : six . advance_iterator ( it ) ) <EOL> self . assertEqual ( sorted ( result ) , [ x * x for x in range ( <NUM_LIT:10> ) ] ) <EOL> def test_imap_random ( self ) : <EOL> it = self . pool . imap ( sqr_random_sleep , range ( <NUM_LIT:10> ) ) <EOL> self . assertEqual ( list ( it ) , list ( map ( sqr , range ( <NUM_LIT:10> ) ) ) ) <EOL> def test_imap_unordered ( self ) : <EOL> it = self . pool . imap_unordered ( sqr , range ( <NUM_LIT:1000> ) ) <EOL> self . assertEqual ( sorted ( it ) , list ( map ( sqr , range ( <NUM_LIT:1000> ) ) ) ) <EOL> it = self . pool . imap_unordered ( sqr , range ( <NUM_LIT:1000> ) ) <EOL> self . assertEqual ( sorted ( it ) , list ( map ( sqr , range ( <NUM_LIT:1000> ) ) ) ) <EOL> def test_imap_unordered_random ( self ) : <EOL> it = self . pool . imap_unordered ( sqr_random_sleep , range ( <NUM_LIT:10> ) ) <EOL> self . assertEqual ( sorted ( it ) , list ( map ( sqr , range ( <NUM_LIT:10> ) ) ) ) <EOL> def test_terminate ( self ) : <EOL> size = self . size or <NUM_LIT:10> <EOL> result = self . pool . map_async ( sleep , [ <NUM_LIT:0.1> ] * ( size * <NUM_LIT:2> ) ) <EOL> gevent . sleep ( <NUM_LIT:0.1> ) <EOL> kill = TimingWrapper ( self . pool . kill ) <EOL> kill ( ) <EOL> assert kill . elapsed < <NUM_LIT:0.1> * self . size + <NUM_LIT:0.5> , kill . elapsed <EOL> result . join ( ) <EOL> def sleep ( self , x ) : <EOL> sleep ( float ( x ) / <NUM_LIT> ) <EOL> return str ( x ) <EOL> def test_imap_unordered_sleep ( self ) : <EOL> result = list ( self . pool . imap_unordered ( self . sleep , [ <NUM_LIT:10> , <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> if self . pool . size == <NUM_LIT:1> : <EOL> expected = [ '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT:2>' ] <EOL> else : <EOL> expected = [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT>' ] <EOL> self . assertEqual ( result , expected ) <EOL> class TestPool2 ( TestPool ) : <EOL> size = <NUM_LIT:2> <EOL> def test_recursive_apply ( self ) : <EOL> p = self . pool <EOL> def a ( ) : <EOL> return p . apply ( b ) <EOL> def b ( ) : <EOL> gevent . sleep ( ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> return "<STR_LIT:B>" <EOL> result = p . apply ( a ) <EOL> self . assertEqual ( result , "<STR_LIT:B>" ) <EOL> test_recursive_apply . ignore_leakcheck = True <EOL> class TestPool3 ( TestPool ) : <EOL> size = <NUM_LIT:3> <EOL> class TestPool10 ( TestPool ) : <EOL> size = <NUM_LIT:10> <EOL> __timeout__ = <NUM_LIT:5> <EOL> class TestJoinEmpty ( TestCase ) : <EOL> switch_expected = False <EOL> def test ( self ) : <EOL> self . pool = ThreadPool ( <NUM_LIT:1> ) <EOL> self . pool . join ( ) <EOL> class TestSpawn ( TestCase ) : <EOL> switch_expected = True <EOL> def test ( self ) : <EOL> self . pool = pool = ThreadPool ( <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( pool ) , <NUM_LIT:0> ) <EOL> log = [ ] <EOL> sleep_n_log = lambda item , seconds : [ sleep ( seconds ) , log . append ( item ) ] <EOL> pool . spawn ( sleep_n_log , '<STR_LIT:a>' , <NUM_LIT:0.1> ) <EOL> self . assertEqual ( len ( pool ) , <NUM_LIT:1> ) <EOL> pool . spawn ( sleep_n_log , '<STR_LIT:b>' , <NUM_LIT:0.1> ) <EOL> self . assertEqual ( len ( pool ) , <NUM_LIT:2> ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( log , [ '<STR_LIT:a>' ] ) <EOL> self . assertEqual ( len ( pool ) , <NUM_LIT:1> ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( log , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) <EOL> self . assertEqual ( len ( pool ) , <NUM_LIT:0> ) <EOL> def error_iter ( ) : <EOL> yield <NUM_LIT:1> <EOL> yield <NUM_LIT:2> <EOL> raise greentest . ExpectedException <EOL> class TestErrorInIterator ( TestCase ) : <EOL> error_fatal = False <EOL> def test ( self ) : <EOL> self . pool = ThreadPool ( <NUM_LIT:3> ) <EOL> self . assertRaises ( greentest . ExpectedException , self . pool . map , lambda x : None , error_iter ( ) ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> def test_unordered ( self ) : <EOL> self . pool = ThreadPool ( <NUM_LIT:3> ) <EOL> def unordered ( ) : <EOL> return list ( self . pool . imap_unordered ( lambda x : None , error_iter ( ) ) ) <EOL> self . assertRaises ( greentest . ExpectedException , unordered ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> class TestMaxsize ( TestCase ) : <EOL> def test_inc ( self ) : <EOL> self . pool = ThreadPool ( <NUM_LIT:0> ) <EOL> done = [ ] <EOL> gevent . spawn ( self . pool . spawn , done . append , <NUM_LIT:1> ) <EOL> gevent . spawn_later ( <NUM_LIT> , self . pool . spawn , done . append , <NUM_LIT:2> ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( done , [ ] ) <EOL> self . pool . maxsize = <NUM_LIT:1> <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( done , [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> def test_setzero ( self ) : <EOL> pool = self . pool = ThreadPool ( <NUM_LIT:3> ) <EOL> pool . spawn ( sleep , <NUM_LIT:0.1> ) <EOL> pool . spawn ( sleep , <NUM_LIT> ) <EOL> pool . spawn ( sleep , <NUM_LIT> ) <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( pool . size , <NUM_LIT:3> ) <EOL> pool . maxsize = <NUM_LIT:0> <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( pool . size , <NUM_LIT:0> ) <EOL> class TestSize ( TestCase ) : <EOL> def test ( self ) : <EOL> pool = self . pool = ThreadPool ( <NUM_LIT:2> ) <EOL> self . assertEqual ( pool . size , <NUM_LIT:0> ) <EOL> pool . size = <NUM_LIT:1> <EOL> self . assertEqual ( pool . size , <NUM_LIT:1> ) <EOL> pool . size = <NUM_LIT:2> <EOL> self . assertEqual ( pool . size , <NUM_LIT:2> ) <EOL> pool . size = <NUM_LIT:1> <EOL> self . assertEqual ( pool . size , <NUM_LIT:1> ) <EOL> def set_neg ( ) : <EOL> pool . size = - <NUM_LIT:1> <EOL> self . assertRaises ( ValueError , set_neg ) <EOL> def set_too_big ( ) : <EOL> pool . size = <NUM_LIT:3> <EOL> self . assertRaises ( ValueError , set_too_big ) <EOL> pool . size = <NUM_LIT:0> <EOL> self . assertEqual ( pool . size , <NUM_LIT:0> ) <EOL> pool . size = <NUM_LIT:2> <EOL> self . assertEqual ( pool . size , <NUM_LIT:2> ) <EOL> class TestRef ( TestCase ) : <EOL> def test ( self ) : <EOL> pool = self . pool = ThreadPool ( <NUM_LIT:2> ) <EOL> refs = [ ] <EOL> obj = SomeClass ( ) <EOL> obj . refs = refs <EOL> func = obj . func <EOL> del obj <EOL> with greentest . disabled_gc ( ) : <EOL> result = pool . apply ( func , ( Object ( ) , ) , { '<STR_LIT>' : Object ( ) } ) <EOL> assert isinstance ( result , Object ) , repr ( result ) <EOL> gevent . sleep ( <NUM_LIT:0.1> ) <EOL> refs . append ( weakref . ref ( func ) ) <EOL> del func , result <EOL> if PYPY : <EOL> gc . collect ( ) <EOL> gc . collect ( ) <EOL> for index , r in enumerate ( refs ) : <EOL> assert r ( ) is None , ( index , r ( ) , greentest . getrefcount ( r ( ) ) , refs ) <EOL> assert len ( refs ) == <NUM_LIT:4> , refs <EOL> class Object ( object ) : <EOL> pass <EOL> class SomeClass ( object ) : <EOL> def func ( self , arg1 , kwarg1 = None ) : <EOL> result = Object ( ) <EOL> self . refs . extend ( [ weakref . ref ( x ) for x in [ arg1 , kwarg1 , result ] ] ) <EOL> return result <EOL> def func ( ) : <EOL> pass <EOL> class TestRefCount ( TestCase ) : <EOL> def test ( self ) : <EOL> pool = ThreadPool ( <NUM_LIT:1> ) <EOL> pool . spawn ( func ) <EOL> gevent . sleep ( <NUM_LIT:0> ) <EOL> pool . kill ( ) <EOL> if hasattr ( gevent . threadpool , '<STR_LIT>' ) : <EOL> from concurrent . futures import TimeoutError as FutureTimeoutError <EOL> from concurrent . futures import wait as cf_wait <EOL> from concurrent . futures import as_completed as cf_as_completed <EOL> from gevent import monkey <EOL> class TestTPE ( _AbstractPoolTest ) : <EOL> size = <NUM_LIT:1> <EOL> MAP_IS_GEN = True <EOL> ClassUnderTest = gevent . threadpool . ThreadPoolExecutor <EOL> MONKEY_PATCHED = False <EOL> @ greentest . ignores_leakcheck <EOL> def test_future ( self ) : <EOL> self . assertEqual ( monkey . is_module_patched ( '<STR_LIT>' ) , <EOL> self . MONKEY_PATCHED ) <EOL> pool = self . pool <EOL> calledback = [ ] <EOL> def fn ( ) : <EOL> gevent . sleep ( <NUM_LIT:0.5> ) <EOL> return <NUM_LIT> <EOL> def callback ( future ) : <EOL> future . calledback += <NUM_LIT:1> <EOL> raise Exception ( "<STR_LIT>" ) <EOL> future = pool . submit ( fn ) <EOL> future . calledback = <NUM_LIT:0> <EOL> future . add_done_callback ( callback ) <EOL> self . assertRaises ( FutureTimeoutError , future . result , timeout = <NUM_LIT> ) <EOL> def spawned ( ) : <EOL> return <NUM_LIT> <EOL> spawned_greenlet = gevent . spawn ( spawned ) <EOL> self . assertEqual ( future . result ( ) , <NUM_LIT> ) <EOL> self . assertTrue ( future . done ( ) ) <EOL> self . assertFalse ( future . cancelled ( ) ) <EOL> gevent . sleep ( ) <EOL> self . assertEqual ( future . calledback , <NUM_LIT:1> ) <EOL> self . assertTrue ( spawned_greenlet . ready ( ) ) <EOL> self . assertEqual ( spawned_greenlet . value , <NUM_LIT> ) <EOL> future . add_done_callback ( lambda f : calledback . append ( True ) ) <EOL> self . assertEqual ( calledback , [ True ] ) <EOL> done , _not_done = cf_wait ( ( future , ) ) <EOL> self . assertEqual ( list ( done ) , [ future ] ) <EOL> self . assertEqual ( list ( cf_as_completed ( ( future , ) ) ) , [ future ] ) <EOL> self . assertEqual ( future . calledback , <NUM_LIT:1> ) <EOL> gevent . sleep ( ) <EOL> self . assertEqual ( future . calledback , <NUM_LIT:1> ) <EOL> pool . kill ( ) <EOL> del future <EOL> del pool <EOL> del self . pool <EOL> @ greentest . ignores_leakcheck <EOL> def test_future_wait_module_function ( self ) : <EOL> self . assertEqual ( monkey . is_module_patched ( '<STR_LIT>' ) , <EOL> self . MONKEY_PATCHED ) <EOL> pool = self . pool <EOL> def fn ( ) : <EOL> gevent . sleep ( <NUM_LIT:0.5> ) <EOL> return <NUM_LIT> <EOL> future = pool . submit ( fn ) <EOL> if self . MONKEY_PATCHED : <EOL> _done , not_done = cf_wait ( ( future , ) , timeout = <NUM_LIT> ) <EOL> self . assertEqual ( list ( not_done ) , [ future ] ) <EOL> def spawned ( ) : <EOL> return <NUM_LIT> <EOL> spawned_greenlet = gevent . spawn ( spawned ) <EOL> done , _not_done = cf_wait ( ( future , ) ) <EOL> self . assertEqual ( list ( done ) , [ future ] ) <EOL> self . assertTrue ( spawned_greenlet . ready ( ) ) <EOL> self . assertEqual ( spawned_greenlet . value , <NUM_LIT> ) <EOL> else : <EOL> self . assertRaises ( AttributeError , cf_wait , ( future , ) ) <EOL> pool . kill ( ) <EOL> del future <EOL> del pool <EOL> del self . pool <EOL> @ greentest . ignores_leakcheck <EOL> def test_future_wait_gevent_function ( self ) : <EOL> self . assertEqual ( monkey . is_module_patched ( '<STR_LIT>' ) , <EOL> self . MONKEY_PATCHED ) <EOL> pool = self . pool <EOL> def fn ( ) : <EOL> gevent . sleep ( <NUM_LIT:0.5> ) <EOL> return <NUM_LIT> <EOL> future = pool . submit ( fn ) <EOL> def spawned ( ) : <EOL> return <NUM_LIT> <EOL> spawned_greenlet = gevent . spawn ( spawned ) <EOL> done = gevent . wait ( ( future , ) ) <EOL> self . assertEqual ( list ( done ) , [ future ] ) <EOL> self . assertTrue ( spawned_greenlet . ready ( ) ) <EOL> self . assertEqual ( spawned_greenlet . value , <NUM_LIT> ) <EOL> pool . kill ( ) <EOL> del future <EOL> del pool <EOL> del self . pool <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> greentest . main ( ) </s>
