File: //usr/local/lib/python3.10/test/__pycache__/_test_eintr.cpython-310.opt-2.pyc
o
�i�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
d dlmZ d dlmZ e jdd� �Ze �eed�d�G d d
� d
e j��Ze �eed�d�G d d � d e��Ze �eed�d�G d
d� de��Ze �eed�d�G dd� de��Ze �eed�d�e �eed�d�G dd� de���Ze �eed�d�G dd� de��ZG dd� de�Z e dkr�e � � dS dS )� N)�support)� os_helper)�
socket_helperc c sJ � | � z| V W n | � � � W d � d S 1 s w Y d S �N)�kill)�proc� r �-/usr/local/lib/python3.10/test/_test_eintr.py�
kill_on_error s �
�"�r
� setitimerzrequires setitimer()c @ sF e Zd Z dZdZdZdd� 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 �
sighandler3 � zEINTRBaseTest.sighandlerc C sB d| _ t�tj| j�| _t�tj| j| j� t j
ddt j d� d S )Nr iX T)�exit�file)
r �signal�SIGALRMr � orig_handlerr � ITIMER_REAL� signal_delay�
signal_period� faulthandlerZdump_traceback_later�sys�
__stderr__�r r r r �setUp6 s �
�zEINTRBaseTest.setUpc C s t �t jdd� d S �Nr )r r r r r r r �
stop_alarmA s zEINTRBaseTest.stop_alarmc C s$ | � � t�tj| j� t�� d S r )r# r r r r Zcancel_dump_traceback_laterr r r r �tearDownE s zEINTRBaseTest.tearDownc O s t jdf| }tj|fi |��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 @ sz e Zd Z dd� 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 ]}|�� qd 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` � zOSEINTRTest.test_wait�wait3zrequires wait3()c C � | � dd� � d S )Nc S s
t �d�S r" )r< r? r r r r �<lambda>e s
z(OSEINTRTest.test_wait3.<locals>.<lambda>)r; r r r r �
test_wait3c � 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 r@ )Nc S � t �| d�S r" )r< �waitpid�rD r r r rA n � z*OSEINTRTest.test_waitpid.<locals>.<lambda>�rE r r r r � test_waitpidm r zOSEINTRTest.test_waitpid�wait4zrequires wait4()c C r@ )Nc S rF r" )r< rL rH r r r rA r rI z(OSEINTRTest.test_wait4.<locals>.<lambda>rJ r r r r �
test_wait4p rC zOSEINTRTest.test_wait4c C s� t �� \}}| �t j|� g d�}d�dddd| d| j ddd d
d f
�}| j|t|�|gd �}t|��' t �|� |D ]}| � |t �
|t |��� q<| � |� � d
� W d � d S 1 s^w Y d S )N)s 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
|��2 t �|� d}|t |�k ra|t � |t
|�|d � �7 }|t |�k sM| �|�� d� W d � d S 1 stw Y d S )N� xrN zimport io, os, sys, timerO zrd = int(sys.argv[1])rQ 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))rS r )r< rU rV rW r Z
PIPE_MAX_SIZErX r- r& rY r
r\ �write�
memoryviewrZ r7 )r r] r^ r` r0 r �writtenr r r �
test_write� sB
�
�"�zOSEINTRTest.test_writeN)r* r+ r, r1 r; r= �unittest�
skipUnless�hasattrr< rB rE rK rM ra rh r r r r r/ O s
r/ c @ s� e Zd Z e�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� g d�}d�ddddt|j� dt|j� d| d | j dd
d dd d
dddf�}|� � }| j
|t |�|gd�}t |��% |�� |D ]
}| �
|||t|��� qN| �
|�� d� W d � d S 1 sow Y d S )N)rb � y� zrN � import os, socket, sys, timerO �fd = int(sys.argv[1])� family = %s�sock_type = %srP rQ 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)rS r )�socketrm rV rW rX �int�family�typer- �filenor& rY r
rZ r\ r7 ) r Z recv_funcr] r^ r_ r0 �fdr r` r r r �
_test_recv� s: �
"�zSocketEINTRTest._test_recvc C � | � tjj� d S r )r{ ru �recvr r r r � test_recv� r zSocketEINTRTest.test_recv�recvmsgzneeds recvmsg()c C r@ )Nc S s | � |�d S r" )r ��sockr` r r r rA � s z.SocketEINTRTest.test_recvmsg.<locals>.<lambda>)r{ r r r r � test_recvmsg� rC z SocketEINTRTest.test_recvmsgc
C s2 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 u rvt|�n| 7 }|t|�k sb| �|�� d� W d � d S 1 s�w Y d S )Ns xyzr2 rN rp rO rq rr rs rQ zdata = b"xyz" * %src z)rd = socket.fromfd(fd, family, sock_type)rt zwith rd:z$ # let the parent block on send()rR 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))rS r )ru rm rV rW r Z
SOCK_MAX_SIZErX rv rw rx r- ry r& rY r
r\ rf rZ r7 )
r Z send_funcr] r^ r` r0 rz r rg �sentr r r �
_test_send� sP �
�"�zSocketEINTRTest._test_sendc C r| r )r� ru �sendr r r r � test_send r zSocketEINTRTest.test_sendc C r| r )r� ru �sendallr r r r � test_sendall r z SocketEINTRTest.test_sendall�sendmsgzneeds sendmsg()c C r@ )Nc S s | � |g�S r )r� r� r r r rA rI z.SocketEINTRTest.test_sendmsg.<locals>.<lambda>)r� r r r r � test_sendmsg rC z SocketEINTRTest.test_sendmsgc C s� t �tjdf�}| �|j� |�� d }d�dddtj d| d| j dd d
d d f
�}| � |�}t
|�� |� � \}}|�� | � |�
� d� W d � d S 1 sRw Y d S )
Nr r rN zimport socket, timerO z host = %rz port = %srQ z # let parent block on accept()rd z,with socket.create_connection((host, port)):rR )ru �
create_serverr ZHOSTrV rW � getsocknamerX r- r& r
�acceptrZ 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 n ty( } z
| �d| � W Y d }~nd }~ww | �t j|� d�ddd| d| j dddd|f �}| �
|�}t |�� ||� | � |�
� d � W d � d S 1 sdw Y d S )
Nzos.mkfifo(): %srN �import os, timerO z path = %arQ z# let the parent blockrd r )r �TESTFN�unlinkr< r� �PermissionErrorZskipTestrV rX r- r& r
rZ r7 )r Zdo_open_close_readerZdo_open_close_writer�filename�er0 r r r r �
_test_open; s2
���
"�zSocketEINTRTest._test_openc C s t |d�}|�� d S )N�w)�openrW )r �path�fpr r r � python_openZ s
zSocketEINTRTest.python_open�darwinz+hangs under macOS; see bpo-25234, bpo-35363c C � | � d| j� d S )Nz fp = open(path, 'r')
fp.close())r� r� r r r r � test_open^ � �zSocketEINTRTest.test_openc C s t �|t j�}t �|� d S r )r< r� �O_WRONLYrW )r r� rz r r r �os_opend s zSocketEINTRTest.os_openc C r� )Nz,fd = os.open(path, os.O_RDONLY)
os.close(fd))r� r� r r r r � test_os_openh r� z SocketEINTRTest.test_os_openN)r* r+ r, ri rj rk ru r{ r~ r� r� r� r� r� r� r Zrequires_freebsd_versionr< r� r� �skipIfr �platformr� r� r� r r r r rl � s4
#
.
�
�rl c @ s e Zd Z dd� 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 @ sJ e Zd Z dd� 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 � n1 saw Y | �|�� d
� d S ) Nc W s d S r r )r( r r r rA � s z/SignalEINTRTest.check_sigwait.<locals>.<lambda>rN r� zpid = %sz signum = %srQ rd zos.kill(pid, signum)r )r �SIGUSR1r< �getpidrV rX rv r- r� � SIG_BLOCK� SIG_UNBLOCKr� r� r& r
rZ r7 )
r r8 r rD Z old_handlerr0 Zold_maskr� r r� r r r �
check_sigwait� s*
�
�z SignalEINTRTest.check_sigwait� sigwaitinfozneed signal.sigwaitinfo()c C � dd� }| � |� d S )Nc S s t �| g� d S r )r r� �r r r r r8 � r> z3SignalEINTRTest.test_sigwaitinfo.<locals>.wait_func�r� �r r8 r r r �test_sigwaitinfo� � z SignalEINTRTest.test_sigwaitinfo� sigtimedwaitc C r� )Nc S s t �| gd� d S )Ng ^@)r r� r� r r r r8 � r z4SignalEINTRTest.test_sigtimedwait.<locals>.wait_funcr� r� r r r �test_sigtimedwait� r� z!SignalEINTRTest.test_sigtimedwaitN)
r* r+ r, r� ri rj rk r r� r� r r r r r� { s �
�r� c @ s� e Zd Z dd� 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� rV rW 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� rV rW r� r� Zcontrolr- r# r� )r r� r� r� r r r � test_kqueue� � zSelectEINTRTest.test_kqueue�devpollzneed select.devpollc C sR t �� }| �|j� t�� }|�| jd � t�� | }| �� | � || j� d S r� )
r� r� rV rW r� r� r� r- r# r� r� r r r � test_devpoll� r� z SelectEINTRTest.test_devpollN)r* r+ r, r� ri r� r r� rj rk 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 s8 | � tjtj� d�ddtj d| d| j f�}t�� }| �|�}t |��l t
tjd��Q} t�� | }|dkr?t d | ��z||t j
t jB � ||t j� t�d
� W n ty^ Y nw q/||t j
� t�� | }| �|| j� | �� W d � n1 s�w Y |�� W d � d S 1 s�w Y d S ) NrN zimport fcntl, timezwith open('%s', 'wb') as f:z fcntl.%s(f, fcntl.LOCK_EX)z time.sleep(%s)�wbTg N@z failed to sync child in %.1f secg{�G�z�?)rV r r� r� rX r- r� r� r& r
r� � Exception�fcntl�LOCK_EX�LOCK_NB�LOCK_UNr� �BlockingIOErrorr� r# r7 )r Z lock_funcZ lock_namer0 Z
start_timer �fr� r r r �_lock� s< �
��
�
"�zFNTLEINTRTest._lockZAIXzAIX returns PermissionErrorc C � | � tjd� d S )N�lockf)r� r� r� r r r r �
test_lockf rC zFNTLEINTRTest.test_lockfc C r� )N�flock)r� r� r� r r r r �
test_flock
r zFNTLEINTRTest.test_flockN)
r* r+ r, r� ri r� r� �systemr� r� r r r r r� � s
r� �__main__) �
contextlibr r� r< r� r� r ru r& r r� ri �testr Z test.supportr r �contextmanagerr
rj rk ZTestCaser r/ rl r� r� r� r� r* �mainr r r r �<module> sL
'h 7 �/8) �