
    \d"                         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e          Z ee           G d	 d
                      Z G d de          Z G 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 Zd Zd ZdS )FakeTransportTestsz%
    Tests for L{FakeTransport}.
    c                 6   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bs      7lib/python3.11/site-packages/twisted/test/test_iosim.pytest_connectionSerialz(FakeTransportTests.test_connectionSerial   sz    
 &((D))&((E**ah,,,ah,,,AHah/////    c                     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   dr   s   abcdN)r   r   writewriteSequenceassertEqualjoinstreamr   r   s     r   test_writeSequencez%FakeTransportTests.test_writeSequence!   si    
 &((E**		***+++!(++W55555r   c                    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!   s     r   test_writeAfterClosez'FakeTransportTests.test_writeAfterClose-   sw    
 &((E**				!(++Y77777r   N)__name__
__module____qualname____doc__r   r"   r%    r   r   r
   r
      sK         	0 	0 	0
6 
6 
6
8 
8 
8 
8 
8r   r
   c                   (    e Zd ZdZdZd Zd Zd ZdS )StrictPushProducerz
    An L{IPushProducer} implementation which produces nothing but enforces
    preconditions on its state transition methods.
    runningc                 H    | j         dk    rt          d          d| _         d S )Nstoppedz)Cannot stop already-stopped IPushProducer_state
ValueErrorr   s    r   stopProducingz StrictPushProducer.stopProducingC   s)    ;)##HIIIr   c                 Z    | j         dk    rt          d| j          d          d| _         d S )Nr-   zCannot pause  IPushProducerpausedr0   r3   s    r   pauseProducingz!StrictPushProducer.pauseProducingH   s6    ;)##HT[HHHIIIr   c                 Z    | j         dk    rt          d| j          d          d| _         d S )Nr7   zCannot resume r6   r-   r0   r3   s    r   resumeProducingz"StrictPushProducer.resumeProducingM   s6    ;(""IdkIIIJJJr   N)r&   r'   r(   r)   r1   r4   r8   r:   r*   r   r   r,   r,   :   sR         
 F     
  
         r   r,   c                       e Zd ZdZd Zd Zd Zd Zd Zd Z	d Z
d	 Zd
 Zd Zd Zd Zd Zd Zd Zd Zd Zd Zd ZdS )StrictPushProducerTestsz*
    Tests for L{StrictPushProducer}.
    c                     t                      S )zp
        @return: A new L{StrictPushProducer} which has not been through any state
            changes.
        )r,   r3   s    r   _initialz StrictPushProducerTests._initialX   s    
 "###r   c                 J    t                      }|                                 |S )z@
        @return: A new, stopped L{StrictPushProducer}.
        )r,   r4   r   producers     r   _stoppedz StrictPushProducerTests._stopped_   s&     &''   r   c                 J    t                      }|                                 |S )z?
        @return: A new, paused L{StrictPushProducer}.
        )r,   r8   r@   s     r   _pausedzStrictPushProducerTests._pausedg   s&     &''!!!r   c                 r    t                      }|                                 |                                 |S )zY
        @return: A new L{StrictPushProducer} which has been paused and resumed.
        )r,   r8   r:   r@   s     r   _resumedz StrictPushProducerTests._resumedo   s8     &''!!!  """r   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   r1   r@   s     r   assertStoppedz%StrictPushProducerTests.assertStoppedx   "     	)44444r   c                 <    |                      |j        d           dS )z
        Assert that the given producer is in the paused state.

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

        @param producer: The producer to verify.
        @type producer: L{StrictPushProducer}
        r-   NrH   r@   s     r   assertRunningz%StrictPushProducerTests.assertRunning   rJ   r   c                 j    |                      t          |                                 j                   dS )zz
        L{StrictPushProducer.stopProducing} raises L{ValueError} if called when
        the producer is stopped.
        N)assertRaisesr2   rB   r4   r3   s    r   test_stopThenStopz)StrictPushProducerTests.test_stopThenStop   s+    
 	*dmmoo&CDDDDDr   c                 j    |                      t          |                                 j                   dS )z{
        L{StrictPushProducer.pauseProducing} raises L{ValueError} if called when
        the producer is stopped.
        N)rP   r2   rB   r8   r3   s    r   test_stopThenPausez*StrictPushProducerTests.test_stopThenPause   s+    
 	*dmmoo&DEEEEEr   c                 j    |                      t          |                                 j                   dS )z|
        L{StrictPushProducer.resumeProducing} raises L{ValueError} if called when
        the producer is stopped.
        N)rP   r2   rB   r:   r3   s    r   test_stopThenResumez+StrictPushProducerTests.test_stopThenResume   s+    
 	*dmmoo&EFFFFFr   c                     |                                  }|                                 |                     |           dS )zn
        L{StrictPushProducer} is stopped if C{stopProducing} is called on a paused
        producer.
        N)rD   r4   rI   r@   s     r   test_pauseThenStopz*StrictPushProducerTests.test_pauseThenStop   s=    
 <<>>   8$$$$$r   c                 n    |                                  }|                     t          |j                   dS )zs
        L{StrictPushProducer.pauseProducing} raises L{ValueError} if called on a
        paused producer.
        N)rD   rP   r2   r8   r@   s     r   test_pauseThenPausez+StrictPushProducerTests.test_pauseThenPause   s0    
 <<>>*h&=>>>>>r   c                     |                                  }|                                 |                     |           dS )zp
        L{StrictPushProducer} is resumed if C{resumeProducing} is called on a
        paused producer.
        N)rD   r:   rN   r@   s     r   test_pauseThenResumez,StrictPushProducerTests.test_pauseThenResume   s=    
 <<>>  """8$$$$$r   c                     |                                  }|                                 |                     |           dS )zo
        L{StrictPushProducer} is stopped if C{stopProducing} is called on a
        resumed producer.
        N)rF   r4   rI   r@   s     r   test_resumeThenStopz+StrictPushProducerTests.test_resumeThenStop   =    
 ==??   8$$$$$r   c                     |                                  }|                                 |                     |           dS )zo
        L{StrictPushProducer} is paused if C{pauseProducing} is called on a
        resumed producer.
        N)rF   r8   rL   r@   s     r   test_resumeThenPausez,StrictPushProducerTests.test_resumeThenPause   =    
 ==??!!!(#####r   c                 n    |                                  }|                     t          |j                   dS )zu
        L{StrictPushProducer.resumeProducing} raises L{ValueError} if called on a
        resumed producer.
        N)rF   rP   r2   r:   r@   s     r   test_resumeThenResumez-StrictPushProducerTests.test_resumeThenResume   0    
 ==??*h&>?????r   c                     |                                  }|                                 |                     |           dS )zn
        L{StrictPushProducer} is stopped if C{stopProducing} is called in the
        initial state.
        N)r>   r4   rI   r@   s     r   	test_stopz!StrictPushProducerTests.test_stop   r^   r   c                     |                                  }|                                 |                     |           dS )zn
        L{StrictPushProducer} is paused if C{pauseProducing} is called in the
        initial state.
        N)r>   r8   rL   r@   s     r   
