Vhb*dZddlmZddlmZddlmZddlmZddl m Z m Z m Z ddl mZddlmZdd lmZmZmZmZmZdd lmZGd d eZeeGd dZGddeZGddeZGddeZy)z" Tests for L{twisted.test.iosim}. ) annotations)Literal) implementer) IPushProducer) ClientFactoryFactoryProtocol)Clock) MemoryReactor)ConnectionCompleter FakeTransportIOPumpconnectconnectedServerAndClient)TestCasec(eZdZdZddZddZddZy)FakeTransportTestsz% Tests for L{FakeTransport}. c ttd}ttd}|j|jt|j|jt|j |j|jy)ze Each L{FakeTransport} receives a serial number that uniquely identifies it. TFN)r objectassertIsInstanceserialintassertNotEqual)selfabs G/home/dcms/DCMS/lib/python3.12/site-packages/twisted/test/test_iosim.pytest_connectionSerialz(FakeTransportTests.test_connectionSerial s` &(D ) &(E * ahh, ahh, AHHahh/cttd}|jd|jgd|j dj |j dy)zl L{FakeTransport.writeSequence} will write a sequence of L{bytes} to the transport. Fa)bcdrsabcdN)r rwrite writeSequence assertEqualjoinstreamrrs rtest_writeSequencez%FakeTransportTests.test_writeSequence+sG &(E *   *+ !((+W5rcttd}|jd|j|jd|j dj |j dy)z L{FakeTransport.write} will accept writes after transport was closed, but the data will be silently discarded. FsbeforesafterrN)r rr%loseConnectionr'r(r)r*s rtest_writeAfterClosez'FakeTransportTests.test_writeAfterClose7sS &(E *     !((+Y7rNreturnNone)__name__ __module__ __qualname____doc__rr+r.rrrrs 0 6 8rrc,eZdZdZdZddZddZddZy)StrictPushProducerz An L{IPushProducer} implementation which produces nothing but enforces preconditions on its state transition methods. runningcF|jdk(r tdd|_y)Nstoppedz)Cannot stop already-stopped IPushProducer_state ValueErrorrs r stopProducingz StrictPushProducer.stopProducingMs! ;;) #HI I rcb|jdk7rtd|jdd|_y)Nr9z Cannot pause  IPushProducerpausedr<r?s rpauseProducingz!StrictPushProducer.pauseProducingRs- ;;) #}T[[MHI I rcb|jdk7rtd|jdd|_y)NrCzCannot resume rBr9r<r?s rresumeProducingz"StrictPushProducer.resumeProducingWs- ;;( "~dkk].IJ J rNr/)r2r3r4r5r=r@rDrFr6rrr8r8Ds F   rr8ceZdZdZddZddZddZddZddZddZ ddZ dd Z dd Z dd Z dd Zdd ZddZddZddZddZddZddZddZy)StrictPushProducerTestsz* Tests for L{StrictPushProducer}. ctS)zp @return: A new L{StrictPushProducer} which has not been through any state changes. )r8r?s r_initialz StrictPushProducerTests._initialbs "##rc:t}|j|S)z@ @return: A new, stopped L{StrictPushProducer}. )r8r@rproducers r_stoppedz StrictPushProducerTests._stoppedis&' rc:t}|j|S)z? @return: A new, paused L{StrictPushProducer}. )r8rDrLs r_pausedzStrictPushProducerTests._pausedqs&'!rcZt}|j|j|S)zY @return: A new L{StrictPushProducer} which has been paused and resumed. )r8rDrFrLs r_resumedz StrictPushProducerTests._resumedys)&'!  "rc<|j|jdy)z Assert that the given producer is in the stopped state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r;Nr'r=rLs r assertStoppedz%StrictPushProducerTests.assertStopped )4rc<|j|jdy)z Assert that the given producer is in the paused state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} rCNrTrLs r assertPausedz$StrictPushProducerTests.assertPauseds (3rc<|j|jdy)z Assert that the given producer is in the running state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r9NrTrLs r assertRunningz%StrictPushProducerTests.assertRunningrVrc`|jt|jjy)zz L{StrictPushProducer.stopProducing} raises L{ValueError} if called when the producer is stopped. N) assertRaisesr>rNr@r?s rtest_stopThenStopz)StrictPushProducerTests.test_stopThenStops *dmmo&C&CDrc`|jt|jjy)z{ L{StrictPushProducer.pauseProducing} raises L{ValueError} if called when the producer is stopped. N)r\r>rNrDr?s rtest_stopThenPausez*StrictPushProducerTests.test_stopThenPauses *dmmo&D&DErc`|jt|jjy)z| L{StrictPushProducer.resumeProducing} raises L{ValueError} if called when the producer is stopped. N)r\r>rNrFr?s rtest_stopThenResumez+StrictPushProducerTests.test_stopThenResumes *dmmo&E&EFrcf|j}|j|j|y)zn L{StrictPushProducer} is stopped if C{stopProducing} is called on a paused producer. N)rPr@rUrLs rtest_pauseThenStopz*StrictPushProducerTests.test_pauseThenStops) <<>  8$rcd|j}|jt|jy)zs L{StrictPushProducer.pauseProducing} raises L{ValueError} if called on a paused producer. N)rPr\r>rDrLs rtest_pauseThenPausez+StrictPushProducerTests.test_pauseThenPauses% <<> *h&=&=>rcf|j}|j|j|y)zp L{StrictPushProducer} is resumed if C{resumeProducing} is called on a paused producer. N)rPrFrZrLs rtest_pauseThenResumez,StrictPushProducerTests.test_pauseThenResumes) <<>  " 8$rcf|j}|j|j|y)zo L{StrictPushProducer} is stopped if C{stopProducing} is called on a resumed producer. N)rRr@rUrLs rtest_resumeThenStopz+StrictPushProducerTests.test_resumeThenStop) ==?  8$rcf|j}|j|j|y)zo L{StrictPushProducer} is paused if C{pauseProducing} is called on a resumed producer. N)rRrDrXrLs rtest_resumeThenPausez,StrictPushProducerTests.test_resumeThenPause) ==?! (#rcd|j}|jt|jy)zu L{StrictPushProducer.resumeProducing} raises L{ValueError} if called on a resumed producer. N)rRr\r>rFrLs rtest_resumeThenResumez-StrictPushProducerTests.test_resumeThenResume% ==? *h&>&>?rcf|j}|j|j|y)zn L{StrictPushProducer} is stopped if C{stopProducing} is called in the initial state. N)rJr@rUrLs r test_stopz!StrictPushProducerTests.test_stoprjrcf|j}|j|j|y)zn L{StrictPushProducer} is paused if C{pauseProducing} is called in the initial state. N)rJrDrXrLs r test_pausez"StrictPushProducerTests.test_pausermrcd|j}|jt|jy)zz L{StrictPushProducer} raises L{ValueError} if C{resumeProducing} is called in the initial state. N)rJr\r>rFrLs r test_resumez#StrictPushProducerTests.test_resumerprN)r0r8)rMr8r0r1r/)r2r3r4r5rJrNrPrRrUrXrZr]r_rarcrergrirlrorrrtrvr6rrrHrH]sk$545EFG%?%%$@%$@rrHc0eZdZdZddZddZddZddZy) IOPumpTestsz Tests for L{IOPump}. c"t}t|d}t}t|d}t||||d}t}||d|}|j |d|j |j d|jy) 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)serverclient) streamingr9N)r r rr8registerProducerpumpr'r=) rmode serverProtoserverTransport clientProtoclientTransportrrMvictims r_testStreamingProducerz"IOPumpTests._testStreamingProducersj ' dCj ' eD      &'%%   D9  HOO4rc(|jdy)z L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer} (stream producer) registered with the server transport. r|rNrr?s rtest_serverStreamingProducerz(IOPumpTests.test_serverStreamingProducer) ###2rc(|jdy)z L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer} (stream producer) registered with the client transport. r}rNrr?s rtest_clientStreamingProducerz(IOPumpTests.test_clientStreamingProducer0rrcgt}ttt|\}}}|jdfd|j |j |j y)zE L{IOPump.pump} advances time in the given L{Clock}. )clockrc&jdS)NT)append) time_passedsrz/IOPumpTests.test_timeAdvances..>s;#5#5d#;rN)r rr callLater assertFalser assertTrue)rr_rrs @rtest_timeAdvanceszIOPumpTests.test_timeAdvances7sW -hN 1d ;< %   $rN)rzLiteral['server', 'client']r0r1r/)r2r3r4r5rrrrr6rrrxrxs!5F33 %rrxc eZdZdZddZddZy)ConnectionCompleterTestszr L{ConnectionCompleter} can connect a connecting client and a listening server in a L{MemoryReactor}. cjtt}|j|jdy)z| When no connection for its reactor is pending, L{ConnectionCompleter.succeedOnce} returns C{None}. N)r r assertIs succeedOnce)rccs rtest_noConnectionz*ConnectionCompleterTests.test_noConnectionJs% ! 1 bnn&-rct}|jdtjt|j ddt jtt|}|j|jty)z When a connection to a matched host and port is pending, L{ConnectionCompleter} returns an L{IOPump} connecting them. Pz 127.0.0.1N) r listenTCPr forProtocolr connectTCPrr rrr)rmemrs rtest_someConnectionz,ConnectionCompleterTests.test_someConnectionRs` o b'--h78 {B (A(A((KL  % bnn.7rNr/)r2r3r4r5rrr6rrrrDs . 8rrN)r5 __future__rtypingrzope.interfacertwisted.internet.interfacesrtwisted.internet.protocolrrr twisted.internet.taskr twisted.internet.testingr twisted.test.iosimr r rrrtwisted.trial.unittestrrr8rHrxrr6rrrs#&5FF'2,&8&8R ]   0a@ha@H@%(@%F8x8r