<s> from irctk . threadpool import ThreadPool , Worker <EOL> from base import IrcTkTestCase <EOL> class ThreadPoolTestCase ( IrcTkTestCase ) : <EOL> def setUp ( self ) : <EOL> IrcTkTestCase . setUp ( self ) <EOL> self . min_workers = <NUM_LIT:3> <EOL> self . tp = ThreadPool ( self . min_workers ) <EOL> self . assertEqual ( self . tp . min_workers , <NUM_LIT:3> ) <EOL> self . assertEqual ( self . tp . logger , None ) <EOL> self . assertEqual ( self . tp . wait , <NUM_LIT> ) <EOL> self . assertTrue ( self . tp . daemon is not None ) <EOL> self . worker = Worker ( self . tp . tasks , None ) <EOL> self . assertEqual ( self . tp . tasks , self . worker . tasks ) <EOL> self . assertEqual ( None , self . worker . logger ) <EOL> def foo ( self , * args , ** kwargs ) : <EOL> return args or kwargs or '<STR_LIT:bar>' <EOL> def bad_foo ( self ) : <EOL> raise Exception <EOL> def test_enqueue_task ( self ) : <EOL> self . tp . enqueue_task ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> task = self . tp . tasks . get ( ) <EOL> self . assertEqual ( ( '<STR_LIT:foo>' , ( '<STR_LIT:bar>' , ) , { } ) , task ) <EOL> self . tp . enqueue_task ( '<STR_LIT:foo>' , '<STR_LIT:bar>' , test = True ) <EOL> task = self . tp . tasks . get ( ) <EOL> self . assertEqual ( ( '<STR_LIT:foo>' , ( '<STR_LIT:bar>' , ) , { '<STR_LIT:test>' : True } ) , task ) <EOL> def test_spawn_worker ( self ) : <EOL> self . tp . _spawn_worker ( ) <EOL> self . assertTrue ( self . tp . too_few_workers ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . tp . _spawn_worker ( ) <EOL> self . assertFalse ( self . tp . too_few_workers ) </s>
