File: //usr/lib/python3/dist-packages/twisted/test/__pycache__/test_fdesc.cpython-38.pyc
U
��W[� � @ s� d Z ddlZddlZddlZz ddlZW n ek
r@ dZY nX ddlmZ ddl m
Z
ddl m Z G dd� de j
�ZG d d
� d
e j
�ZG d d � d e j
�ZdS )
z&
Tests for L{twisted.internet.fdesc}.
� Nz not supported on this platform)�fdesc)�untilConcludes)�unittestc @ s e Zd ZdZdd� Zdd� ZdS )�NonBlockingTestszE
Tests for L{fdesc.setNonBlocking} and L{fdesc.setBlocking}.
c C sj t �� \}}| �t j|� | �t j|� | �t�|tj�t j@ � t� |� | �
t�|tj�t j@ � dS )zR
L{fdesc.setNonBlocking} sets a file description to non-blocking.
N) �os�pipe�
addCleanup�close� assertFalse�fcntl�F_GETFL�
O_NONBLOCKr �setNonBlocking�
assertTrue��self�r�w� r �9/usr/lib/python3/dist-packages/twisted/test/test_fdesc.py�test_setNonBlocking s
z$NonBlockingTests.test_setNonBlockingc C sZ t �� \}}| �t j|� | �t j|� t�|� t�|� | �t�|tj �t j
@ � dS )zK
L{fdesc.setBlocking} sets a file description to blocking.
N) r r r r r r Z setBlockingr
r r r
r r r r �test_setBlocking( s
z!NonBlockingTests.test_setBlockingN)�__name__�
__module__� __qualname__�__doc__r r r r r r r s r c @ sp e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� ZdS )�ReadWriteTestsz<
Tests for L{fdesc.readFromFD}, L{fdesc.writeToFD}.
c C s, t �� \| _| _t�| j� t�| j� dS )zG
Create a non-blocking pipe that can be used in tests.
N)r r r r r r �r r r r �setUp: s zReadWriteTests.setUpc C sP zt �| j� W n tk
r$ Y nX zt �| j� W n tk
rJ Y nX dS )z
Close pipes.
N)r r r �OSErrorr r r r r �tearDownC s zReadWriteTests.tearDownc C s t �| j|�S )z)
Write data to the pipe.
)r Z writeToFDr )r �dr r r �writeQ s zReadWriteTests.writec C s6 g }t �| j|j�}|dkr.|r(|d S dS n|S dS )z*
Read data from the pipe.
Nr � )r �
readFromFDr �append)r �l�resr r r �readX s zReadWriteTests.readc C sH | � d�}| �|dk� | �� }| �t|�|� | �dd|� |� dS )z�
Test that the number of bytes L{fdesc.writeToFD} reports as written
with its return value are seen by L{fdesc.readFromFD}.
s hellor N)r" r r( � assertEqual�len)r �n�sr r r �test_writeAndReadg s
z ReadWriteTests.test_writeAndReadc C s� dd }| � |�}| �|dk� g }d}d}||k s<|dk rd|�| �� � |t|d �7 }|d7 }q,d�|�}| �t|�|� | �|d|� |� dS ) z}
Similar to L{test_writeAndRead}, but use a much larger string to verify
the behavior for that case.
s
0123456879i' r �2 ���� r# N)r" r r% r( r* �joinr) )r ZorigZwritten�resultZ resultlength�ir r r �test_writeAndReadLarges s
z%ReadWriteTests.test_writeAndReadLargec C s. g }t �| j|j�}| �|g � | �|� dS )z�
Verify that reading from a file descriptor with no data does not raise
an exception and does not result in the callback function being called.
N)r r$ r r% r) Z assertIsNone)r r& r2 r r r �test_readFromEmpty� s z!ReadWriteTests.test_readFromEmptyc C s" t �| j� | �| �� tj� dS )z�
Test that using L{fdesc.readFromFD} on a cleanly closed file descriptor
returns a connection done indicator.
N)r r r r) r( r ZCONNECTION_DONEr r r r �test_readFromCleanClose� s z&ReadWriteTests.test_readFromCleanClosec C s$ t �| j� | �| �d�tj� dS )z�
Verify that writing with L{fdesc.writeToFD} when the read end is closed
results in a connection lost indicator.
� sN)r r r r) r" r �CONNECTION_LOSTr r r r �test_writeToClosed� s z!ReadWriteTests.test_writeToClosedc C s" t �| j� | �| �� tj� dS )z�
Verify that reading with L{fdesc.readFromFD} when the read end is
closed results in a connection lost indicator.
N)r r r r) r( r r8 r r r r �test_readFromInvalid� s z#ReadWriteTests.test_readFromInvalidc C s$ t �| j� | �| �d�tj� dS )z�
Verify that writing with L{fdesc.writeToFD} when the write end is
closed results in a connection lost indicator.
r7 N)r r r r) r" r r8 r r r r �test_writeToInvalid� s z"ReadWriteTests.test_writeToInvalidc C sf t j}dd� }|t _z| �| �d�d� W 5 |t _X dd� }|t _z| �| �d�d� W 5 |t _X dS )z8
Test error path for L{fdesc.writeTod}.
c S s t � }tj|_|�d S �N)r �errnoZEAGAIN��fd�data�errr r r � eagainWrite� s z4ReadWriteTests.test_writeErrors.<locals>.eagainWriter7 r c S s t � }tj|_|�d S r< )r r= ZEINTRr> r r r �
eintrWrite� s z3ReadWriteTests.test_writeErrors.<locals>.eintrWriteN)r r" r) )r Z
oldOsWriterB rC r r r �test_writeErrors� s z ReadWriteTests.test_writeErrorsN)r r r r r r r" r( r- r4 r5 r6 r9 r: r; rD r r r r r 5 s r c @ s, e Zd ZdZdZdd� Zdd� Zdd� Zd S )
�CloseOnExecTestszL
Tests for L{fdesc._setCloseOnExec} and L{fdesc._unsetCloseOnExec}.
z�
import os, errno
try:
os.write(%d, b'lul')
except OSError as e:
if e.errno == errno.EBADF:
os._exit(0)
os._exit(5)
except:
os._exit(10)
else:
os._exit(20)
c C sv t �� }|dkr`z&t �tjtjd| j|�� f g� W qr dd l}|�� t � d� Y qrX nt
t j |d�d S d S )Nr z-c� r0 ) r �fork�execv�sys�
executable�program�fileno� traceback� print_exc�_exitr �waitpid)r �fObj�pidrM r r r �_execWithFileDescriptor� s &z(CloseOnExecTests._execWithFileDescriptorc C sX t | �� d��@}t�|�� � | �|�}| �t�|�� | � t�
|�d� W 5 Q R X dS )z�
A file descriptor passed to L{fdesc._setCloseOnExec} is not inherited
by a new process image created with one of the exec family of
functions.
�wbr N) �open�mktempr �_setCloseOnExecrL rS r r � WIFEXITEDr) � WEXITSTATUS�r rQ Zstatusr r r �test_setCloseOnExec� s
z$CloseOnExecTests.test_setCloseOnExecc C sf t | �� d��N}t�|�� � t�|�� � | �|�}| �t� |�� | �
t� |�d� W 5 Q R X dS )z�
A file descriptor passed to L{fdesc._unsetCloseOnExec} is inherited by
a new process image created with one of the exec family of functions.
rT � N) rU rV r rW rL Z_unsetCloseOnExecrS r r rX r) rY rZ r r r �test_unsetCloseOnExec s
z&CloseOnExecTests.test_unsetCloseOnExecN)r r r r rK rS r[ r] r r r r rE � s
rE )r r rI r= r � ImportError�skipZtwisted.internetr Ztwisted.python.utilr Z
twisted.trialr ZSynchronousTestCaser r rE r r r r �<module> s