python (3.12.0)

Browse
Build Log
Usage
Build Options
Dependencies
Source
Versions
Version Info
(root)/
lib/
python3.12/
test/
__pycache__/
test_threading.cpython-312.pyc

ΑeJdZddlZddlmZmZddlmZmZmZddlm	Z	ddl
mZmZddl
Z
ddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlmZddlmZdd	lmZej:d
dZd
ZGdde Z!GddejDZ#GddejHZ%Gdde%Z&Gdde%Z'Gdde%Z(Gdde%Z)GddejDZ*Gdde%Z+Gd d!e%Z,Gd"d#ejZZ-Gd$d%ej\Z/ej`ejbdud&Gd'd(ej\Z2Gd)d*ejfZ3Gd+d,ej\Z4Gd-d.ejjZ5Gd/d0ejlZ6Gd1d2ejnZ7Gd3d4ejpZ8Gd5d6ejHZ9Gd7d8ejHZ:Gd9d:ejHZ;e<d;k(rejzyy)<z!
Tests for the threading module.
N)threading_helperrequires_subprocess)verbosecpython_only	os_helper)
import_module)assert_python_okassert_python_failure)mock)
lock_tests)supportT)module)netbsd5zhp-ux11c|jttdtjtjt_y)N
excepthook)
addCleanupsetattr	threadingr__excepthook__)testcases G/BuggyBox/python/3.12.0/bootstrap/lib/python3.12/test/test_threading.pyrestore_default_excepthookr&s*L):N:NO$33Ic$eZdZdZdZdZdZy)Countercd|_y)Nrvalueselfs r__init__zCounter.__init__-s	
rc.|xjdz
c_yNrrs rinczCounter.inc/

a
rc.|xjdzc_yr#rrs rdeczCounter.dec1r&rc|jSNrrs rgetzCounter.get3szzrN)__name__
__module____qualname__r!r%r(r+rrrr,srrceZdZdZdZy)
TestThreadc~tjj||||_||_||_||_y)Nname)rThreadr!rsemamutexnrunning)r r4rr6r7r8s      rr!zTestThread.__init__7s7!!$T!2 
	
 
rctjdz}trtd|j|dzfz|j5|j
5|jjtr$t|jjd|jj|jjddddtj|trtd|jd|j
5|jj|jj|jjdtr2td	|j|jjfzddddddy#1swYxYw#1swYxYw#1swYyxYw)
Ng@ztask %s will run for %.1f usecg.Aztasks are runningtaskdonerz$%s is finished. %d tasks are running)randomrprintr4r6r7r8r%r+rassertLessEqualtimesleepr(assertGreaterEqual)r delays  rrunzTestThread.run>s^

')299eck*+
,YY	<
F

!!#$--++-/BC

--dmm.?.?.A1E	
F
JJufdii0
<

!!#

001B1B1DaH@99dmm&7&7&9:;<	
<	<	<
F
F
<
<	<	<s?
GA9F+
AGBF7G+F4	0G7G	<GGN)r,r-r.r!rDr/rrr1r16s!<rr1ceZdZdZdZy)BaseTestCasec6tj|_yr*)rthreading_setup_threadsrs rsetUpzBaseTestCase.setUpXs(88:
rcxtj|jtjjyr*)rthreading_cleanuprItestr

reap_childrenrs rtearDownzBaseTestCase.tearDown[s$**DMM:""$rN)r,r-r.rJrOr/rrrFrFWs;%rrFcveZdZedZdZedZdZdZdZ	dZ
dZd	Zd
Z
dZdZd
ZdZdZdZdZdZej.dZej.dZdZej.ej8eedddZej@e!jDe#vdej.ej8eedddZ$dZ%dZ&dZ'dZ(dZ)ed Z*d!Z+d"Z,d#Z-d$Z.ed%Z/d&Z0d'Z1d(Z2d)Z3d*Z4y+),ThreadTestscd}tjd}|j|jdtjd}|j|jdtj|d}|j|jdtj
j
tdd	
5tjd}|j|jddddtj
j
tdd
5tj}|j|jddddtj
j
tdd
5tj|}|j|jddddy#1swYxYw#1swYwxYw#1swYyxYw)Ncyr*r/r/rrfuncz#ThreadTests.test_name.<locals>.funcdrmyname1r3{123myname2)targetr4_counter)return_valuezThread-2r:zThread-3rZzThread-5 (func))rr5assertEqualr4rpatchobject)r rTthreads   r	test_namezThreadTests.test_namebsu!!y1i0!!s+e,!!I>i0
ZZ

y*1

