<s> import sys <EOL> import warnings <EOL> from tests import LimitedTestCase , main , skip_on_windows <EOL> warnings . simplefilter ( '<STR_LIT:ignore>' , DeprecationWarning ) <EOL> from eventlet import processes , api <EOL> warnings . simplefilter ( '<STR_LIT:default>' , DeprecationWarning ) <EOL> class TestEchoPool ( LimitedTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestEchoPool , self ) . setUp ( ) <EOL> self . pool = processes . ProcessPool ( '<STR_LIT>' , [ "<STR_LIT:hello>" ] ) <EOL> @ skip_on_windows <EOL> def test_echo ( self ) : <EOL> result = None <EOL> proc = self . pool . get ( ) <EOL> try : <EOL> result = proc . read ( ) <EOL> finally : <EOL> self . pool . put ( proc ) <EOL> self . assertEquals ( result , '<STR_LIT>' ) <EOL> @ skip_on_windows <EOL> def test_read_eof ( self ) : <EOL> proc = self . pool . get ( ) <EOL> try : <EOL> proc . read ( ) <EOL> self . assertRaises ( processes . DeadProcess , proc . read ) <EOL> finally : <EOL> self . pool . put ( proc ) <EOL> @ skip_on_windows <EOL> def test_empty_echo ( self ) : <EOL> p = processes . Process ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( '<STR_LIT>' , p . read ( ) ) <EOL> self . assertRaises ( processes . DeadProcess , p . read ) <EOL> class TestCatPool ( LimitedTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestCatPool , self ) . setUp ( ) <EOL> api . sleep ( <NUM_LIT:0> ) <EOL> self . pool = processes . ProcessPool ( '<STR_LIT>' ) <EOL> @ skip_on_windows <EOL> def test_cat ( self ) : <EOL> result = None <EOL> proc = self . pool . get ( ) <EOL> try : <EOL> proc . write ( '<STR_LIT>' ) <EOL> proc . close_stdin ( ) <EOL> result = proc . read ( ) <EOL> finally : <EOL> self . pool . put ( proc ) <EOL> self . assertEquals ( result , '<STR_LIT>' ) <EOL> @ skip_on_windows <EOL> def test_write_to_dead ( self ) : <EOL> result = None <EOL> proc = self . pool . get ( ) <EOL> try : <EOL> proc . write ( '<STR_LIT>' ) <EOL> proc . close_stdin ( ) <EOL> result = proc . read ( ) <EOL> self . assertRaises ( processes . DeadProcess , proc . write , '<STR_LIT:foo>' ) <EOL> finally : <EOL> self . pool . put ( proc ) <EOL> @ skip_on_windows <EOL> def test_close ( self ) : <EOL> result = None <EOL> proc = self . pool . get ( ) <EOL> try : <EOL> proc . write ( '<STR_LIT:hello>' ) <EOL> proc . close ( ) <EOL> self . assertRaises ( processes . DeadProcess , proc . write , '<STR_LIT>' ) <EOL> finally : <EOL> self . pool . put ( proc ) <EOL> class TestDyingProcessesLeavePool ( LimitedTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestDyingProcessesLeavePool , self ) . setUp ( ) <EOL> self . pool = processes . ProcessPool ( '<STR_LIT>' , [ '<STR_LIT:hello>' ] , max_size = <NUM_LIT:1> ) <EOL> @ skip_on_windows <EOL> def test_dead_process_not_inserted_into_pool ( self ) : <EOL> proc = self . pool . get ( ) <EOL> try : <EOL> try : <EOL> result = proc . read ( ) <EOL> self . assertEquals ( result , '<STR_LIT>' ) <EOL> result = proc . read ( ) <EOL> except processes . DeadProcess : <EOL> pass <EOL> finally : <EOL> self . pool . put ( proc ) <EOL> proc2 = self . pool . get ( ) <EOL> self . assert_ ( proc is not proc2 ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
