Server : LiteSpeed
System : Linux server51.dnsbootclub.com 4.18.0-553.62.1.lve.el8.x86_64 #1 SMP Mon Jul 21 17:50:35 UTC 2025 x86_64
User : nandedex ( 1060)
PHP Version : 8.1.33
Disable Function : NONE
Directory :  /opt/cppython/lib/python3.8/test/eintrdata/__pycache__/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]


Current File : //opt/cppython/lib/python3.8/test/eintrdata/__pycache__/eintr_tester.cpython-38.opt-2.pyc
U

?��g�F�@s�ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl	Z	ddl
Z
ddlZddlm
Z
ejdd��Ze�eed�d�Gdd�dej��Ze�eed�d�Gd	d
�d
e��Ze�eed�d�Gdd�de��Ze�eed�d�Gd
d�de��Ze�eed�d�e�eed�d�Gdd�de���Ze�eed�d�Gdd�de��ZGdd�de�Zedk�r|e��dS)�N)�supportc	cs6|�(z
|VWn|���YnXW5QRXdS�N)�kill)�proc�r�:/opt/cppython/lib/python3.8/test/eintrdata/eintr_tester.py�
kill_on_errors
r�	setitimerzrequires setitimer()c@sDeZdZdZdZdZdd�Zdd�Zedd��Z	d	d
�Z
dd�Zd
S)�
EINTRBaseTestg�������?g�������?cCs|jd7_dS�N�)�signals)�self�signum�framerrr�
sighandler1szEINTRBaseTest.sighandlercCsLd|_t�tj|j�|_t�tj|j|j�t	t
d�rHt
jddtj
d�dS)Nr�dump_traceback_lateriXT)�exit�file)r
�signal�SIGALRMr�orig_handlerr	�ITIMER_REAL�signal_delay�
signal_period�hasattr�faulthandlerr�sys�
__stderr__�rrrr�setUp4s�
�zEINTRBaseTest.setUpcCst�tjdd�dS�Nr)rr	rrrrr�
stop_alarm@szEINTRBaseTest.stop_alarmcCs.|��t�tj|j�ttd�r*t��dS)N�cancel_dump_traceback_later)r"rrrrrr#rrrr�tearDownDs
zEINTRBaseTest.tearDowncOstjdf|}tj|f|�S)Nz-c)r�
executable�
subprocess�Popen)r�args�kwZcmd_argsrrrr&JszEINTRBaseTest.subprocessN)�__name__�
__module__�__qualname__rr�
sleep_timerr �staticmethodr"r$r&rrrrr
%s
r
c@sxeZdZdd�Zdd�Zdd�Ze�ee	d�d�d	d
��Z
dd�Zd
d�Ze�ee	d�d�dd��Z
dd�Zdd�ZdS)�OSEINTRTestcCsd|j}|�|�S)Nzimport time; time.sleep(%r))r-r&)r�coderrr�new_sleep_processSs
zOSEINTRTest.new_sleep_processcsDd}�fdd�t|�D�}t|�D]
}|�q"|D]}|��q2dS)N�csg|]}����qSr)r1)�.0�_rrr�
<listcomp>Ysz3OSEINTRTest._test_wait_multiple.<locals>.<listcomp>)�range�wait)r�	wait_func�num�	processesr4rrrr�_test_wait_multipleWszOSEINTRTest._test_wait_multiplecCs|�tj�dSr)r;�osr7rrrr�	test_wait`szOSEINTRTest.test_wait�wait3zrequires wait3()cCs|�dd��dS)NcSs
t�d�Sr!)r<r>rrrr�<lambda>e�z(OSEINTRTest.test_wait3.<locals>.<lambda>)r;rrrr�
test_wait3cszOSEINTRTest.test_wait3cCs|��}||j�|��dSr)r1�pidr7)rr8rrrr�_test_wait_singlegs
zOSEINTRTest._test_wait_singlecCs|�dd��dS)NcSst�|d�Sr!)r<�waitpid�rBrrrr?nr@z*OSEINTRTest.test_waitpid.<locals>.<lambda>�rCrrrr�test_waitpidmszOSEINTRTest.test_waitpid�wait4zrequires wait4()cCs|�dd��dS)NcSst�|d�Sr!)r<rHrErrrr?rr@z(OSEINTRTest.test_wait4.<locals>.<lambda>rFrrrr�
test_wait4pszOSEINTRTest.test_wait4cCs�t��\}}|�tj|�dddg}d�dddd|d	|jdd
ddd
f
�}|j|t|�|gd�}t|��Bt�|�|D]}|�	|t�
|t|���qz|�	|��d�W5QRXdS)Nshellosworldsspam�
zimport os, sys, time�zwr = int(sys.argv[1])�
datas = %r�sleep_time = %rzfor data in datas:z$    # let the parent block on read()�    time.sleep(sleep_time)z    os.write(wr, data)��pass_fdsr)
r<�pipe�
addCleanup�close�joinr-r&�strr�assertEqual�read�lenr7)r�rd�wr�datasr0r�datarrr�	test_readts*
�


zOSEINTRTest.test_readcCs�t��\}}|�tj|�dtj}d�dddd|jdtjddd	d
dddd
ddddddf�}|j|t	|�|gd�}t
|��Nt�|�d}|t|�kr�|t�|t
|�|d��7}q�|�|��d�W5QRXdS)N�xrJzimport io, os, sys, timerKzrd = int(sys.argv[1])rMzdata = b"x" * %s�data_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)zvalue = read_data.getvalue()zif value != data:z0    raise Exception("read error: %s vs %s bytes"z-                    % (len(value), data_len))rOr)r<rQrRrSrZ
PIPE_MAX_SIZErTr-r&rUrrX�write�
memoryviewrVr7)rrYrZr\r0r�writtenrrr�
test_write�s>
�

zOSEINTRTest.test_writeN)r*r+r,r1r;r=�unittest�
skipUnlessrr<rArCrGrIr]rdrrrrr/Os	

r/c@s�eZdZe�eed�d�dd��Zdd�Ze�eejd�d�d	d
��Z	dd�Z
d
d�Zdd�Ze�eejd�d�dd��Z
dd�Ze�dd�e�eed�d�dd���Zdd�Ze�ejdkd �d!d"��Zd#d$�Ze�ejdkd �d%d&��Zd'S)(�SocketEINTRTest�
socketpairzneeds socketpair()c	Cs�t��\}}|�|j�dddg}d�ddddt|j�d	t|j�d
|d|jddd
ddddddf�}|�	�}|j
|t|�|gd�}t|��>|��|D]}|�
|||t|���q�|�
|��d�W5QRXdS)Nr^�y�zrJ�import os, socket, sys, timerK�fd = int(sys.argv[1])�family = %s�sock_type = %srLrMz)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)rOr)�socketrhrRrSrT�int�family�typer-�filenor&rUrrVrXr7)	rZ	recv_funcrYrZr[r0�fdrr\rrr�
_test_recv�s8
�
zSocketEINTRTest._test_recvcCs|�tjj�dSr)rvrp�recvrrrr�	test_recv�szSocketEINTRTest.test_recv�recvmsgzneeds recvmsg()cCs|�dd��dS)NcSs|�|�dSr!)ry��sockr\rrrr?�r@z.SocketEINTRTest.test_recvmsg.<locals>.<lambda>)rvrrrr�test_recvmsg�szSocketEINTRTest.test_recvmsgc
Cst��\}}|�|j�dtjd}d�ddddt|j�dt|j	�d	|j
d
tjddddd
dddddddddddddf�}|��}|j|t
|�|gd�}t|��^|��d}|t|�kr�||t|�|d��}	||	dkr�t|�n|	7}q�|�|��d�W5QRXdS)Nsxyzr2rJrkrKrlrmrnrMzdata = b"xyz" * %sr_z)rd = socket.fromfd(fd, family, sock_type)rozwith rd:z$    # let the parent block on send()rNz'    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))rOr)rprhrRrSrZ
SOCK_MAX_SIZErTrqrrrsr-rtr&rUrrXrbrVr7)
rZ	send_funcrYrZr\r0rurrc�sentrrr�
_test_send�sL�
zSocketEINTRTest._test_sendcCs|�tjj�dSr)r~rp�sendrrrr�	test_sendszSocketEINTRTest.test_sendcCs|�tjj�dSr)r~rp�sendallrrrr�test_sendallszSocketEINTRTest.test_sendall�sendmsgzneeds sendmsg()cCs|�dd��dS)NcSs|�|g�Sr)r�rzrrrr?r@z.SocketEINTRTest.test_sendmsg.<locals>.<lambda>)r~rrrr�test_sendmsgszSocketEINTRTest.test_sendmsgcCs�t�tjdf�}|�|j�|��d}d�dddtjd|d|jdd	d
ddf
�}|�	|�}t
|��*|��\}}|��|�|�
�d�W5QRXdS)
NrrrJzimport socket, timerKz	host = %rz	port = %srMz# let parent block on accept()r`z,with socket.create_connection((host, port)):rN)rp�
create_serverrZHOSTrRrS�getsocknamerTr-r&r�acceptrVr7)rr{�portr0rZclient_sockr4rrr�test_accepts(�

zSocketEINTRTest.test_accept�
r2�mkfifozneeds mkfifo()cCs�tj}t�|�zt�|�Wn0tk
rN}z|�d|�W5d}~XYnX|�tj|�d�ddd|d|j	dddd|f	�}|�
|�}t|��||�|�|�
�d	�W5QRXdS)
Nzos.mkfifo(): %srJ�import os, timerKz	path = %arMz# let the parent blockr`r)r�TESTFN�unlinkr<r��PermissionErrorZskipTestrRrTr-r&rrVr7)rZdo_open_close_readerZdo_open_close_writer�filename�er0rrrr�
_test_open;s,
 �

