<s> import time <EOL> import testify as T <EOL> import vimap . pool <EOL> import vimap . worker_process <EOL> @ vimap . worker_process . worker <EOL> def worker_proc ( seq , init = <NUM_LIT:0> , sleep_s = None ) : <EOL> for x in seq : <EOL> if sleep_s is not None : <EOL> time . sleep ( sleep_s ) <EOL> yield x + init <EOL> class FuzzTest ( T . TestCase ) : <EOL> def fork_pool ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return vimap . pool . fork ( * args , ** kwargs ) <EOL> def test_fuzz ( self ) : <EOL> for n in xrange ( <NUM_LIT:1> , <NUM_LIT:100> ) : <EOL> processes = self . fork_pool ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ) <EOL> res = list ( processes . imap ( list ( range ( <NUM_LIT:1> , n ) ) ) . zip_in_out ( ) ) <EOL> T . assert_sets_equal ( set ( out for in_ , out in res ) , set ( range ( <NUM_LIT:2> , n + <NUM_LIT:1> ) ) ) <EOL> def test_try_overwhelm_output_queue ( self ) : <EOL> processes = self . fork_pool ( worker_proc . init_args ( init = i ) for i in [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ) <EOL> processes . imap ( xrange ( <NUM_LIT> ) ) . block_ignore_output ( ) <EOL> def test_slow_consumer ( self ) : <EOL> processes = self . fork_pool ( worker_proc . init_args ( <EOL> init = i , sleep_s = <NUM_LIT> ) for i in [ <NUM_LIT:1> ] * <NUM_LIT:10> ) <EOL> for _ , output in processes . imap ( xrange ( <NUM_LIT:100> ) ) . zip_in_out ( ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> class ChunkedFuzzTest ( FuzzTest ) : <EOL> def fork_pool ( self , * args , ** kwargs ) : <EOL> return vimap . pool . fork_chunked ( * args , ** kwargs ) </s>
