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