<s> """<STR_LIT>""" <EOL> import _dummy_thread as _thread <EOL> import time <EOL> import queue <EOL> import random <EOL> import unittest <EOL> from test import support <EOL> DELAY = <NUM_LIT:0> <EOL> class LockTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . lock = _thread . allocate_lock ( ) <EOL> def test_initlock ( self ) : <EOL> self . assertTrue ( not self . lock . locked ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_release ( self ) : <EOL> self . lock . acquire ( ) <EOL> self . lock . release ( ) <EOL> self . assertTrue ( not self . lock . locked ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_improper_release ( self ) : <EOL> self . assertRaises ( RuntimeError , self . lock . release ) <EOL> def test_cond_acquire_success ( self ) : <EOL> self . assertTrue ( self . lock . acquire ( <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> def test_cond_acquire_fail ( self ) : <EOL> self . lock . acquire ( <NUM_LIT:0> ) <EOL> self . assertTrue ( not self . lock . acquire ( <NUM_LIT:0> ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_uncond_acquire_success ( self ) : <EOL> self . lock . acquire ( ) <EOL> self . assertTrue ( self . lock . locked ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_uncond_acquire_return_val ( self ) : <EOL> self . assertTrue ( self . lock . acquire ( <NUM_LIT:1> ) is True , <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( self . lock . acquire ( ) is True ) <EOL> def test_uncond_acquire_blocking ( self ) : <EOL> def delay_unlock ( to_unlock , delay ) : <EOL> """<STR_LIT>""" <EOL> time . sleep ( delay ) <EOL> to_unlock . release ( ) <EOL> self . lock . acquire ( ) <EOL> start_time = int ( time . time ( ) ) <EOL> _thread . start_new_thread ( delay_unlock , ( self . lock , DELAY ) ) <EOL> if support . verbose : <EOL> print ( ) <EOL> print ( "<STR_LIT>" "<STR_LIT>" % DELAY ) <EOL> self . lock . acquire ( ) <EOL> end_time = int ( time . time ( ) ) <EOL> if support . verbose : <EOL> print ( "<STR_LIT>" ) <EOL> self . assertTrue ( ( end_time - start_time ) >= DELAY , <EOL> "<STR_LIT>" ) <EOL> class MiscTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_exit ( self ) : <EOL> self . assertRaises ( SystemExit , _thread . exit ) <EOL> def test_ident ( self ) : <EOL> self . assertIsInstance ( _thread . get_ident ( ) , int , <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( _thread . get_ident ( ) != <NUM_LIT:0> , <EOL> "<STR_LIT>" ) <EOL> def test_LockType ( self ) : <EOL> self . assertIsInstance ( _thread . allocate_lock ( ) , _thread . LockType , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_interrupt_main ( self ) : <EOL> def call_interrupt ( ) : <EOL> _thread . interrupt_main ( ) <EOL> self . assertRaises ( KeyboardInterrupt , _thread . start_new_thread , <EOL> call_interrupt , tuple ( ) ) <EOL> def test_interrupt_in_main ( self ) : <EOL> self . assertRaises ( KeyboardInterrupt , _thread . interrupt_main ) <EOL> class ThreadTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_arg_passing ( self ) : <EOL> def arg_tester ( queue , arg1 = False , arg2 = False ) : <EOL> """<STR_LIT>""" <EOL> queue . put ( ( arg1 , arg2 ) ) <EOL> testing_queue = queue . Queue ( <NUM_LIT:1> ) <EOL> _thread . start_new_thread ( arg_tester , ( testing_queue , True , True ) ) <EOL> result = testing_queue . get ( ) <EOL> self . assertTrue ( result [ <NUM_LIT:0> ] and result [ <NUM_LIT:1> ] , <EOL> "<STR_LIT>" ) <EOL> _thread . start_new_thread ( arg_tester , tuple ( ) , { '<STR_LIT>' : testing_queue , <EOL> '<STR_LIT>' : True , '<STR_LIT>' : True } ) <EOL> result = testing_queue . get ( ) <EOL> self . assertTrue ( result [ <NUM_LIT:0> ] and result [ <NUM_LIT:1> ] , <EOL> "<STR_LIT>" ) <EOL> _thread . start_new_thread ( arg_tester , ( testing_queue , True ) , { '<STR_LIT>' : True } ) <EOL> result = testing_queue . get ( ) <EOL> self . assertTrue ( result [ <NUM_LIT:0> ] and result [ <NUM_LIT:1> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_multi_creation ( self ) : <EOL> def queue_mark ( queue , delay ) : <EOL> """<STR_LIT>""" <EOL> time . sleep ( delay ) <EOL> queue . put ( _thread . get_ident ( ) ) <EOL> thread_count = <NUM_LIT:5> <EOL> testing_queue = queue . Queue ( thread_count ) <EOL> if support . verbose : <EOL> print ( ) <EOL> print ( "<STR_LIT>" "<STR_LIT>" % ( DELAY , thread_count ) ) <EOL> for count in range ( thread_count ) : <EOL> if DELAY : <EOL> local_delay = round ( random . random ( ) , <NUM_LIT:1> ) <EOL> else : <EOL> local_delay = <NUM_LIT:0> <EOL> _thread . start_new_thread ( queue_mark , <EOL> ( testing_queue , local_delay ) ) <EOL> time . sleep ( DELAY ) <EOL> if support . verbose : <EOL> print ( '<STR_LIT>' ) <EOL> self . assertTrue ( testing_queue . qsize ( ) == thread_count , <EOL> "<STR_LIT>" % <EOL> ( thread_count , DELAY ) ) <EOL> def test_main ( imported_module = None ) : <EOL> global _thread , DELAY <EOL> if imported_module : <EOL> _thread = imported_module <EOL> DELAY = <NUM_LIT:2> <EOL> if support . verbose : <EOL> print ( ) <EOL> print ( "<STR_LIT>" % _thread ) <EOL> support . run_unittest ( LockTests , MiscTests , ThreadTests ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_main ( ) </s>
