| Server IP : 209.38.156.173 / Your IP : 216.73.216.122 [ Web Server : Apache/2.4.52 (Ubuntu) System : Linux lakekumayuhotel 5.15.0-136-generic #147-Ubuntu SMP Sat Mar 15 15:53:30 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.1.2-1ubuntu2.22 Disable Function : NONE Domains : 2 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
o
�bo? � @ s� d Z ddlmZ ddlmZ ddlmZmZmZm Z m
Z
m Z m Z m
Z
mZ ddlmZmZ ddlmZmZmZmZ ddlmZ ddlmZ G d d
� d
e�ZG d d � d e�Z G d
d� d�Z G dd� de�Z G dd� de�Z dS )z(
Tests for L{twisted.internet.testing}.
� )� verifyObject)� IPv4Address) �IAddress�
IConnector� IConsumer�IListeningPort�
IPushProducer� IReactorSSL� IReactorTCP� IReactorUNIX�
ITransport)�
ClientFactory�Factory)�
MemoryReactor�NonStreamingProducer�RaisingMemoryReactor�StringTransport)�namedAny)�TestCasec @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd d � Zd d � Zd d!� Zd"d#� Zd$d%� Zd&d'� Zd(S ))�StringTransportTestsz@
Tests for L{twisted.internet.testing.StringTransport}.
c C s t � | _d S �N)r � transport��self� r �D/usr/lib/python3/dist-packages/twisted/internet/test/test_testing.py�setUp& s zStringTransportTests.setUpc C s: | � tt| j�� | � tt| j�� | � tt| j�� dS )zq
L{StringTransport} instances provide L{ITransport}, L{IPushProducer},
and L{IConsumer}.
N)�
assertTruer r r r r r r r r �test_interfaces) s z$StringTransportTests.test_interfacesc C s> t � }t � }| j�||� | �| jj|� | �| jj|� dS )zz
L{StringTransport.registerProducer} records the arguments supplied to
it as instance attributes.
N)�objectr �registerProducer�assertIs�producer� streaming�r r" r# r r r �test_registerProducer2 s
z*StringTransportTests.test_registerProducerc C sL t � }| j�|d� | �t| jjt � d� | �| jj|� | �| jj� dS )zy
L{StringTransport.registerProducer} raises L{RuntimeError} if a
producer is already registered.
TFN) r r r � assertRaises� RuntimeErrorr! r" r r# )r r" r r r � test_disallowedRegisterProducer= s �z4StringTransportTests.test_disallowedRegisterProducerc C sb t � }t � }| j�|d� | j�� | �| jj� | j�|d� | �| jj|� | �| jj� dS )z�
L{StringTransport.unregisterProducer} causes the transport to forget
about the registered producer and makes it possible to register a new
one.
FTN) r r r �unregisterProducer� assertIsNoner" r! r r# )r � oldProducer� newProducerr r r �test_unregisterProducerJ s
z,StringTransportTests.test_unregisterProducerc C s | � t| jj� dS )z�
L{StringTransport.unregisterProducer} raises L{RuntimeError} if called
when no producer is registered.
N)r& r' r r) r r r r � test_invalidUnregisterProducerY s z3StringTransportTests.test_invalidUnregisterProducerc C s | � | jjd� dS )zO
L{StringTransport.producerState} is initially C{'producing'}.
� producingN)� assertEqualr �
producerStater r r r �test_initialProducerState` s z.StringTransportTests.test_initialProducerStatec C � | j �� | �| j jd� dS )zy
L{StringTransport.pauseProducing} changes the C{producerState} of the
transport to C{'paused'}.
�pausedN)r �pauseProducingr0 r1 r r r r �test_pauseProducingf �
z(StringTransportTests.test_pauseProducingc C s( | j �� | j �� | �| j jd� dS )z}
L{StringTransport.resumeProducing} changes the C{producerState} of the
transport to C{'producing'}.
r/ N)r r5 �resumeProducingr0 r1 r r r r �test_resumeProducingn s
z)StringTransportTests.test_resumeProducingc C r3 )z{
L{StringTransport.stopProducing} changes the C{'producerState'} of the
transport to C{'stopped'}.
�stoppedN)r �
stopProducingr0 r1 r r r r �test_stopProducingw r7 z'StringTransportTests.test_stopProducingc C � | j �� | �t| j j� dS )zu
L{StringTransport.pauseProducing} raises L{RuntimeError} if the
transport has been stopped.
N)r r; r& r' r5 r r r r � test_stoppedTransportCannotPause r7 z5StringTransportTests.test_stoppedTransportCannotPausec C r= )zv
L{StringTransport.resumeProducing} raises L{RuntimeError} if the
transport has been stopped.
N)r r; r& r' r8 r r r r �!test_stoppedTransportCannotResume� r7 z6StringTransportTests.test_stoppedTransportCannotResumec C r= )zz
L{StringTransport.pauseProducing} raises L{RuntimeError} if the
transport is being disconnected.
N)r �loseConnectionr& r' r5 r r r r �&test_disconnectingTransportCannotPause� r7 z;StringTransportTests.test_disconnectingTransportCannotPausec C r= )z{
L{StringTransport.resumeProducing} raises L{RuntimeError} if the
transport is being disconnected.
N)r r@ r& r' r8 r r r r �'test_disconnectingTransportCannotResume� r7 z<StringTransportTests.test_disconnectingTransportCannotResumec C s* | � | jj� | j�� | �| jj� dS )zv
L{StringTransport.loseConnection} toggles the C{disconnecting} instance
variable to C{True}.
N)� assertFalser �
disconnectingr@ r r r r r �$test_loseConnectionSetsDisconnecting� s
z9StringTransportTests.test_loseConnectionSetsDisconnectingc C s t � }| �t|��� |� dS )z�
If a host address is passed to L{StringTransport.__init__}, that
value is returned from L{StringTransport.getHost}.
N)r r! r �getHost�r �addressr r r �test_specifiedHostAddress� s z.StringTransportTests.test_specifiedHostAddressc C s t � }| �t|d��� |� dS )z�
If a peer address is passed to L{StringTransport.__init__}, that
value is returned from L{StringTransport.getPeer}.
)� peerAddressN)r r! r �getPeerrG r r r �test_specifiedPeerAddress� s z.StringTransportTests.test_specifiedPeerAddressc C � t � �� }| �|t� dS )z�
If no host address is passed to L{StringTransport.__init__}, an
L{IPv4Address} is returned from L{StringTransport.getHost}.
N)r rF �assertIsInstancer rG r r r �test_defaultHostAddress� �
z,StringTransportTests.test_defaultHostAddressc C rM )z�
If no peer address is passed to L{StringTransport.__init__}, an
L{IPv4Address} is returned from L{StringTransport.getPeer}.
N)r rK rN r rG r r r �test_defaultPeerAddress� rP z,StringTransportTests.test_defaultPeerAddressN)�__name__�
__module__� __qualname__�__doc__r r r% r( r- r. r2 r6 r9 r< r> r? rA rB rE rI rL rO rQ r r r r r ! s*
r c @ s@ e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dS )� ReactorTestszA
Tests for L{MemoryReactor} and L{RaisingMemoryReactor}.
c C �( t � }tt|� tt|� tt|� dS )zt
L{MemoryReactor} provides all of the attributes described by the
interfaces it advertises.
N)r r r
r r )r �
memoryReactorr r r �test_memoryReactorProvides� �
z'ReactorTests.test_memoryReactorProvidesc C rW )z{
L{RaisingMemoryReactor} provides all of the attributes described by the
interfaces it advertises.
N)r r r
r r )r �raisingReactorr r r �test_raisingReactorProvides� rZ z(ReactorTests.test_raisingReactorProvidesc C s� t � }|�ddt� �|�ddt� d�fD ] }tt|� |�� }tt|� | �|j d� | �|j
d� q|� dt� �}tt|� |�� }tt|� | �|j d� dS )a
L{MemoryReactor.connectTCP}, L{MemoryReactor.connectSSL}, and
L{MemoryReactor.connectUNIX} will return an L{IConnector} whose
C{getDestination} method returns an L{IAddress} with attributes which
reflect the values passed.
ztest.example.comi� Ns
/fake/path)
r �
connectTCPr
�
connectSSLr r �getDestinationr r0 �host�port� connectUNIX�name)r rX � connectorrH r r r �test_connectDestination� s �
z$ReactorTests.test_connectDestinationc C s� t � }|�dt� �|�dt� d�fD ] }tt|� |�� }tt|� | �|j d� | �|j
d� q|� dt� �}tt|� |�� }tt|� | �|j d� dS )a�
L{MemoryReactor.listenTCP}, L{MemoryReactor.listenSSL} and
L{MemoryReactor.listenUNIX} will return an L{IListeningPort} whose
C{getHost} method returns an L{IAddress}; C{listenTCP} and C{listenSSL}
will have a default host of C{'0.0.0.0'}, and a port that reflects the
value passed, and C{listenUNIX} will have a name that reflects the path
passed.
i2 Nz0.0.0.0s /path/to/socket)
r � listenTCPr � listenSSLr r rF r r0 r` ra �
listenUNIXrc )r rX ra rH r r r �test_listenDefaultHost� s �
z#ReactorTests.test_listenDefaultHostc C �P t � }t� }|�|� |�|� | �|�� |g� |�|� | �|�� g � dS )z>
Adding, removing, and listing readers works.
N)r r � addReaderr0 �
getReaders� removeReader)r �reader�reactorr r r � test_readers �
zReactorTests.test_readersc C rj )z>
Adding, removing, and listing writers works.
N)r r � addWriterr0 �
getWriters� removeWriter)r �writerro r r r � test_writers" rq zReactorTests.test_writersN)
rR rS rT rU rY r\ re ri rp rv r r r r rV � s
rV c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) � TestConsumerzP
A very basic test consumer for use with the NonStreamingProducerTests.
c C s g | _ d | _d | _d S r )�writesr" �producerStreamingr r r r �__init__8 s
zTestConsumer.__init__c C s || _ || _dS )z�
Registers a single producer with this consumer. Just keeps track of it.
@param producer: The producer to register.
@param streaming: Whether the producer is a streaming one or not.
N�r" ry r$ r r r r = s
z TestConsumer.registerProducerc C s d| _ d| _dS )zC
Forget the producer we had previously registered.
Nr{ r r r r r) G s
z TestConsumer.unregisterProducerc C s | j �|� dS )zz
Some data was written to the consumer: stores it for later use.
@param data: The data to write.
N)rx �append)r �datar r r �writeN s zTestConsumer.writeN)rR rS rT rU rz r r) r~ r r r r rw 3 s
rw c @ s e Zd ZdZdd� Zdd� ZdS )�NonStreamingProducerTestszF
Tests for the L{NonStreamingProducer} to validate behaviour.
c C s� t � }t|�}|�|d� | �|j|� | �|j|� | �|j� td�D ]}|� � q%g d�}| �
|j� | �
|j� | �
|j� | � |j |� | �
t|j � dS )z�
When the L{NonStreamingProducer} has resumeProducing called 10 times,
it writes the counter each time and then fails.
F�
)
� 0� 1� 2� 3� 4� 5� 6� 7� 8� 9N)rw r r r! r" �consumerrC ry �ranger8 r* r0 rx r& r' )r r� r" �_�expectedWritesr r r �test_producesOnly10Times\ s
z2NonStreamingProducerTests.test_producesOnly10Timesc C s4 t � }t|�}|�|d� |�� | �t|j� dS )zb
When the L{NonStreamingProducer} is paused, it raises a
L{RuntimeError}.
FN)rw r r r8 r& r' r5 )r r� r" r r r �test_cannotPauseProductionv s
z4NonStreamingProducerTests.test_cannotPauseProductionN)rR rS rT rU r� r� r r r r r W s r c @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd d � Zd d � Zd S )!�DeprecationTestsz8
Deprecations in L{twisted.test.proto_helpers}.
c C sd d|j � �}| �|g�}| �t|d d � | �dt|�� | �||d d � | �|t|�� d S )Nztwisted.internet.testing.r �category� �message)rR �
flushWarningsr0 �DeprecationWarning�len�assertInr! r )r �test�obj�new_path�warningsr r r �helper� s zDeprecationTests.helperc C � ddl m} | �| j|� d S )Nr )�AccumulatingProtocol)�twisted.test.proto_helpersr� r� �test_accumulatingProtocol)r r� r r r r� � � z*DeprecationTests.test_accumulatingProtocolc C r� )Nr )�LineSendingProtocol)r� r� r� �test_lineSendingProtocol)r r� r r r r� � r� z)DeprecationTests.test_lineSendingProtocolc C r� )Nr )�FakeDatagramTransport)r� r� r� �test_fakeDatagramTransport)r r� r r r r� � r� z+DeprecationTests.test_fakeDatagramTransportc C r� )Nr )r )r� r r� �test_stringTransport)r r r r r r� � r� z%DeprecationTests.test_stringTransportc C r� )Nr )� StringTransportWithDisconnection)r� r� r� �%test_stringTransportWithDisconnection)r r� r r r r� � s �z6DeprecationTests.test_stringTransportWithDisconnectionc C r� )Nr )�StringIOWithoutClosing)r� r� r� �test_stringIOWithoutClosing)r r� r r r r� � r� z,DeprecationTests.test_stringIOWithoutClosingc C r� )Nr )�_FakeConnector)r� r� r� �test__fakeConnector)r r� r r r r� � r� z$DeprecationTests.test__fakeConnectorc C r� )Nr )� _FakePort)r� r� r� �test__fakePort)r r� r r r r� � r� z DeprecationTests.test__fakePortc C r� )Nr )r )r� r r� �test_memoryReactor)r r r r r r� � r� z#DeprecationTests.test_memoryReactorc C r� )Nr )�MemoryReactorClock)r� r� r� �test_memoryReactorClock)r r� r r r r� � r� z(DeprecationTests.test_memoryReactorClockc C r� )Nr )r )r� r r� �test_raisingMemoryReactor)r r r r r r� � r� z*DeprecationTests.test_raisingMemoryReactorc C r� )Nr )r )r� r r� �test_nonStreamingProducer)r r r r r r� � r� z*DeprecationTests.test_nonStreamingProducerc C r� )Nr )�waitUntilAllDisconnected)r� r� r� � test_waitUntilAllDisconnected)r r� r r r r� � r� z.DeprecationTests.test_waitUntilAllDisconnectedc C r� )Nr )�EventLoggingObserver)r� r� r� �test_eventLoggingObserver)r r� r r r r� � r� z*DeprecationTests.test_eventLoggingObserverN)rR rS rT rU r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r r r r r� � s" r� N) rU �zope.interface.verifyr �twisted.internet.addressr �twisted.internet.interfacesr r r r r r r
r r �twisted.internet.protocolr
r �twisted.internet.testingr r r r �twisted.python.reflectr �twisted.trial.unittestr r rV rw r r� r r r r �<module> s , )j$.