E	6%%2.FV[[*5	6ZZ

y*1

E	6%%'FV[[*5	6ZZ

y*1

E	=%%T2FV[[*;<	=	=	6	6	6	6	=	=s$3F="1G	3G=G	GGcFdg}d}dg}d}dgf}dg}|fdf|fdf|fdf|fdf|fd	f|fd
ff}|D]X\}}	j|	|5tj|	|}
|
j|
j	dddZy#1swYexYw)Nr$r$strrhc(j|dSr#raargr s r<lambda>z0ThreadTests.test_args_argument.<locals>.<lambda>s4#3#3C#;rc(j|dSr#rkrls rrnz0ThreadTests.test_args_argument.<locals>.<lambda>sD$4$4S!$<rc(j|dSNrhrkrls rrnz0ThreadTests.test_args_argument.<locals>.<lambda>s4#3#3C#?rc(j|dSrqrkrls rrnz0ThreadTests.test_args_argument.<locals>.<lambda>sD$4$4S%$@rc*j|dgSr#rkrls rrnz0ThreadTests.test_args_argument.<locals>.<lambda>s(8(8qc(Brc(j|dS)Nrgrkrls rrnz0ThreadTests.test_args_argument.<locals>.<lambda>s(8(8d(CrrZargs)subTestrr5startjoin)r num_list	num_tuplestr_list	str_tuple
list_in_tuple
tuple_in_list
test_casesrvrZts`          rtest_args_argumentzThreadTests.test_args_argument}s3	7	

;<
<=
?@
@A
BC
CD


'	LD&V$7
$$F>	

	

s8BB 	c~tj}tjj	|t|yr*)rLockrMr
check_disallow_instantiationtyper locks  rtest_disallow_instantiationz'ThreadTests.test_disallow_instantiations(~~11$T
Crcd}tjd}tj}t}g}t	|D]k}td|z||||}|j
||j|j|jt|d|jmttdrXtd|Dtjhz}|jd||j!t#||dzt$rt'd	|D]}|j)|j+|j-|j/|jd
|j1|j|jt|dt$rt'd|j!|j3d
y)
N
r:rz<thread %d>z^<TestThread\(.*, initial\)>$
get_native_idc34K|]}|jywr*)	native_id).0rs  r	<genexpr>z/ThreadTests.test_various_ops.<locals>.<genexpr>s:QQ[[:sr$z!waiting for all tasks to completerz#^<TestThread\(.*, stopped -?\d+\)>$zall tasks done)rBoundedSemaphoreRLockrranger1appendassertIsNoneidentassertRegexreprrxhasattrsetrassertNotInralenrr>ryassertFalseis_aliveassertNotEqualassertIsNotNoner+)	r NUMTASKSr6r7
numrunningthreadsir
native_idss	         rtest_various_opszThreadTests.test_various_opss))2!Y
x	A=?D$zJANN1agg&T!W&FG
GGI	9o.:'::i>U>U>W=XXJT:.S_hl;56	NA
FFHQZZ\*+  )T!W&LM	N"#)1-rc|jtjjfd}tjgtj5tj|d}j|jd|dddtjd=y#1swYxYw)Ncjtjjj	yr*)rrcurrent_threadrr)r<rsrfz9ThreadTests.test_ident_of_no_threading_threads.<locals>.fs'LL11399:HHJrr/r)rrrrEventrwait_threads_exit_threadstart_new_threadwaitra_active)r rtidr<rs   @@r"test_ident_of_no_threading_threadsz.ThreadTests.test_ident_of_no_threading_threadssY557==>	 

/
/
1	,**1b1CIIKU1Xs+	,

eAh'	,	,s <B99Cctrtd	tjd|jtjdy#tj
$rt
jdwxYw)Nz!with 256 KiB thread stack size...i4platform does not support changing thread stack sizer	rr>r
stack_sizererrorunittestSkipTestrrs rtest_various_ops_small_stackz(ThreadTests.test_various_ops_small_stacksm56	H  (	
Q	}}	H##FH
H	HA)A7ctrtd	tjd|jtjdy#tj
$rt
jdwxYw)Nzwith 1 MiB thread stack size...irrrrs rtest_various_ops_large_stackz(ThreadTests.test_various_ops_large_stacksm34	H  *	
Q	}}	H##FH
H	Hrcnd}tj}|jtj5tj||f}|jddd|jtj|jtj|tj|jtj|j|jttj|dtj|=y#1swYxYw)NcLtj|jyr*)rrrelease)r7s rrz*ThreadTests.test_foreign_thread.<locals>.fs
$$&MMOr_DummyThread)rracquirerrrrassertInrassertIsInstancer
assertTruerrr)r rr7rs    rtest_foreign_threadzThreadTests.test_foreign_threads	 




/
/
1	**1uh7CMMO		


c9,,-i//4i6L6LM	))#.779:i//45~Fc"		s(D++D4c	
td}|jj}|j|jf|_Gddt	|j		}tj}|j|t|j|d	|||}		#	$rYnwxYw	|jdn#t$rYnwxYwtjtj
G	
fddtj }|}d|_|j%t&rt)d	t&rt)d
|d|}|j|dt&rt)dj+}|j-|t&rt)d
|j/|j0t&rt)d||j2|}|j|dt&rt)d
j+t4j6|j-|j0t&rt)d|j0r|j9yy)NctypesceZdZy)<ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.AsyncExcN)r,r-r.r/rrAsyncExcrsrrrTr$ceZdZfdZy):ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Workerctj|_d|_		j	tjd&#$rd|_j	YywxYw)NFTg?)r	get_identidfinishedrr@rA)r rworker_saw_exceptionworker_starteds rrDz>ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Worker.run,sb#--/ %
/&**,

3 /$(DM(,,./s'A

A)(A)Nr,r-r.rD)rrrsrWorkerr+s

/rrz    started worker threadz     trying nonsensical thread idz,    waiting for worker thread to get startedz"    verifying worker hasn't exitedz2    attempting to raise asynch exception in workerz5    waiting for worker to say it caught the exceptiontimeoutz    all OK -- joining worker)r	pythonapiPyThreadState_SetAsyncExcc_ulong	py_objectargtypes	Exceptionrrrint
assertGreaterfailraUnboundLocalErrorrr5daemonrxrr>rrrrrr

SHORT_TIMEOUTry)r r
set_async_exc	exceptionrresultrrretrrrs         @@@rtest_PyThreadState_SetAsyncExcz*ThreadTests.test_PyThreadState_SetAsyncExcs!x(((BB
"(..&2B2B!C
	y	$$X.	!!#c3'3"	-"3	2F			VQ' 		#*(0	/	/Y%%	/
H		-.45r9-#@A!!#67$FGqttY/#IJ!!'*?*?!@

#01::
FFHs$B++B32B37C

	CCc>d}tj}|t_	tjd}|jtj|j
|j
|tjvd|t_y#|t_wxYw)Nc*tjr*)rThreadErrorrvs rfail_new_threadz7ThreadTests.test_limbo_cleanup.<locals>.fail_new_thread\s''))rcyr*r/r/rrrnz0ThreadTests.test_limbo_cleanup.<locals>.<lambda>arUrr`z:Failed to cleanup _limbo map on failure of Thread.start().)r_start_new_threadr5assertRaisesrrxr_limbo)r rrrs    rtest_limbo_cleanupzThreadTests.test_limbo_cleanupZs~	*%77&5	#	<  5Ai33QWW=Y%%%L
N+<I'*;I'sA#B
Bc^tdtdd\}}}|j|dy)Nr-caNif 1:
            import ctypes, sys, time, _thread

            # This lock is used as a simple event variable.
            ready = _thread.allocate_lock()
            ready.acquire()

            # Module globals are cleared before __del__ is run
            # So we save the functions in class dict
            class C:
                ensure = ctypes.pythonapi.PyGILState_Ensure
                release = ctypes.pythonapi.PyGILState_Release
                def __del__(self):
                    state = self.ensure()
                    self.release(state)

            def waitingThread():
                x = C()
                ready.release()
                time.sleep(100)

            _thread.start_new_thread(waitingThread, ())
            ready.acquire()  # Be sure the other thread is waiting.
            sys.exit(42)
            *)rr
