<s> import mock <EOL> import testify as T <EOL> import vimap . pool <EOL> import vimap . testing <EOL> import vimap . worker_process <EOL> @ vimap . worker_process . worker <EOL> def worker_proc ( seq , init ) : <EOL> for x in seq : <EOL> yield x + init <EOL> class DebugPoolTest ( T . TestCase ) : <EOL> @ vimap . testing . mock_debug_pool ( ) <EOL> def test_debug_pool ( self ) : <EOL> T . assert_equal ( vimap . pool . VimapPool , vimap . testing . DebugPool ) <EOL> processes = vimap . pool . fork ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ) <EOL> processes . imap ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) . block_ignore_output ( ) <EOL> for i in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] : <EOL> T . assert_equal ( processes . output_for_input [ i ] , i + <NUM_LIT:1> ) <EOL> class SerialPoolTest ( T . TestCase ) : <EOL> @ T . setup_teardown <EOL> def mock_pool ( self ) : <EOL> with vimap . testing . mock_serial_pool ( ) : <EOL> T . assert_equal ( vimap . pool . VimapPool , vimap . testing . SerialPool ) <EOL> yield <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_serial_pool_doesnt_fork ( self , start ) : <EOL> processes = vimap . pool . fork ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ) <EOL> processes . imap ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> processes . block_ignore_output ( ) <EOL> for i in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] : <EOL> T . assert_equal ( processes . output_for_input [ i ] , i + <NUM_LIT:1> ) <EOL> assert not start . called <EOL> def test_zip_in_out ( self ) : <EOL> processes = vimap . pool . fork ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> results = list ( processes . imap ( [ <NUM_LIT:4> , <NUM_LIT:4> , <NUM_LIT:4> ] ) . zip_in_out ( close_if_done = False ) ) <EOL> T . assert_sets_equal ( set ( results ) , set ( [ ( <NUM_LIT:4> , <NUM_LIT:5> ) , ( <NUM_LIT:4> , <NUM_LIT:6> ) , ( <NUM_LIT:4> , <NUM_LIT:7> ) ] ) ) <EOL> results = list ( processes . imap ( [ <NUM_LIT:4> , <NUM_LIT:4> , <NUM_LIT:4> ] ) . zip_in_out ( close_if_done = True ) ) <EOL> T . assert_sets_equal ( set ( results ) , set ( [ ( <NUM_LIT:4> , <NUM_LIT:5> ) , ( <NUM_LIT:4> , <NUM_LIT:6> ) , ( <NUM_LIT:4> , <NUM_LIT:7> ) ] ) ) <EOL> def test_zip_in_out_lots_of_input ( self ) : <EOL> processes = vimap . pool . fork ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> results = list ( processes . imap ( [ <NUM_LIT:4> , <NUM_LIT:4> , <NUM_LIT:4> ] * <NUM_LIT:3> ) . zip_in_out ( ) ) <EOL> T . assert_equal ( set ( results [ <NUM_LIT:0> : <NUM_LIT:3> ] ) , set ( [ ( <NUM_LIT:4> , <NUM_LIT:5> ) , ( <NUM_LIT:4> , <NUM_LIT:6> ) , ( <NUM_LIT:4> , <NUM_LIT:7> ) ] ) ) <EOL> T . assert_equal ( set ( results [ <NUM_LIT:3> : <NUM_LIT:6> ] ) , set ( [ ( <NUM_LIT:4> , <NUM_LIT:5> ) , ( <NUM_LIT:4> , <NUM_LIT:6> ) , ( <NUM_LIT:4> , <NUM_LIT:7> ) ] ) ) <EOL> T . assert_equal ( set ( results [ <NUM_LIT:6> : <NUM_LIT:9> ] ) , set ( [ ( <NUM_LIT:4> , <NUM_LIT:5> ) , ( <NUM_LIT:4> , <NUM_LIT:6> ) , ( <NUM_LIT:4> , <NUM_LIT:7> ) ] ) ) <EOL> def test_zip_in_out_more_workers_than_input ( self ) : <EOL> processes = vimap . pool . fork ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , - <NUM_LIT:1> , <NUM_LIT:0> ] ) <EOL> results = list ( processes . imap ( [ <NUM_LIT:4> , <NUM_LIT:4> ] ) . zip_in_out ( ) ) <EOL> T . assert_equal ( set ( results ) , set ( [ ( <NUM_LIT:4> , <NUM_LIT:5> ) , ( <NUM_LIT:4> , <NUM_LIT:3> ) ] ) ) <EOL> def test_zip_in_out_no_input ( self ) : <EOL> processes = vimap . pool . fork ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , - <NUM_LIT:1> , <NUM_LIT:0> ] ) <EOL> results = processes . imap ( [ ] ) . zip_in_out ( ) <EOL> T . assert_equal ( set ( results ) , set ( [ ] ) ) <EOL> class NoWarningsTest ( T . TestCase ) : <EOL> def test_no_warnings ( self ) : <EOL> with mock . patch ( '<STR_LIT>' ) as stderr : <EOL> vimap . exception_handling . print_warning ( '<STR_LIT>' ) <EOL> T . assert_equal ( True , stderr . write . called ) <EOL> T . assert_equal ( True , stderr . flush . called ) <EOL> with T . assert_raises ( AssertionError ) : <EOL> with vimap . testing . no_warnings ( ) : <EOL> vimap . exception_handling . print_warning ( '<STR_LIT>' ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> T . run ( ) </s>
