<s> """<STR_LIT>""" <EOL> import coro <EOL> from coro . test import coro_unittest <EOL> import os <EOL> import unittest <EOL> class Test ( unittest . TestCase ) : <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> def test_wait_for_interrupt_new ( self ) : <EOL> proc = coro_process . spawn_job_bg ( '<STR_LIT>' ) <EOL> def waiter ( ) : <EOL> proc . wait ( ) <EOL> waiter_thread = coro . spawn ( waiter ) <EOL> def killer ( ) : <EOL> waiter_thread . shutdown ( ) <EOL> coro . spawn ( killer ) <EOL> coro . yield_slice ( ) <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> def test_wait_for_interrupt_submitted ( self ) : <EOL> proc = coro_process . spawn_job_bg ( '<STR_LIT>' ) <EOL> def waiter ( ) : <EOL> proc . wait ( ) <EOL> waiter_thread = coro . spawn ( waiter ) <EOL> coro . yield_slice ( ) <EOL> waiter_thread . shutdown ( ) <EOL> coro . yield_slice ( ) <EOL> def test_wait_for_interrupted_fired ( self ) : <EOL> read_fd1 , write_fd1 = os . pipe ( ) <EOL> read_fd2 , write_fd2 = os . pipe ( ) <EOL> try : <EOL> read_sock1 = coro . sock ( fd = read_fd1 ) <EOL> read_sock2 = coro . sock ( fd = read_fd2 ) <EOL> sleeper1_thread = None <EOL> sleeper2_thread = None <EOL> sleeper1_data = [ ] <EOL> sleeper2_data = [ ] <EOL> def sleeper1 ( ) : <EOL> data = read_sock1 . read ( <NUM_LIT:10> ) <EOL> sleeper1_data . append ( data ) <EOL> sleeper2_thread . shutdown ( ) <EOL> def sleeper2 ( ) : <EOL> data = read_sock2 . read ( <NUM_LIT:10> ) <EOL> sleeper2_data . append ( data ) <EOL> sleeper1_thread . shutdown ( ) <EOL> sleeper1_thread = coro . spawn ( sleeper1 ) <EOL> sleeper2_thread = coro . spawn ( sleeper2 ) <EOL> coro . yield_slice ( ) <EOL> os . write ( write_fd1 , '<STR_LIT>' ) <EOL> os . write ( write_fd2 , '<STR_LIT>' ) <EOL> coro . yield_slice ( ) <EOL> coro . yield_slice ( ) <EOL> self . assertTrue ( len ( sleeper1_data ) == <NUM_LIT:1> or <EOL> len ( sleeper2_data ) == <NUM_LIT:1> ) <EOL> self . assertTrue ( sleeper1_thread . dead ) <EOL> self . assertTrue ( sleeper2_thread . dead ) <EOL> finally : <EOL> os . close ( read_fd1 ) <EOL> os . close ( write_fd1 ) <EOL> os . close ( read_fd2 ) <EOL> os . close ( write_fd2 ) <EOL> def test_with_timeout ( self ) : <EOL> def serve ( port ) : <EOL> s = coro . tcp_sock ( ) <EOL> s . bind ( ( '<STR_LIT>' , port ) ) <EOL> s . listen ( <NUM_LIT:5> ) <EOL> with self . assertRaises ( coro . TimeoutError ) : <EOL> conn , addr = coro . with_timeout ( <NUM_LIT:1> , s . accept ) <EOL> with self . assertRaises ( coro . TimeoutError ) : <EOL> conn , addr = coro . with_timeout ( <NUM_LIT:1> , s . accept ) <EOL> coro . spawn ( serve , <NUM_LIT> ) <EOL> coro . yield_slice ( ) <EOL> coro . sleep_relative ( <NUM_LIT:3> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> coro_unittest . run_tests ( ) </s>