zSocketEINTRTest._test_opencCst|d�}|��dS)N�w)�openrS)r�path�fprrr�python_openZs
zSocketEINTRTest.python_open�darwinz+hangs under macOS; see bpo-25234, bpo-35363cCs|�d|j�dS)Nzfp = open(path, 'r')
fp.close())r�r�rrrr�	test_open^s�zSocketEINTRTest.test_opencCst�|tj�}t�|�dSr)r<r��O_WRONLYrS)rr�rurrr�os_opendszSocketEINTRTest.os_opencCs|�d|j�dS)Nz,fd = os.open(path, os.O_RDONLY)
os.close(fd))r�r�rrrr�test_os_openhs�zSocketEINTRTest.test_os_openN)r*r+r,rerfrrprvrxr|r~r�r�r�r�rZrequires_freebsd_versionr<r�r��skipIfr�platformr�r�r�rrrrrg�s0
#
.

�
�rgc@seZdZdd�ZdS)�
TimeEINTRTestcCs:t��}t�|j�|��t��|}|�||j�dSr)�time�	monotonic�sleepr-r"�assertGreaterEqual�r�t0�dtrrr�
test_sleepss
zTimeEINTRTest.test_sleepN)r*r+r,r�rrrrr�osr��pthread_sigmaskzneed signal.pthread_sigmask()c@sHeZdZdd�Ze�eed�d�dd��Ze�eed�d�dd	��Z	d
S)�SignalEINTRTestc
	Cs�tj}t��}t�|dd��}|�tj||�d�ddt��dt|�d|jdd	f�}t�tj	|g�}|�tjtj
