|
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__/ |
U
?��g�F � @ s� d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl
Z
d dl Z d dl m
Z
e jdd� �Ze �eed�d�G dd� de j��Ze �eed�d�G d d
� d
e��Ze �eed�d�G d d � d e��Ze �eed�d�G d
d� de��Ze �eed�d�e �eed�d�G dd� de���Ze �eed�d�G dd� de��ZG dd� de�Zedk�r|e �� dS )� N)�supportc c s6 | �( z
| V W n | � � � Y nX W 5 Q R X d S �N)�kill)�proc� r �:/opt/cppython/lib/python3.8/test/eintrdata/eintr_tester.py�
kill_on_error s
r � setitimerzrequires setitimer()c @ sD e Zd ZdZdZdZdd� Zdd� Zedd� �Z d d
� Z
d d � Z d
S )�
EINTRBaseTestg�������?g�������?c C s | j d7 _ d S �N� )�signals)�self�signum�framer r r �
sighandler1 s zEINTRBaseTest.sighandlerc C sL d| _ t�tj| j�| _t�tj| j| j� t t
d�rHt
j ddt j
d� d S )Nr �dump_traceback_lateriX T)�exit�file)r
�signal�SIGALRMr � orig_handlerr � ITIMER_REAL� signal_delay�
signal_period�hasattr� faulthandlerr �sys�
__stderr__�r r r r �setUp4 s �
�zEINTRBaseTest.setUpc C s t �t jdd� d S �Nr )r r r r r r r �
stop_alarm@ s zEINTRBaseTest.stop_alarmc C s. | � � t�tj| j� ttd�r*t�� d S )N�cancel_dump_traceback_later)r" r r r r r r# r r r r �tearDownD s
zEINTRBaseTest.tearDownc O s t jdf| }tj|f|�S )Nz-c)r �
executable�
subprocess�Popen)r �args�kwZcmd_argsr r r r& J s zEINTRBaseTest.subprocessN) �__name__�
__module__� __qualname__r r �
sleep_timer r � staticmethodr" r$ r& r r r r r
% s
r
c @ sx e Zd Zdd� Zdd� Zdd� Ze�ee d�d�d d
� �Z
d d � Z d
d� Z e�ee d�d�dd� �Z
dd� Zdd� ZdS )� OSEINTRTestc C s d| j }| �|�S )Nzimport time; time.sleep(%r))r- r& )r �coder r r �new_sleep_processS s
z OSEINTRTest.new_sleep_processc sD d}� fdd�t |�D �}t |�D ]
}|� q"|D ] }|�� q2d S )N� c s g | ] }� � � �qS r )r1 )�.0�_r r r �
<listcomp>Y s z3OSEINTRTest._test_wait_multiple.<locals>.<listcomp>)�range�wait)r � wait_func�num� processesr4 r r r r �_test_wait_multipleW s z OSEINTRTest._test_wait_multiplec C s | � tj� d S r )r; �osr7 r r r r � test_wait` s zOSEINTRTest.test_wait�wait3zrequires wait3()c C s | � dd� � d S )Nc S s
t �d�S r! )r< r> r r r r �<lambda>e � z(OSEINTRTest.test_wait3.<locals>.<lambda>)r; r r r r �
test_wait3c s zOSEINTRTest.test_wait3c C s | � � }||j� |�� d S r )r1 �pidr7 )r r8 r r r r �_test_wait_singleg s
z OSEINTRTest._test_wait_singlec C s | � dd� � d S )Nc S s t �| d�S r! )r< �waitpid�rB r r r r? n r@ z*OSEINTRTest.test_waitpid.<locals>.<lambda>�rC r r r r � test_waitpidm s zOSEINTRTest.test_waitpid�wait4zrequires wait4()c C s | � dd� � d S )Nc S s t �| d�S r! )r< rH rE r r r r? r r@ z(OSEINTRTest.test_wait4.<locals>.<lambda>rF r r r r �
test_wait4p s zOSEINTRTest.test_wait4c C s� t �� \}}| �t j|� dddg}d�dddd| d | j dd
d d d
f
�}| j|t|�|gd�}t|��B t �|� |D ] }| � |t �
|t |��� qz| � |� � d� W 5 Q R X d S )Ns hellos worlds spam�
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�datasr0 r �datar r r � test_readt s*
�
zOSEINTRTest.test_readc C s� t �� \}}| �t j|� dtj }d�dddd| j dtj ddd d
dd d d
ddddddf�}| j|t |�|gd�}t
|��N t �|� d}|t |�k r�|t � |t
|�|d � �7 }q�| �|�� d� W 5 Q R X d S )N� xrJ zimport io, os, sys, timerK zrd = int(sys.argv[1])rM zdata = 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)z value = read_data.getvalue()zif value != data:z0 raise Exception("read error: %s vs %s bytes"z- % (len(value), data_len))rO r )r< rQ rR rS r Z
PIPE_MAX_SIZErT r- r& rU r rX �write�
memoryviewrV r7 )r rY rZ r\ r0 r �writtenr r r �
test_write� s>
�
zOSEINTRTest.test_writeN)r* r+ r, r1 r; r= �unittest�
skipUnlessr r< rA rC rG rI r] rd r r r r r/ O s
r/ c @ s� e Zd Ze�eed�d�dd� �Zdd� Ze�eejd�d�d d
� �Z d d � Z
d
d� Z dd� Z e�eejd�d�dd� �Z
dd� Ze�dd�e�eed�d�dd � ��Zd d � Ze�ejd kd �d!d"� �Zd#d$� Ze�ejd kd �d%d&� �Zd'S )(�SocketEINTRTest�
socketpairzneeds socketpair()c C s� t �� \}}| �|j� dddg}d�ddddt|j� d t|j� d
| d | j dd d
ddddddf�}|� � }| j
|t |�|gd�}t |��> |�� |D ]}| �
|||t|��� q�| �
|�� d� W 5 Q R X d S )Nr^ � y� zrJ � import os, socket, sys, timerK �fd = int(sys.argv[1])� family = %s�sock_type = %srL rM 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)rO r )�socketrh rR rS rT �int�family�typer- �filenor& rU r rV rX r7 ) r Z recv_funcrY rZ r[ r0 �fdr r\ r r r �
_test_recv� s8
�
zSocketEINTRTest._test_recvc C s | � tjj� d S r )rv rp �recvr r r r � test_recv� s zSocketEINTRTest.test_recv�recvmsgzneeds recvmsg()c C s | � dd� � d S )Nc S s | � |�d S r! )ry ��sockr\ r r r r? � r@ z.SocketEINTRTest.test_recvmsg.<locals>.<lambda>)rv r r r r � test_recvmsg� s z SocketEINTRTest.test_recvmsgc
C s t �� \}}| �|j� dtjd }d�ddddt|j� dt|j � d | j
d
tjd d dd d
dddddddddddddf�}|� � }| j |t
|�|gd�}t|��^ |�� d}|t|�k r�||t|�|d � �} || d kr�t|�n| 7 }q�| �|�� d� W 5 Q R X d S )Ns xyzr2 rJ rk rK rl rm rn rM zdata = b"xyz" * %sr_ z)rd = socket.fromfd(fd, family, sock_type)ro zwith rd:z$ # let the parent block on send()rN 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))rO r )rp rh rR rS r Z
SOCK_MAX_SIZErT rq rr rs r- rt r& rU r rX rb rV r7 )
r Z send_funcrY rZ r\ r0 ru r rc �sentr r r �
_test_send� sL �
zSocketEINTRTest._test_sendc C s | � tjj� d S r )r~ rp �sendr r r r � test_send s zSocketEINTRTest.test_sendc C s | � tjj� d S r )r~ rp �sendallr r r r � test_sendall s z SocketEINTRTest.test_sendall�sendmsgzneeds sendmsg()c C s | � dd� � d S )Nc S s | � |g�S r )r� rz r r r r? r@ z.SocketEINTRTest.test_sendmsg.<locals>.<lambda>)r~ r r r r � test_sendmsg s z SocketEINTRTest.test_sendmsgc C s� t �tjdf�}| �|j� |�� d }d�dddtj d| d| j dd d
d d f
�}| � |�}t
|��* |� � \}}|�� | � |�
� d� W 5 Q R X d S )
Nr r rJ zimport socket, timerK z host = %rz port = %srM z # let parent block on accept()r` z,with socket.create_connection((host, port)):rN )rp �
create_serverr ZHOSTrR rS � getsocknamerT r- r& r �acceptrV r7 )r r{ �portr0 r Z client_sockr4 r r r � test_accept s( �
zSocketEINTRTest.test_accept�
r2 �mkfifozneeds mkfifo()c C s� t j}t �|� zt�|� W n0 tk
rN } z| �d| � W 5 d }~X Y nX | �t j|� d�ddd| d| j dddd|f �}| �
|�}t |�� ||� | � |�
� d � W 5 Q R X d S )
Nzos.mkfifo(): %srJ �import os, timerK z path = %arM z# let the parent blockr` r )r �TESTFN�unlinkr< r� �PermissionErrorZskipTestrR rT r- r& r rV r7 )r Zdo_open_close_readerZdo_open_close_writer�filename�er0 r r r r �
_test_open; s,
�
zSocketEINTRTest._test_openc C s t |d�}|�� d S )N�w)�openrS )r �path�fpr r r � python_openZ s
zSocketEINTRTest.python_open�darwinz+hangs under macOS; see bpo-25234, bpo-35363c C s | � d| j� d S )Nz fp = open(path, 'r')
fp.close())r� r� r r r r � test_open^ s �zSocketEINTRTest.test_openc C s t �|t j�}t �|� d S r )r< r� �O_WRONLYrS )r r� ru r r r �os_opend s zSocketEINTRTest.os_openc C s | � d| j� d S )Nz,fd = os.open(path, os.O_RDONLY)
os.close(fd))r� r� r r r r � test_os_openh s �z SocketEINTRTest.test_os_openN)r* r+ r, re rf r rp rv rx r| r~ r� r� r� r� r Zrequires_freebsd_versionr< r� r� �skipIfr �platformr� r� r� r r r r rg � s0
#
.
�
�rg c @ s e Zd Zdd� ZdS )�
TimeEINTRTestc C s: t �� }t �| j� | �� t �� | }| �|| j� d S r )�time� monotonic�sleepr- r" �assertGreaterEqual�r �t0�dtr r r �
test_sleeps s
zTimeEINTRTest.test_sleepN)r* r+ r, r� r r r r r� o s r� �pthread_sigmaskz need signal.pthread_sigmask()c @ sH e Zd Zdd� Ze�eed�d�dd� �Ze�eed�d�dd � �Z d
S ) �SignalEINTRTestc
C s� t j}t�� }t � |dd� �}| �t j ||� d�ddt�� dt|� d| j dd f�}t �t j |g�}| �t jt j
|g� t � � }| �
|�}t|�� ||� t � � | } W 5 Q R X | �|�� d
� d S ) Nc W s d S r r )r( r r r r? � r@ z/SignalEINTRTest.check_sigwait.<locals>.<lambda>rJ r� zpid = %sz signum = %srM r` zos.kill(pid, signum)r )r �SIGUSR1r< �getpidrR rT rq r- r� � SIG_BLOCK� SIG_UNBLOCKr� r� r& r rV r7 )
r r8 r rB Z old_handlerr0 Zold_maskr� r r� r r r �
check_sigwait� s(
�
z SignalEINTRTest.check_sigwait� sigwaitinfozneed signal.sigwaitinfo()c C s dd� }| � |� d S )Nc S s t �| g� d S r )r r� �r r r r r8 � s z3SignalEINTRTest.test_sigwaitinfo.<locals>.wait_func�r� �r r8 r r r �test_sigwaitinfo� s z SignalEINTRTest.test_sigwaitinfo� sigtimedwaitc C s dd� }| � |� d S )Nc S s t �| gd� d S )Ng ^@)r r� r� r r r r8 � s z4SignalEINTRTest.test_sigtimedwait.<locals>.wait_funcr� r� r r r �test_sigtimedwait� s z!SignalEINTRTest.test_sigtimedwaitN)
r* r+ r, r� re rf r r r� r� r r r r r� { s �
�r� c @ s� e Zd Zdd� Ze�ejdkd�e�e e
d�d�dd� ��Z e�e e
d �d
�d d � �Z e�e e
d
�d�dd� �Z
e�e e
d�d�dd� �ZdS )�SelectEINTRTestc C s@ t �� }t�g g g | j� t �� | }| �� | �|| j� d S r )r� r� �selectr- r" r� r� r r r � test_select� s
zSelectEINTRTest.test_selectr� z(poll may fail on macOS; see issue #28087�pollzneed select.pollc C sF t �� }t�� }|�| jd � t�� | }| �� | �|| j� d S �Ng @�@)r� r� r� r� r- r" r� �r Zpollerr� r� r r r � test_poll� s zSelectEINTRTest.test_poll�epollzneed select.epollc C sN t �� }| �|j� t�� }|�| j� t�� | }| �� | � || j� d S r )
r� r� rR rS r� r� r� r- r" r� r� r r r �
test_epoll� s zSelectEINTRTest.test_epoll�kqueuezneed select.kqueuec C sR t �� }| �|j� t�� }|�d d| j� t�� | }| �� | � || j� d S r )
r� r� rR rS r� r� Zcontrolr- r" r� )r r� r� r� r r r � test_kqueue� s zSelectEINTRTest.test_kqueue�devpollzneed select.devpollc C sR t �� }| �|j� t�� }|�| jd � t�� | }| �� | � || j� d S r� )
r� r� rR rS r� r� r� r- r" r� r� r r r � test_devpoll� s z SelectEINTRTest.test_devpollN)r* r+ r, r� re r� r r� rf r r� r� r� r� r� r r r r r� � s �
r� c @ s8 e Zd Zdd� Ze�e�� dkd�dd� �Zdd� Z d S )
�
FNTLEINTRTestc
C s | � tjtj� d�ddtj d| d| j f�}t�� }| �|�}t |��� t
tjd���}t�� | }|dkr|t d| ��z,||t j
t jB � ||t j� t�d � W q\ tk
r� Y q�Y q\X q\||t j
� t�� | }| �|| j� | �� W 5 Q R X |�� W 5 Q R X d S )
NrJ zimport fcntl, timezwith open('%s', 'wb') as f:z fcntl.%s(f, fcntl.LOCK_EX)z time.sleep(%s)�wbg N@z failed to sync child in %.1f secg{�G�z�?)rR r r� r� rT r- r� r� r& r r� � Exception�fcntlZLOCK_EXZLOCK_NBZLOCK_UNr� �BlockingIOErrorr� r" r7 )r Z lock_funcZ lock_namer0 Z
start_timer �fr� r r r �_lock� s2 �
zFNTLEINTRTest._lockZAIXzAIX returns PermissionErrorc C s | � tjd� d S )N�lockf)r� r� r� r r r r �
test_lockf s zFNTLEINTRTest.test_lockfc C s | � tjd� d S )N�flock)r� r� r� r r r r �
test_flock
s zFNTLEINTRTest.test_flockN)
r* r+ r, r� re r� r� �systemr� r� r r r r r� � s
r� �__main__) �
contextlibr r� r< r� r� r rp r& r r� re �testr �contextmanagerr rf r ZTestCaser
r/ rg r� r� r� r� r* �mainr r r r �<module> sB
)h 7 �/8)