
    ϪfW&                        d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl	m
Z
 ddlmZ ddlmZmZmZ dd	lmZ  G d
 de      Z ee       G d d             Z G d de      Z G d de      Zy)z"
Tests for L{twisted.test.iosim}.
    )annotations)Literal)implementer)IPushProducer)Protocol)Clock)FakeTransportconnectconnectedServerAndClient)TestCasec                  (    e Zd ZdZddZddZddZy)FakeTransportTestsz%
    Tests for L{FakeTransport}.
    c                    t        t               d      }t        t               d      }| j                  |j                  t               | j                  |j                  t               | j                  |j                  |j                         y)ze
        Each L{FakeTransport} receives a serial number that uniquely identifies
        it.
        TFN)r	   objectassertIsInstanceserialintassertNotEqual)selfabs      9/usr/lib/python3/dist-packages/twisted/test/test_iosim.pytest_connectionSerialz(FakeTransportTests.test_connectionSerial   s`    
 &(D)&(E*ahh,ahh,AHHahh/    c                    t        t               d      }|j                  d       |j                  g d       | j	                  dj                  |j                        d       y)zl
        L{FakeTransport.writeSequence} will write a sequence of L{bytes} to the
        transport.
        F   a)   b   c   dr   s   abcdN)r	   r   writewriteSequenceassertEqualjoinstreamr   r   s     r   test_writeSequencez%FakeTransportTests.test_writeSequence$   sG    
 &(E*		*+!((+W5r   c                    t        t               d      }|j                  d       |j                          |j                  d       | j	                  dj                  |j                        d       y)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%   s     r   test_writeAfterClosez'FakeTransportTests.test_writeAfterClose0   sS    
 &(E*				!((+Y7r   NreturnNone)__name__
__module____qualname____doc__r   r&   r)    r   r   r   r      s    	0
6
8r   r   c                  ,    e Zd ZdZdZddZddZddZy)StrictPushProducerz
    An L{IPushProducer} implementation which produces nothing but enforces
    preconditions on its state transition methods.
    runningc                F    | j                   dk(  rt        d      d| _         y )Nstoppedz)Cannot stop already-stopped IPushProducer_state
ValueErrorr   s    r   stopProducingz StrictPushProducer.stopProducingF   s!    ;;)#HIIr   c                b    | j                   dk7  rt        d| j                    d      d| _         y )Nr4   zCannot pause  IPushProducerpausedr7   r:   s    r   pauseProducingz!StrictPushProducer.pauseProducingK   s-    ;;)#}T[[MHIIr   c                b    | j                   dk7  rt        d| j                    d      d| _         y )Nr>   zCannot resume r=   r4   r7   r:   s    r   resumeProducingz"StrictPushProducer.resumeProducingP   s-    ;;("~dkk].IJJr   Nr*   )r-   r.   r/   r0   r8   r;   r?   rA   r1   r   r   r3   r3   =   s    
 F 

 r   r3   c                      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y)StrictPushProducerTestsz*
    Tests for L{StrictPushProducer}.
    c                    t               S )zp
        @return: A new L{StrictPushProducer} which has not been through any state
            changes.
        )r3   r:   s    r   _initialz StrictPushProducerTests._initial[   s    
 "##r   c                :    t               }|j                          |S )z@
        @return: A new, stopped L{StrictPushProducer}.
        )r3   r;   r   producers     r   _stoppedz StrictPushProducerTests._stoppedb   s     &' r   c                :    t               }|j                          |S )z?
        @return: A new, paused L{StrictPushProducer}.
        )r3   r?   rG   s     r   _pausedzStrictPushProducerTests._pausedj   s     &'!r   c                Z    t               }|j                          |j                          |S )zY
        @return: A new L{StrictPushProducer} which has been paused and resumed.
        )r3   r?   rA   rG   s     r   _resumedz StrictPushProducerTests._resumedr   s)     &'!  "r   c                <    | j                  |j                  d       y)z
        Assert that the given producer is in the stopped state.

        @param producer: The producer to verify.
        @type producer: L{StrictPushProducer}
        r6   Nr"   r8   rG   s     r   assertStoppedz%StrictPushProducerTests.assertStopped{        	)4r   c                <    | j                  |j                  d       y)z
        Assert that the given producer is in the paused state.

        @param producer: The producer to verify.
        @type producer: L{StrictPushProducer}
        r>   NrO   rG   s     r   assertPausedz$StrictPushProducerTests.assertPaused   s     	(3r   c                <    | j                  |j                  d       y)z
        Assert that the given producer is in the running state.

        @param producer: The producer to verify.
        @type producer: L{StrictPushProducer}
        r4   NrO   rG   s     r   assertRunningz%StrictPushProducerTests.assertRunning   rQ   r   c                `    | j                  t        | j                         j                         y)zz
        L{StrictPushProducer.stopProducing} raises L{ValueError} if called when
        the producer is stopped.
        N)assertRaisesr9   rI   r;   r:   s    r   test_stopThenStopz)StrictPushProducerTests.test_stopThenStop   s     
 	*dmmo&C&CDr   c                `    | j                  t        | j                         j                         y)z{
        L{StrictPushProducer.pauseProducing} raises L{ValueError} if called when
        the producer is stopped.
        N)rW   r9   rI   r?   r:   s    r   test_stopThenPausez*StrictPushProducerTests.test_stopThenPause   s     
 	*dmmo&D&DEr   c                `    | j                  t        | j                         j                         y)z|
        L{StrictPushProducer.resumeProducing} raises L{ValueError} if called when
        the producer is stopped.
        N)rW   r9   rI   rA   r:   s    r   test_stopThenResumez+StrictPushProducerTests.test_stopThenResume   s     
 	*dmmo&E&EFr   c                f    | j                         }|j                          | j                  |       y)zn
        L{StrictPushProducer} is stopped if C{stopProducing} is called on a paused
        producer.
        N)rK   r;   rP   rG   s     r   test_pauseThenStopz*StrictPushProducerTests.test_pauseThenStop   s)    
 <<> 8$r   c                d    | j                         }| j                  t        |j                         y)zs
        L{StrictPushProducer.pauseProducing} raises L{ValueError} if called on a
        paused producer.
        N)rK   rW   r9   r?   rG   s     r   test_pauseThenPausez+StrictPushProducerTests.test_pauseThenPause   s%    
 <<>*h&=&=>r   c                f    | j                         }|j                          | j                  |       y)zp
        L{StrictPushProducer} is resumed if C{resumeProducing} is called on a
        paused producer.
        N)rK   rA   rU   rG   s     r   test_pauseThenResumez,StrictPushProducerTests.test_pauseThenResume   s)    
 <<>  "8$r   c                f    | j                         }|j                          | j                  |       y)zo
        L{StrictPushProducer} is stopped if C{stopProducing} is called on a
        resumed producer.
        N)rM   r;   rP   rG   s     r   test_resumeThenStopz+StrictPushProducerTests.test_resumeThenStop   )    
 ==? 8$r   c                f    | j                         }|j                          | j                  |       y)zo
        L{StrictPushProducer} is paused if C{pauseProducing} is called on a
        resumed producer.
        N)rM   r?   rS   rG   s     r   test_resumeThenPausez,StrictPushProducerTests.test_resumeThenPause   )    
 ==?!(#r   c                d    | j                         }| j                  t        |j                         y)zu
        L{StrictPushProducer.resumeProducing} raises L{ValueError} if called on a
        resumed producer.
        N)rM   rW   r9   rA   rG   s     r   test_resumeThenResumez-StrictPushProducerTests.test_resumeThenResume   %    
 ==?*h&>&>?r   c                f    | j                         }|j                          | j                  |       y)zn
        L{StrictPushProducer} is stopped if C{stopProducing} is called in the
        initial state.
        N)rE   r;   rP   rG   s     r   	test_stopz!StrictPushProducerTests.test_stop   re   r   c                f    | j                         }|j                          | j                  |       y)zn
        L{StrictPushProducer} is paused if C{pauseProducing} is called in the
        initial state.
        N)rE   r?   rS   rG   s     r   
