<s> import sys <EOL> import unittest <EOL> import warnings <EOL> warnings . simplefilter ( '<STR_LIT:ignore>' , DeprecationWarning ) <EOL> from eventlet import proc <EOL> warnings . simplefilter ( '<STR_LIT:default>' , DeprecationWarning ) <EOL> from eventlet import coros <EOL> from eventlet import event as _event <EOL> from eventlet import Timeout , sleep , getcurrent , with_timeout <EOL> from tests import LimitedTestCase , skipped , silence_warnings <EOL> DELAY = <NUM_LIT> <EOL> class ExpectedError ( Exception ) : <EOL> pass <EOL> class TestLink_Signal ( LimitedTestCase ) : <EOL> @ silence_warnings <EOL> def test_send ( self ) : <EOL> s = proc . Source ( ) <EOL> q1 , q2 , q3 = coros . queue ( ) , coros . queue ( ) , coros . queue ( ) <EOL> s . link_value ( q1 ) <EOL> self . assertRaises ( Timeout , s . wait , <NUM_LIT:0> ) <EOL> assert s . wait ( <NUM_LIT:0> , None ) is None <EOL> assert s . wait ( <NUM_LIT> , None ) is None <EOL> self . assertRaises ( Timeout , s . wait , <NUM_LIT> ) <EOL> s . send ( <NUM_LIT:1> ) <EOL> assert not q1 . ready ( ) <EOL> assert s . wait ( ) == <NUM_LIT:1> <EOL> sleep ( <NUM_LIT:0> ) <EOL> assert q1 . ready ( ) <EOL> s . link_exception ( q2 ) <EOL> s . link ( q3 ) <EOL> assert not q2 . ready ( ) <EOL> sleep ( <NUM_LIT:0> ) <EOL> assert q3 . ready ( ) <EOL> assert s . wait ( ) == <NUM_LIT:1> <EOL> @ silence_warnings <EOL> def test_send_exception ( self ) : <EOL> s = proc . Source ( ) <EOL> q1 , q2 , q3 = coros . queue ( ) , coros . queue ( ) , coros . queue ( ) <EOL> s . link_exception ( q1 ) <EOL> s . send_exception ( OSError ( '<STR_LIT:hello>' ) ) <EOL> sleep ( <NUM_LIT:0> ) <EOL> assert q1 . ready ( ) <EOL> s . link_value ( q2 ) <EOL> s . link ( q3 ) <EOL> assert not q2 . ready ( ) <EOL> sleep ( <NUM_LIT:0> ) <EOL> assert q3 . ready ( ) <EOL> self . assertRaises ( OSError , q1 . wait ) <EOL> self . assertRaises ( OSError , q3 . wait ) <EOL> self . assertRaises ( OSError , s . wait ) <EOL> class TestProc ( LimitedTestCase ) : <EOL> def test_proc ( self ) : <EOL> p = proc . spawn ( lambda : <NUM_LIT:100> ) <EOL> receiver = proc . spawn ( sleep , <NUM_LIT:1> ) <EOL> p . link ( receiver ) <EOL> self . assertRaises ( proc . LinkedCompleted , receiver . wait ) <EOL> receiver2 = proc . spawn ( sleep , <NUM_LIT:1> ) <EOL> p . link ( receiver2 ) <EOL> self . assertRaises ( proc . LinkedCompleted , receiver2 . wait ) <EOL> def test_event ( self ) : <EOL> p = proc . spawn ( lambda : <NUM_LIT:100> ) <EOL> event = _event . Event ( ) <EOL> p . link ( event ) <EOL> self . assertEqual ( event . wait ( ) , <NUM_LIT:100> ) <EOL> for i in xrange ( <NUM_LIT:3> ) : <EOL> event2 = _event . Event ( ) <EOL> p . link ( event2 ) <EOL> self . assertEqual ( event2 . wait ( ) , <NUM_LIT:100> ) <EOL> def test_current ( self ) : <EOL> p = proc . spawn ( lambda : <NUM_LIT:100> ) <EOL> p . link ( ) <EOL> self . assertRaises ( proc . LinkedCompleted , sleep , <NUM_LIT:0.1> ) <EOL> class TestCase ( LimitedTestCase ) : <EOL> def link ( self , p , listener = None ) : <EOL> getattr ( p , self . link_method ) ( listener ) <EOL> def tearDown ( self ) : <EOL> LimitedTestCase . tearDown ( self ) <EOL> self . p . unlink ( ) <EOL> def set_links ( self , p , first_time , kill_exc_type ) : <EOL> event = _event . Event ( ) <EOL> self . link ( p , event ) <EOL> proc_flag = [ ] <EOL> def receiver ( ) : <EOL> sleep ( DELAY ) <EOL> proc_flag . append ( '<STR_LIT>' ) <EOL> receiver = proc . spawn ( receiver ) <EOL> self . link ( p , receiver ) <EOL> queue = coros . queue ( <NUM_LIT:1> ) <EOL> self . link ( p , queue ) <EOL> try : <EOL> self . link ( p ) <EOL> except kill_exc_type : <EOL> if first_time : <EOL> raise <EOL> else : <EOL> assert first_time , '<STR_LIT>' <EOL> callback_flag = [ '<STR_LIT>' ] <EOL> self . link ( p , lambda * args : callback_flag . remove ( '<STR_LIT>' ) ) <EOL> for _ in range ( <NUM_LIT:10> ) : <EOL> self . link ( p , _event . Event ( ) ) <EOL> self . link ( p , coros . queue ( <NUM_LIT:1> ) ) <EOL> return event , receiver , proc_flag , queue , callback_flag <EOL> def set_links_timeout ( self , link ) : <EOL> event = _event . Event ( ) <EOL> link ( event ) <EOL> proc_finished_flag = [ ] <EOL> def myproc ( ) : <EOL> sleep ( <NUM_LIT:10> ) <EOL> proc_finished_flag . append ( '<STR_LIT>' ) <EOL> return <NUM_LIT> <EOL> myproc = proc . spawn ( myproc ) <EOL> link ( myproc ) <EOL> queue = coros . queue ( <NUM_LIT:0> ) <EOL> link ( queue ) <EOL> return event , myproc , proc_finished_flag , queue <EOL> def check_timed_out ( self , event , myproc , proc_finished_flag , queue ) : <EOL> X = object ( ) <EOL> assert with_timeout ( DELAY , event . wait , timeout_value = X ) is X <EOL> assert with_timeout ( DELAY , queue . wait , timeout_value = X ) is X <EOL> assert with_timeout ( DELAY , proc . waitall , [ myproc ] , timeout_value = X ) is X <EOL> assert proc_finished_flag == [ ] , proc_finished_flag <EOL> class TestReturn_link ( TestCase ) : <EOL> link_method = '<STR_LIT>' <EOL> def test_return ( self ) : <EOL> def return25 ( ) : <EOL> return <NUM_LIT> <EOL> p = self . p = proc . spawn ( return25 ) <EOL> self . _test_return ( p , True , <NUM_LIT> , proc . LinkedCompleted , lambda : sleep ( <NUM_LIT:0> ) ) <EOL> for _ in xrange ( <NUM_LIT:3> ) : <EOL> self . _test_return ( p , False , <NUM_LIT> , proc . LinkedCompleted , lambda : sleep ( <NUM_LIT:0> ) ) <EOL> def _test_return ( self , p , first_time , result , kill_exc_type , action ) : <EOL> event , receiver , proc_flag , queue , callback_flag = self . set_links ( p , first_time , kill_exc_type ) <EOL> xxxxx = self . set_links_timeout ( p . link_exception ) <EOL> try : <EOL> sleep ( DELAY * <NUM_LIT:2> ) <EOL> except kill_exc_type : <EOL> assert first_time , '<STR_LIT>' <EOL> else : <EOL> assert not first_time , '<STR_LIT>' <EOL> assert not p , p <EOL> self . assertEqual ( event . wait ( ) , result ) <EOL> self . assertEqual ( queue . wait ( ) , result ) <EOL> self . assertRaises ( kill_exc_type , receiver . wait ) <EOL> self . assertRaises ( kill_exc_type , proc . waitall , [ receiver ] ) <EOL> sleep ( DELAY ) <EOL> assert not proc_flag , proc_flag <EOL> assert not callback_flag , callback_flag <EOL> self . check_timed_out ( * xxxxx ) <EOL> class TestReturn_link_value ( TestReturn_link ) : <EOL> sync = False <EOL> link_method = '<STR_LIT>' <EOL> class TestRaise_link ( TestCase ) : <EOL> link_method = '<STR_LIT>' <EOL> def _test_raise ( self , p , first_time , kill_exc_type ) : <EOL> event , receiver , proc_flag , queue , callback_flag = self . set_links ( p , first_time , kill_exc_type ) <EOL> xxxxx = self . set_links_timeout ( p . link_value ) <EOL> try : <EOL> sleep ( DELAY ) <EOL> except kill_exc_type : <EOL> assert first_time , '<STR_LIT>' <EOL> else : <EOL> assert not first_time , '<STR_LIT>' <EOL> assert not p , p <EOL> self . assertRaises ( ExpectedError , event . wait ) <EOL> self . assertRaises ( ExpectedError , queue . wait ) <EOL> self . assertRaises ( kill_exc_type , receiver . wait ) <EOL> self . assertRaises ( kill_exc_type , proc . waitall , [ receiver ] ) <EOL> sleep ( DELAY ) <EOL> assert not proc_flag , proc_flag <EOL> assert not callback_flag , callback_flag <EOL> self . check_timed_out ( * xxxxx ) <EOL> @ silence_warnings <EOL> def test_raise ( self ) : <EOL> p = self . p = proc . spawn ( lambda : getcurrent ( ) . throw ( ExpectedError ( '<STR_LIT>' ) ) ) <EOL> self . _test_raise ( p , True , proc . LinkedFailed ) <EOL> for _ in xrange ( <NUM_LIT:3> ) : <EOL> self . _test_raise ( p , False , proc . LinkedFailed ) <EOL> def _test_kill ( self , p , first_time , kill_exc_type ) : <EOL> event , receiver , proc_flag , queue , callback_flag = self . set_links ( p , first_time , kill_exc_type ) <EOL> xxxxx = self . set_links_timeout ( p . link_value ) <EOL> p . kill ( ) <EOL> try : <EOL> sleep ( DELAY ) <EOL> except kill_exc_type : <EOL> assert first_time , '<STR_LIT>' <EOL> else : <EOL> assert not first_time , '<STR_LIT>' <EOL> assert not p , p <EOL> self . assertRaises ( proc . ProcExit , event . wait ) <EOL> self . assertRaises ( proc . ProcExit , queue . wait ) <EOL> self . assertRaises ( kill_exc_type , proc . waitall , [ receiver ] ) <EOL> self . assertRaises ( kill_exc_type , receiver . wait ) <EOL> sleep ( DELAY ) <EOL> assert not proc_flag , proc_flag <EOL> assert not callback_flag , callback_flag <EOL> self . check_timed_out ( * xxxxx ) <EOL> @ silence_warnings <EOL> def test_kill ( self ) : <EOL> p = self . p = proc . spawn ( sleep , DELAY ) <EOL> self . _test_kill ( p , True , proc . LinkedKilled ) <EOL> for _ in xrange ( <NUM_LIT:3> ) : <EOL> self . _test_kill ( p , False , proc . LinkedKilled ) <EOL> class TestRaise_link_exception ( TestRaise_link ) : <EOL> link_method = '<STR_LIT>' <EOL> class TestStuff ( LimitedTestCase ) : <EOL> def test_wait_noerrors ( self ) : <EOL> x = proc . spawn ( lambda : <NUM_LIT:1> ) <EOL> y = proc . spawn ( lambda : <NUM_LIT:2> ) <EOL> z = proc . spawn ( lambda : <NUM_LIT:3> ) <EOL> self . assertEqual ( proc . waitall ( [ x , y , z ] ) , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> e = _event . Event ( ) <EOL> x . link ( e ) <EOL> self . assertEqual ( e . wait ( ) , <NUM_LIT:1> ) <EOL> x . unlink ( e ) <EOL> e = _event . Event ( ) <EOL> x . link ( e ) <EOL> self . assertEqual ( e . wait ( ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( [ proc . waitall ( [ X ] ) for X in [ x , y , z ] ] , [ [ <NUM_LIT:1> ] , [ <NUM_LIT:2> ] , [ <NUM_LIT:3> ] ] ) <EOL> @ skipped <EOL> def test_wait_error ( self ) : <EOL> def x ( ) : <EOL> sleep ( DELAY ) <EOL> return <NUM_LIT:1> <EOL> x = proc . spawn ( x ) <EOL> z = proc . spawn ( lambda : <NUM_LIT:3> ) <EOL> y = proc . spawn ( lambda : getcurrent ( ) . throw ( ExpectedError ( '<STR_LIT>' ) ) ) <EOL> y . link ( x ) <EOL> x . link ( y ) <EOL> y . link ( z ) <EOL> z . link ( y ) <EOL> self . assertRaises ( ExpectedError , proc . waitall , [ x , y , z ] ) <EOL> self . assertRaises ( proc . LinkedFailed , proc . waitall , [ x ] ) <EOL> self . assertEqual ( proc . waitall ( [ z ] ) , [ <NUM_LIT:3> ] ) <EOL> self . assertRaises ( ExpectedError , proc . waitall , [ y ] ) <EOL> def test_wait_all_exception_order ( self ) : <EOL> def first ( ) : <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> a = proc . spawn ( first ) <EOL> b = proc . spawn ( lambda : getcurrent ( ) . throw ( ExpectedError ( '<STR_LIT>' ) ) ) <EOL> try : <EOL> proc . waitall ( [ a , b ] ) <EOL> except ExpectedError , ex : <EOL> assert '<STR_LIT>' in str ( ex ) , repr ( str ( ex ) ) <EOL> sleep ( <NUM_LIT> ) <EOL> def test_multiple_listeners_error ( self ) : <EOL> p = proc . spawn ( lambda : <NUM_LIT:5> ) <EOL> results = [ ] <EOL> def listener1 ( * args ) : <EOL> results . append ( <NUM_LIT:10> ) <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> def listener2 ( * args ) : <EOL> results . append ( <NUM_LIT:20> ) <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> def listener3 ( * args ) : <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> p . link ( listener1 ) <EOL> p . link ( listener2 ) <EOL> p . link ( listener3 ) <EOL> sleep ( DELAY * <NUM_LIT:10> ) <EOL> assert results in [ [ <NUM_LIT:10> , <NUM_LIT:20> ] , [ <NUM_LIT:20> , <NUM_LIT:10> ] ] , results <EOL> p = proc . spawn ( lambda : getcurrent ( ) . throw ( ExpectedError ( '<STR_LIT>' ) ) ) <EOL> results = [ ] <EOL> p . link ( listener1 ) <EOL> p . link ( listener2 ) <EOL> p . link ( listener3 ) <EOL> sleep ( DELAY * <NUM_LIT:10> ) <EOL> assert results in [ [ <NUM_LIT:10> , <NUM_LIT:20> ] , [ <NUM_LIT:20> , <NUM_LIT:10> ] ] , results <EOL> def _test_multiple_listeners_error_unlink ( self , p ) : <EOL> results = [ ] <EOL> def listener1 ( * args ) : <EOL> p . unlink ( listener2 ) <EOL> results . append ( <NUM_LIT:5> ) <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> def listener2 ( * args ) : <EOL> p . unlink ( listener1 ) <EOL> results . append ( <NUM_LIT:5> ) <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> def listener3 ( * args ) : <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> p . link ( listener1 ) <EOL> p . link ( listener2 ) <EOL> p . link ( listener3 ) <EOL> sleep ( DELAY * <NUM_LIT:10> ) <EOL> assert results == [ <NUM_LIT:5> ] , results <EOL> def test_multiple_listeners_error_unlink_Proc ( self ) : <EOL> p = proc . spawn ( lambda : <NUM_LIT:5> ) <EOL> self . _test_multiple_listeners_error_unlink ( p ) <EOL> def test_multiple_listeners_error_unlink_Source ( self ) : <EOL> p = proc . Source ( ) <EOL> proc . spawn ( p . send , <NUM_LIT:6> ) <EOL> self . _test_multiple_listeners_error_unlink ( p ) <EOL> def test_killing_unlinked ( self ) : <EOL> e = _event . Event ( ) <EOL> def func ( ) : <EOL> try : <EOL> raise ExpectedError ( '<STR_LIT>' ) <EOL> except : <EOL> e . send_exception ( * sys . exc_info ( ) ) <EOL> p = proc . spawn_link ( func ) <EOL> try : <EOL> try : <EOL> e . wait ( ) <EOL> except ExpectedError : <EOL> pass <EOL> finally : <EOL> p . unlink ( ) <EOL> sleep ( DELAY ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
