<s> """<STR_LIT>""" <EOL> import coro <EOL> import unittest <EOL> from coro . test import coro_unittest <EOL> class Test ( unittest . TestCase ) : <EOL> def test_isem_interrupt_schedule ( self ) : <EOL> """<STR_LIT>""" <EOL> s = coro . inverted_semaphore ( ) <EOL> s . acquire ( <NUM_LIT:1> ) <EOL> self . _resume_count = <NUM_LIT:0> <EOL> threads = [ ] <EOL> for unused in xrange ( <NUM_LIT:5> ) : <EOL> threads . append ( coro . spawn ( self . _isem_block , s ) ) <EOL> no_interrupt_thread = coro . spawn ( self . _isem_block , s ) <EOL> coro . yield_slice ( ) <EOL> for t in threads : <EOL> t . shutdown ( ) <EOL> s . release ( <NUM_LIT:1> ) <EOL> coro . yield_slice ( ) <EOL> self . assertEqual ( self . _resume_count , <NUM_LIT:1> ) <EOL> def _isem_block ( self , s ) : <EOL> s . block_till_zero ( ) <EOL> self . _resume_count += <NUM_LIT:1> <EOL> def test_isem_schedule_interrupt ( self ) : <EOL> """<STR_LIT>""" <EOL> s = coro . inverted_semaphore ( ) <EOL> s . acquire ( <NUM_LIT:1> ) <EOL> self . _resume_count = <NUM_LIT:0> <EOL> threads = [ ] <EOL> for unused in xrange ( <NUM_LIT:5> ) : <EOL> threads . append ( coro . spawn ( self . _isem_block , s ) ) <EOL> no_interrupt_thread = coro . spawn ( self . _isem_block , s ) <EOL> coro . yield_slice ( ) <EOL> s . release ( <NUM_LIT:1> ) <EOL> for t in threads : <EOL> t . shutdown ( ) <EOL> coro . yield_slice ( ) <EOL> self . assertEqual ( self . _resume_count , <NUM_LIT:1> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> coro_unittest . run_tests ( ) </s>
