ó [³XMc@sHdZddlmZddlmZddlmZddlm Z m Z ddl m Z ddl mZddlmZdd lmZmZmZdd l mZmZd e jfd „ƒYZd efd„ƒYZdfd„ƒYZdeejfd„ƒYZdeejfd„ƒYZdeejfd„ƒYZdS(s. Test case for L{twisted.protocols.loopback}. iÿÿÿÿ(t implements(tunittest(tsuppress(tbasictloopback(tdefer(tProtocol(tDeferred(tIAddresst IPushProducert IPullProducer(treactort interfacestSimpleProtocolcBs,eZd„Zd„Zd„Zd„ZRS(cCs%tjƒ|_g|_g|_dS(N(RRtconntlinestconnLost(tself((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt__init__s cCs|jjdƒdS(N(RtcallbacktNone(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pytconnectionMadescCs|jj|ƒdS(N(Rtappend(Rtline((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt lineReceivedscCs|jj|ƒdS(N(RR(Rtreason((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pytconnectionLost s(t__name__t __module__RRRR(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR s   t DoomProtocolcBseZdZd„ZRS(icCsl|jd7_|jdkr5|jd|jƒntj||ƒ|jddkrh|jjƒndS(NiisHello %diÿÿÿÿsHello 3(titsendLineR RRt transporttloseConnection(RR((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR&s (RRRR(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR$stLoopbackTestCaseMixincBseZd„Zd„ZRS(csotƒ‰tƒ‰‡fd†}ˆjj|ƒ‡‡‡fd†}tjˆjˆˆƒ}|j|ƒ|S(NcsˆjdƒˆjjƒdS(NsTHIS IS LINE ONE!(RR R!(tresult(ts(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt sendALine6s csLˆjˆjdgƒˆjtˆjƒdƒˆjtˆjƒdƒdS(NsTHIS IS LINE ONE!i(t assertEqualsRtlenR(tignored(RR$tc(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pytcheck;s(R Rt addCallbackRt maybeDeferredt loopbackFunc(RR%R*td((R)RR$s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttestRegularFunction2s   csotƒ‰tƒ‰‡fd†}ˆjj|ƒ‡‡‡fd†}tjˆjˆˆƒ}|j|ƒ|S(NcsˆjdƒdS(Ns DOOM LINE(R(R#(R$(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR%GscsqˆjˆjdddgƒˆjˆjddddgƒˆjtˆjƒdƒˆjtˆjƒdƒdS(NsHello 1sHello 2sHello 3s DOOM LINEi(R&RR'R(R((RR)R$(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR*Ks(RRR+RR,R-(RR%R*R.((R)RR$s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttestSneakyHiddenDoomCs   (RRR/R0(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR"1s tLoopbackAsyncTestCasecBs¿eZeejƒZd„Zd„Zd„Zd„Z dd„Z d„Z d„Z d„Z d „Zd d „Zd „Zd „Zd„Zd„Zd„Zd„Zd„Zd„ZRS(cCsbdtfd„ƒY}|ƒ}|ƒ}tj||ƒ|j|jdƒ|j|jdƒdS(s} Test that the client and server protocol both have makeConnection invoked on them by loopbackAsync. t TestProtocolcBseZdZd„ZRS(cSs ||_dS(N(R (RR ((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pytmakeConnectionasN(RRRR R3(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR2_sN(RRt loopbackAsynct failIfEqualR R(RR2tservertclient((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_makeConnectionZs   cs’dtfd„ƒY}|rA|ƒ}tƒ}|_tƒ}n"tƒ}|ƒ}tƒ}|_tj||ƒ‡‡fd†}|j|ƒS(sJ Test one of the permutations of client/server host/peer. R2cBseZd„ZRS(cSs$tj||ƒ|jj|ƒdS(N(RR3t onConnectionR(RR ((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR3ps(RRR3(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR2oscs,t|ˆƒƒ}ˆjtj|ƒƒdS(N(tgetattrt failUnlessRt providedBy(R thost(tgetR(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt connecteds(RRR9RR4R+(RR>t testServerR2R6R.R7R?((RR>s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt _hostpeertestks    cCs|jdtƒS(s„ Test that the server gets a transport with a properly functioning implementation of L{ITransport.getHost}. tgetHost(RAtTrue(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_serverHost†scCs|jdtƒS(sG Like C{test_serverHost} but for L{ITransport.getPeer} tgetPeer(RARC(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_serverPeerŽsRBcCs|jdtƒS(s„ Test that the client gets a transport with a properly functioning implementation of L{ITransport.getHost}. RB(RAtFalse(RR>((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_clientHost•scCs|jdtƒS(sH Like C{test_clientHost} but for L{ITransport.getPeer}. RE(RARG(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_clientPeerscs“dtfd„ƒY}dtf‡fd†ƒY}|r]|ƒ}|ƒ}tƒ}|_n"|ƒ}tƒ}|_|ƒ}tj||ƒ|S(sT Test one of the permutations of write/writeSequence client/server. tGreeteeProtocolcBseZdZd„ZRS(tcSs5|j|7_|jdkr1|jjdƒndS(Ntbytes(RLtreceivedRR(RRL((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt dataReceivedªs(RRRLRN(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRJ¨stGreeterProtocolcseZ‡fd†ZRS(cst|jˆƒdƒdS(NRL(R:R (R(twrite(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR°s(RRR((RP(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRO¯s(RRRMRR4(RRPR@RJROR6R7R.((RPs>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt _greetingtest¤s    cCs|jdtƒS(s„ Test that on a connection where the client speaks first, the server receives the bytes sent by the client. RP(RQRG(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_clientGreetingÀscCs|jdtƒS(sz Like C{test_clientGreeting}, but use C{writeSequence} instead of C{write} to issue the greeting. t writeSequence(RQRG(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_clientGreetingSequenceÈsRPcCs|jdtƒS(s„ Test that on a connection where the server speaks first, the client receives the bytes sent by the server. RP(RQRC(RRP((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_serverGreetingÐscCs|jdtƒS(sz Like C{test_serverGreeting}, but use C{writeSequence} instead of C{write} to issue the greeting. RS(RQRC(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_serverGreetingSequenceØscsŽtttddƒƒ‰dtf‡‡fd†ƒY}dtf‡‡‡fd†ƒY}|ƒ‰|ƒ‰tƒˆ_tjˆˆƒˆjS(Nii tProducingProtocolcseZ‡‡fd†ZRS(cs,ˆtˆƒƒ|_|jj|jƒdS(N(tlisttproducertstartR (R(t toProducet producerClass(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRäs(RRR((R[R\(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRWãstReceivingProtocolcs#eZdZ‡‡‡fd†ZRS(RKcsD|j|7_|jdjˆƒkr@|jjˆˆfƒndS(NRK(RLtjoinRMR(RRL(R[R7R6(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRNês(RRRLRN((R[R7R6(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR]ès(tmaptstrtrangeRRRMRR4(RR\RWR]((R\R6R[R7s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt _producertestàs"   csEdtfd„ƒY}ˆj|ƒ}‡fd†}|j|ƒ|S(sO Test a push producer registered against a loopback transport. t PushProducercBs<eZeeƒeZd„Zd„Zd„Zd„Z RS(cSs ||_dS(N(R[(RR[((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRÿscSs t|_dS(N(RCtresumed(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pytresumeProducingscSs'||_|j|tƒ|jƒdS(N(tconsumertregisterProducerRCt_produceAndSchedule(RRf((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRZs cSsL|jr;|jj|jjdƒƒtjd|jƒn |jjƒdS(Ni(R[RfRPtpopR t callLaterRhtunregisterProducer(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRh s ( RRRR RGRdRReRZRh(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRcûs     cs&|\}}ˆj|jjdƒdS(Ns0Streaming producer should not have been resumed.(tfailIfRYRd(t.0R7R6(R(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pytfinisheds  (tobjectRbR+(RRcR.Rn((Rs>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_pushProducer÷s  cCs#dtfd„ƒY}|j|ƒS(sO Test a pull producer registered against a loopback transport. t PullProducercBs-eZeeƒd„Zd„Zd„ZRS(cSs ||_dS(N(R[(RR[((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR!scSs ||_|jj|tƒdS(N(RfRgRG(RRf((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRZ$s cSs9|jj|jjdƒƒ|js5|jjƒndS(Ni(RfRPR[RiRk(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRe(s (RRRR RRZRe(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRqs   (RoRb(RRq((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_pullProducerscssdtfd„ƒY}dtfd„ƒY}|ƒ}|ƒ‰tjˆ|ƒ}‡‡fd†}|j|ƒ|S(sº L{loopback.loopbackAsync} does not call a protocol's C{dataReceived} method while that protocol's transport's C{write} method is higher up on the stack. tServercBseZd„ZRS(cSs|jjdƒdS(NRL(R RP(RRL((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRN6s(RRRN(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRs5stClientcBs)eZeZd„Zd„Zd„ZRS(cSstjd|jƒdS(Ni(R Rjtgo(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR<scSs|jjdƒt|_dS(Ntfoo(R RPRCtready(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRu?scSs|j|_|jjƒdS(N(RwtwasReadyR R!(RRL((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRNCs (RRRGRwRRuRN(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRt9s  csˆjˆjƒdS(N(t assertTrueRx(R((RR7(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt cbFinishedKs(RRR4R+(RRsRtR6R.Rz((RR7s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_writeNotReentrant/s   csÂg‰‡fd†}tƒ‰tƒ‰tjˆˆ|ƒ}ˆjˆgƒˆjjdƒˆjjdƒˆjjdƒˆjjdƒˆjjƒ‡‡‡‡fd†}|j|ƒ|S(sÜ The callable passed as the value for the C{pumpPolicy} parameter to L{loopbackAsync} is called with a L{_LoopbackQueue} of pending bytes and a protocol to which they should be delivered. cs=g}x|r%|j|jƒƒq Wˆj||fƒdS(N(RR>(tqueuettargetRL(t pumpCalls(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt dummyPolicyXs Rvtbartbaztquuxcs5ˆjˆˆdddgfˆddgfgƒdS(NRR‚RvR€(R&R(R((RR~R6R7(s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyt cbCompletejs(RRR4R&R RPR!R+(RRRnRƒ((RR~R6R7s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_pumpPolicyQs    cCsxg}tƒ}|j|_tjƒ}|jdƒ|jdƒ|jdƒtj||ƒ|j|ddgƒdS(sŸ L{identityPumpPolicy} is a pump policy which calls the target's C{dataReceived} method one for each string in the queue passed to it. RvR€N( RRRNRt_LoopbackQueuetputRtidentityPumpPolicyR&(RRLR7R|((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_identityPumpPolicyus      cCsug}tƒ}|j|_tjƒ}|jdƒ|jdƒ|jdƒtj||ƒ|j|dgƒdS(sÀ L{collapsingPumpPolicy} is a pump policy which calls the target's C{dataReceived} only once with all of the strings in the queue passed to it joined together. RvR€tfoobarN( RRRNRR…R†RtcollapsingPumpPolicyR&(RRLR7R|((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyttest_collapsingPumpPolicy‡s      (RRt staticmethodRR4R-R8RARDRFRHRIRQRRRTRURVRbRpRrR{R„RˆR‹(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR1Vs&             #  " $ tLoopbackTCPTestCasecBseZeejƒZRS((RRRŒRt loopbackTCPR-(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyR›stLoopbackUNIXTestCasecBs8eZeejƒZejedƒdkr6dZ nRS(s-Current reactor does not support UNIX socketsN( RRRŒRt loopbackUNIXR-R t IReactorUNIXR Rtskip(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyRŸsN(t__doc__tzope.interfaceRt twisted.trialRttwisted.trial.utilRtSUPPRESSttwisted.protocolsRRttwisted.internetRttwisted.internet.protocolRttwisted.internet.deferRttwisted.internet.interfacesRR R R R t LineReceiverR RR"tTestCaseR1RR(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_loopback.pyts  %ÿF