|g�t��}|�
|�}t|��||�t��|}	W5QRX|�|��d
�dS)NcWsdSrr)r(rrrr?�r@z/SignalEINTRTest.check_sigwait.<locals>.<lambda>rJr�zpid = %szsignum = %srMr`zos.kill(pid, signum)r)r�SIGUSR1r<�getpidrRrTrqr-r��	SIG_BLOCK�SIG_UNBLOCKr�r�r&rrVr7)
rr8rrBZold_handlerr0Zold_maskr�rr�rrr�
check_sigwait�s(

�	

zSignalEINTRTest.check_sigwait�sigwaitinfozneed signal.sigwaitinfo()cCsdd�}|�|�dS)NcSst�|g�dSr)rr��rrrrr8�sz3SignalEINTRTest.test_sigwaitinfo.<locals>.wait_func�r��rr8rrr�test_sigwaitinfo�sz SignalEINTRTest.test_sigwaitinfo�sigtimedwaitcCsdd�}|�|�dS)NcSst�|gd�dS)Ng^@)rr�r�rrrr8�sz4SignalEINTRTest.test_sigtimedwait.<locals>.wait_funcr�r�rrr�test_sigtimedwait�sz!SignalEINTRTest.test_sigtimedwaitN)
r*r+r,r�rerfrrr�r�rrrrr�{s�
�r�c@s�eZdZdd�Ze�ejdkd�e�e	e
d�d�dd���Ze�e	e
d	�d
�dd��Ze�e	e
d
�d�dd��Z
e�e	e
d�d�dd��ZdS)�SelectEINTRTestcCs@t��}t�ggg|j�t��|}|��|�||j�dSr)r�r��selectr-r"r�r�rrr�test_select�s
zSelectEINTRTest.test_selectr�z(poll may fail on macOS; see issue #28087�pollzneed select.pollcCsFt��}t��}|�|jd�t��|}|��|�||j�dS�Ng@�@)r�r�r�r�r-r"r��rZpollerr�r�rrr�	test_poll�szSelectEINTRTest.test_poll�epollzneed select.epollcCsNt��}|�|j�t��}|�|j�t��|}|��|�	||j�dSr)
r�r�rRrSr�r�r�r-r"r�r�rrr�
test_epoll�szSelectEINTRTest.test_epoll�kqueuezneed select.kqueuecCsRt��}|�|j�t��}|�dd|j�t��|}|��|�	||j�dSr)
r�r�rRrSr�r�Zcontrolr-r"r�)rr�r�r�rrr�test_kqueue�szSelectEINTRTest.test_kqueue�devpollzneed select.devpollcCsRt��}|�|j�t��}|�|jd�t��|}|��|�	||j�dSr�)
r�r�rRrSr�r�r�r-r"r�r�rrr�test_devpoll�szSelectEINTRTest.test_devpollN)r*r+r,r�rer�rr�rfrr�r�r�r�r�rrrrr��s�	



r�c@s8eZdZdd�Ze�e��dkd�dd��Zdd�Z	d	S)
�
FNTLEINTRTestc
Cs|�tjtj�d�ddtjd|d|jf�}t��}|�|�}t	|���t
tjd���}t��|}|dkr|td|��z,||tj
tjB�||tj�t�d	�Wq\tk
r�Yq�Yq\Xq\||tj
�t��|}|�||j�|��W5QRX|��W5QRXdS)
NrJzimport fcntl, timezwith open('%s', 'wb') as f:z   fcntl.%s(f, fcntl.LOCK_EX)z   time.sleep(%s)�wbgN@z failed to sync child in %.1f secg{�G�z�?)rRrr�r�rTr-r�r�r&rr��	Exception�fcntlZLOCK_EXZLOCK_NBZLOCK_UNr��BlockingIOErrorr�r"r7)rZ	lock_funcZ	lock_namer0Z
start_timer�fr�rrr�_lock�s2�

zFNTLEINTRTest._lockZAIXzAIX returns PermissionErrorcCs|�tjd�dS)N�lockf)r�r�r�rrrr�
test_lockf	szFNTLEINTRTest.test_lockfcCs|�tjd�dS)N�flock)r�r�r�rrrr�
test_flock
szFNTLEINTRTest.test_flockN)
r*r+r,r�rer�r��systemr�r�rrrrr��s 
r��__main__)�
contextlibrr�r<r�r�rrpr&rr�re�testr�contextmanagerrrfrZTestCaser
r/rgr�r�r�r�r*�mainrrrr�<module>sB

)h7�/8)


F1le Man4ger