test_pausez"StrictPushProducerTests.test_pause   ra   r   c                 n    |                                  }|                     t          |j                   dS )zz
        L{StrictPushProducer} raises L{ValueError} if C{resumeProducing} is called
        in the initial state.
        N)r>   rP   r2   r:   r@   s     r   test_resumez#StrictPushProducerTests.test_resume   rd   r   N)r&   r'   r(   r)   r>   rB   rD   rF   rI   rL   rN   rQ   rS   rU   rW   rY   r[   r]   r`   rc   rf   rh   rj   r*   r   r   r<   r<   S   sL        $ $ $      5 5 54 4 45 5 5E E EF F FG G G% % %? ? ?% % %% % %$ $ $@ @ @% % %$ $ $@ @ @ @ @r   r<   c                   $    e Zd ZdZd Zd Zd ZdS )IOPumpTestsz
    Tests for L{IOPump}.
    c                 h   t                      }t          |d          }t                      }t          |d          }t          ||||d          }t                      }||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   r1   )	r   modeserverProtoserverTransportclientProtoclientTransportrt   rA   victims	            r   _testStreamingProducerz"IOPumpTests._testStreamingProducer   s     jj'dCCCjj'eDDD
 
 
 &''%%
 
  	D999		HO44444r   c                 2    |                      d           dS )z
        L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer}
        (stream producer) registered with the server transport.
        rp   ru   Nr{   r3   s    r   test_serverStreamingProducerz(IOPumpTests.test_serverStreamingProducer  !    
 	###22222r   c                 2    |                      d           dS )z
        L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer}
        (stream producer) registered with the client transport.
        rq   r}   Nr~   r3   s    r   test_clientStreamingProducerz(IOPumpTests.test_clientStreamingProducer&  r   r   N)r&   r'   r(   r)   r{   r   r   r*   r   r   rl   rl      sL         !5 !5 !5F3 3 33 3 3 3 3r   rl   N)r)   zope.interfacer   twisted.internet.interfacesr   twisted.internet.protocolr   twisted.test.iosimr   r   twisted.trial.unittestr   r
   r,   r<   rl   r*   r   r   <module>r      sJ   
 ' & & & & & 5 5 5 5 5 5 . . . . . . 5 5 5 5 5 5 5 5 + + + + + +&8 &8 &8 &8 &8 &8 &8 &8R ]               0a@ a@ a@ a@ a@h a@ a@ a@H43 43 43 43 43( 43 43 43 43 43r   