← Back
Editing: 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
Save File
Cancel