<s> import test . test_support <EOL> from test . test_support import verbose , is_jython <EOL> import random <EOL> import re <EOL> import sys <EOL> thread = test . test_support . import_module ( '<STR_LIT>' ) <EOL> threading = test . test_support . import_module ( '<STR_LIT>' ) <EOL> import time <EOL> import unittest <EOL> import weakref <EOL> import os <EOL> import subprocess <EOL> from test import lock_tests <EOL> class Counter ( object ) : <EOL> def __init__ ( self ) : <EOL> self . value = <NUM_LIT:0> <EOL> def inc ( self ) : <EOL> self . value += <NUM_LIT:1> <EOL> def dec ( self ) : <EOL> self . value -= <NUM_LIT:1> <EOL> def get ( self ) : <EOL> return self . value <EOL> class TestThread ( threading . Thread ) : <EOL> def __init__ ( self , name , testcase , sema , mutex , nrunning ) : <EOL> threading . Thread . __init__ ( self , name = name ) <EOL> self . testcase = testcase <EOL> self . sema = sema <EOL> self . mutex = mutex <EOL> self . nrunning = nrunning <EOL> def run ( self ) : <EOL> delay = random . random ( ) / <NUM_LIT> <EOL> if verbose : <EOL> print '<STR_LIT>' % ( <EOL> self . name , delay * <NUM_LIT> ) <EOL> with self . sema : <EOL> with self . mutex : <EOL> self . nrunning . inc ( ) <EOL> if verbose : <EOL> print self . nrunning . get ( ) , '<STR_LIT>' <EOL> self . testcase . assertTrue ( self . nrunning . get ( ) <= <NUM_LIT:3> ) <EOL> time . sleep ( delay ) <EOL> if verbose : <EOL> print '<STR_LIT>' , self . name , '<STR_LIT>' <EOL> with self . mutex : <EOL> self . nrunning . dec ( ) <EOL> self . testcase . assertTrue ( self . nrunning . get ( ) >= <NUM_LIT:0> ) <EOL> if verbose : <EOL> print '<STR_LIT>' % ( <EOL> self . name , self . nrunning . get ( ) ) <EOL> class BaseTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . _threads = test . test_support . threading_setup ( ) <EOL> def tearDown ( self ) : <EOL> test . test_support . threading_cleanup ( * self . _threads ) <EOL> test . test_support . reap_children ( ) <EOL> class ThreadTests ( BaseTestCase ) : <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_various_ops ( self ) : <EOL> NUMTASKS = <NUM_LIT:10> <EOL> sema = threading . BoundedSemaphore ( value = <NUM_LIT:3> ) <EOL> mutex = threading . RLock ( ) <EOL> numrunning = Counter ( ) <EOL> threads = [ ] <EOL> for i in range ( NUMTASKS ) : <EOL> t = TestThread ( "<STR_LIT>" % i , self , sema , mutex , numrunning ) <EOL> threads . append ( t ) <EOL> self . assertEqual ( t . ident , None ) <EOL> self . assertTrue ( re . match ( '<STR_LIT>' , repr ( t ) ) ) <EOL> t . start ( ) <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> for t in threads : <EOL> t . join ( NUMTASKS ) <EOL> self . assertTrue ( not t . is_alive ( ) ) <EOL> self . assertNotEqual ( t . ident , <NUM_LIT:0> ) <EOL> self . assertFalse ( t . ident is None ) <EOL> self . assertTrue ( re . match ( '<STR_LIT>' , repr ( t ) ) ) <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> self . assertEqual ( numrunning . get ( ) , <NUM_LIT:0> ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_ident_of_no_threading_threads ( self ) : <EOL> self . assertFalse ( threading . currentThread ( ) . ident is None ) <EOL> def f ( ) : <EOL> ident . append ( threading . currentThread ( ) . ident ) <EOL> done . set ( ) <EOL> done = threading . Event ( ) <EOL> ident = [ ] <EOL> thread . start_new_thread ( f , ( ) ) <EOL> done . wait ( ) <EOL> self . assertFalse ( ident [ <NUM_LIT:0> ] is None ) <EOL> del threading . _active [ ident [ <NUM_LIT:0> ] ] <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_various_ops_small_stack ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> try : <EOL> threading . stack_size ( <NUM_LIT> ) <EOL> except thread . error : <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> return <EOL> self . test_various_ops ( ) <EOL> threading . stack_size ( <NUM_LIT:0> ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_various_ops_large_stack ( self ) : <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> try : <EOL> threading . stack_size ( <NUM_LIT> ) <EOL> except thread . error : <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> return <EOL> self . test_various_ops ( ) <EOL> threading . stack_size ( <NUM_LIT:0> ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_foreign_thread ( self ) : <EOL> def f ( mutex ) : <EOL> threading . current_thread ( ) <EOL> mutex . release ( ) <EOL> mutex = threading . Lock ( ) <EOL> mutex . acquire ( ) <EOL> tid = thread . start_new_thread ( f , ( mutex , ) ) <EOL> mutex . acquire ( ) <EOL> self . assertIn ( tid , threading . _active ) <EOL> self . assertIsInstance ( threading . _active [ tid ] , threading . _DummyThread ) <EOL> del threading . _active [ tid ] <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_PyThreadState_SetAsyncExc ( self ) : <EOL> try : <EOL> import ctypes <EOL> except ImportError : <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> return <EOL> set_async_exc = ctypes . pythonapi . PyThreadState_SetAsyncExc <EOL> class AsyncExc ( Exception ) : <EOL> pass <EOL> exception = ctypes . py_object ( AsyncExc ) <EOL> tid = thread . get_ident ( ) <EOL> try : <EOL> result = set_async_exc ( ctypes . c_long ( tid ) , exception ) <EOL> while True : <EOL> pass <EOL> except AsyncExc : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> try : <EOL> self . assertEqual ( result , <NUM_LIT:1> ) <EOL> except UnboundLocalError : <EOL> pass <EOL> worker_started = threading . Event ( ) <EOL> worker_saw_exception = threading . Event ( ) <EOL> class Worker ( threading . Thread ) : <EOL> def run ( self ) : <EOL> self . id = thread . get_ident ( ) <EOL> self . finished = False <EOL> try : <EOL> while True : <EOL> worker_started . set ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> except AsyncExc : <EOL> self . finished = True <EOL> worker_saw_exception . set ( ) <EOL> t = Worker ( ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> result = set_async_exc ( ctypes . c_long ( - <NUM_LIT:1> ) , exception ) <EOL> self . assertEqual ( result , <NUM_LIT:0> ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> ret = worker_started . wait ( ) <EOL> self . assertTrue ( ret ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> self . assertTrue ( not t . finished ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> result = set_async_exc ( ctypes . c_long ( t . id ) , exception ) <EOL> self . assertEqual ( result , <NUM_LIT:1> ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> worker_saw_exception . wait ( timeout = <NUM_LIT:10> ) <EOL> self . assertTrue ( t . finished ) <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> if t . finished : <EOL> t . join ( ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_limbo_cleanup ( self ) : <EOL> def fail_new_thread ( * args ) : <EOL> raise thread . error ( ) <EOL> _start_new_thread = threading . _start_new_thread <EOL> threading . _start_new_thread = fail_new_thread <EOL> try : <EOL> t = threading . Thread ( target = lambda : None ) <EOL> self . assertRaises ( thread . error , t . start ) <EOL> self . assertFalse ( <EOL> t in threading . _limbo , <EOL> "<STR_LIT>" ) <EOL> finally : <EOL> threading . _start_new_thread = _start_new_thread <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_finalize_runnning_thread ( self ) : <EOL> try : <EOL> import ctypes <EOL> except ImportError : <EOL> if verbose : <EOL> print ( "<STR_LIT>" ) <EOL> return <EOL> rc = subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] ) <EOL> self . assertEqual ( rc , <NUM_LIT> ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_finalize_with_trace ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> stdout , stderr = p . communicate ( ) <EOL> rc = p . returncode <EOL> self . assertFalse ( rc == <NUM_LIT:2> , "<STR_LIT>" ) <EOL> self . assertTrue ( rc == <NUM_LIT:0> , <EOL> "<STR_LIT>" + repr ( stderr ) ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_join_nondaemon_on_shutdown ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> stdout , stderr = p . communicate ( ) <EOL> self . assertEqual ( stdout . strip ( ) , <EOL> "<STR_LIT>" ) <EOL> stderr = re . sub ( r"<STR_LIT>" , "<STR_LIT>" , stderr , re . MULTILINE ) . strip ( ) <EOL> self . assertEqual ( stderr , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_enumerate_after_join ( self ) : <EOL> enum = threading . enumerate <EOL> old_interval = sys . getcheckinterval ( ) <EOL> try : <EOL> for i in xrange ( <NUM_LIT:1> , <NUM_LIT:100> ) : <EOL> sys . setcheckinterval ( i // <NUM_LIT:5> ) <EOL> t = threading . Thread ( target = lambda : None ) <EOL> t . start ( ) <EOL> t . join ( ) <EOL> l = enum ( ) <EOL> self . assertNotIn ( t , l , <EOL> "<STR_LIT>" % ( i , l ) ) <EOL> finally : <EOL> sys . setcheckinterval ( old_interval ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_no_refcycle_through_target ( self ) : <EOL> class RunSelfFunction ( object ) : <EOL> def __init__ ( self , should_raise ) : <EOL> self . should_raise = should_raise <EOL> self . thread = threading . Thread ( target = self . _run , <EOL> args = ( self , ) , <EOL> kwargs = { '<STR_LIT>' : self } ) <EOL> self . thread . start ( ) <EOL> def _run ( self , other_ref , yet_another ) : <EOL> if self . should_raise : <EOL> raise SystemExit <EOL> cyclic_object = RunSelfFunction ( should_raise = False ) <EOL> weak_cyclic_object = weakref . ref ( cyclic_object ) <EOL> cyclic_object . thread . join ( ) <EOL> del cyclic_object <EOL> self . assertEqual ( None , weak_cyclic_object ( ) , <EOL> msg = ( '<STR_LIT>' % <EOL> sys . getrefcount ( weak_cyclic_object ( ) ) ) ) <EOL> raising_cyclic_object = RunSelfFunction ( should_raise = True ) <EOL> weak_raising_cyclic_object = weakref . ref ( raising_cyclic_object ) <EOL> raising_cyclic_object . thread . join ( ) <EOL> del raising_cyclic_object <EOL> self . assertEqual ( None , weak_raising_cyclic_object ( ) , <EOL> msg = ( '<STR_LIT>' % <EOL> sys . getrefcount ( weak_raising_cyclic_object ( ) ) ) ) <EOL> class ThreadJoinOnShutdown ( BaseTestCase ) : <EOL> platforms_to_skip = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def _run_and_join ( self , script ) : <EOL> script = """<STR_LIT>""" + script <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , script ] , stdout = subprocess . PIPE ) <EOL> rc = p . wait ( ) <EOL> data = p . stdout . read ( ) . replace ( '<STR_LIT:\r>' , '<STR_LIT>' ) <EOL> p . stdout . close ( ) <EOL> self . assertEqual ( data , "<STR_LIT>" ) <EOL> self . assertFalse ( rc == <NUM_LIT:2> , "<STR_LIT>" ) <EOL> self . assertTrue ( rc == <NUM_LIT:0> , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_1_join_on_shutdown ( self ) : <EOL> script = """<STR_LIT>""" <EOL> self . _run_and_join ( script ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform in platforms_to_skip , "<STR_LIT>" ) <EOL> def test_2_join_in_forked_process ( self ) : <EOL> script = """<STR_LIT>""" <EOL> self . _run_and_join ( script ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform in platforms_to_skip , "<STR_LIT>" ) <EOL> def test_3_join_in_forked_from_thread ( self ) : <EOL> script = """<STR_LIT>""" <EOL> self . _run_and_join ( script ) <EOL> def assertScriptHasOutput ( self , script , expected_output ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , script ] , <EOL> stdout = subprocess . PIPE ) <EOL> rc = p . wait ( ) <EOL> data = p . stdout . read ( ) . decode ( ) . replace ( '<STR_LIT:\r>' , '<STR_LIT>' ) <EOL> self . assertEqual ( rc , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> self . assertEqual ( data , expected_output ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform in platforms_to_skip , "<STR_LIT>" ) <EOL> def test_4_joining_across_fork_in_worker_thread ( self ) : <EOL> script = """<STR_LIT>""" <EOL> self . assertScriptHasOutput ( script , "<STR_LIT>" ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform in platforms_to_skip , "<STR_LIT>" ) <EOL> def test_5_clear_waiter_locks_to_avoid_crash ( self ) : <EOL> script = """<STR_LIT>""" <EOL> output = "<STR_LIT>" <EOL> self . assertScriptHasOutput ( script , output ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform in platforms_to_skip , "<STR_LIT>" ) <EOL> def test_reinit_tls_after_fork ( self ) : <EOL> def do_fork_and_wait ( ) : <EOL> pid = os . fork ( ) <EOL> if pid > <NUM_LIT:0> : <EOL> os . waitpid ( pid , <NUM_LIT:0> ) <EOL> else : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:16> ) : <EOL> t = threading . Thread ( target = do_fork_and_wait ) <EOL> threads . append ( t ) <EOL> t . start ( ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> class ThreadingExceptionTests ( BaseTestCase ) : <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_start_thread_again ( self ) : <EOL> thread = threading . Thread ( ) <EOL> thread . start ( ) <EOL> self . assertRaises ( RuntimeError , thread . start ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_joining_current_thread ( self ) : <EOL> current_thread = threading . current_thread ( ) <EOL> self . assertRaises ( RuntimeError , current_thread . join ) ; <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_joining_inactive_thread ( self ) : <EOL> thread = threading . Thread ( ) <EOL> self . assertRaises ( RuntimeError , thread . join ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> def test_daemonize_active_thread ( self ) : <EOL> thread = threading . Thread ( ) <EOL> thread . start ( ) <EOL> self . assertRaises ( RuntimeError , setattr , thread , "<STR_LIT>" , True ) <EOL> class LockTests ( lock_tests . LockTests ) : <EOL> locktype = staticmethod ( threading . Lock ) <EOL> class RLockTests ( lock_tests . RLockTests ) : <EOL> locktype = staticmethod ( threading . RLock ) <EOL> class EventTests ( lock_tests . EventTests ) : <EOL> eventtype = staticmethod ( threading . Event ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> class ConditionAsRLockTests ( lock_tests . RLockTests ) : <EOL> locktype = staticmethod ( threading . Condition ) <EOL> class ConditionTests ( lock_tests . ConditionTests ) : <EOL> condtype = staticmethod ( threading . Condition ) <EOL> class SemaphoreTests ( lock_tests . SemaphoreTests ) : <EOL> semtype = staticmethod ( threading . Semaphore ) <EOL> @ unittest . skipIf ( is_jython , "<STR_LIT>" ) <EOL> class BoundedSemaphoreTests ( lock_tests . BoundedSemaphoreTests ) : <EOL> semtype = staticmethod ( threading . BoundedSemaphore ) <EOL> @ unittest . skipUnless ( sys . platform == '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_recursion_limit ( self ) : <EOL> script = """<STR_LIT>""" <EOL> expected_output = "<STR_LIT>" <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , script ] , <EOL> stdout = subprocess . PIPE ) <EOL> stdout , stderr = p . communicate ( ) <EOL> data = stdout . decode ( ) . replace ( '<STR_LIT:\r>' , '<STR_LIT>' ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> self . assertEqual ( data , expected_output ) <EOL> def test_main ( ) : <EOL> test . test_support . run_unittest ( LockTests , RLockTests , EventTests , <EOL> ConditionAsRLockTests , ConditionTests , <EOL> SemaphoreTests , BoundedSemaphoreTests , <EOL> ThreadTests , <EOL> ThreadJoinOnShutdown , <EOL> ThreadingExceptionTests , <EOL> ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) </s>
