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   ) %