o 9Bib*@sdZddlmZddlmZddlmZddlmZddl m Z m Z m Z ddl mZddlmZdd lmZmZmZmZmZdd lmZGd d d eZeeGd ddZGdddeZGdddeZGdddeZdS)z" Tests for L{twisted.test.iosim}. ) annotations)Literal) implementer) IPushProducer) ClientFactoryFactoryProtocol)Clock) MemoryReactor)ConnectionCompleter FakeTransportIOPumpconnectconnectedServerAndClient)TestCasec@s.eZdZdZd ddZd ddZd dd Zd S) FakeTransportTestsz% Tests for L{FakeTransport}. returnNonecCsHttd}ttd}||jt||jt||j|jdS)ze Each L{FakeTransport} receives a serial number that uniquely identifies it. TFN)r objectassertIsInstanceserialintassertNotEqual)selfabrW/var/www/html/Trade-python/venv/lib/python3.10/site-packages/twisted/test/test_iosim.pytest_connectionSerial s  z(FakeTransportTests.test_connectionSerialcCs<ttd}|d|gd|d|jddS)zl L{FakeTransport.writeSequence} will write a sequence of L{bytes} to the transport. Fa)bcdsabcdN)r rwrite writeSequence assertEqualjoinstreamrrrrrtest_writeSequence+s  z%FakeTransportTests.test_writeSequencecCs@ttd}|d||d|d|jddS)z L{FakeTransport.write} will accept writes after transport was closed, but the data will be silently discarded. Fsbeforesafterr#N)r rr$loseConnectionr&r'r(r)rrrtest_writeAfterClose7s   z'FakeTransportTests.test_writeAfterCloseNrr)__name__ __module__ __qualname____doc__rr*r,rrrrrs    rc@s2eZdZdZdZd ddZd ddZd d d Zd S) StrictPushProducerz An L{IPushProducer} implementation which produces nothing but enforces preconditions on its state transition methods. runningrrcCs|jdkr tdd|_dS)Nstoppedz)Cannot stop already-stopped IPushProducer_state ValueErrorrrrr stopProducingMs  z StrictPushProducer.stopProducingcC&|jdkrtd|jdd|_dS)Nr3z Cannot pause  IPushProducerpausedr5r8rrrpauseProducingR  z!StrictPushProducer.pauseProducingcCr:)Nr<zCannot resume r;r3r5r8rrrresumeProducingWr>z"StrictPushProducer.resumeProducingNr-)r.r/r0r1r6r9r=r?rrrrr2Ds   r2c@seZdZdZd-ddZd-ddZd-dd Zd-d d Zd.ddZd.ddZ d.ddZ d/ddZ d/ddZ d/ddZ d/ddZd/ddZd/ddZd/d d!Zd/d"d#Zd/d$d%Zd/d&d'Zd/d(d)Zd/d*d+Zd,S)0StrictPushProducerTestsz* Tests for L{StrictPushProducer}. rr2cCstS)zp @return: A new L{StrictPushProducer} which has not been through any state changes. )r2r8rrr_initialbsz StrictPushProducerTests._initialcCt}||S)z@ @return: A new, stopped L{StrictPushProducer}. )r2r9rproducerrrr_stoppediz StrictPushProducerTests._stoppedcCrB)z? @return: A new, paused L{StrictPushProducer}. )r2r=rCrrr_pausedqrFzStrictPushProducerTests._pausedcCst}|||S)zY @return: A new L{StrictPushProducer} which has been paused and resumed. )r2r=r?rCrrr_resumedysz StrictPushProducerTests._resumedrDrcC||jddS)z Assert that the given producer is in the stopped state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r4Nr&r6rCrrr assertStoppedz%StrictPushProducerTests.assertStoppedcCrI)z Assert that the given producer is in the paused state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r<NrJrCrrr assertPausedrLz$StrictPushProducerTests.assertPausedcCrI)z Assert that the given producer is in the running state. @param producer: The producer to verify. @type producer: L{StrictPushProducer} r3NrJrCrrr assertRunningrLz%StrictPushProducerTests.assertRunningcC|t|jdS)zz L{StrictPushProducer.stopProducing} raises L{ValueError} if called when the producer is stopped. N) assertRaisesr7rEr9r8rrrtest_stopThenStopz)StrictPushProducerTests.test_stopThenStopcCrO)z{ L{StrictPushProducer.pauseProducing} raises L{ValueError} if called when the producer is stopped. N)rPr7rEr=r8rrrtest_stopThenPauserRz*StrictPushProducerTests.test_stopThenPausecCrO)z| L{StrictPushProducer.resumeProducing} raises L{ValueError} if called when the producer is stopped. N)rPr7rEr?r8rrrtest_stopThenResumerRz+StrictPushProducerTests.test_stopThenResumecC|}|||dS)zn L{StrictPushProducer} is stopped if C{stopProducing} is called on a paused producer. N)rGr9rKrCrrrtest_pauseThenStopz*StrictPushProducerTests.test_pauseThenStopcC|}|t|jdS)zs L{StrictPushProducer.pauseProducing} raises L{ValueError} if called on a paused producer. N)rGrPr7r=rCrrrtest_pauseThenPausez+StrictPushProducerTests.test_pauseThenPausecCrU)zp L{StrictPushProducer} is resumed if C{resumeProducing} is called on a paused producer. N)rGr?rNrCrrrtest_pauseThenResumerWz,StrictPushProducerTests.test_pauseThenResumecCrU)zo L{StrictPushProducer} is stopped if C{stopProducing} is called on a resumed producer. N)rHr9rKrCrrrtest_resumeThenStoprWz+StrictPushProducerTests.test_resumeThenStopcCrU)zo L{StrictPushProducer} is paused if C{pauseProducing} is called on a resumed producer. N)rHr=rMrCrrrtest_resumeThenPauserWz,StrictPushProducerTests.test_resumeThenPausecCrX)zu L{StrictPushProducer.resumeProducing} raises L{ValueError} if called on a resumed producer. N)rHrPr7r?rCrrrtest_resumeThenResumerZz-StrictPushProducerTests.test_resumeThenResumecCrU)zn L{StrictPushProducer} is stopped if C{stopProducing} is called in the initial state. N)rAr9rKrCrrr test_stoprWz!StrictPushProducerTests.test_stopcCrU)zn L{StrictPushProducer} is paused if C{pauseProducing} is called in the initial state. N)rAr=rMrCrrr test_pauserWz"StrictPushProducerTests.test_pausecCrX)zz L{StrictPushProducer} raises L{ValueError} if C{resumeProducing} is called in the initial state. N)rArPr7r?rCrrr test_resumerZz#StrictPushProducerTests.test_resumeN)rr2)rDr2rrr-)r.r/r0r1rArErGrHrKrMrNrQrSrTrVrYr[r\r]r^r_r`rarrrrr@]s*           r@c@s8eZdZdZdddZddd Zdd d Zdd d ZdS) IOPumpTestsz Tests for L{IOPump}. modeLiteral['server', 'client']rrc Csrt}t|dd}t}t|dd}t||||dd}t}||d|}|j|dd||d|jdS) 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) streamingr3N)rr rr2registerProducerpumpr&r6) rrc serverProtoserverTransport clientProtoclientTransportrkrDvictimrrr_testStreamingProducers(  z"IOPumpTests._testStreamingProducercC|jdddS)z L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer} (stream producer) registered with the server transport. rgrcNrqr8rrrtest_serverStreamingProducer)z(IOPumpTests.test_serverStreamingProducercCrr)z L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer} (stream producer) registered with the client transport. rhrsNrtr8rrrtest_clientStreamingProducer0rvz(IOPumpTests.test_clientStreamingProducercsRgt}ttt|d\}}}|dfdd|||dS)zE L{IOPump.pump} advances time in the given L{Clock}. )clockrcs dS)NT)appendr time_passedrr>s z/IOPumpTests.test_timeAdvances..N)r rr callLater assertFalserk assertTrue)rrx_rkrrzrtest_timeAdvances7s zIOPumpTests.test_timeAdvancesN)rcrdrrr-)r.r/r0r1rqrurwrrrrrrbs   # rbc@s$eZdZdZd ddZd ddZdS) ConnectionCompleterTestszr L{ConnectionCompleter} can connect a connecting client and a listening server in a L{MemoryReactor}. rrcCstt}||ddS)z| When no connection for its reactor is pending, L{ConnectionCompleter.succeedOnce} returns C{None}. N)r r assertIs succeedOnce)rccrrrtest_noConnectionJs z*ConnectionCompleterTests.test_noConnectioncCsHt}|dtt|ddttt|}|| t dS)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 )rmemrrrrtest_someConnectionRs z,ConnectionCompleterTests.test_someConnectionNr-)r.r/r0r1rrrrrrrDs rN)r1 __future__rtypingrzope.interfacertwisted.internet.interfacesrtwisted.internet.protocolrrrtwisted.internet.taskr twisted.internet.testingr twisted.test.iosimr r r rrtwisted.trial.unittestrrr2r@rbrrrrrs"       )%C