rar rcouterrs    rtest_finalize_running_threadz(ThreadTests.test_finalize_running_threadis5	h,T4C2	
R rctddy)NraPif 1:
            import sys, threading

            # A deadlock-killer, to prevent the
            # testsuite to hang forever
            def killer():
                import os, time
                time.sleep(2)
                print('program blocked; aborting')
                os._exit(2)
            t = threading.Thread(target=killer)
            t.daemon = True
            t.start()

            # This is the trace function
            def func(frame, event, arg):
                threading.current_thread()
                return func

            sys.settrace(func)
            )r	rs rtest_finalize_with_tracez$ThreadTests.test_finalize_with_traces	 	rctdd\}}}|j|jd|j|dy)Nraif 1:
                import threading
                from time import sleep

                def child():
                    sleep(1)
                    # As a non-daemon thread we SHOULD wake up and nothing
                    # should be torn down yet
                    print("Woke up, sleep function is:", sleep)

                threading.Thread(target=child).start()
                raise SystemExit
            s5Woke up, sleep function is: <built-in function sleep>r)r	rastriprs    rtest_join_nondaemon_on_shutdownz+ThreadTests.test_join_nondaemon_on_shutdownsG(/C	
D	Fc"rctj}tj}	t	ddD]p}tj
|dztjd}|j|j|}|j||d||fzr	tj
|y#tj
|wxYw)Nr$dg-C6*?cyr*r/r/rrrnz7ThreadTests.test_enumerate_after_join.<locals>.<lambda>rUrr`z&#1703448 triggered after %d trials: %s)
r	enumeratesysgetswitchintervalrsetswitchintervalr5rxryr)r enumold_intervalrrls      rtest_enumerate_after_joinz%ThreadTests.test_enumerate_after_joins"",,.
	01c]
G%%a&j1$$L9	F  A<1vEG

G
!!,/C!!,/sA?B<<CcGddt}t||d}tj|}|jj~|j
|dtj|z|d}tj|}|jj~|j
|dtj|zy)NceZdZdZdZy)DThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunctionc||_tj|j|fd|i|_|jjy)Nyet_another)rZrvkwargs)should_raiserr5_runrdrx)r rs  rr!zMThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction.__init__sD%1!'..dii59G7DT6JL!!#rc(|jrtyr*)r
SystemExit)r 	other_refrs   rrzIThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction._runs$$$$%rN)r,r-r.r!rr/rrRunSelfFunctionrs
$
%rrF)rz%d references still around)msgT)	rcrweakrefrefrdryrrgetrefcount)r r
cyclic_objectweak_cyclic_objectraising_cyclic_objectweak_raising_cyclic_objects      rtest_no_refcycle_through_targetz+ThreadTests.test_no_refcycle_through_targets	%f	%	#4('U;
$[[7!!#,.:!oo.@.BCD		F!0T B%,[[1F%G"$$))+!46:!oo.H.JKL		Nrchtj}|jtd5|j	ddd|jtd5|jdddd|jtd5|j
ddd|jtd5|jddddtj}|jtd5|jdddtj}|j|jtd5|jddd|jtd	5tjddd|jtd
5tjdddy#1swYxYw#1swYsxYw#1swYQxYw#1swY.xYw#1swYxYw#1swYxYw#1swYxYw#1swYyxYw)Nzget the daemon attributezset the daemon attributeTzget the name attributezset the name attributer4zuse is_set()zuse notify_all()zuse active_count()zuse current_thread())rr5assertWarnsRegexDeprecationWarningisDaemon	setDaemongetNamesetNamerisSet	Conditionr	notifyAllactiveCount
currentThread)r reconds    rtest_old_threading_apiz"ThreadTests.test_old_threading_apis


"
"#5#>@	
JJL	
"
"#5#>@	
KK	
"
"#5#<>	
IIK	
"
"#5#<>	
IIf	
OO

"
"#5~
F	
GGI	""$

"
"#57I
J	NN	
"
"#57K
L	$!!#	$

"
"#57M
N	&##%	&	&1								
		
			$	$	&	&s_GG
G*9G7=HH?H2H(GG'*G47HH
HH%(H1ctj}|jdt|d|_|jdt|yNrT)rr5rrrrr rs  rtest_repr_daemonzThreadTests.test_repr_daemons=47+

hQ(rc&tj}|j|jtjd}|j|jtjd}|j	|jy)NFrT)rr5rrrr2s  rtest_daemon_paramzThreadTests.test_daemon_paramsd"E*"D)!rcpd}td|\}}}|j|d|j|dy)Naif 1:
            import _thread, threading, os, time, warnings

            def background_thread(evt):
                # Creates and registers the _DummyThread instance
                threading.current_thread()
                evt.set()
                time.sleep(10)

            evt = threading.Event()
            _thread.start_new_thread(background_thread, (evt,))
            evt.wait()
            assert threading.active_count() == 2, threading.active_count()
            with warnings.catch_warnings(record=True) as ws:
                warnings.filterwarnings(
                        "always", category=DeprecationWarning)
                if os.fork() == 0:
                    assert threading.active_count() == 1, threading.active_count()
                    os._exit(0)
                else:
                    assert ws[0].category == DeprecationWarning, ws[0]
                    assert 'fork' in str(ws[0].message), ws[0]
                    os.wait()
        rrr	rar code_rrs     rtest_dummy_thread_after_forkz(ThreadTests.test_dummy_thread_after_forks=0'tT23c"c"rcHtj}|jtj|tj
jdt
dD]}tjd}|jtjtd5tjx}dk(r(tj|j!rdnd	n'|j#tj$|d	
dddy#1swYxYw)Ngư>cyr*r/r/rrrnz6ThreadTests.test_is_alive_after_fork.<locals>.<lambda>ArUrr`ignorecategoryactionrrexitcode)rrrrrMr
rrr5rxwarningscatch_warningsr#osfork_exitrrywait_process)r r	rrpids     rtest_is_alive_after_forkz$ThreadTests.test_is_alive_after_fork6s,,.--|<	
&&t,r	;A  5A
GGI((2D08:
;779$C*HH1::<RR8FFH((r:
;
;		;
;
;s
$A)DD!	ctj}j|jdj|jtj
jj|jtjfd}tj|}|j|jy)N
MainThreadcjtjjtjjyr*)rrmain_threadrrrsrrz'ThreadTests.test_main_thread.<locals>.fSs5	 5 5 7 = = ) 8 8 : @ @
Brr`)
rrRrar4rrrr5rxry)r mainrths`   rtest_main_threadzThreadTests.test_main_threadMs$$&L1Y%=%=%?%E%EFY%8%8%:;	B

Q
'


	rwaitpidztest needs os.waitpid()cd}td|\}}}|jjdd}|j|d|j|dy)Naif 1:
            import os, threading
            from test import support

            pid = os.fork()
            if pid == 0:
                main = threading.main_thread()
                print(main.name)
                print(main.ident == threading.current_thread().ident)
                print(main.ident == threading.get_ident())
            else:
                support.wait_process(pid, exitcode=0)
        r
r^rzMainThread
True
True
r	decodereplacerar r:r;rrdatas      rtest_main_thread_after_forkz'ThreadTests.test_main_thread_after_forkZsV'tT23zz|##D"-c"9:rdue to known OS bugcd}td|\}}}|jjdd}|j|jdd|j|dy)NaOif 1:
            import os, threading, sys, warnings
            from test import support

            def func():
                with warnings.catch_warnings(record=True) as ws:
                    warnings.filterwarnings(
                            "always", category=DeprecationWarning)
                    pid = os.fork()
                    if pid == 0:
                        main = threading.main_thread()
                        print(main.name)
                        print(main.ident == threading.current_thread().ident)
                        print(main.ident == threading.get_ident())
                        # stdout is fully buffered because not a tty,
                        # we have to flush before exit.
                        sys.stdout.flush()
                    else:
                        assert ws[0].category == DeprecationWarning, ws[0]
                        assert 'fork' in str(ws[0].message), ws[0]
                        support.wait_process(pid, exitcode=0)

            th = threading.Thread(target=func)
            th.start()
            th.join()
        rrXr^utf-8zThread-1 (func)
True
True
rYr\s      r/test_main_thread_after_fork_from_nonmain_threadz;ThreadTests.test_main_thread_after_fork_from_nonmain_threados_4'tT23zz|##D"-G,b1>?rcd}td|\}}}|j}|j|d|j|jdgdzy)Naif 1:
            import gc, threading

            main_thread = threading.current_thread()
            assert main_thread is threading.main_thread()  # sanity check

            class RefCycle:
                def __init__(self):
                    self.cycle = self

                def __del__(self):
                    print("GC:",
                          threading.current_thread() is main_thread,
                          threading.main_thread() is main_thread,
                          threading.enumerate() == [main_thread])

            RefCycle()
            gc.collect()  # sanity check
            x = RefCycle()
        rrzGC: True True Truer\)r	rZra