test_pausez"StrictPushProducerTests.test_pause   rh   r   c                d    | j                         }| j                  t        |j                         y)zz
        L{StrictPushProducer} raises L{ValueError} if C{resumeProducing} is called
        in the initial state.
        N)rE   rW   r9   rA   rG   s     r   test_resumez#StrictPushProducerTests.test_resume   rk   r   N)r+   r3   )rH   r3   r+   r,   r*   )r-   r.   r/   r0   rE   rI   rK   rM   rP   rS   rU   rX   rZ   r\   r^   r`   rb   rd   rg   rj   rm   ro   rq   r1   r   r   rC   rC   V   sk    $545EFG%?%%$@%$@r   rC   c                  0    e Zd ZdZddZddZddZddZy)	IOPumpTestsz
    Tests for L{IOPump}.
    c                "   t               }t        |d      }t               }t        |d      }t        ||||d      }t               }||d|   }|j	                  |d       |j                          | j                  d|j                         y)	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)	streamingr4   N)r   r	   r
   r3   registerProducerpumpr"   r8   )	r   modeserverProtoserverTransportclientProtoclientTransportr{   rH   victims	            r   _testStreamingProducerz"IOPumpTests._testStreamingProducer   s     j'dCj'eD
 &'%%
  	D9		HOO4r   c                (    | j                  d       y)z
        L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer}
        (stream producer) registered with the server transport.
        rw   r|   Nr   r:   s    r   test_serverStreamingProducerz(IOPumpTests.test_serverStreamingProducer"      
 	###2r   c                (    | j                  d       y)z
        L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer}
        (stream producer) registered with the client transport.
        rx   r   Nr   r:   s    r   test_clientStreamingProducerz(IOPumpTests.test_clientStreamingProducer)  r   r   c                    g t               }t        t        t        |      \  }}}|j                  dfd       | j	                         |j                          | j                         y)zE
        L{IOPump.pump} advances time in the given L{Clock}.
        )clockr   c                 &     j                  d      S )NT)append)time_passeds   r   <lambda>z/IOPumpTests.test_timeAdvances.<locals>.<lambda>7  s    ;#5#5d#; r   N)r   r   r   	callLaterassertFalser{   
assertTrue)r   r   _r{   r   s       @r   test_timeAdvanceszIOPumpTests.test_timeAdvances0  sW     -hN
1d;<%		$r   N)r|   zLiteral['server', 'client']r+   r,   r*   )r-   r.   r/   r0   r   r   r   r   r1   r   r   rs   rs      s    !5F33
%r   rs   N)r0   
__future__r   typingr   zope.interfacer   twisted.internet.interfacesr   twisted.internet.protocolr   twisted.internet.taskr   twisted.test.iosimr	   r
   r   twisted.trial.unittestr   r   r3   rC   rs   r1   r   r   <module>r      sq    #  & 5 . ' O O +&8 &8R ]     0a@h a@H@%( @%r   