File: //usr/lib/python3/dist-packages/twisted/test/__pycache__/test_stringtransport.cpython-38.pyc
U
W[�3 � @ s� d Z ddlmZ ddlmZmZmZmZmZm Z m
Z
m Z m Z ddl
mZ ddlmZ ddlmZmZmZmZ ddlmZmZ G dd � d e�ZG d
d � d e�ZG d d
� d
e�Z G dd� de�Z dS )z*
Tests for L{twisted.test.proto_helpers}.
� )� verifyObject) �
ITransport�
IPushProducer� IConsumer� IReactorTCP� IReactorSSL� IReactorUNIX�IAddress�IListeningPort�
IConnector)� IPv4Address)�TestCase)�StringTransport�
MemoryReactor�RaisingMemoryReactor�NonStreamingProducer)�
ClientFactory�Factoryc @ 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 ))�StringTransportTestszB
Tests for L{twisted.test.proto_helpers.StringTransport}.
c C s t � | _d S �N)r � transport��self� r �C/usr/lib/python3/dist-packages/twisted/test/test_stringtransport.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_registerProducer& 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_disallowedRegisterProducer2 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 Z oldProducerZ newProducerr r r �test_unregisterProducer? 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_invalidUnregisterProducerO s z3StringTransportTests.test_invalidUnregisterProducerc C s | � | jjd� dS )zO
L{StringTransport.producerState} is initially C{'producing'}.
� producingN)� assertEqualr �
producerStater r r r �test_initialProducerStateW s z.StringTransportTests.test_initialProducerStatec C s | j �� | �| j jd� dS )zy
L{StringTransport.pauseProducing} changes the C{producerState} of the
transport to C{'paused'}.
ZpausedN)r �pauseProducingr- r. r r r r �test_pauseProducing^ s
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 r0 �resumeProducingr- r. r r r r �test_resumeProducingg s
z)StringTransportTests.test_resumeProducingc C s | j �� | �| j jd� dS )z{
L{StringTransport.stopProducing} changes the C{'producerState'} of the
transport to C{'stopped'}.
ZstoppedN)r �
stopProducingr- r. r r r r �test_stopProducingq s
z'StringTransportTests.test_stopProducingc C s | j �� | �t| j j� dS )zu
L{StringTransport.pauseProducing} raises L{RuntimeError} if the
transport has been stopped.
N)r r4 r% r& r0 r r r r � test_stoppedTransportCannotPausez s
z5StringTransportTests.test_stoppedTransportCannotPausec C s | j �� | �t| j j� dS )zv
L{StringTransport.resumeProducing} raises L{RuntimeError} if the
transport has been stopped.
N)r r4 r% r& r2 r r r r �!test_stoppedTransportCannotResume� s
z6StringTransportTests.test_stoppedTransportCannotResumec C s | j �� | �t| j j� dS )zz
L{StringTransport.pauseProducing} raises L{RuntimeError} if the
transport is being disconnected.
N)r �loseConnectionr% r& r0 r r r r �&test_disconnectingTransportCannotPause� s
z;StringTransportTests.test_disconnectingTransportCannotPausec C s | j �� | �t| j j� dS )z{
L{StringTransport.resumeProducing} raises L{RuntimeError} if the
transport is being disconnected.
N)r r8 r% r& r2 r r r r �'test_disconnectingTransportCannotResume� s
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 Z
disconnectingr8 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}.
)Z peerAddressN)r r r �getPeerr> r r r �test_specifiedPeerAddress� s
�z.StringTransportTests.test_specifiedPeerAddressc C s 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 r= �assertIsInstancer r> r r r �test_defaultHostAddress� s
z,StringTransportTests.test_defaultHostAddressc C s t � �� }| �|t� dS )z�
If no peer address is passed to L{StringTransport.__init__}, an
L{IPv4Address} is returned from L{StringTransport.getPeer}.
N)r rA rC r r> r r r �test_defaultPeerAddress� s
z,StringTransportTests.test_defaultPeerAddressN)�__name__�
__module__� __qualname__�__doc__r r r$ r' r* r+ r/ r1 r3 r5 r6 r7 r9 r: r< r@ rB rD rE 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 s( 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� s
z'ReactorTests.test_memoryReactorProvidesc C s( t � }tt|� tt|� tt|� dS )z{
L{RaisingMemoryReactor} provides all of the attributes described by the
interfaces it advertises.
N)r r r r r )r ZraisingReactorr r r �test_raisingReactorProvides� s
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 Z
connectTCPr Z
connectSSLr r ZgetDestinationr r- �host�portZ connectUNIX�name)r rK Z connectorr? 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 Z listenTCPr Z listenSSLr r
r= r r- rN rO Z
listenUNIXrP )r rK rO r? r r r �test_listenDefaultHost s �
z#ReactorTests.test_listenDefaultHostc C sP t � }t� }|�|� |�|� | �|�� |g� |�|� | �|�� g � dS )z>
Adding, removing, and listing readers works.
N)r r Z addReaderr- Z
getReadersZ removeReader)r �reader�reactorr r r � test_readers s
zReactorTests.test_readersc C sP t � }t� }|�|� |�|� | �|�� |g� |�|� | �|�� g � dS )z>
Adding, removing, and listing writers works.
N)r r Z addWriterr- Z
getWritersZ removeWriter)r �writerrT r r r � test_writers+ s
zReactorTests.test_writersN)
rF rG rH rI rL rM rQ rR rU rW r r r r rJ � s rJ 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__A 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! rZ r# r r r r G s z TestConsumer.registerProducerc C s d| _ d| _dS )zC
Forget the producer we had previously registered.
Nr\ r r r r r( R 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)rY �append)r �datar r r �writeZ s zTestConsumer.writeN)rF rG rH rI r[ r r( r_ r r r r rX = s
rX 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 ] }|� � qJddddddd d
d d g
}| �
|j� | �
|j� | �
|j� | � |j |� | �
t|j � d
S )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)rX r r r r! �consumerr; rZ �ranger2 r) r- rY r% r& )r rl r! �_ZexpectedWritesr r r �test_producesOnly10Timesh s0
� z2NonStreamingProducerTests.test_producesOnly10Timesc C s4 t � }t|�}|�|d� |�� | �t|j� dS )zb
When the L{NonStreamingProducer} is paused, it raises a
L{RuntimeError}.
FN)rX r r r2 r% r& r0 )r rl r! r r r �test_cannotPauseProduction� s
z4NonStreamingProducerTests.test_cannotPauseProductionN)rF rG rH rI ro rp r r r r r` d s r` N) rI Zzope.interface.verifyr Ztwisted.internet.interfacesr r r r r r r r
r Ztwisted.internet.addressr Ztwisted.trial.unittestr
Ztwisted.test.proto_helpersr r r r Ztwisted.internet.protocolr r r rJ r rX r` r r r r �<module> s , ;o'