File manager - Edit - /usr/lib/python3/dist-packages/twisted/test/__pycache__/test_iosim.cpython-310.pyc
Back
o �b�" � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZm Z ddl m Z G dd� de �Z ee�G d d � d ��Z G d d � d e �ZG d d� de �ZdS )z" Tests for L{twisted.test.iosim}. � )� implementer)� IPushProducer)�Protocol)� FakeTransport�connect)�TestCasec @ �( e Zd ZdZdd� Zdd� Zdd� ZdS ) �FakeTransportTestsz% Tests for L{FakeTransport}. c C sH t t� d�}t t� d�}| �|jt� | �|jt� | �|j|j� dS )ze Each L{FakeTransport} receives a serial number that uniquely identifies it. TFN)r �object�assertIsInstance�serial�int�assertNotEqual)�self�a�b� r �9/usr/lib/python3/dist-packages/twisted/test/test_iosim.py�test_connectionSerial s z(FakeTransportTests.test_connectionSerialc C s< t t� d�}|�d� |�g d�� | �d�|j�d� dS )zl L{FakeTransport.writeSequence} will write a sequence of L{bytes} to the transport. F� a)� b� c� d� s abcdN)r r �write� writeSequence� assertEqual�join�stream�r r r r r �test_writeSequence! s z%FakeTransportTests.test_writeSequencec C s@ t t� d�}|�d� |�� |�d� | �d�|j�d� dS )z� L{FakeTransport.write} will accept writes after transport was closed, but the data will be silently discarded. Fs befores afterr N)r r r �loseConnectionr r r r r r r �test_writeAfterClose- s z'FakeTransportTests.test_writeAfterCloseN)�__name__� __module__� __qualname__�__doc__r r r" r r r r r s r c @ s, e Zd ZdZdZdd� Zdd� Zdd� Zd S ) �StrictPushProducerz� An L{IPushProducer} implementation which produces nothing but enforces preconditions on its state transition methods. �runningc C s | j dkr td��d| _ d S )N�stoppedz)Cannot stop already-stopped IPushProducer��_state� ValueError�r r r r � stopProducingC s z StrictPushProducer.stopProducingc C �& | j dkrtd| j � d���d| _ d S )Nr( z Cannot pause � IPushProducer�pausedr* r- r r r �pauseProducingH � z!StrictPushProducer.pauseProducingc C r/ )Nr1 zCannot resume r0 r( r* r- r r r �resumeProducingM r3 z"StrictPushProducer.resumeProducingN)r# r$ r% r& r+ r. r2 r4 r r r r r' : 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 d!� Zd"d#� Zd$d%� Zd&d'� Zd(S ))�StrictPushProducerTestsz* Tests for L{StrictPushProducer}. c C s t � S )zp @return: A new L{StrictPushProducer} which has not been through any state changes. )r' r- r r r �_initialX s z StrictPushProducerTests._initialc C � t � }|�� |S )z@ @return: A new, stopped L{StrictPushProducer}. )r' r. �r �producerr r r �_stopped_ � z StrictPushProducerTests._stoppedc C r7 )z? @return: A new, paused L{StrictPushProducer}. )r' r2 r8 r r r �_pausedg r; z StrictPushProducerTests._pausedc C s t � }|�� |�� |S )zY @return: A new L{StrictPushProducer} which has been paused and resumed. )r' r2 r4 r8 r r r �_resumedo s z StrictPushProducerTests._resumedc C � | � |jd� dS )z� Assert that the given producer is in the stopped state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r) N�r r+ r8 r r r � assertStoppedx � z%StrictPushProducerTests.assertStoppedc C r> )z� Assert that the given producer is in the paused state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r1 Nr? r8 r r r � assertPaused� rA z$StrictPushProducerTests.assertPausedc C r> )z� Assert that the given producer is in the running state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r( Nr? r8 r r r � assertRunning� rA z%StrictPushProducerTests.assertRunningc C � | � t| �� j� dS )zz L{StrictPushProducer.stopProducing} raises L{ValueError} if called when the producer is stopped. N)� assertRaisesr, r: r. r- r r r �test_stopThenStop� � z)StrictPushProducerTests.test_stopThenStopc C rD )z{ L{StrictPushProducer.pauseProducing} raises L{ValueError} if called when the producer is stopped. N)rE r, r: r2 r- r r r �test_stopThenPause� rG z*StrictPushProducerTests.test_stopThenPausec C rD )z| L{StrictPushProducer.resumeProducing} raises L{ValueError} if called when the producer is stopped. N)rE r, r: r4 r- r r r �test_stopThenResume� rG z+StrictPushProducerTests.test_stopThenResumec C � | � � }|�� | �|� dS )zn L{StrictPushProducer} is stopped if C{stopProducing} is called on a paused producer. N)r< r. r@ r8 r r r �test_pauseThenStop� � z*StrictPushProducerTests.test_pauseThenStopc C � | � � }| �t|j� dS )zs L{StrictPushProducer.pauseProducing} raises L{ValueError} if called on a paused producer. N)r< rE r, r2 r8 r r r �test_pauseThenPause� � z+StrictPushProducerTests.test_pauseThenPausec C rJ )zp L{StrictPushProducer} is resumed if C{resumeProducing} is called on a paused producer. N)r< r4 rC r8 r r r �test_pauseThenResume� rL z,StrictPushProducerTests.test_pauseThenResumec C rJ )zo L{StrictPushProducer} is stopped if C{stopProducing} is called on a resumed producer. N)r= r. r@ r8 r r r �test_resumeThenStop� rL z+StrictPushProducerTests.test_resumeThenStopc C rJ )zo L{StrictPushProducer} is paused if C{pauseProducing} is called on a resumed producer. N)r= r2 rB r8 r r r �test_resumeThenPause� rL z,StrictPushProducerTests.test_resumeThenPausec C rM )zu L{StrictPushProducer.resumeProducing} raises L{ValueError} if called on a resumed producer. N)r= rE r, r4 r8 r r r �test_resumeThenResume� rO z-StrictPushProducerTests.test_resumeThenResumec C rJ )zn L{StrictPushProducer} is stopped if C{stopProducing} is called in the initial state. N)r6 r. r@ r8 r r r � test_stop� rL z!StrictPushProducerTests.test_stopc C rJ )zn L{StrictPushProducer} is paused if C{pauseProducing} is called in the initial state. N)r6 r2 rB r8 r r r � test_pause� rL z"StrictPushProducerTests.test_pausec C rM )zz L{StrictPushProducer} raises L{ValueError} if C{resumeProducing} is called in the initial state. N)r6 rE r, r4 r8 r r r � test_resume� rO z#StrictPushProducerTests.test_resumeN)r# r$ r% r& r6 r: r< r= r@ rB rC rF rH rI rK rN rP rQ rR rS rT rU rV r r r r r5 S s* r5 c @ r ) � IOPumpTestsz Tests for L{IOPump}. c C sr t � }t|dd�}t � }t|dd�}t||||dd�}t� }||d�| }|j|dd� |�� | �d|j� dS ) a� Connect a couple protocol/transport pairs to an L{IOPump} and then pump it. Verify that a streaming producer registered with one of the transports does not receive invalid L{IPushProducer} method calls and ends in the right state. @param mode: C{u"server"} to test a producer registered with the server transport. C{u"client"} to test a producer registered with the client transport. T)�isServerF)�greet)�server�client)� streamingr( N)r r r r' �registerProducer�pumpr r+ ) r �mode� serverProto�serverTransport� clientProto�clientTransportr^ r9 �victimr r r �_testStreamingProducer� s( ���z"IOPumpTests._testStreamingProducerc C � | j dd� dS )z� L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer} (stream producer) registered with the server transport. rZ �r_ N�re r- r r r � test_serverStreamingProducer � z(IOPumpTests.test_serverStreamingProducerc C rf )z� L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer} (stream producer) registered with the client transport. r[ rg Nrh r- r r r � test_clientStreamingProducer& rj z(IOPumpTests.test_clientStreamingProducerN)r# r$ r% r& re ri rk r r r r rW � s # rW N)r& �zope.interfacer �twisted.internet.interfacesr �twisted.internet.protocolr �twisted.test.iosimr r �twisted.trial.unittestr r r' r5 rW r r r r �<module> s ) %
| ver. 1.4 |
Github
|
.
| PHP 8.2.9 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings