python (3.11.7)
e d Z ddlZddlmZmZ ddlmZmZmZ ddlm Z ddl
m Z m Z ddl
Z
ddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlmZ ddlmZ dd lmZ ej d
d Z e ed
Z d Z d Z! G d de" Z# G d dej$ Z% G d dej& Z' G d de' Z( G d de' Z) G d de' Z* G d d e' Z+ G d d ej$ Z, G d d!e' Z- G d" d#e' Z. G d$ d%ej/ Z/ G d& d'ej0 Z1 ej2 ej3 du d( G d) d*ej0 Z4 G d+ d,ej5 Z5 G d- d.ej0 Z6 G d/ d0ej7 Z7 G d1 d2ej8 Z8 G d3 d4ej9 Z9 G d5 d6ej: Z: G d7 d8ej& Z; G d9 d:ej& Z< G d; d<ej& Z=e>d=k r ej? dS dS )>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-ux11gettotalrefcountc
t j s t j d | S t j t v r t j d | S t j r t j d | S | S )Nzrequires working os.fork()z*due to known OS bug related to thread+forkz?libasan has a pthread_create() dead lock related to thread+fork)r
has_fork_supportunittestskipsysplatformplatforms_to_skipHAVE_ASAN_FORK_BUG)tests G/BuggyBox/python/3.11.7/bootstrap/lib/python3.11/test/test_threading.pyskip_unless_reliable_forkr ( s # A:x} 9::4@@@
|(((Jx} IJJ4PPP! f_x} ^__`deee K c | t t dt j t j t _ d S )N
excepthook)
addCleanupsetattr threadingr __excepthook__)testcases r restore_default_excepthookr$ 2 s0 L):NOOO$3Ir c & e Zd Zd Zd Zd Zd ZdS )Counterc d| _ d S )Nr valueselfs r __init__zCounter.__init__9 s
r c & | xj dz
c_ d S N r( r* s r incz Counter.inc;
a
r c & | xj dz c_ d S r. r( r* s r decz Counter.dec= r1 r c | j S Nr( r* s r getz Counter.get? s
zr N)__name__
__module__ __qualname__r, r0 r3 r6 r r r& r& 8 sP r r& c e Zd Zd Zd ZdS )
TestThreadc t j | | || _ || _ || _ || _ d S )Nname)r! Threadr, r# semamutexnrunning)r+ r? r# rA rB rC s r r, zTestThread.__init__C s> !!$T!222
r c t j dz }t r t d| j |dz fz | j 5 | j 5 | j t r't | j d | j
| j d d d d n # 1 swxY w Y t j | t rt d| j d | j 5 | j
| j | j d t r0t d | j | j fz d d d n # 1 swxY w Y d d d d S # 1 swxY w Y d S )
Ng @z task %s will run for %.1f usecg .Aztasks are running taskdoner z$%s is finished. %d tasks are running)randomr printr? rA rB rC r0 r6 r# assertLessEqualtimesleepr3 assertGreaterEqual)r+ delays r runzTestThread.runJ s
') , 29eck*+
,
,
, Y < <
F
F
!!### D$-++--/BCCC
--dm.?.?.A.A1EEE
F
F
F
F
F
F
F
F
F
F
F
F
F
F
F
Ju
1fdi000
<
<
!!###
00 1B1B1D1DaHHH <@ 9dm&7&7&9&9:; < < <
<
<
<
<
<
<
<
<
<
<
<
<
<
<
< < < < < < < < < < < < < < < < < < <s\ F9 A:C F9C F9C ;F9BF! F9!F% %F9(F% )F99F= F=N)r7 r8 r9 r, rO r: r r r< r< B s2 ! ! !< < < < <r r< c e Zd Zd Zd ZdS ) BaseTestCasec 6 t j | _ d S r5 )r threading_setup_threadsr* s r setUpzBaseTestCase.setUpd s (8::
r c h t j | j t j d S r5 )r threading_cleanuprT r r
reap_childrenr* s r tearDownzBaseTestCase.tearDowng s, *DM:: ""$$$$$r N)r7 r8 r9 rU rY r: r r rQ rQ c s2 ; ; ;% % % % %r rQ c 4 e Zd Zed Zd Zed Zd Zd Zd Z d Z
d Z d Z d
Z
d Zd Zd
Zd Zd Zd Zd Zd Zed Zed Zed Zd Ze ej e e d d d Z e ej e e d d d Z d Z!d Z"d Z#d Z$d Z%ed Z&d! Z'd" Z(ed# Z)d$ Z*d% Z+d& Z, ej e-d' d( Z.d) Z/d*S )+ ThreadTestsc d }t j d }| |j d t j d }| |j d t j |d }| |j d t j t dd
5 t j d }| |j d d d d n # 1 swxY w Y t j t dd
5 t j }| |j d d d d n # 1 swxY w Y t j t dd
5 t j | }| |j d d d d d S # 1 swxY w Y d S )Nc d S r5 r: r: r r funcz#ThreadTests.test_name.<locals>.funcp r myname1r> { 123myname2)targetr? _counter ) return_value zThread-2rE zThread-3 rd zThread-5 (func))r! r@ assertEqualr? r patchobject)r+ r^ threads r test_namezThreadTests.test_namen sx !y111 i000 !s+++ e,,, !I>>> i000
Z
y*1
E
E 6 6 %2...F V[* 5 5 5 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6 Z
y*1
E
E 6 6 %''F V[* 5 5 5 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6 Z
y*1
E
E = = %T222F V[*; < < < = = = = = = = = = = = = = = = = = =s6 <1C99C= C=+/E&&E*-E*1GG Gc ` dg}d}dg}d}dgf}dg}| fdf| fdf| fdf| fdf| fd f| fd
ff}|D ]r\ }} | | 5 t j | | }
|
|
d d d n # 1 swxY w Y sd S ) Nr/ r/ strrr c 0 | d S r. rk argr+ s r <lambda>z0ThreadTests.test_args_argument.<locals>.<lambda> s 4#3#3C#;#; r c 0 | d S r. ru rv s r rx z0ThreadTests.test_args_argument.<locals>.<lambda> s D$4$4S!$<$< r c 0 | d S Nrr ru rv s r rx z0ThreadTests.test_args_argument.<locals>.<lambda> s 4#3#3C#?#? r c 0 | d S r{ ru rv s r rx z0ThreadTests.test_args_argument.<locals>.<lambda> s D$4$4S%$@$@ r c 2 | dg S r. ru rv s r rx z0ThreadTests.test_args_argument.<locals>.<lambda> s (8(8qc(B(B r c 0 | d S )Nrq ru rv s r rx z0ThreadTests.test_args_argument.<locals>.<lambda> s (8(8d(C(C r rd args)subTestr! r@ startjoin) r+ num_list num_tuplestr_list str_tuple
list_in_tuple
tuple_in_list
test_casesr rd ts ` r test_args_argumentz ThreadTests.test_args_argument s\ 3 7
;;;; <
<<<< =
???? @
@@@@ A
B B B B C
C C C C D
' LD& V$ 77
$F>>>
s ?B""B& )B& c t j }t j | t | d S r5 )r! Lockr r
check_disallow_instantiationtyper+ locks r test_disallow_instantiationz'ThreadTests.test_disallow_instantiation s5 ~ 11$T
CCCCCr c l d}t j d }t j }t }g }t | D ]~}t d|z | ||| }| | | |j | t | d | t t d rjt d |D t j hz }| d | | t# | |dz t$ rt' d |D ]}| | | | |j d
| |j | t | d t$ rt' d | | d
d S )
N
rE r( z <thread %d>z ^<TestThread\(.*, initial\)>$
get_native_idc 3 $ K | ] }|j V d S r5 ) native_id).0r s r <genexpr>z/ThreadTests.test_various_ops.<locals>.<genexpr> s$ : :QQ[ : : : : : :r r/ z!waiting for all tasks to completer z#^<TestThread\(.*, stopped -?\d+\)>$zall tasks done)r! BoundedSemaphoreRLockr& ranger< append assertIsNoneident assertRegexreprr hasattrsetr assertNotInrk lenr rI r assertFalseis_aliveassertNotEqualassertIsNotNoner6 ) r+ NUMTASKSrA rB
numrunningthreadsir
native_idss r test_various_opsz ThreadTests.test_various_ops s )222!! YY
x A=?D$zJJA NN1 ag & & & T!WW&F G G G
GGIIII 9o . . < : :' : : :::i>U>W>W=XXJ T: . . . S__hl ; ; ; 7 5 6 6 6 N NA
FFHHH QZZ\\ * * * + + + ) ) ) T!WW&L M M M M $ " # # # ))1-----r c | t j j fd}t j g t j 5 t j |d } |
d | d d d n # 1 swxY w Y t j d = d S )Nc t j j d S r5 )r r! current_threadr r )rG r s r fz9ThreadTests.test_ident_of_no_threading_threads.<locals>.f s2 LL1339 : : : HHJJJJJr r: r ) r r! r r Eventr wait_threads_exit_threadstart_new_threadwaitrk _active)r+ r tidrG r s @@r "test_ident_of_no_threading_threadsz.ThreadTests.test_ident_of_no_threading_threads s
Y 5 7 7 =>>>
/
1
1 , ,*1b11C IIKKK U1Xs + + + , , , , , , , , , , , , , , ,
eAh ' ' 's AB..B25B2c t rt d t j d n'# t j $ r t
j d w xY w| t j d d S )Nz!with 256 KiB thread stack size...i 4platform does not support changing thread stack sizer r rI r!
stack_sizer errorr SkipTestr r* s r test_various_ops_small_stackz(ThreadTests.test_various_ops_small_stack s 7 5 6 6 6 H ( ( ( (} H H H#FH H
H H
Q - $Ac t rt d t j d n'# t j $ r t
j d w xY w| t j d d S )Nz with 1 MiB thread stack size...i r r r r* s r test_various_ops_large_stackz(ThreadTests.test_various_ops_large_stack s 5 3 4 4 4 H * * * *} H H H#FH H
H H
Q r c d }t j }| t j 5 t j ||f }| d d d n # 1 swxY w Y | |t j | t j | t j
| t j | |
t t j | d t j |= d S )Nc T t j | d S r5 )r! r release)rB s r r z*ThreadTests.test_foreign_thread.<locals>.f s$
$ & & & MMOOOOOr _DummyThread)r! r acquirer r r r assertInr assertIsInstancer
assertTruer r r )r+ r rB r s r test_foreign_threadz ThreadTests.test_foreign_thread sI
/
1
1 *1uh77C MMOOO
c9,--- i / 4i6LMMM )#.7799::: i / 455~FFF c " " "s +A55A9<A9c
t d }|j j }|j |j f|_ G d dt | }t j }| |t | |d ||| } # $ r Y nw xY w |
|d n# t $ r Y nw xY wt j t j
G
fdd t j } | }d|_ | t& rt) d
t& rt) d |d | }|
|d t& rt) d
}| | t& rt) d | |j t& rt) d ||j | }|
|d t& rt) d
t4 j | |j t& rt) d |j r| d S d S )Nctypesc e Zd ZdS )<ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.AsyncExcN)r7 r8 r9 r: r r AsyncExcr s Dr r r TzAsyncExc not raisedr/ c e Zd Z fdZdS ):ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Workerc t j | _ d| _ t j d )# $ r d| _ Y d S w xY w)NFTg?)r! get_identidfinishedr rK rL )r+ r worker_saw_exceptionworker_starteds r rO z>ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Worker.run8 s #-// %
/(&**,,,
3( / / /$(DM(,,....../s *A A0/A0Nr7 r8 r9 rO )r r r s r Workerr 7 s8
/
/
/
/
/
/
/
/
/r r z 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 Exceptionr! r r int
assertGreaterfailrk UnboundLocalErrorr r@ daemonr r rI r r r r r r
SHORT_TIMEOUTr ) r+ r
set_async_exc exceptionr resultr r retr r r s @@@r test_PyThreadState_SetAsyncExcz*ThreadTests.test_PyThreadState_SetAsyncExc s. x(( (B
"(.&2B!C
y $$X.. !## c3''' 3""" -"]3 22F
D VQ ' ' ' ' D # ** ( 0 0 / / / / / / / / /Y% / / /
FHH / - . . . 6 4 5 5 5 r9-- ### B @ A A A !!## 8 6 7 7 7 $$$ H F G G G qtY// ### K I J J J !!'*?!@@@
### 2 0 1 1 1 :
FFHHHHH s$ B. .B65B6:C
C C c 2 d }t j }|t _ t j d }| t j |j | |t j v d |t _ d S # |t _ w xY w)Nc ( t j r5 )r! ThreadErrorr s r fail_new_threadz7ThreadTests.test_limbo_cleanup.<locals>.fail_new_threadh s ')) )r c d S r5 r: r: r r rx z0ThreadTests.test_limbo_cleanup.<locals>.<lambda>m r rj z:Failed to cleanup _limbo map on failure of Thread.start().)r! _start_new_threadr@ assertRaisesr r r _limbo)r+ r r r s r test_limbo_cleanupz ThreadTests.test_limbo_cleanupf s * * * % 7&5 # < 555A i 3QW = = = Y%%L
N
N
N +<I ' ' '*;I ' ; ; ; ;s A B Bc x t d t dd \ }}}| |d d S )Nr -caN if 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)
* )r r
rk r+ rcouterrs r test_finalize_running_threadz(ThreadTests.test_finalize_running_threadu sL h ,T 4 C2
R r c & t dd d S )Nr aP if 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 r* s r test_finalize_with_tracez$ThreadTests.test_finalize_with_trace s' r c t dd \ }}}| | d | |d d S )Nr a if 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
s5 Woke up, sleep function is: <built-in function sleep>r )r rk stripr s r test_join_nondaemon_on_shutdownz+ThreadTests.test_join_nondaemon_on_shutdown sf ( / C
D F F F c"""""r c t j }t j } t dd D ]}}t j |dz t j d }| | | }| ||d||fz ~ t j | d S # t j | w xY w)Nr/ d g-C 6*?c d S r5 r: r: r r rx z7ThreadTests.test_enumerate_after_join.<locals>.<lambda> s D r rj z� triggered after %d trials: %s)
r! enumerater getswitchintervalr setswitchintervalr@ r r r )r+ enum old_intervalr r ls r test_enumerate_after_joinz%ThreadTests.test_enumerate_after_join s ",..
01c]]
G
G%a&j111 $LL999 DFF A<1vEG G G G
G
!, / / / / /C !, / / / /s BC C c " G d dt }t | |d }t j | }|j ~| | dt j | z |d }t j | }|j ~| | dt j | z d S )Nc e Zd Zd Zd ZdS )DThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunctionc || _ t j | j | fd| i | _ | j d S )N yet_another)rd r kwargs) should_raiser! r@ _runrn r )r+ r s r r, zMThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction.__init__ sU %1! ' .di59G7DT6J L L L !!#####r c " | j rt d S r5 )r
SystemExit)r+ other_refr s r r zIThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction._run s $ %$$% %r N)r7 r8 r9 r, r r: r r RunSelfFunctionr s2
$
$
$
%
%
%
%
%r r F)r z%d references still around)msgT) rm r$ weakrefrefrn r r r getrefcount)r+ r
cyclic_objectweak_cyclic_objectraising_cyclic_objectweak_raising_cyclic_objects r test_no_refcycle_through_targetz+ThreadTests.test_no_refcycle_through_target sX % % % % %f % % % #4((('U;;;
$[ 7 7 !!### ,,.. : !o.@.@.B.B C C D F F F !0T B B B %,[1F%G%G" $))+++ ! 4466 : !o.H.H.J.J K K L N N N N Nr c t j }| t d 5 | d d d n # 1 swxY w Y | t d 5 | d d d d n # 1 swxY w Y | t d 5 | d d d n # 1 swxY w Y | t d 5 | d d d d n # 1 swxY w Y t j }| t d 5 | d d d n # 1 swxY w Y t j
}| | t d 5 | d d d n # 1 swxY w Y | t d 5 t j
d d d n # 1 swxY w Y | t d
5 t j d d d d S # 1 swxY w Y d S ) Nzget the daemon attributezset the daemon attributeTzget the name attributezset the name attributer? z use is_set()zuse notify_all()zuse active_count()zuse current_thread())r! r@ assertWarnsRegexDeprecationWarningisDaemon setDaemongetNamesetNamer isSet Conditionr notifyAll activeCount
currentThread)r+ r econds r test_old_threading_apiz"ThreadTests.test_old_threading_api sI
"
"#5#>@ @
JJLLL
"
"#5#>@ @
KK
"
"#5#<> >
IIKKK
"
"#5#<> >
IIf
O
"
"#5~
F
F
GGIII "$$
"
"#57I
J
J NN
"
"#57K
L
L $ $ ! # # # $ $ $ $ $ $ $ $ $ $ $ $ $ $ $
"
"#57M
N
N & & # % % % & & & & & & & & & & & & & & & & & &s AAA6BB B >C C#&C#D''D+.D+ FFFG//G36G3H55H9<H9I<<J J c t j }| dt | d|_ | dt | d S Nr T)r! r@ r r r r r+ r s r test_repr_daemonz ThreadTests.test_repr_daemon sS 477+++
hQ(((((r c t j }| |j t j d }| |j t j d }| |j d S )NFr T)r! r@ r r r r; s r test_daemon_paramz ThreadTests.test_daemon_param s| """ E * * * """ D ) ) ) !!!!!r c t j d }t d| \ }}}| |d | | d d S )Nat
import atexit
import os
import sys
from test.support import wait_process
# Import the threading module to register its "at fork" callback
import threading
def exit_handler():
pid = os.fork()
if not pid:
print("child process ok", file=sys.stderr, flush=True)
# child process
else:
wait_process(pid, exitcode=0)
# exit_handler() will be called after threading._shutdown()
atexit.register(exit_handler)
r r s child process ok)textwrapdedentr rk rstripr+ code_r r s r test_fork_at_exitz ThreadTests.test_fork_at_exit" sl
( 'tT22 3 c""" ':;;;;;r c d}t d| \ }}}| |d | |d d S )Na if 1:
import _thread, threading, os, time
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()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
r r r rk rD s r test_dummy_thread_after_forkz(ThreadTests.test_dummy_thread_after_fork> sR & 'tT22 3 c""" c"""""r c t j }| t j | t j d t
d D ]}t j d }| t j }|dk r+t j |
rdnd p| t j |d d S ) Ngư> c d S r5 r: r: r r rx z6ThreadTests.test_is_alive_after_fork.<locals>.<lambda>d r r rj r r exitcode)r r r r r r
r r! r@ r osfork_exitr r wait_process)r+ r r r pids r test_is_alive_after_forkz$ThreadTests.test_is_alive_after_forkY s ,.. -|<<<
&&t,,,r 7 7A 555A
GGIII'))Caxxqzz||34444$S266666 7 7r c t j } |j d |j t j j |j t j fd}t j | }| | d S )N
MainThreadc t j j t j j d S r5 )r r! main_threadr r r* s r r z'ThreadTests.test_main_thread.<locals>.ft sG 5 7 7 = ) 8 : : @
B
B
B
B
Br rj )
r! rZ rk r? r r r r@ r r )r+ mainr ths ` r test_main_threadz ThreadTests.test_main_threadn s $&& L111 Y%=%?%?%EFFF Y%8%:%:;;; B B B B B
Q
'
'
'
r waitpidztest needs os.waitpid()c d}t d| \ }}}| dd }| |d | |d d S )Na if 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
rh r zMainThread
True
True
r decodereplacerk r+ rE rF r r datas r test_main_thread_after_forkz'ThreadTests.test_main_thread_after_fork{ sp 'tT22 3zz||##D"-- c""" 9:::::r c d}t d| \ }}}| dd }| |d | |d d S )Na if 1:
import os, threading, sys
from test import support
def func():
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:
support.wait_process(pid, exitcode=0)
th = threading.Thread(target=func)
th.start()
th.join()
r r` rh r zThread-1 (func)
True
True
ra rd s r /test_main_thread_after_fork_from_nonmain_threadz;ThreadTests.test_main_thread_after_fork_from_nonmain_thread sp * 'tT22 3zz||##D"-- c""" >?????r c d}t d| \ }}}| }| |d | | dgdz d S )Na if 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()
r r zGC: True True Truerf )r rb rk
splitlinesrd s r test_main_thread_during_shutdownz,ThreadTests.test_main_thread_during_shutdown s| ( 'tT22 3zz|| c""" **./!3 5 5 5 5 5r c ^ d}t d| \ }}}| |d d S )Na if 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()
r r rI r+ rE r r r s r test_finalization_shutdownz&ThreadTests.test_finalization_shutdown s< 4 (d33 C c"""""r c t j t j fd}t j | }| |j d | | | |j }|
| d d | | t j
d | | | |
| | |j | d S )Nc t j d d S )N{Gz?)r r rK rL finishstarteds r r z'ThreadTests.test_tstate_lock.<locals>.f s9 OO NN Jt r rj r r F)r
allocate_lockr r! r@ assertIs _tstate_lockr r r r r r
r r r )r+ r r tstate_lockrs rt s @@r test_tstate_lockz ThreadTests.test_tstate_lock s '))&((
A & & &
and+++
%%% n ,,Q,77???
++G4I+JJERRR
%%% &&& !.))) r c ~ t j t j fd}t j | }| | dt | d}t d D ])}|t | v r nt j d *| |t | | d S )Nc X d S r5 )r r rr s r r z(ThreadTests.test_repr_stopped.<locals>.f s) OO NN r rj rt stoppedi rq )
r ru r r! r@ r r r r r rK rL r )r+ r r LOOKING_FORr rs rt s @@r test_repr_stoppedz ThreadTests.test_repr_stopped s* '))&((
A & & &
ia))) s Ad1gg%% Jt
k477+++ r c t dd D ]}t j | fdt | D }|D ]}| |D ]}| fdt | D }|D ]}| |D ]}| | t j d S )Nr/ r c D g | ] }t j j S rj )r! r@ r r rF bss r
<listcomp>z;ThreadTests.test_BoundedSemaphore_limit.<locals>.<listcomp>/ 9 . . . !'rz::: . . .r c D g | ] }t j j S r )r! r@ r r s r r z;ThreadTests.test_BoundedSemaphore_limit.<locals>.<listcomp>5 r r )r r! r r r r
ValueErrorr )r+ limitr r r s @r test_BoundedSemaphore_limitz'ThreadTests.test_BoundedSemaphore_limit+ s# 1b\\ 6 6E+E22B. . . . %e . . .G
. . . . %e . . .G
j"* 5 5 5 5 6 6r c t fdd fdd _ t j }t j t j dd l}| t d D ] }
t j | d S # t j | w xY w)Nc S r5 r: frameeventrw
noop_traces r r z9ThreadTests.test_frame_tstate_tracing.<locals>.noop_traceE r c 3 K dV )Nr/ generatorr: r: r r r z8ThreadTests.test_frame_tstate_tracing.<locals>.generatorI s
"!!!!
"r c X j _ t j S r5 )gennext)callbackr s r r z7ThreadTests.test_frame_tstate_tracing.<locals>.callbackM s( |# (y{{ %% %r r rE )r r gettracesettracer! _testcapicall_in_temporary_c_threadr )r+ old_tracer r r r r s @@@r test_frame_tstate_tracingz%ThreadTests.test_frame_tstate_tracing= s " " " & & & & & & LNN Z $ z * * *
0 0 : : : a
L # # # # #CL # # # #s
A B! !B7c fdt j } t j t j }| | t j | d S # t j | w xY w)Nc S r5 r: r s r r z-ThreadTests.test_gettrace.<locals>.noop_tracef r r )r! r r rk )r+ r
trace_funcr s @r
test_gettracezThreadTests.test_gettracee s &(( * z * * *"+--J Z
3 3 3 y ) ) ) ) )I y ) ) ) )s =A. .Bc d }t j } t j | | |t j t j | d S # t j | w xY w)Nc d S r5 r: r s r fnz'ThreadTests.test_getprofile.<locals>.fnr r_ r )r!
getprofile
setprofilerk )r+ r old_profiles r test_getprofilezThreadTests.test_getprofileq s| *,, . $ $ $ R!5!7!7 8 8 8 - - - - -I - - - -s ;A) )A?c dD ]}| | 5 t j }t j |j | }| |j }|s!| |t j n | |t j |
| | |t j d d d n # 1 swxY w Y d S )N)FTr> )rd r ) r r! r r@ r r rw r _shutdown_locksr r r )r+ r r rn rx s r test_shutdown_locksz ThreadTests.test_shutdown_locksz sH # I IF V ,,
I
I!))")FKKK $ 1 MMM+y/HIIII$$[)2KLLL
i.GHHH'
I
I
I
I
I
I
I
I
I
I
I
I
I
I
I I Is CC==D D c ~ t dd \ }}}| | d d S )Nr a( 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()
s thread_dict.atexit = 'value')r rk rC r s r test_locals_at_exitz ThreadTests.test_locals_at_exit sG ( /
C
'FGGGGGr c G d dt } | }t j | }| | | |j d S )Nc e Zd Zd Zd Zd ZdS )6ThreadTests.test_boolean_target.<locals>.BooleanTargetc d| _ d S NFranr* s r r, z?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__init__ s
r c dS r r: r* s r __bool__z?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__bool__ s ur c d| _ d S )NTr r* s r __call__z?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__call__ s
r N)r7 r8 r9 r, r r r: r r
BooleanTargetr sA
!
!
!
r r rj )rm r! r@ r r r r )r+ r rd rn s r test_boolean_targetz ThreadTests.test_boolean_target s
F !000
#####r c d }t j 5 t j | d d d d S # 1 swxY w Y d S )Nc d S r5 r: r: r r noopz0ThreadTests.test_leak_without_join.<locals>.noop r_ r rj )r r r! r@ r )r+ r s r test_leak_without_joinz"ThreadTests.test_leak_without_join s
/
1
1 2 2 D ) ) ) / / 1 1 1 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2s (A AAzneed debug build (Py_DEBUG)c d t dddd \ }}}d}| || d S )Nz -Wdefaultr pass1)PYTHONTHREADDEBUGs DeprecationWarning: The threading debug (PYTHONTHREADDEBUG environment variable) is deprecated and will be removed in Python 3.12)r r )r+ r r r r! s r test_debug_deprecationz"ThreadTests.test_debug_deprecation sK ( T6:=? ? ? CC
c3 r c t j d }t d| \ }}}| |d | |d d S )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
r r )rA rB r rk rm s r test_import_from_another_threadz+ThreadTests.test_import_from_another_thread sc
2 (d33 C c""" c"""""r N)0r7 r8 r9 r ro r r r r r r r r r r r r
r r) r8 r< r? r rG rJ rV r] r
skipUnlessr rQ rf rh rk rn ry r~ r r r r r r r r Py_DEBUGr r r: r r r[ r[ l sD = = \=4 6 D D \D". ". ".H
(
(
( # # #.U U Up
<
<
< ! ! !B 2# # #(0 0 0" N N NB & & &>) ) ) " " " < < <6 # # #4 7 7 7( XY//1JKK; ; LK ;& XY//1JKK@ @ LK @65 5 5:!# !# !#F! ! !F 66 6 6$ %$ %$ \%$N
*
*
*. . . I I \I.H H H"$ $ $&2 2 2 X#@AA BA # # # # #r r[ c e Zd Zd Zd Zed Zed Z ej e
j e v d d Z
ed Zed Zd S )
ThreadJoinOnShutdownc d|z }t d| \ }}}| dd }| |d d S )Na if 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()
r r` rh zend of main
end of thread
ra )r+ scriptr r r re s r
_run_and_joinz"ThreadJoinOnShutdown._run_and_join sd
(f55 Czz||##D"-- =>>>>>r c 4 d}| | d S )Nzif 1:
import os
t = threading.Thread(target=joiningfunc,
args=(threading.current_thread(),))
t.start()
time.sleep(0.1)
print('end of main')
r r+ r s r test_1_join_on_shutdownz,ThreadJoinOnShutdown.test_1_join_on_shutdown s%
6"""""r c 4 d}| | d S )Na if 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')
r r s r test_2_join_in_forked_processz2ThreadJoinOnShutdown.test_2_join_in_forked_process s%
6"""""r c 4 d}| | d S )Na if 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()
r r s r !test_3_join_in_forked_from_threadz6ThreadJoinOnShutdown.test_3_join_in_forked_from_thread s%
*
6"""""r zdue to known OS bugc \ d}t d| \ }}}| | d S )Na if 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()
r r r r+ r r r r s r test_4_daemon_threadsz*ThreadJoinOnShutdown.test_4_daemon_threads9 s;
!D (f55 C r c d }g }t d D ]@}t j | }| | | A|D ]}| d S )Nc t j } | dk rt j | d d S t j d d S )Nr 2 rO )rQ rR r
rT rS )rU s r do_fork_and_waitzIThreadJoinOnShutdown.test_reinit_tls_after_fork.<locals>.do_fork_and_waith sC '))CQww$S2666666 r rj )r r! r@ r r r )r+ r r r r s r test_reinit_tls_after_forkz/ThreadJoinOnShutdown.test_reinit_tls_after_forkc s
r A (8999A NN1
GGIIII A
FFHHHH r c g }t d D ]A}t j d }| | | Bt j }|dk rNt t j dk rt j
d n+t j
d nt j |d |D ]}|
d S ) Nr c * t j d S )Ng333333?)rK rL r: r r rx zKThreadJoinOnShutdown.test_clear_threads_states_after_fork.<locals>.<lambda> s C r rj r r/ 3 4 rO )r r! r@ r r rQ rR r r _current_framesrS r
rT r )r+ r r r rU s r $test_clear_threads_states_after_forkz9ThreadJoinOnShutdown.test_clear_threads_states_after_forkz s
r A (@(@AAAA NN1
GGIIIIgii !883&(())Q.. r 2 2 2 2 A
FFHHHH r N)r7 r8 r9 r r r r r r skipIfr r r r r r r: r r r r s ? ? ?"
#
#
# # # #& # # #6 X_S\%668MNN' ' ON' R , r r c 6 e Zd Zd Zd Zd Zed ZdS )SubinterpThreadingTestsc
t j \ }}| t j | | t j | t t d rt j |d ||fS )N set_blockingF)rQ piper closer r )r+ rws r r z SubinterpThreadingTests.pipe sj wyy1 !$$$ !$$$ 2~ & & & OAu % % %1v
r c | \ }}t j d|fz }t j | }| |d | t j |d d d S )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()
r r/ x r rA rB r r
run_in_subinterprk rQ readr+ r r rE r s r test_threads_joinz)SubinterpThreadingTests.test_threads_join s yy{{1 $ d% & l++D11 a A -----r c | \ }}t j d|fz }t j | }| |d | t j |d d d S )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()
r r/ r r r s r test_threads_join_2z+SubinterpThreadingTests.test_threads_join_2 s
yy{{1 2 d3 4 l++D11 a A -----r c dt j j d}d|d}t j 5 t d| \ }}}d d d n # 1 swxY w Y | d| d S )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(z)
r z:Fatal Python error: Py_EndInterpreter: not the last thread)r r
r SuppressCrashReportr
r rb )r+ subinterp_coder r r r s r test_daemon_threads_fatal_errorz7SubinterpThreadingTests.test_daemon_threads_fatal_error s
!L 6
">> $
\
-
-
/
/ ? ?0v>>LBS ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
,-0ZZ\\ ; ; ; ; ;s AA "A N)r7 r8 r9 r r r r r r: r r r r s] . . .8#. #. #.J ; ; \; ; ;r r c p e Zd Zd Zd Zd Zd Zd Z e d Z d Z
d Z d Z d
Z
d Zd S )
ThreadingExceptionTestsc t j }| | t |j | d S r5 )r! r@ r r RuntimeErrorr r+ rn s r test_start_thread_againz/ThreadingExceptionTests.test_start_thread_again sF !## , 555
r c l t j }| t |j d S r5 )r! r r r r )r+ r s r test_joining_current_threadz3ThreadingExceptionTests.test_joining_current_thread s/ "133 ,(;<<<<<r c l t j }| t |j d S r5 )r! r@ r r r r s r test_joining_inactive_threadz4ThreadingExceptionTests.test_joining_inactive_thread s. !## , 44444r c t j }| | t t
|dd | d S r: )r! r@ r r r r r r s r test_daemonize_active_threadz4ThreadingExceptionTests.test_daemonize_active_thread sJ !## ,4HHH
r c l t j }| t |j d S r5 )r! r r r r r s r test_releasing_unacquired_lockz6ThreadingExceptionTests.test_releasing_unacquired_lock s- ~ , 55555r c d}d}t j t j d|gt j t j }| \ }}| dd }| |j dd| z | || d S ) Na if 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 )stdoutstderrr` rh r zUnexpected error: )
subprocessPopenr
executablePIPE communicaterb rc rk
returncode)r+ r expected_outputpr r
re s r test_recursion_limitz,ThreadingExceptionTests.test_recursion_limit s " 1 cndF ;$.OJO
M
M
M }}&&tR00 q*> *PQQQ /////r c 6 d}t d| \ }}}| |d | }| d| | d| | d| | d| d S )Na if 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()
r r Exception in thread"Traceback (most recent call last):ZeroDivisionErrorUnhandled exceptionr rk rb r r r s r test_print_exceptionz,ThreadingExceptionTests.test_print_exception0 s $ (f55 C c"""jjll
+S111
:C@@@
)3/// .44444r c 6 d}t d| \ }}}| |d | }| d| | d| | d| | d| d S )Na if 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()
r r r r r r r r s r %test_print_exception_stderr_is_none_1z=ThreadingExceptionTests.test_print_exception_stderr_is_none_1K s ( (f55 C c"""jjll
+S111
:C@@@
)3/// .44444r c d}t d| \ }}}| |d | d| d S )Na if 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()
r r r )r rk r rb r s r %test_print_exception_stderr_is_none_2z=ThreadingExceptionTests.test_print_exception_stderr_is_none_2h s[ ( (f55 C c""" .
=====r c ( d G fddt j } | }| | | |j | |j t d |_ d S )Nc r5 r: r: r r
bare_raisezOThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.bare_raise s r c e Zd ZdZ fdZdS )OThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558Nc \ d S # t $ r}|| _ Y d }~d S d }~ww xY wr5 )r exc)r+ r% r! s r rO zSThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558.run sI # JLLLLL # # # "DHHHHHHH#s
+&+)r7 r8 r9 r% rO )r! s r
Issue27558r# s3 C
#
#
#
#
#
#
#r r& )r! r@ r r r r% r r )r+ r& rn r! s @r #test_bare_raise_in_brand_new_threadz;ThreadingExceptionTests.test_bare_raise_in_brand_new_thread s # # # # # # #) # # #
VZ((( fj,777
r c d | t j t j fdt d D }|D ]*}| | +d S )Nc t t j dd 5 } | d t j d d d d S # 1 swxY w Y d S )Nr zutf-8)encoding )openr TESTFNwrite traceback format_stack)fps r modify_filezQThreadingExceptionTests.test_multithread_modify_file_noerror.<locals>.modify_file s i&g>>>
)"
&(((
)
)
)
)
)
)
)
)
)
)
)
)
)
)
)
)
)
)s )AAAc : g | ]}t j S r )r! r@ )r r r2 s r r zPThreadingExceptionTests.test_multithread_modify_file_noerror.<locals>.<listcomp> s7
K 0 0 0
r r )r r unlinkr- r r r )r+ r r r2 s @r $test_multithread_modify_file_noerrorz<ThreadingExceptionTests.test_multithread_modify_file_noerror s ) ) )
()*:;;;
3ZZ
A
GGIII
FFHHHH r N)r7 r8 r9 r r r r r r r r r r r' r5 r: r r r r s > > >5 5 5 6 6 6 0 0 0>5 5 565 5 5:> > >2 * r r c e Zd Zd ZdS )
ThreadRunFailc t d )N
run failedr r* s r rO zThreadRunFail.run s &&&r Nr r: r r r7 r7 s# ' ' ' ' 'r r7 c \ e Zd Z fdZd Zej d Zd Zd Z d Z
d Z xZ S )ExceptHookTestsc f t | t d S r5 )r$ superrU )r+ __class__s r rU zExceptHookTests.setUp s& "4(((
r c t j d 5 }t d }| | d d d n # 1 swxY w Y | }| d|j d| | d| | d| | d| d S ) Nr
excepthook threadr> Exception in thread :
#Traceback (most recent call last):
z raise ValueError("run failed")zValueError: run failed) r
captured_outputr7 r r getvaluer r r? )r+ r
rn s r test_excepthookz ExceptHookTests.test_excepthook s
$X
.
. &"(;<<<F LLNNN KKMMM
""((**
=V[===vFFF
<fEEE
8&AAA
.77777 9AA !A c d t j d 5 } t d # t $ rR}t j g t
j d } t j | d }n# d }w xY wY d }~nd }~ww xY w d d d n # 1 swxY w Y |
}| dt j d| | d| | d| | d| d S )Nr
bugrB rC rD z raise ValueError("bug")zValueError: bug)
r
rE r r r! ExceptHookArgsr exc_infor rF r r r )r+ r
r% r s r test_excepthook_thread_Nonez+ExceptHookTests.test_excepthook_thread_None s
$X
.
. &
'''
/0G#,..0G$0GHH (... DD4DKKKKDDDDD
""((**
GY-@-B-BGGGPPP
<fEEE
16:::
'00000sI B&
B)A=A1.A=1A55A=8B=BBBBc B G d dt j }t j d 5 } | }| | d d d n # 1 swxY w Y | | d d S )Nc e Zd Zd ZdS )4ExceptHookTests.test_system_exit.<locals>.ThreadExitc . t j d d S r. )r exitr* s r rO z8ExceptHookTests.test_system_exit.<locals>.ThreadExit.run s r Nr r: r r
ThreadExitrP s#
r rS r
rh )r! r@ r
rE r r rk rF )r+ rS r
rn s r test_system_exitz ExceptHookTests.test_system_exit s )
$X
.
. & Z\\F LLNNN KKMMM
**B/////s 3A,,A03A0c d fd} t j t d| 5 t }| | d d d n # 1 swxY w Y | j t | t j
d | j j
j |
j | d d S # d w xY w)Nc | d S r5 r: ) hook_argsr s r hookz4ExceptHookTests.test_custom_excepthook.<locals>.hook s DDDr r r9 )r
swap_attrr! r7 r r rk exc_typer rr exc_value
exc_traceback
__traceback__rv rn )r+ rX rn r s @r test_custom_excepthookz&ExceptHookTests.test_custom_excepthook s9 "9lDAA
&
T]J 7 7 7 S 0 0, ? ? ? T / 1M N N N MM$+v . . . DDD4DKKKKs/ C? 7A( C? (A,,C? /A,0B C? ?Dc ( d }d fd}t j t d| 5 t j t d| 5 t j d 5 }t }| | d d d n # 1 swxY w Y d d d n # 1 swxY w Y d d d n # 1 swxY w Y | | d | d d S )Nc t d )Nthreading_hook failedr: r s r threading_hookzCExceptHookTests.test_custom_excepthook_fail.<locals>.threading_hook s 455 5r c & t | d S r5 rs )rZ r[ r\ err_strs r sys_hookz=ExceptHookTests.test_custom_excepthook_fail.<locals>.sys_hook s )nnGGGr r r
z#Exception in threading.excepthook:
ra )
r
rY r! r rE r7 r r rk rF )r+ rb re r
rn rd s @r test_custom_excepthook_failz+ExceptHookTests.test_custom_excepthook_fail s 6 6 6 % % % % %
y,
G
G
sL(
;
;
$X
.
. 28"__F LLNNN KKMMM
**? A A A "9:::::sY C B27B B2B B2"B #B2& C 2B6 6C 9B6 :C C
C
c d d }d } | }t j t d| 5 | }t j t _ | }d d d n # 1 swxY w Y | || | || | |d d S )Nc t j d 5 } t d }| | d d d n # 1 swxY w Y | S )Nr
rA r> )r
rE r7 r r rF )outputrn s r
run_threadz<ExceptHookTests.test_original_excepthook.<locals>.run_thread
s (22
f&,?@@@
??$$ $rH c < t dt j d S )NzRunning a thread failed)file)rI r r
r s r rb z@ExceptHookTests.test_original_excepthook.<locals>.threading_hook s +#* = = = = = =r r zRunning a thread failed
)r
rY r! r" r rk r )r+ rj rb default_outputcustom_hook_outputrecovered_outputs r test_original_excepthookz(ExceptHookTests.test_original_excepthook s % % % > > > $
y,
G
G , ,!+ #,#;I )z|| , , , , , , , , , , , , , , ,
)9::: N,>??? +-HIIIIIs +A##A'*A')
r7 r8 r9 rU rG r
r rM rT r^ rf rp
__classcell__)r? s @r r< r< s
8
8
8
1 1 1( 0 0 0 *; ; ;*J J J J J J Jr r< c e Zd Zd Zd Zd ZdS )
TimerTestsc x t | g | _ t j | _ d S r5 )rQ rU
callback_argsr! r callback_eventr* s r rU zTimerTests.setUp! s4 4 'o / /r c ~ t j d| j }| | j |j d d|j d< | j t j d| j }| | j |
t | j d |
| j di fdi fg |
|
d S )Nrq blahbarfoorf r: )r! Timer
_callback_spyr rv r r r r clearrk r ru r )r+ timer1timer2s r test_init_immutable_default_argsz+TimerTests.test_init_immutable_default_args& s t'9:: """ 6""" $
e !!### t'9:: """ T /00!444 +r2hR-ABBB
r c | j |d d | f | j d S r5 )ru r copyrv r )r+ r r s r r| zTimerTests._callback_spy7 sI !!47FKKMM":;;; !!!!!r N)r7 r8 r9 rU r r| r: r r rs rs sA 0 0 0
"" " " " "r rs c . e Zd Z eej ZdS ) LockTestsN)r7 r8 r9 staticmethodr! r locktyper: r r r r ; s |IN++HHHr r c . e Zd Z eej ZdS ) PyRLockTestsN)r7 r8 r9 r r! _PyRLockr r: r r r r > s |I .//HHHr r zRLock not implemented in Cc . e Zd Z eej ZdS ) CRLockTestsN)r7 r8 r9 r r! _CRLockr r: r r r r A s |I -..HHHr r c . e Zd Z eej ZdS )
EventTestsN)r7 r8 r9 r r! r eventtyper: r r r r E s Y_--IIIr r c 4 e Zd Z eej Zd ZdS )ConditionAsRLockTestsc 0 | d d S )Nz,Condition does not expose _recursion_count())skipTestr* s r test_recursion_countz*ConditionAsRLockTests.test_recursion_countL s
DEEEEEr N)r7 r8 r9 r r! r2 r r r: r r r r H s: |I /00HF F F F Fr r c . e Zd Z eej ZdS )ConditionTestsN)r7 r8 r9 r r! r2 condtyper: r r r r O s |I /00HHHr r c . e Zd Z eej ZdS )SemaphoreTestsN)r7 r8 r9 r r! Semaphoresemtyper: r r r r R s l9.//GGGr r c . e Zd Z eej ZdS )BoundedSemaphoreTestsN)r7 r8 r9 r r! r r r: r r r r U s l9566GGGr r c . e Zd Z eej ZdS ) BarrierTestsN)r7 r8 r9 r r! Barrier barriertyper: r r r r X s ,y 011KKKr r c e Zd Zd ZdS ) MiscTestCasec n t | dh}ddh}t j | t d|| d S )Nr r5 r4 )r! r )extra not_exported)r$ r
check__all__r! )r+ r r s r test__all__zMiscTestCase.test__all__] sW "4((( ' 7 T9.F#(| E E E E E Er N)r7 r8 r9 r r: r r r r \ s( E E E E Er r c X e Zd Zd Zd Zd Zd Zd Zd Zd Z e
j d Z d S )
InterruptMainTestsc d }t j || }| t j || | t 5 t j d d d d S # 1 swxY w Y d S )Nc ddz d S )Nr/ r r: )signumr s r handlerzLInterruptMainTests.check_interrupt_main_with_signal_handler.<locals>.handlerh s
aCCCCr )signalr r r r interrupt_main)r+ r r old_handlers r (check_interrupt_main_with_signal_handlerz;InterruptMainTests.check_interrupt_main_with_signal_handlerg s mFG44
v{;;;
0
1
1 % % " $ $ $ % % % % % % % % % % % % % % % % % %s A55A9<A9c Z t j | } t j |t j t j | t j |t j t j | t j || d S # t j || w xY wr5 )r getsignalSIG_IGNr r SIG_DFL)r+ r r s r check_interrupt_main_noerrorz/InterruptMainTests.check_interrupt_main_noerrorq s "6** + M&&. 1 1 1 "6 * * * M&&. 1 1 1 "6 * * *
M&' * * * * *FM&' * * * *s A&B B*c d }t j | }| t 5 | | d d d n # 1 swxY w Y | d S )Nc , t j d S r5 )r r r: r r call_interruptzHInterruptMainTests.test_interrupt_main_subthread.<locals>.call_interrupt s " $ $ $ $ $r rj )r! r@ r KeyboardInterruptr r )r+ r r s r test_interrupt_main_subthreadz0InterruptMainTests.test_interrupt_main_subthread~ s % % % N 3 3 3
0
1
1
GGIII
FFHHH
s )A((A,/A,c | t 5 t j d d d d S # 1 swxY w Y d S r5 )r r r r r* s r test_interrupt_main_mainthreadz1InterruptMainTests.test_interrupt_main_mainthread s
0
1
1 % % " $ $ $ % % % % % % % % % % % % % % % % % %s <A A c | t j | t j d S r5 )r r SIGINTSIGTERMr* s r 'test_interrupt_main_with_signal_handlerz:InterruptMainTests.test_interrupt_main_with_signal_handler s4 55fmDDD 55fnEEEEEr c | t j | t j d S r5 )r r r r r* s r test_interrupt_main_noerrorz.InterruptMainTests.test_interrupt_main_noerror s4 ))&-888 ))&.99999r c | t t j d | t t j t j | t t j d d S )Nr i@B )r r r r r NSIGr* s r "test_interrupt_main_invalid_signalz5InterruptMainTests.test_interrupt_main_invalid_signal s[ *g&<bAAA *g&<fkJJJ *g&<gFFFFFr c dg}dg}dg}d }t j ||||f }| |d s |d d|d< | | |d d S )NTFc T d}d| d< |d r|r|dz }nd S |d d|d< d S )Ni Tr r/ r: )rt cont interrupted
iterationss r workerzAInterruptMainTests.test_can_interrupt_tight_loops.<locals>.worker sT $J GAJq'
!OJJF q'
"KNNNr r r )r! r@ r r r )r+ r rt r r r s r test_can_interrupt_tight_loopsz1InterruptMainTests.test_can_interrupt_tight_loops s v' g " " "
F$ 0L M M M !* !* Q A'''''r N)
r7 r8 r9 r r r r r r r r reap_threadsr r: r r r r f s % % % + + + % % % F F F: : :G G G
"( ( #"( ( (r r c e Zd Zd Zd Zd ZdS ) AtexitTestsc t dd \ }}}| | | | d d S )Nr zif True:
import threading
def run_last():
print('parrot')
threading._register_atexit(run_last)
s parrot)r r rk r r s r test_atexit_outputz AtexitTests.test_atexit_output sX ' /
C
i00000r c X t dd \ }}}| | d S )Nr aN if 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()
r r s r test_atexit_called_oncez#AtexitTests.test_atexit_called_once s: '
/
C
r c t dd \ }}}| | | d| d S )Nr zif 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 r r rb r s r test_atexit_after_shutdownz&AtexitTests.test_atexit_after_shutdown sb (
/
C
J
r N)r7 r8 r9 r r r r: r r r r sA 1 1 1
r r __main__)@__doc__ test.supportr r r r r r test.support.import_helperr test.support.script_helperr r
rH r r r! rK r r" rQ r r rA r/ r r r
requires_working_threadingr r r r r$ rm r& r@ r< TestCaserQ r[ r r r r7 r< rs r
RLockTestsr r r r r r r r r r r r r r7 r[ r: r r <module>r s > > > > > > > > 9 9 9 9 9 9 9 9 9 9 4 4 4 4 4 4 N N N N N N N N
+ +4 8 8 8 8 + 73* + + 4 4 4 f < < < < <! < < <B% % % % %8$ % % %{
# {
# {
# {
# {
#, {
# {
# {
#|f f f f f< f f fR_; _; _; _; _;l _; _; _;Do o o o ol o o od' ' ' ' 'I$ ' ' '
pJ pJ pJ pJ pJl pJ pJ pJf" " " " " " " "8, , , , ,
$ , , ,0 0 0 0 0:( 0 0 0 "d*,HII/ / / / /*' / / JI/. . . . .& . . .F F F F FJ 1 F F F1 1 1 1 1Z. 1 1 10 0 0 0 0Z. 0 0 07 7 7 7 7J < 7 7 72 2 2 2 2:* 2 2 2E E E E E8$ E E EM( M( M( M( M(* M( M( M(`/ / / / / (# / / / d zHMOOOOO r 
Browse
Usage
Versions