splitlinesr\s      r test_main_thread_during_shutdownz,ThreadTests.test_main_thread_during_shutdownsZ('tT23zz|c"*./!3	5rcLd}td|\}}}|j|dy)Naif 1:
            import os
            import threading
            import time
            import random

            def random_sleep():
                seconds = random.random() * 0.010
                time.sleep(seconds)

            class Sleeper:
                def __del__(self):
                    random_sleep()

            tls = threading.local()

            def f():
                # Sleep a bit so that the thread is still running when
                # Py_Finalize() is called.
                random_sleep()
                tls.x = Sleeper()
                random_sleep()

            threading.Thread(target=f).start()
            random_sleep()
        rrr8r r:rrrs     rtest_finalization_shutdownz&ThreadTests.test_finalization_shutdowns-4(d3Cc"rcTtjtjjjfd}tj|}|j|jd|jj|j|j|j}|j|jddj|j|jtjd|j|j|j|j|j|j|j|jy)Ncpjjtjdy)N{Gz?)rrr@rAfinishstartedsrrz'ThreadTests.test_tstate_lock.<locals>.fs#OONNJJtrr`rrF)r
allocate_lockrrr5assertIs_tstate_lockrxrrrrr
rrry)r rrtstate_lockrmrns    @@rtest_tstate_lockzThreadTests.test_tstate_locks''')&&(	

A&

annd+		

%nn,,Q,7?	
++G4I4I+JER	


%&!..)	rc(tjtjjjfd}tj|}|jj|j
dt|jd}tdD]&}|t|vrntjd(|j
|t||jy)NcFjjyr*)rrrlsrrz(ThreadTests.test_repr_stopped.<locals>.fsOONNrr`rnstoppedirk)
rrorrr5rxrrrrr@rAry)r rrLOOKING_FORrrmrns     @@rtest_repr_stoppedzThreadTests.test_repr_stoppeds'')&&(	
A&		

ia) s	Ad1g%JJt		


k47+	rc8tddD]}tj|}t|Dcgc]"}tj|j$}}|D]}|j|D]}|j
t|Dcgc]"}tj|j$}}|D]}|j|D]}|j
|jt|jycc}wcc}w)Nr$rr`)
rrrr5rrxryrr
ValueError)r limitbsr;rrs      rtest_BoundedSemaphore_limitz'ThreadTests.test_BoundedSemaphore_limits1b\	6E++E2B %e.!''rzz:.G.
	


!&e.!''rzz:.G.
	


j"**5	6..s'D'Dcfddfdd_tj}tj	t	jddl}|j
tdD]	}	tj|t	j|y#tj|t	j|wxYw)NcSr*r/frameeventrm
noop_traces   rrz9ThreadTests.test_frame_tstate_tracing.<locals>.noop_trace*	rc3K	dw)N	generatorr/r/rrrz8ThreadTests.test_frame_tstate_tracing.<locals>.generator.s!!s	c^j_tjSr*)gennext)callbackrsrrz7ThreadTests.test_frame_tstate_tracing.<locals>.callback2s%||#({%%rrr:)rrgettracesettracer	_testcapicall_in_temporary_c_threadr)r 	old_tracerrMrrrs    @@@rtest_frame_tstate_tracingz%ThreadTests.test_frame_tstate_tracing"s		"	&LLN	Z 	*z*
00:a



LL#y)
LL#y)s
AB00,Ccfdtj}	tjtj}|j|tj|y#tj|wxYw)NcSr*r/rs   rrz-ThreadTests.test_gettrace.<locals>.noop_traceLrr)rrrra)r r
trace_funcrs   @r
test_gettracezThreadTests.test_gettraceKsf	&&(		*z*"++-JZ
3y)Iy)s;A--Bcd}tj}tjtjgfd}	tj|}|j	jtj|j|j|jd|g|jtj||jtj|tj||jtj||jtj|y#tj|wxYw)Ncyr*r/rs rfnz1ThreadTests.test_gettrace_all_threads.<locals>.fnXrUrcjtjjj	jtjyr*)rrrrr)first_checksecond_checktrace_funcssrcheckerz6ThreadTests.test_gettrace_all_threads.<locals>.checker^sAs||~.OOs||~.rr`)rrrr5rxrsettrace_all_threadsrryrar)r rrrrrrrs     @@@rtest_gettrace_all_threadsz%ThreadTests.test_gettrace_all_threadsWs&&(	oo' (	/	6  0A
GGI**2.
FFH[4*5Y//126S\\^R0**95++-y93
**95

CE22F	cd}tj}	tj||j|tjtj|y#tj|wxYw)Ncyr*r/rs rrz'ThreadTests.test_getprofile.<locals>.fnurUr)r
getprofile
setprofilera)r rold_profiles   rtest_getprofilezThreadTests.test_getprofilets\**,	.  $R!5!5!78  -I  -s9A((A?cd}tj}tjtjgfd}	tj|}|j	jtj|j|j|jd|g|jtj||jtj|tj||jtj||jtj|y#tj|wxYw)Ncyr*r/rs rrz3ThreadTests.test_getprofile_all_threads.<locals>.fn~rUrcjtjjj	jtjyr*)rrrrr)r
profile_funcsrsrrz8ThreadTests.test_getprofile_all_threads.<locals>.checkersC  !12OO  !12rr`)rrrr5rxrsetprofile_all_threadsrryrar)r rrrrrrrs     @@@rtest_getprofile_all_threadsz'ThreadTests.test_getprofile_all_threads}s**,oo' (
	3	:  0A
GGI,,R0
FFH]T2J7Y113R8S^^-r2,,[9--/=);7
,,[9rcdD]}|j|5tj}tj|j|}|j|j}|s!|j|tjn |j|tj|j|j|j|tjdddy#1swYxYw)N)FTr5)rZr)rwrrr5rrxrqr_shutdown_locksrrry)r rrrdrrs     rtest_shutdown_lockszThreadTests.test_shutdown_lockss#	IFV,
I!)"))FK$11MM+y/H/HI$$[)2K2KL		
  i.G.GH'
I
I	I
I
IsCC88D	cdtdd\}}}|j|jdy)Nra(if 1:
            import threading

            class Atexit:
                def __del__(self):
                    print("thread_dict.atexit = %r" % thread_dict.atexit)

            thread_dict = threading.local()
            thread_dict.atexit = "value"

            atexit = Atexit()
        sthread_dict.atexit = 'value')r	rarstriprs    rtest_locals_at_exitzThreadTests.test_locals_at_exits4(/
C	
'FGrcGddt}|}tj|}|j|j	|j|jy)NceZdZdZdZdZy)6ThreadTests.test_boolean_target.<locals>.BooleanTargetcd|_yNFranrs rr!z?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__init__s	 rcyrr/rs r__bool__z?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__bool__srcd|_y)NTrrs r__call__z?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__call__s	rN)r,r-r.r!rrr/rr
BooleanTargetrs
!

 rrr`)rcrr5rxryrr)r rrZrds    rtest_boolean_targetzThreadTests.test_boolean_targetsI
	 F	 !!0


#rcd}tj5tj|j	dddy#1swYyxYw)Ncyr*r/r/rrnoopz0ThreadTests.test_leak_without_join.<locals>.nooprUrr`)rrrr5rx)r rs  rtest_leak_without_joinz"ThreadTests.test_leak_without_joins?	

/
/
1	2D)//1	2	2	2s%AActjd}td|\}}}|j|d|j|dy)Na
            import _thread
            import sys

            event = _thread.allocate_lock()
            event.acquire()

            def import_threading():
                import threading
                event.release()

            if 'threading' in sys.modules:
                raise Exception('threading is already imported')

            _thread.start_new_thread(import_threading, ())

            # wait until the threading module is imported
            event.acquire()
            event.release()

            if 'threading' not in sys.modules:
                raise Exception('threading is not imported')

            # don't wait until the thread completes
        rr)textwrapdedentr	rargs     rtest_import_from_another_threadz+ThreadTests.test_import_from_another_threadsK 
2(d3Cc"c"rcpd}td|\}}}|j|d|jd|y)Naif 1:
            import atexit
            import _thread

            def f():
                print("shouldn't be printed")

            def exit_handler():
                _thread.start_new_thread(f, ())

            atexit.register(exit_handler)
        rrs/can't create new thread at interpreter shutdown)r	rarr9s     rtest_start_new_thread_at_exitz)ThreadTests.test_start_new_thread_at_exits<'tT23c"

H#NrN)5r,r-r.rrerrrrrrrrrrrrrr r/r3r6r

requires_forkr<rNrUr
skipUnlessrrIr^skipIfrplatformplatforms_to_skiprbrerhrsrxr}rrrrrrrrrrrr/rrrQrQ`s==46DD".H
( 	 	 #.Up
<!B2#(0"NB&>)"W##>W;;,WXY/1JK;L;&X__S\\%668MNWXY/1JK@LO@@5:!#F!F66$&*&*P
*4:.8:II.H"$&2#BOrrQceZdZdZdZejeje	jevddZejeje	jevddZ
eje	jevddZejeje	jevddZejdZy	)
ThreadJoinOnShutdowncd|z}td|\}}}|jjdd}|j|dy)Naif 1:
            import sys, os, time, threading

            # a thread, which waits for the main program to terminate
            def joiningfunc(mainthread):
                mainthread.join()
                print('end of thread')
                # stdout is fully buffered because not a tty, we have to flush
                # before exit.
                sys.stdout.flush()
        
rrXr^zend of main
end of thread
rY)r scriptrrrr]s      r
_run_and_joinz"ThreadJoinOnShutdown._run_and_joinsP

(f5Czz|##D"-=>rc*d}|j|y)Nzif 1:
            import os
            t = threading.Thread(target=joiningfunc,
                                 args=(threading.current_thread(),))
            t.start()
            time.sleep(0.1)
            print('end of main')
            rr rs  rtest_1_join_on_shutdownz,ThreadJoinOnShutdown.test_1_join_on_shutdown#s	
6"rr_c*d}|j|y)Naif 1:
            from test import support

            childpid = os.fork()
            if childpid != 0:
                # parent process
                support.wait_process(childpid, exitcode=0)
                sys.exit(0)

            # child process
            t = threading.Thread(target=joiningfunc,
                                 args=(threading.current_thread(),))
            t.start()
            print('end of main')
            rrs  rtest_2_join_in_forked_processz2ThreadJoinOnShutdown.test_2_join_in_forked_process/s	
6"rc*d}|j|y)Naif 1:
            from test import support

            main_thread = threading.current_thread()
            def worker():
                childpid = os.fork()
                if childpid != 0:
                    # parent process
                    support.wait_process(childpid, exitcode=0)
                    sys.exit(0)

                # child process
                t = threading.Thread(target=joiningfunc,
                                     args=(main_thread,))
                print('end of main')
                t.start()
                t.join() # Should not block: main_thread is already stopped

            w = threading.Thread(target=worker)
            w.start()
            rrs  r!test_3_join_in_forked_from_threadz6ThreadJoinOnShutdown.test_3_join_in_forked_from_threadDs*	
6"rcJd}td|\}}}|j|y)Naif True:
            import os
            import random
            import sys
            import time
            import threading

            thread_has_run = set()

            def random_io():
                '''Loop for a while sleeping random tiny amounts and doing some I/O.'''
                import test.test_threading as mod
                while True:
                    with open(mod.__file__, 'rb') as in_f:
                        stuff = in_f.read(200)
                        with open(os.devnull, 'wb') as null_f:
                            null_f.write(stuff)
                            time.sleep(random.random() / 1995)
                    thread_has_run.add(threading.current_thread())

            def main():
                count = 0
                for _ in range(40):
                    new_thread = threading.Thread(target=random_io)
                    new_thread.daemon = True
                    new_thread.start()
                    count += 1
                while len(thread_has_run) < count:
                    time.sleep(0.001)
                # Trigger process shutdown
                sys.exit(0)

            main()
            rr	rr rrrrs     rtest_4_daemon_threadsz*ThreadJoinOnShutdown.test_4_daemon_threadsas,
!D(f5Crc,d}tjtd5g}tdD]9}t	j
|}|j
||j;|D]}|j	dddy#1swYyxYw)Nctj}|dkDrtj|dytjdy)Nr2rE)rIrJr
rLrK)rMs rdo_fork_and_waitzIThreadJoinOnShutdown.test_reinit_tls_after_fork.<locals>.do_fork_and_waits/'')CQw$$S26rr@rAr`)	rGrHr#rrr5rrxry)r rrrrs     rtest_reinit_tls_after_forkz/ThreadJoinOnShutdown.test_reinit_tls_after_forks	
$
$.@,46
	G2Y
$$,<=q!	




	
	
	sA!B

BcXg}tdD]:}tjd}|j||j	<	tjtd5tj}|dk(rLttjdk(rtjdn-tjd	ntj|d
ddd|D]}|j!y#1swY!xYw#|D]}|j!wxYw)Nrc,tjdS)Ng333333?)r@rAr/rrrnzKThreadJoinOnShutdown.test_clear_threads_states_after_fork.<locals>.<lambda>sCrr`r@rArr$34rE)rrr5rrxrGrHr#rIrJrr_current_framesrKr
rLry)r rrrrMs     r$test_clear_threads_states_after_forkz9ThreadJoinOnShutdown.test_clear_threads_states_after_forks
r	A  (@AANN1
GGI	
	((2D08:

;ggi!83..01Q6((r:

;



;

;

s%D'A=D$DD
	DD)N)r,r-r.rrr
rrrrrrrrrrrr/rrrrs?"
#WX__S\\%668MN#O#&WX__S\\%668MN#O#6X__S\\%668MN'O'RWX__S\\%668MNO2Wrrc`eZdZdZdZdZedZddddddZed	Z	ed
Z
y)
SubinterpThreadingTestsctj\}}|jtj||jtj|t	tdrtj
|d||fS)Nset_blockingF)rIpipercloserr)r rws   rrzSubinterpThreadingTests.pipesVwwy1!$!$2~&OOAu%1v
rc
|j\}}tjd|fz}tjj|}|j
|d|j
tj|ddy)Na
            import os
            import random
            import threading
            import time

            def random_sleep():
                seconds = random.random() * 0.010
                time.sleep(seconds)

            def f():
                # Sleep a bit so that the thread is still running when
                # Py_EndInterpreter is called.
                random_sleep()
                os.write(%d, b"x")

            threading.Thread(target=f).start()
            random_sleep()
        rr$x	rrrrMr
run_in_subinterprarIreadr rrr:rs     rtest_threads_joinz)SubinterpThreadingTests.test_threads_joinsryy{1 $d% &ll++D1a A-rc
|j\}}tjd|fz}tjj|}|j
|d|j
tj|ddy)Na
            import os
            import random
            import threading
            import time

            def random_sleep():
                seconds = random.random() * 0.010
                time.sleep(seconds)

            class Sleeper:
                def __del__(self):
                    random_sleep()

            tls = threading.local()

            def f():
                # Sleep a bit so that the thread is still running when
                # Py_EndInterpreter is called.
                random_sleep()
                tls.x = Sleeper()
                os.write(%d, b"x")

            threading.Thread(target=f).start()
            random_sleep()
        rr$rrrs     rtest_threads_join_2z+SubinterpThreadingTests.test_threads_join_2sr
yy{1 2d3 4ll++D1a A-rcdtjjd}d|d}tjj5t	d|\}}}ddd|jdj
y#1swY*xYw)Nzif 1:
            import os
            import threading
            import time

            def f():
                # Make sure the daemon thread is still running when
                # Py_EndInterpreter is called.
                time.sleep(zJ)
            threading.Thread(target=f, daemon=True).start()
            zKif 1:
            import _testcapi

            _testcapi.run_in_subinterp()
            rz:Fatal Python error: Py_EndInterpreter: not the last thread)rMr
rSuppressCrashReportr
rrZ)r subinterp_coderrrrs      rtest_daemon_threads_fatal_errorz7SubinterpThreadingTests.test_daemon_threads_fatal_error
s!LL6678

"	$
\\
-
-
/	?0v>LBS	?

,-0ZZ\	;	?	?sA==BTFalloweddaemon_allowedrc	<tjd|d|d}tjd|d|d|d}tjj	5td|\}}}ddd|j
S#1swYj
SxYw)	Nz
            import test.support
            import threading
            def func():
                print('this should not have run!')
            t = threading.Thread(target=func, daemon=rz#
            t.start()
            zh
            import test.support
            test.support.run_in_subinterp_with_config(
                z,
                use_main_obmalloc=True,
                allow_fork=True,
                allow_exec=True,
                allow_threads=z',
                allow_daemon_threads=zp,
                check_multi_interp_extensions=False,
                own_gil=False,
            )
            r)rrrMr
rr	rZ)	r before_startrrrrrr;rs	         r_check_allowedz&SubinterpThreadingTests._check_allowed$s
".6
7=X>

N

*& "# 'i(&&4%56
"\\
-
-
/	7(v6IAq#	7zz|	7zz|sBBcP|jddd}|jd|y)NFrRuntimeError)rrr rs  rtest_threads_not_allowedz0SubinterpThreadingTests.test_threads_not_allowedCs0!! "

	


nc*rc6|jd5|jddd}|jd|ddd|jd5|jdddd}|jd|dddy#1swYKxYw#1swYyxYw)Nzvia Thread()TFrrzvia Thread.daemon setterzt.daemon = True)rwrrr	s  rtest_daemon_threads_not_allowedz7SubinterpThreadingTests.test_daemon_threads_not_allowedLs
\\.
)	/%%$&C

MM.#.
	/\\4
5	/%%!$	&C
MM.#.	/	/	/	/	/	/s'B(BBBN)r^)r,r-r.rrrrrrr
rr/rrrrsa.8#.J;;,#&*#>++//rrcheZdZdZdZdZdZdZedZ	dZ
dZd	Zd
Z
dZdZy
)ThreadingExceptionTestsctj}|j|jt|j|jyr*)rr5rxrrryr rds  rtest_start_thread_againz/ThreadingExceptionTests.test_start_thread_againcs6!!#,5
rcltj}|jt|jyr*)rrrrry)r rs  rtest_joining_current_threadz3ThreadingExceptionTests.test_joining_current_threadis&"113,(;(;<rcltj}|jt|jyr*)rr5rrryrs  rtest_joining_inactive_threadz4ThreadingExceptionTests.test_joining_inactive_threadms$!!#,4rctj}|j|jtt
|dd|j
yr1)rr5rxrrrryrs  rtest_daemonize_active_threadz4ThreadingExceptionTests.test_daemonize_active_threadqs8!!#,4H
rcltj}|jt|jyr*)rrrrrrs  rtest_releasing_unacquired_lockz6ThreadingExceptionTests.test_releasing_unacquired_lockws"~~,5rc|d}d}tjtjd|gtjtj}|j\}}|j
jdd}|j|jdd|j
z|j||y)	Naif True:
            import threading

            def recurse():
                return recurse()

            def outer():
                try:
                    recurse()
                except RecursionError:
                    pass

            w = threading.Thread(target=outer)
            w.start()
            w.join()
            print('end of main thread')
            zend of main thread
r)stdoutstderrrXr^rzUnexpected error: )

subprocessPopenr
executablePIPEcommunicaterZr[ra
returncode)r rexpected_outputprrr]s       rtest_recursion_limitz,ThreadingExceptionTests.test_recursion_limit{s"1cnndF;$.OOJOO
M}}&&tR0q*>*PQ/rcd}td|\}}}|j|d|j}|jd||jd||jd||j	d|y)Naif True:
            import threading
            import time

            running = False
            def run():
                global running
                running = True
                while running:
                    time.sleep(0.01)
                1/0
            t = threading.Thread(target=run)
            t.start()
            while not running:
                time.sleep(0.01)
            running = False
            t.join()
            rrException in thread"Traceback (most recent call last):ZeroDivisionErrorUnhandled exceptionr	rarZrrrs     rtest_print_exceptionz,ThreadingExceptionTests.test_print_exceptionsv$(f5Cc"jjl

+S1

:C@

)3/.4rcd}td|\}}}|j|d|j}|jd||jd||jd||j	d|y)Naif True:
            import sys
            import threading
            import time

            running = False
            def run():
                global running
                running = True
                while running:
                    time.sleep(0.01)
                1/0
            t = threading.Thread(target=run)
            t.start()
            while not running:
                time.sleep(0.01)
            sys.stderr = None
            running = False
            t.join()
            rrr'r(r)r*r+rs     r%test_print_exception_stderr_is_none_1z=ThreadingExceptionTests.test_print_exception_stderr_is_none_1sv((f5Cc"jjl

+S1

:C@

)3/.4rcd}td|\}}}|j|d|jd|jy)Naif True:
            import sys
            import threading
            import time

            running = False
            def run():
                global running
                running = True
                while running:
                    time.sleep(0.01)
                1/0
            sys.stderr = None
            t = threading.Thread(target=run)
            t.start()
            while not running:
                time.sleep(0.01)
            running = False
            t.join()
            rrr*)r	rarrZrs     r%test_print_exception_stderr_is_none_2z=ThreadingExceptionTests.test_print_exception_stderr_is_none_2sD((f5Cc".

=rc d}td|y)Na9if True:
            import time
            import threading
            import _thread

            def f():
                try:
                    f()
                except RecursionError:
                    f()

            def g():
                try:
                    raise ValueError()
                except* ValueError:
                    f()

            def h():
                time.sleep(1)
                _thread.interrupt_main()

            t = threading.Thread(target=h)
            t.start()
            g()
            t.join()
            r)r
rs  rtest_print_exception_gh_102056z6ThreadingExceptionTests.test_print_exception_gh_102056s6	dF+rcdGfddtj}|}|j|j|j	|j
|j
|j
td|_y)Ncr*r/r/rr
bare_raisezOThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.bare_raisesrceZdZdZfdZy)OThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558NcP	y#t$r}||_Yd}~yd}~wwxYwr*)rexc)r r9r5s  rrDzSThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558.runs&#L #"DHH#s	% %)r,r-r.r9rD)r5sr
Issue27558r7s
C
#rr:)rr5rxryrr9rr)r r:rdr5s   @r#test_bare_raise_in_brand_new_threadz;ThreadingExceptionTests.test_bare_raise_in_brand_new_threadsb		#))	#
VZZ(fjj,7
rcd}|jtjtjt	dDcgc]}tj|}}|D]"}|j|j$ycc}w)Ncttjdd5}|jdt	j
dddy#1swYyxYw)Nrra)encoding )openrTESTFNwrite	tracebackformat_stack)fps rmodify_filezQThreadingExceptionTests.test_multithread_modify_file_noerror.<locals>.modify_file"sBi&&g>
)"
&&(
)
)
)s&AArr`)	rrunlinkrArrr5rxry)r rFrrrs     r$test_multithread_modify_file_noerrorz<ThreadingExceptionTests.test_multithread_modify_file_noerror sy	)
	
	(()*:*:;3Z

K0

	A
GGI
FFH		
sBN)r,r-r.rrrrrrr%r,r.r0r2r;rHr/rrrr`sP>5600>565:>2,>*rrceZdZdZy)
ThreadRunFailctd)N
run failedrzrs rrDzThreadRunFail.run2s
&&rNrr/rrrJrJ1s'rrJc`eZdZfdZdZejdZdZdZ	dZ
dZxZS)ExceptHookTestsc8t|t|	yr*)rsuperrJ)r 	__class__s rrJzExceptHookTests.setUp7s"4(

rctjd5}td}|j|j	dddjj
}|jdjd||jd||jd||jd|y#1swY~xYw)	Nrexcepthook threadr3Exception in thread :
#Traceback (most recent call last):
z   raise ValueError("run failed")zValueError: run failed)	r
captured_outputrJrxrygetvaluerrr4)r rrds   rtest_excepthookzExceptHookTests.test_excepthook;s

$
$X
.	&"(;<FLLNKKM	
"((*

,V[[M=vF

<fE

8&A

.7		s-CC	c.tjd5}	td#t$rU}t	j
gt
jd}	t	j|d}n#d}wxYwYd}~nd}~wwxYw	dddn#1swYnxYwjj}|jdt	jd||jd||jd||jd|y)NrbugrUrVrWz  raise ValueError("bug")zValueError: bug)
r
rXrzrrExceptHookArgsrexc_inforrYrrr)r rr9rvs    rtest_excepthook_thread_Nonez+ExceptHookTests.test_excepthook_thread_NoneGs
$
$X
.		 &
  ''
  //0G#,,.0G$0GH ((. D4DD

 		 		 		 "((*

,Y-@-@-B,C3GP

<fE

16:

'0sCB#	B+A<A0-A<0A44A<7B<BBBcGddtj}tjd5}|}|j	|jddd|j
jdy#1swY*xYw)NceZdZdZy)4ExceptHookTests.test_system_exit.<locals>.ThreadExitc.tjdyr#)rexitrs rrDz8ExceptHookTests.test_system_exit.<locals>.ThreadExit.run^s
rNrr/rr
ThreadExitrb]s
rrerr^)rr5r
rXrxryrarY)r rerrds    rtest_system_exitz ExceptHookTests.test_system_exit\sm	))	

$
$X
.	&\FLLNKKM	
	
*B/		s(BB	cdfd}	tjtd|5t}|j	|jddd|j
jt|j
tjd|j
jjj|jjdy#1swYxYw#dwxYw)Nc
|yr*r/)	hook_argsrvs rhookz4ExceptHookTests.test_custom_excepthook.<locals>.hooklsDrrrL)r
	swap_attrrrJrxryraexc_typerzrh	exc_value
exc_traceback
__traceback__rprd)r rjrdrvs   @rtest_custom_excepthookz&ExceptHookTests.test_custom_excepthookis		""9lDA
&



T]]J7S0,?T//1M1MNMM$++v.D

Ds#C8+C,BC8,C51C88C<cd}dfd}tjtd|5tjtd|5tjd5}t}|j
|jddddddddd|jjd|jdy#1swYLxYw#1swYPxYw#1swYTxYw)Nctd)Nthreading_hook failedrMrs rthreading_hookzCExceptHookTests.test_custom_excepthook_fail.<locals>.threading_hooks455rct|yr*ri)rlrmrnerr_strs   rsys_hookz=ExceptHookTests.test_custom_excepthook_fail.<locals>.sys_hooks)nGrrrz#Exception in threading.excepthook:
rs)
r
rkrrrXrJrxryrarY)r rtrwrrdrvs     @rtest_custom_excepthook_failz+ExceptHookTests.test_custom_excepthook_fail~s	6	%

y,
G	


sL(
;	

$
$X
.	28"_FLLNKKM				
*?	A"9:						s;C'C+CCC'CCC$	 C''C0c8d}d}|}tjtd|5|}tjt_|}ddd|j||j
||j|dy#1swY@xYw)Nctjd5}td}|j|j	ddd|jS#1swYjSxYw)NrrTr3)r
rXrJrxryrY)outputrds  r
run_threadz<ExceptHookTests.test_original_excepthook.<locals>.run_threads]((2
f&,?@

??$$	
??$$s-AA3c:tdtjy)NzRunning a thread failed)file)r>rrrs rrtz@ExceptHookTests.test_original_excepthook.<locals>.threading_hooks+#**=rrzRunning a thread failed
)r
rkrrrrar)r r|rtdefault_outputcustom_hook_outputrecovered_outputs      rtest_original_excepthookz(ExceptHookTests.test_original_excepthooks	%	>$


y,
G	,!+#,#;#;I )|	,
	
)9:N,>?+-HI	,	,s(BB)
r,r-r.rJrZr
rr_rfrprxr
__classcell__)rRs@rrOrO6s<
8
11(0*;*JrrOceZdZdZdZdZy)
TimerTestscntj|g|_tj|_yr*)rFrJ
callback_argsrrcallback_eventrs rrJzTimerTests.setUps'4 'oo/rctjd|j}|j|jj|jjdd|jd<|jjtjd|j}|j|jj|jt|jd|j|jdifdifg|j|jy)Nrkblahbarfoor\r/)rTimer
_callback_spyrxrrrvrrclearrarrry)r timer1timer2s   r test_init_immutable_default_argsz+TimerTests.test_init_immutable_default_argsst'9'9:  "6"$

e!!#t'9'9:  "T//0!4++r2hR-AB

rc|jj|dd|jf|jj	yr*)rrcopyrr)r rvrs   rrzTimerTests._callback_spys7!!47FKKM":;!rN)r,r-r.rJrrr/rrrrs0
""rrc0eZdZeej
Zy)	LockTestsN)r,r-r.staticmethodrrlocktyper/rrrrsINN+Hrrc0eZdZeej
Zy)PyRLockTestsN)r,r-r.rr_PyRLockrr/rrrrsI../HrrzRLock not implemented in Cc0eZdZeej
Zy)CRLockTestsN)r,r-r.rr_CRLockrr/rrrrsI--.Hrrc0eZdZeej
Zy)
EventTestsN)r,r-r.rrr	eventtyper/rrrrsY__-Irrc0eZdZeej
Zy)ConditionAsRLockTestsN)r,r-r.rrr)rr/rrrrsI//0Hrrc0eZdZeej
Zy)ConditionTestsN)r,r-r.rrr)condtyper/rrrrsI//0Hrrc0eZdZeej
Zy)SemaphoreTestsN)r,r-r.rr	Semaphoresemtyper/rrrrs9../Grrc0eZdZeej
Zy)BoundedSemaphoreTestsN)r,r-r.rrrrr/rrrrs9556Grrc0eZdZeej
Zy)BarrierTestsN)r,r-r.rrBarrierbarriertyper/rrrrsy001KrrceZdZdZy)MiscTestCasecdt|dh}ddh}tj|td||y)Nrr,r+)rr)extranot_exported)rr
check__all__r)r rrs   rtest__all__zMiscTestCase.test__all__s6"4('7T9.F#(|	ErN)r,r-r.rr/rrrrsErrcZeZdZdZdZdZdZdZdZdZ	e
jdZy	)
InterruptMainTestscd}tj||}|jtj|||jt5t	j
dddy#1swYyxYw)Ncddzy)Nr$rr/)signumrs  rhandlerzLInterruptMainTests.check_interrupt_main_with_signal_handler.<locals>.handlers	
aCr)signalrrr)rinterrupt_main)r rrold_handlers    r(check_interrupt_main_with_signal_handlerz;InterruptMainTests.check_interrupt_main_with_signal_handlers]	mmFG4

v{;


0
1	%""$	%	%	%sA..A7cvtj|}	tj|tjtj|tj|tj
tj|tj||y#tj||wxYwr*)r	getsignalSIG_IGNrrSIG_DFL)r rrs   rcheck_interrupt_main_noerrorz/InterruptMainTests.check_interrupt_main_noerrorsv""6*		+MM&&..1""6*MM&&..1""6*
MM&'*FMM&'*sA2B  B8cd}tj|}|jt5|j	|jddd|jy#1swYxYw)Nc,tjyr*)rrr/rrcall_interruptzHInterruptMainTests.test_interrupt_main_subthread.<locals>.call_interrupts""$rr`)rr5rKeyboardInterruptrxry)r rrs   rtest_interrupt_main_subthreadz0InterruptMainTests.test_interrupt_main_subthreadsX	%N3


0
1	
GGI
FFH		
		s!A))A2c|jt5tjdddy#1swYyxYwr*)rrrrrs rtest_interrupt_main_mainthreadz1InterruptMainTests.test_interrupt_main_mainthreads4

0
1	%""$	%	%	%s4=c|jtj|jtjyr*)rrSIGINTSIGTERMrs r'test_interrupt_main_with_signal_handlerz:InterruptMainTests.test_interrupt_main_with_signal_handlers&55fmmD55fnnErc|jtj|jtjyr*)rrrrrs rtest_interrupt_main_noerrorz.InterruptMainTests.test_interrupt_main_noerrors&))&--8))&..9rc|jttjd|jttjtj
|jttjdy)Nri@B)rrzrrrNSIGrs r"test_interrupt_main_invalid_signalz5InterruptMainTests.test_interrupt_main_invalid_signalsN*g&<&<bA*g&<&<fkkJ*g&<&<gFrcdg}dg}dg}d}tj||||f}|j|ds	|dsd|d<|j|j	|dy)NTFcFd}d|d<|dr|r|dz}ny	|drd|d<y)NiTrr$r/)rncontinterrupted
iterationss    rworkerzAInterruptMainTests.test_can_interrupt_tight_loops.<locals>.worker)s=$JGAJq'!OJq'"KNrrur)rr5rxryr)r rrnrrrs      rtest_can_interrupt_tight_loopsz1InterruptMainTests.test_can_interrupt_tight_loops#suv'g		"
F$0LM		!*!*Q	A'rN)
r,r-r.rrrrrrrrreap_threadsrr/rrrrsA%+	%F:G
""(#(rrceZdZdZdZdZy)AtexitTestsctdd\}}}|j||j|jdy)Nrzif True:
            import threading

            def run_last():
                print('parrot')

            threading._register_atexit(run_last)
        sparrot)r	rrarrs    rtest_atexit_outputzAtexitTests.test_atexit_output?s?'/
C	
i0rcFtdd\}}}|j|y)NraNif True:
            import threading
            from unittest.mock import Mock

            mock = Mock()
            threading._register_atexit(mock)
            mock.assert_not_called()
            # force early shutdown to ensure it was called once
            threading._shutdown()
            mock.assert_called_once()
        rrs    rtest_atexit_called_oncez#AtexitTests.test_atexit_called_onceLs)'
/

C	
rctdd\}}}|j||jd|jy)Nrzif True:
            import threading

            def func():
                pass

            def run_last():
                threading._register_atexit(func)

            threading._register_atexit(run_last)
        z2RuntimeError: can't register atexit after shutdown)r	rrrZrs    rtest_atexit_after_shutdownz&AtexitTests.test_atexit_after_shutdown[sA(
/

C	


J

	rN)r,r-r.rrrr/rrrr=s1
rr__main__)>__doc__test.supportrMrrrrrtest.support.import_helperrtest.support.script_helperr	r
r=rrrr@rrrIrrrrCrGrrr
requires_working_threadingrrrcrr5r1TestCaserFrQrrrrJrOrr
RLockTestsrrrrrrrrrrrrrr,rSr/rr<module>rs+>994N

	
+++48+4f<!!<B%8$$%nO,nO`p<pfZ/lZ/zNlNb'I$$'
pJlpJf""8,
$$,0:((0""d*,HI/*''/J/.&&.1J1111Z..10Z..07J<<72:**2E8$$EM(**M(`/(##/dzHMMOr