File: //usr/local/lib/python3.7/test/eintrdata/__pycache__/eintr_tester.cpython-37.opt-2.pyc
B
��gG � @ 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 | �( y
| V W n | � � � Y nX W d Q R X d S )N)�kill)�proc� r �7/usr/local/lib/python3.7/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 sL d}� fdd�t |�D �}xt |�D ]
}|� q$W x|D ] }|�� q8W d S )N� c s g | ] }� � � �qS r )r- )�.0�_)r r r �
<listcomp>Y s z3OSEINTRTest._test_wait_multiple.<locals>.<listcomp>)�range�wait)r � wait_funcZnumZ processesr0 r r )r r �_test_wait_multipleW s
z OSEINTRTest._test_wait_multiplec C s | � tj� d S )N)r5 �osr3 )r r r r � test_wait` s zOSEINTRTest.test_wait�wait3zrequires wait3()c C s | � dd� � d S )Nc S s
t �d�S )Nr )r6 r8 r r r r �<lambda>e � z(OSEINTRTest.test_wait3.<locals>.<lambda>)r5 )r r r r �
test_wait3c s zOSEINTRTest.test_wait3c C s | � � }||j� |�� d S )N)r- �pidr3 )r r4 r r r r �_test_wait_singleg s
z OSEINTRTest._test_wait_singlec C s | � dd� � d S )Nc S s t �| d�S )Nr )r6 �waitpid)r< r r r r9 n r: z*OSEINTRTest.test_waitpid.<locals>.<lambda>)r= )r r r r � test_waitpidm s zOSEINTRTest.test_waitpid�wait4zrequires wait4()c C s | � dd� � d S )Nc S s t �| d�S )Nr )r6 r@ )r< r r r r9 r r: z(OSEINTRTest.test_wait4.<locals>.<lambda>)r= )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|��F t �|� x$|D ] }| � |t �
|t |��� q|W | � |� � d� W d Q R X d S )Ns hellos worlds spam�
zimport os, sys, time� zwr = int(sys.argv[1])z
datas = %rzsleep_time = %rzfor data in datas:z$ # let the parent block on read()z time.sleep(sleep_time)z os.write(wr, data))�pass_fdsr )
r6 �pipe�
addCleanup�close�joinr) r" �strr � assertEqual�read�lenr3 )r �rd�wr�datasr, 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
|��R t �|� d}x,|t |�k r�|t � |t
|�|d � �7 }q�W | �|�� d� W d Q R X d S )N� xrB zimport io, os, sys, timerC zrd = int(sys.argv[1])zsleep_time = %rzdata = b"x" * %szdata_len = len(data)z!# let the parent block on write()ztime.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)))rD r )r6 rE rF rG r Z
PIPE_MAX_SIZErH r) r" rI r rL �write�
memoryviewrJ r3 )r rM rN rP r, r �writtenr r r �
test_write� s<
zOSEINTRTest.test_writeN)r& r' r( r- r5 r7 �unittest�
skipUnlessr r6 r; r= r? rA rQ rV 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 |��B |�� x"|D ]}| �
|||t|��� q�W | �
|�� d� W d Q R X d S )NrR � y� zrB z import os, socket, sys, timerC zfd = int(sys.argv[1])z family = %szsock_type = %sz
datas = %rzsleep_time = %rz)wr = socket.fromfd(fd, family, sock_type)z 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))rD r )�socketrZ rF rG rH �int�family�typer) �filenor" rI r rJ rL r3 ) r Z recv_funcrM rN rO r, �fdr rP r r r �
_test_recv� s6
zSocketEINTRTest._test_recvc C s | � tjj� d S )N)rc r] Zrecv)r r r r � test_recv� s zSocketEINTRTest.test_recv�recvmsgzneeds recvmsg()c C s | � dd� � d S )Nc S s | � |�d S )Nr )re )�sockrP r r r r9 � r: z.SocketEINTRTest.test_recvmsg.<locals>.<lambda>)rc )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|��b |�� d}x>|t|�k r�||t|�|d � �} || d kr�t|�n| 7 }q�W | �|�� d� W d Q R X d S )Ns xyzr. rB z import os, socket, sys, timerC zfd = int(sys.argv[1])z family = %szsock_type = %szsleep_time = %rzdata = b"xyz" * %szdata_len = len(data)z)rd = socket.fromfd(fd, family, sock_type)z os.close(fd)zwith rd:z$ # let the parent block on send()z time.sleep(sleep_time)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)))rD r )r] rZ rF rG r Z
SOCK_MAX_SIZErH r^ r_ r` r) ra r" rI r rL rT rJ r3 )
r Z send_funcrM rN rP r, rb r rU Zsentr r r �
_test_send� sJ
zSocketEINTRTest._test_sendc C s | � tjj� d S )N)rh r] �send)r r r r � test_send s zSocketEINTRTest.test_sendc C s | � tjj� d S )N)rh r] Zsendall)r r r r � test_sendall s z SocketEINTRTest.test_sendall�sendmsgzneeds sendmsg()c C s | � dd� � d S )Nc S s | � |g�S )N)rl )rf rP r r r r9 r: z.SocketEINTRTest.test_sendmsg.<locals>.<lambda>)rh )r r r r � test_sendmsg s z SocketEINTRTest.test_sendmsgc C s� t � t jt j�}| �|j� |�tjdf� |�� d }|� � d�
dddtj d| d| j dd d
d d f
�}| � |�}t
|��* |�� \}}|�� | �|�� d� W d Q R X d S )
Nr r
rB zimport socket, timerC z host = %rz port = %szsleep_time = %rz # let parent block on accept()ztime.sleep(sleep_time)z,with socket.create_connection((host, port)):z time.sleep(sleep_time))r] ZAF_INETZ SOCK_STREAMrF rG Zbindr ZHOSTZ getsocknameZlistenrH r) r" r ZacceptrJ r3 )r rf Zportr, r Z client_sockr0 r r r � test_accept s*
zSocketEINTRTest.test_accept�
r. �mkfifozneeds mkfifo()c C s� t j}t �|� yt�|� W n0 tk
rN } z| �d| � W d d }~X Y nX | �t j|� d�ddd| d| j dddd|f �}| �
|�}t |�� ||� | � |�
� d � W d Q R X d S )
Nzos.mkfifo(): %srB zimport os, timerC z path = %azsleep_time = %rz# let the parent blockztime.sleep(sleep_time)r )r �TESTFN�unlinkr6 rp �PermissionErrorZskipTestrF rH r) r" r rJ r3 )r Zdo_open_close_readerZdo_open_close_writer�filename�er, r r r r �
_test_open> s*
zSocketEINTRTest._test_openc C s t |d�}|�� d S )N�w)�openrG )r �path�fpr r r � python_open] 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())rv r{ )r r r r � test_opena s zSocketEINTRTest.test_openc C s t �|t j�}t �|� d S )N)r6 rx �O_WRONLYrG )r ry rb r r r �os_openg s zSocketEINTRTest.os_openc C s | � d| j� d S )Nz,fd = os.open(path, os.O_RDONLY)
os.close(fd))rv r )r r r r � test_os_openk s z SocketEINTRTest.test_os_openN)r& r' r( rW rX r r] rc rd rg rh rj rk rm rn r Zrequires_freebsd_versionr6 rv r{ �skipIfr �platformr} r r� r r r r rY � s $ .
rY c @ s e Zd Zdd� ZdS )�
TimeEINTRTestc C s: t �� }t �| j� | �� t �� | }| �|| j� d S )N)�time� monotonic�sleepr) r �assertGreaterEqual)r �t0�dtr r r �
test_sleepv s
zTimeEINTRTest.test_sleepN)r& r' r( r� r r r r r� r 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 d Q R X | �|�� d
� d S ) Nc W s d S )Nr )r$ r r r r9 � r: z/SignalEINTRTest.check_sigwait.<locals>.<lambda>rB zimport os, timezpid = %sz signum = %szsleep_time = %rztime.sleep(sleep_time)zos.kill(pid, signum)r )r �SIGUSR1r6 �getpidrF rH r^ r) r� � SIG_BLOCK� SIG_UNBLOCKr� r� r" r rJ r3 )
r r4 r
r< Z old_handlerr, 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 )N)r r� )r
r r r r4 � s z3SignalEINTRTest.test_sigwaitinfo.<locals>.wait_func)r� )r r4 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 r4 � s z4SignalEINTRTest.test_sigtimedwait.<locals>.wait_func)r� )r r4 r r r �test_sigtimedwait� s z!SignalEINTRTest.test_sigtimedwaitN)
r& r' r( r� rW rX 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 )N)r� r� �selectr) r r� )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 �pollerr� r� r r r � test_poll� s zSelectEINTRTest.test_poll�epollzneed select.epollc C sN t �� }| �|j� t�� }|�| j� t�� | }| �� | � || j� d S )N)
r� r� rF rG r� r� r� 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 )Nr
)
r� r� rF rG 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 )Ng @�@)
r� r� rF rG r� r� r� r) r r� )r r� r� r� r r r � test_devpoll� s z SelectEINTRTest.test_devpollN)r& r' r( r� rW r� r r� rX 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���}xht�� | }|dkr~t d| ��y,||t j
t jB � ||t j� t�d � W q^ tk
r� P Y q^X q^W ||t j
� t�� | }| �|| j� | �� W d Q R X |�� W d Q R X d S )
NrB 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�?)rF r rr rq rH r) r� r� r" r rx � Exception�fcntlZLOCK_EXZLOCK_NBZLOCK_UNr� �BlockingIOErrorr� r r3 )r Z lock_funcZ lock_namer, 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� rW r� r� �systemr� r� r r r r r� � s r� �__main__) �
contextlibr r� r6 r� r� r r] r" r r� rW Ztestr �contextmanagerr rX r ZTestCaser r+ rY r� r� r� r� r& �mainr r r r �<module> s>
)h :
/8)