o
    b"                     @   s   d Z ddlmZ ddlmZ ddlmZ ddlmZm	Z	 ddl
mZ G dd deZeeG d	d
 d
ZG dd deZG dd deZdS )z"
Tests for L{twisted.test.iosim}.
    )implementer)IPushProducer)Protocol)FakeTransportconnect)TestCasec                   @   (   e Zd ZdZdd Zdd Zdd ZdS )	FakeTransportTestsz%
    Tests for L{FakeTransport}.
    c                 C   sH   t t d}t t d}| |jt | |jt | |j|j dS )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 dS )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 dS )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 Zd ZdZdZdd Zdd Zdd Zd	S )
StrictPushProducerz
    An L{IPushProducer} implementation which produces nothing but enforces
    preconditions on its state transition methods.
    runningc                 C   s   | j dkr	tdd| _ d S )Nstoppedz)Cannot stop already-stopped IPushProducer_state
ValueErrorr   r   r   r   stopProducingC   s   

z StrictPushProducer.stopProducingc                 C   &   | j dkrtd| j  dd| _ d S )Nr(   zCannot pause  IPushProducerpausedr*   r-   r   r   r   pauseProducingH      

z!StrictPushProducer.pauseProducingc                 C   r/   )Nr1   zCannot 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 Zd ZdZdd Zdd Zdd Zdd	 Zd
d Zdd Z	dd Z
dd Zdd Zdd Zdd Zdd Zdd Zdd Zdd Zd d! Zd"d# Zd$d% Zd&d' Zd(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      |  |jd dS )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 dS )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      |   }|  | | dS )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 dS )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|dd}t  }t|dd}t||||dd}t }||d| }|j|dd |  | d|j dS )	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 dd dS )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   ) %