python (3.12.0)
͑eF d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl
Z
d dl Z d dl m
Z
d dlmZ d dlmZ e j" d Z e j& eed d G d d e j* Z e j& eed d G d
d e Z e j& eed d G d d
e Z e j& eed d G d de Z e j& eed d e j& eed d G d de Z e j& eed d G d de Z G d de Z e dk( r e j< yy) N)support) os_helper)
socket_helperc # t K | 5 | d d d y # | j xY w# 1 sw Y y xY wwN)kill)procs D/BuggyBox/python/3.12.0/bootstrap/lib/python3.12/test/_test_eintr.py
kill_on_errorr sB Q
J IIK s 8, 8),58 setitimerzrequires setitimer()c B e Zd Z dZdZdZd Zd Zed Z d Z
d Z y)
EINTRBaseTestg?g?c . | xj dz
c_ y N )signals)selfsignumframes r
sighandlerzEINTRBaseTest.sighandler3 s c 6 d| _ t j t j | j | _ t j
t j | j | j t j ddt j y )Nr iX T)exitfile)
r signalSIGALRMr orig_handlerr ITIMER_REAL signal_delay
signal_period faulthandlerdump_traceback_latersys
__stderr__r s r
setUpzEINTRBaseTest.setUp6 sf "MM&..$// J++T->-> ++ -
))'/2~~ ?r c N t j t j dd y Nr )r r r r r
stop_alarmzEINTRBaseTest.stop_alarmA s ++Q2r c | j t j t j | j t j
y r )r* r r r r! cancel_dump_traceback_laterr% s r
tearDownzEINTRBaseTest.tearDownE s0
fnnd&7&78002r c X t j df|z }t j |fi |S )Nz-c)r#
executable
subprocessPopen)r argskwcmd_argss r
r0 zEINTRBaseTest.subprocessJ s+ NND)D0 /B//r N) __name__
__module__ __qualname__r r
sleep_timer r& staticmethodr* r- r0 r) r r
r r ' s@ ' LM J ? 3 33
0r r c e Zd Z d Zd Zd Z ej ee d d d Z
d Z d Z ej ee d d
d Z
d Zd
Zy) OSEINTRTestc B d| j z }| j | S )Nzimport time; time.sleep(%r))r8 r0 )r codes r
new_sleep_processz OSEINTRTest.new_sleep_processS s ,t>t$$r c d}t | D cg c] }| j }}t | D ] } | |D ] }|j y c c}w )N )ranger> wait)r wait_funcnum_ processesr s r
_test_wait_multiplez OSEINTRTest._test_wait_multipleW s_ 7<SzB!T++-B Bs A K D IIK Cs Ac B | j t j y r )rG osrB r% s r
test_waitzOSEINTRTest.test_wait` s )r wait3zrequires wait3()c ( | j d y )Nc , t j d S r( )rI rK r) r r
<lambda>z(OSEINTRTest.test_wait3.<locals>.<lambda>e s ! r )rG r% s r
test_wait3zOSEINTRTest.test_wait3c s !45r c h | j } ||j |j y r )r> pidrB )r rC r s r
_test_wait_singlez OSEINTRTest._test_wait_singleg s% %%'$(( r c ( | j d y )Nc . t j | d S r( )rI waitpidrQ s r
rN z*OSEINTRTest.test_waitpid.<locals>.<lambda>n s 2::c1+= r rR r% s r
test_waitpidzOSEINTRTest.test_waitpidm s =>r wait4zrequires wait4()c ( | j d y )Nc . t j | d S r( )rI rY rV s r
rN z(OSEINTRTest.test_wait4.<locals>.<lambda>r s 288C+; r rW r% s r
test_wait4zOSEINTRTest.test_wait4p s ;<r c " t j \ }}| j t j | g d}dj dddd|z d| j
z ddd d
d f
}| j
|t | |g }t | 5 t j | |D ]1 }| j |t j |t | 3 | j |j d
d d d y # 1 sw Y y xY w)N)s hellos worlds spam
zimport os, sys, time zwr = int(sys.argv[1])
datas = %rsleep_time = %rzfor data in datas:z$ # let the parent block on read() time.sleep(sleep_time)z os.write(wr, data)pass_fdsr )
rI pipe
addCleanupclosejoinr8 r0 strr assertEqualreadlenrB )r rdwrdatasr= r datas r
test_readzOSEINTRTest.test_readt s B "%
.yy " # 5 / 2 ( $
tSWt<
4
- HHRL
? rwwr3t9'=>
? TYY[! , - - -s
A,DDc t j \ }}| j t j | dt j
z }dj
dddd| j z dt j
z ddd d
dd d d
ddddddf }| j |t | |g }t | 5 t j | d}|t | k r4|t j |t | |d z
}|t | k r4| j |j d d d d y # 1 sw Y y xY w)N xr^ zimport io, os, sys, timer_ zrd = int(sys.argv[1])ra zdata = b"x" * %sdata_len = len(data)z!# let the parent block on write()time.sleep(sleep_time)zread_data = io.BytesIO()z+while len(read_data.getvalue()) < data_len:z% chunk = os.read(rd, 2 * data_len)z read_data.write(chunk)z value = read_data.getvalue()zif value != data:z0 raise Exception("read error: %s vs %s bytes"z- % (len(value), data_len))rc r )rI re rf rg r
PIPE_MAX_SIZErh r8 r0 ri r rl write
memoryviewrj rB )r rm rn rp r= r writtens r
test_writezOSEINTRTest.test_write s/ B "% g+++yy & # / !6!6 6 " / $ & 9 3 ( * > ;'
, tSWt<
4
- HHRLGCI%288B
4(8(BCC CI% TYY[! , - - -s 6AD9 D99EN)r5 r6 r7 r> rG rJ unittest
skipUnlesshasattrrI rO rR rX r\ rq rz r) r r
r; r; O s *%* XW-/AB6 C6 ? XW-/AB= C=-:$-r r; c $ e Zd Z ej eed d d Zd Z ej eej d d d Z d Z
d Z d
Z ej eej d d d
Z
d Z ej dd ej eed d d Zd Z ej( ej, dk( d d Zd Z ej( ej, dk( d d Zy)SocketEINTRTest
socketpairzneeds socketpair()c x t j \ }}| j |j g d}dj ddddt |j z dt |j z d|z d | j z dd
d dd d
dddf }|j }| j |t | |g }t | 5 |j |D ]$ }| j | ||t | & | j |j d d d d y # 1 sw Y y xY w)N)rs y zr^ import os, socket, sys, timer_ fd = int(sys.argv[1]) family = %ssock_type = %sr` ra z)wr = socket.fromfd(fd, family, sock_type) os.close(fd)zwith wr:z for data in datas:z( # let the parent block on recv()z time.sleep(sleep_time)z wr.sendall(data)rc r )socketr rf rg rh intfamilytyper8 filenor0 ri r rj rl rB ) r recv_funcrm rn ro r= fdr rp s r
_test_recvzSocketEINTRTest._test_recv s% ""$B ! #yy * # C N * s277| + 5 / 7 $ 6 , &!
& YY[tSWt<
4
- HHJ
A ySY'?@
A TYY[! , - - -s
AD00D9c V | j t j j y r )r r recvr% s r
test_recvzSocketEINTRTest.test_recv
**+r recvmsgzneeds recvmsg()c ( | j d y )Nc * | j | d S r( )r sockrp s r
rN z.SocketEINTRTest.test_recvmsg.<locals>.<lambda> s 4<<+=a+@ r )r r% s r
test_recvmsgz SocketEINTRTest.test_recvmsg s @Ar c t j \ }}| j |j dt j
dz z }dj
ddddt |j z dt |j z d | j z d
t j
dz z d dd d
dddddddddddddf }|j }| j |t | |g }t | 5 |j d}|t | k r6 ||t! | |d } || t | n| z
}|t | k r6| j# |j% d d d d y # 1 sw Y y xY w)Ns xyzr@ r^ r r_ r r r ra zdata = b"xyz" * %srt z)rd = socket.fromfd(fd, family, sock_type)r zwith rd:z$ # let the parent block on send()rb z' received_data = bytearray(data_len)z n = 0z while n < data_len:z8 n += rd.recv_into(memoryview(received_data)[n:])zif received_data != data:z0 raise Exception("recv error: %s vs %s bytes"z5 % (len(received_data), data_len))rc r )r r rf rg r
SOCK_MAX_SIZErh r r r r8 r r0 ri r rl rx rj rB )
r send_funcrm rn rp r= r r ry sents
r
_test_sendzSocketEINTRTest._test_send sy ""$B ! ..!34yy * # C N * s277| + / G$9$9Q$> ? " 7 2 ( 5 % F ' > C1
6 YY[tSWt<
4
- HHJGCI% Z%5gh%?@ 3t9$> CI%
TYY[! , - - -s 8AE8 E88Fc V | j t j j y r )r r sendr% s r
test_sendzSocketEINTRTest.test_send r r c V | j t j j y r )r r sendallr% s r
test_sendallz SocketEINTRTest.test_sendall s
--.r sendmsgzneeds sendmsg()c ( | j d y )Nc & | j |g S r )r r s r
rN z.SocketEINTRTest.test_sendmsg.<locals>.<lambda> s 4<<+? r )r r% s r
test_sendmsgz SocketEINTRTest.test_sendmsg s ?@r c
t j t j df }| j |j
|j
d }dj dddt j z d|z d| j z dd d
d d f
}| j | }t | 5 |j \ }}|j | j |j d d d d y # 1 sw Y y xY w)
Nr r r^ zimport socket, timer_ z host = %rz port = %sra z # let parent block on accept()ru z,with socket.create_connection((host, port)):rb )r
create_serverr HOSTrf rg getsocknamerh r8 r0 r acceptrj rB )r r portr= r client_sockrE s r
test_acceptzSocketEINTRTest.test_accept s ##]%7%7$;<
# !!$yy ! -,, , $ / , $ : (
t$
4
- ![[]NK TYY[! , - - -s
,AC99D
r@ mkfifozneeds mkfifo()c t j }t j | t j | | j t j | dj ddd|z d| j z dddd|f }| j | }t | 5 || | j |j d d d d y # t
$ r }| j
d|z Y d }~d }~ww xY w# 1 sw Y y xY w)
Nzos.mkfifo(): %sr^ import os, timer_ z path = %ara z# let the parent blockru r )r TESTFNunlinkrI r PermissionErrorskipTestrf rh r8 r0 r rj rB )r do_open_close_readerdo_open_close_writerfilenameer= r s r
_test_openzSocketEINTRTest._test_open; s ## " 1 IIh
(((3yy ( " / $ $
t$
4
- * TYY[! , - -# 1 MM+a/ 0 0 1" - -s# C ")C> C; C66C;>Dc < t |d }|j y )Nw)openrg )r pathfps r
python_openzSocketEINTRTest.python_openZ s
$_
r darwinz+hangs under macOS; see bpo-25234, bpo-35363c < | j d| j y )Nz fp = open(path, 'r')
fp.close())r r r% s r
test_openzSocketEINTRTest.test_open^ s
: (( *r c v t j |t j }t j | y r )rI r O_WRONLYrg )r r r s r
os_openzSocketEINTRTest.os_opend s!
WWT2;;
'
r c < | j d| j y )Nz,fd = os.open(path, os.O_RDONLY)
os.close(fd))r r r% s r
test_os_openz SocketEINTRTest.test_os_openh s
G &r N)r5 r6 r7 r{ r| r} r r r r r r r r r r requires_freebsd_versionrI r r skipIfr# platformr r r r) r r
r r sY .X 68LM!- N!-F, X :<MNB OB,-\,/ X :<MNA OA-8 &W%%b!,XX.0@A- B --: X__S\\X-BD*D* X__S\\X-BD&D&r r c e Zd Z d Zy)
TimeEINTRTestc t j }t j | j | j t j |z
}| j || j y r )time monotonicsleepr8 r* assertGreaterEqualr t0dts r
test_sleepzTimeEINTRTest.test_sleeps sM
^^
4??#
^^
" DOO4r N)r5 r6 r7 r r) r r
r r o s
,5r r pthread_sigmaskz need signal.pthread_sigmask()c e Zd Z d Z ej
eed d d Z ej
eed d d Z y)SignalEINTRTestc t j }t j }t j |d }| j t j || dj ddt j z dt
| z d| j z ddf }t j t j |g }| j t j t j |g | j | }t | 5 || d d d | j |j d y # 1 sw Y *xY w)
Nc y r r) )r2 s r
rN z/SignalEINTRTest.check_sigwait.<locals>.<lambda> s r r^ r zpid = %sz signum = %sra ru zos.kill(pid, signum)r )r SIGUSR1rI getpidrf rh r r8 r SIG_BLOCK SIG_UNBLOCKr0 r rj rB )r rC r rQ old_handlerr= old_maskr s r
check_sigwaitz SignalEINTRTest.check_sigwait s iik mmF,>?
v{;yy $ CK ' / $ "
))&*:*:VHE ..0B0BVHMt$
4
f
a( s EE sigwaitinfozneed signal.sigwaitinfo()c , d }| j | y )Nc 0 t j | g y r )r r r s r
rC z3SignalEINTRTest.test_sigwaitinfo.<locals>.wait_func s x (r r r rC s r
test_sigwaitinfoz SignalEINTRTest.test_sigwaitinfo s )
9%r sigtimedwaitc , d }| j | y )Nc 2 t j | gd y )Ng ^@)r r r s r
rC z4SignalEINTRTest.test_sigtimedwait.<locals>.wait_func s % 0r r r s r
test_sigtimedwaitz!SignalEINTRTest.test_sigtimedwait s 1
9%r N)
r5 r6 r7 r r{ r| r} r r r r) r r
r r { sf /)2 X 746&6& X 846&6&r r c Z e Zd Z d Z ej
ej dk( d ej e e
d d d Z ej e e
d d d Z ej e e
d
d d Z
ej e e
d
d d Zy)SelectEINTRTestc t j }t j g g g | j t j |z
}| j | j || j y r )r r selectr8 r* r r s r
test_selectzSelectEINTRTest.test_select sS
^^
b"b$//2
^^
" DOO4r r z(poll may fail on macOS; see issue #28087pollzneed select.pollc t j }t j }|j | j dz t j |z
}| j | j
|| j y Ng @@)r r r r r8 r* r r pollerr r s r
test_pollzSelectEINTRTest.test_poll s_
^^
DOOc)*
^^
" DOO4r epollzneed select.epollc F t j }| j |j t j
}|j
| j t j
|z
}| j | j || j y r )
r r rf rg r r r r8 r* r r s r
test_epollzSelectEINTRTest.test_epoll sh %
^^
DOO$
^^
" DOO4r kqueuezneed select.kqueuec J t j }| j |j t j
}|j
d d| j t j
|z
}| j | j || j y r )
r r rf rg r r controlr8 r* r )r r r r s r
test_kqueuezSelectEINTRTest.test_kqueue sl %
^^
tQ0
^^
" DOO4r devpollzneed select.devpollc L t j }| j |j t j
}|j
| j dz t j
|z
}| j | j || j y r )
r r rf rg r r r r8 r* r r s r
test_devpollz SelectEINTRTest.test_devpoll sn ! %
^^
DOOc)*
^^
" DOO4r N)r5 r6 r7 r r{ r r# r r| r} r r r r r r) r r
r r s .5 X__S\\X-?AX02DE5 FA5 X13FG5 H5 X24HI5 J5 X35JK5 L5r r c n e Zd Zd Z ej
ej dk( d d Zd Z y)
FNTLEINTRTestc | j t j t j dj ddt j z d|z d| j
z f }t
j }| j | }t | 5 t t j d 5 }t
j }t j t j d D ]B } ||t j t j z ||t j" D t
j |z
}t' d |z # t$ $ r Y nw xY w ||t j t
j |z
}| j) || j
| j+ d d d n # 1 sw Y nxY w|j- d d d y # 1 sw Y y xY w)
Nr^ zimport fcntl, timezwith open('%s', 'wb') as f:z fcntl.%s(f, fcntl.LOCK_EX)z time.sleep(%s)wbF)errorz failed to sync child in %.1f sec)rf r r r rh r8 r r r0 r r r sleeping_retry LONG_TIMEOUTfcntlLOCK_EXLOCK_NBLOCK_UNBlockingIOError Exceptionr r* rB ) r lock_func lock_namer=
start_timer frE r s r
_lockzFNTLEINTRTest._lock s (()*:*:;yy )I,<,< < +i 7 $// 1 3 4
^^%
t$
4
i&&-
" !^^ -
//0D0DER MA !!U]]U]]%BC!!U]]3M )J6B#$F$KLL + !U]]+^^%
2''DOO<!'
"
"
"(
IIK+ sI G,0=G.?E-'G E! G E! !A G> G,G G,,G5AIXzAIX returns PermissionErrorc D | j t j d y )Nlockf)r r r r% s r
test_lockfzFNTLEINTRTest.test_lockf s
5;;(r c D | j t j d y )Nflock)r r r r% s r
test_flockzFNTLEINTRTest.test_flock
s
5;;(r N)
r5 r6 r7 r r{ r r systemr r r) r r
r r s> D X___X__&%/1NO) P))r r __main__)
contextlibr! r rI r r r r r0 r# r r{ testr test.supportr r contextmanagerr r| r} TestCaser r; r r r r r r5 mainr) r r
<module>r( s
" & WV[13IJ$0H%% $0 K$0N WV[13IJe-- e- Ke-P WV[13IJs&m s& Ks&l WV[13IJ5M 5 K5 WV[13IJ WV%6746*&m *&6 K
*&Z WV[13IJ55m 55 K55p()M ()V zHMMO r 
Browse
Usage
Versions