| Server IP : 138.197.176.125 / Your IP : 216.73.217.54 Web Server : Apache/2.4.41 (Ubuntu) System : Linux SuiteCRM-8 5.4.0-216-generic #236-Ubuntu SMP Fri Apr 11 19:53:21 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.3.19 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /proc/self/root/lib/python3/dist-packages/twisted/test/__pycache__/ |
Upload File : |
U
W[d3 � @ s$ d Z ddlmZmZ ddlZddlZddlZddlmZ ddl m
Z
m Z ddl m
Z
ddlmZ ddlmZmZmZ dd lmZmZmZmZmZ dd
lmZ d Z dZ e� � r�e
d �dkr�d
Z e ej �Z!ej"�#ej$�e!d< n"e� Z!ej"�#ej$��%e�&� �e!d< G dd� dej'�Z(G dd� dej)�Z*dS )z�
Tests for L{twisted.internet.stdio}.
@var properEnv: A copy of L{os.environ} which has L{bytes} keys/values on POSIX
platforms and native L{str} keys/values on Windows.
� )�absolute_import�divisionN)�unittest)�filepath�log)�
requireModule)�platform)�range�
intToBytes� bytesEnviron)�error�defer�protocol�stdio�reactor)� ConnectionLostNotifyingProtocols xyz123abc Twisted is great!Z win32processzIOn windows, spawnProcess is not available in the absence of win32process.�
PYTHONPATHs
PYTHONPATHc @ s4 e Zd ZdZdZdd� Zdd� Zdd� Zd d
� ZdS ) � StandardIOTestProcessProtocola�
Test helper for collecting output from a child process and notifying
something when it exits.
@ivar onConnection: A L{defer.Deferred} which will be called back with
L{None} when the connection to the child process is established.
@ivar onCompletion: A L{defer.Deferred} which will be errbacked with the
failure associated with the child process exiting when it exits.
@ivar onDataReceived: A L{defer.Deferred} which will be called back with
this instance whenever C{childDataReceived} is called, or L{None} to
suppress these callbacks.
@ivar data: A C{dict} mapping file descriptors to strings containing all
bytes received from the child process on each file descriptor.
Nc C s t �� | _t �� | _i | _d S �N)r
�Deferred� onConnection� onCompletion�data��self� r �9/usr/lib/python3/dist-packages/twisted/test/test_stdio.py�__init__B s
z&StandardIOTestProcessProtocol.__init__c C s | j �d � d S r )r �callbackr r r r �connectionMadeH s z,StandardIOTestProcessProtocol.connectionMadec C s> | j �|d�| | j |< | jdk r:| jd }| _|�| � dS )z�
Record all bytes received from the child process in the C{data}
dictionary. Fire C{onDataReceived} if it is not L{None}.
� N)r �get�onDataReceivedr )r �name�bytes�dr r r �childDataReceivedL s
z/StandardIOTestProcessProtocol.childDataReceivedc C s | j �|� d S r )r r )r �reasonr r r � processEndedW s z*StandardIOTestProcessProtocol.processEnded) �__name__�
__module__� __qualname__�__doc__r" r r r& r( r r r r r . s r c @ s~ e Zd ZeZdd� Zdd� Zdd� Zdd� Zd d
� Z d d � Z
d
d� Z dd� Z dd� Z
dd� Zdd� Zdd� Ze�� rzde_dS )�StandardInputOutputTestsc O s: t jdd| tjjgt|� }tj|t j|fdti|��S )a_
Launch a child Python process and communicate with it using the
given ProcessProtocol.
@param proto: A L{ProcessProtocol} instance which will be connected
to the child process.
@param sibling: The basename of a file containing the Python program
to run in the child process.
@param *args: strings which will be passed to the child process on
the command line as C{argv[2:]}.
@param **kw: additional arguments to pass to L{reactor.spawnProcess}.
@return: The L{IProcessTransport} provider for the spawned process.
s -ms
twisted.test.�env)�sys�
executabler � __class__r* �listZ spawnProcess� properEnv)r �protoZsibling�args�kwr r r �
_spawnProcess` s �����z&StandardInputOutputTests._spawnProcessc s$ �fdd�}� fdd�}|� ||�S )Nc s � � d| f � d S )Nz'Process terminated with non-Failure: %r)Zfail)�resultr r r �cb~ s z4StandardInputOutputTests._requireFailure.<locals>.cbc s � | �S r r )�err)r r r �eb� s z4StandardInputOutputTests._requireFailure.<locals>.eb)Z addCallbacks)r r% r r9 r; r )r r r �_requireFailure} s z(StandardInputOutputTests._requireFailurec sL �� � � t�d� � t� ��j}���d� � � ��fdd�}��||�S )z�
Verify that a protocol connected to L{StandardIO} can disconnect
itself using C{transport.loseConnection}.
�Child process logging to s stdio_test_loseconnc sP t � d��"}|D ]}t�d|�� � qW 5 Q R X ��d�j� | �tj� d S )N�rzChild logged: � ) �openr �msg�rstripZfailIfInr �trapr � ProcessDone)r' �f�line�� errorLogFile�pr r r r( � s
zBStandardInputOutputTests.test_loseConnection.<locals>.processEnded)�mktempr rA r r r7 r< �r r% r( r rG r �test_loseConnection� s z,StandardInputOutputTests.test_loseConnectionc sf | � � }t�d| � t� � t�� � _� fdd�}� j�|� dd� }| �� j|�}| � � d|� |S )z�
When stdin is closed and the protocol connected to it implements
L{IHalfCloseableProtocol}, the protocol's C{readConnectionLost} method
is called.
r= c s � j }� j�� |S r )r Z transportZ
closeStdin)Zignoredr% �rI r r �cbBytes� s
zAStandardInputOutputTests.test_readConnectionLost.<locals>.cbBytesc S s | � tj� d S r )rC r rD �r' r r r r( � s zFStandardInputOutputTests.test_readConnectionLost.<locals>.processEndeds stdio_test_halfclose)
rJ r rA r r
r r" � addCallbackr< r7 )r rH rN r( r% r rM r �test_readConnectionLost� s
�z0StandardInputOutputTests.test_readConnectionLostc
sj t � � z�j� dtdd� W n0 tk
rL } zt�t|���W 5 d}~X Y nX � �fdd�}��� j|�S )z�
Verify that a write made directly to stdout using L{os.write}
after StandardIO has finished is reliably received by the
process reading that stdout.
s stdio_test_lastwriteT)ZusePTYNc s0 �� � jd �t�d� jf � | �tj� dS )z�
Asserts that the parent received the bytes written by the child
immediately after the child starts.
r? z4Received %r from child, did not find expected bytes.N)�
assertTruer �endswith�UNIQUE_LAST_WRITE_STRINGrC r rD rO �rI r r r r( � s ��zEStandardInputOutputTests.test_lastWriteReceived.<locals>.processEnded) r r7 rT �
ValueErrorr ZSkipTest�strr< r )r �er( r rU r �test_lastWriteReceived� s �
z/StandardInputOutputTests.test_lastWriteReceivedc s2 t � � � j}��� d� � �fdd�}��||�S )z�
Verify that the transport of a protocol connected to L{StandardIO}
has C{getHost} and C{getPeer} methods.
s stdio_test_hostpeerc s6 � j d �� \}}��|� ��|� | �tj� d S )Nr? )r �
splitlinesrR rC r rD )r' ZhostZpeerrU r r r( � s
z?StandardInputOutputTests.test_hostAndPeer.<locals>.processEnded�r r r7 r< rK r rU r �test_hostAndPeer� s
z)StandardInputOutputTests.test_hostAndPeerc s2 t � � � j}��� d� � �fdd�}��||�S )z�
Verify that the C{write} method of the transport of a protocol
connected to L{StandardIO} sends bytes to standard out.
s stdio_test_writec s" �� � jd d� | �tj� d S �Nr? s ok!�� assertEqualr rC r rD rO rU r r r( � s z9StandardInputOutputTests.test_write.<locals>.processEndedr[ rK r rU r �
test_write� s
z#StandardInputOutputTests.test_writec s2 t � � � j}��� d� � �fdd�}��||�S )z�
Verify that the C{writeSequence} method of the transport of a
protocol connected to L{StandardIO} sends bytes to standard out.
s stdio_test_writeseqc s" �� � jd d� | �tj� d S r] r^ rO rU r r r( s zAStandardInputOutputTests.test_writeSequence.<locals>.processEndedr[ rK r rU r �test_writeSequence� s
z+StandardInputOutputTests.test_writeSequencec C sB | � � }t|d��&}td�D ]}|�t|�d � q W 5 Q R X |S )N�wbi �
)rJ r@ r �writer
)r �junkPathZjunkFile�ir r r � _junkPath
s
z"StandardInputOutputTests._junkPathc sd t � ��j}g �ttd���� ���fdd�� ���d���j�� � ����fdd�}��||�S )z�
Verify that the transport of a protocol connected to L{StandardIO}
is a working L{IProducer} provider.
�d c s: �r6�� t��� �d � ���d � t�d� d � d S )Nrc ���g{�G�z�?)�appendr
�poprd r � callLater)Zign)r �proc�toWrite�writtenr r r s z>StandardInputOutputTests.test_producer.<locals>.connectionMades stdio_test_producerc s> �� � jd d���� ���dt��f � | �tj� d S )Nr? r z*Connection lost with %d writes left to go.)r_ r �joinZ assertFalse�lenrC r rD rO )rI r rn ro r r r( * s �z<StandardInputOutputTests.test_producer.<locals>.processEnded)r r r2 r r7 r rP r< rK r )r rI rm r rn ro r �
test_producer s z&StandardInputOutputTests.test_producerc s> t � ��j}��� � ���d� � � ��fdd�}��||�S )z�
Verify that the transport of a protocol connected to L{StandardIO}
is a working L{IConsumer} provider.
s stdio_test_consumerc s< t � d�� }���jd |�� � W 5 Q R X | �tj� d S )N�rbr? )r@ r_ r �readrC r rD )r' rE �re rI r r r r( ? s z<StandardInputOutputTests.test_consumer.<locals>.processEnded)r r rg r7 r< rK r ru r �
test_consumer3 s z&StandardInputOutputTests.test_consumerc s� t �� }t|�}t���� ����d� �_}��|j � t
|� � d�}t �
� s�t�� \}}��tj |� ��tj |� ||d< tj|f|�� d�t�� �� ���fdd��t�d�� ����fdd �}|�|� |S )
aE
If L{StandardIO} is created with a file descriptor which refers to a
normal file (ie, a file from the filesystem), L{StandardIO.write}
writes bytes to that file. In particular, it does not immediately
consider the file closed or call its protocol's C{connectionLost}
method.
rb )�stdout�stdin� c sB �D ],} | �kr � � � d S � �t| �� q2qt�d�� d S )Nr )ZloseConnectionrd r
r rl )�value)�
connection�count�howMany�spinr r r~ e s zAStandardInputOutputTests.test_normalFileStandardOut.<locals>.spinr c s8 �� t� ��d � �� ��� d�ttt����� d S )Nr? r )r_ �nextZ
getContentrp �mapr
r rO )r| r} �pathr r r �cbLostq s
�zCStandardInputOutputTests.test_normalFileStandardOut.<locals>.cbLost)r
r r r ZFilePathrJ r@ �normalZ
addCleanup�close�dict�filenor � isWindows�os�piper Z
StandardIO� itertoolsr| r rl rP )r Z
onConnLostr4 r� �kwargsr> �wr� r )r{ r| r} r� r r~ r �test_normalFileStandardOutF s&
z3StandardInputOutputTests.test_normalFileStandardOutzpStandardIO does not accept stdout as an argument to Windows. Testing redirection to a file is therefore harder.N)r) r* r+ �skipWindowsNopywin32�skipr7 r< rL rQ rY r\ r` ra rg rr rv r� r r� r r r r r- \ s ' 3�r- )+r, Z
__future__r r r� r/ r� Z
twisted.trialr Ztwisted.pythonr r Ztwisted.python.reflectr Ztwisted.python.runtimer Ztwisted.python.compatr r
r Ztwisted.internetr r
r r r Ztwisted.test.test_tcpr rT r� r� r� �environr3 �pathseprp r� �encode�getfilesystemencodingZProcessProtocolr ZTestCaser- r r r r �<module> s0
�.