ó [³XMc@s1dZddlmZmZmZddlmZddlmZddlm Z ddlm Z ddl m Z m Z mZmZmZddlmZd e jfd „ƒYZd e jfd „ƒYZd e jfd„ƒYZde jfd„ƒYZdejfd„ƒYZdejfd„ƒYZdejfd„ƒYZdej fd„ƒYZ dejfd„ƒYZ!dejfd„ƒYZ"de jej#fd„ƒYZ$dejfd „ƒYZ%d!ejfd"„ƒYZ&d#efd$„ƒYZ'd%ej(fd&„ƒYZ)d'ejfd(„ƒYZ*d)S(*s Test code for policies. iÿÿÿÿ(t Interfacet implementst implementedBy(tStringIO(tunittest(tStringTransport(t StringTransportWithDisconnection(tprotocoltreactortaddresstdeferttask(tpoliciestSimpleProtocolcBs<eZdZZdZd„Zd„Zd„Zd„ZRS(itcCs"tjƒ|_tjƒ|_dS(N(R tDeferredt dConnectedt dDisconnected(tself((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt__init__scCsd|_|jjdƒdS(NiR(t connectedRtcallback(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pytconnectionMades cCsd|_|jjdƒdS(NiR(t disconnectedRR(Rtreason((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pytconnectionLost"s cCs|j|7_dS(N(tbuffer(Rtdata((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt dataReceived&s( t__name__t __module__RRRRRRR(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR s     t SillyFactorycBseZd„Zd„ZRS(cCs ||_dS(N(tp(RR ((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR-scCs|jS(N(R (Rtaddr((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt buildProtocol0s(RRRR"(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR+s t EchoProtocolcBs2eZeZd„Zd„Zd„Zd„ZRS(cCs t|_dS(N(tTruetpaused(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pytpauseProducing7scCs t|_dS(N(tFalseR%(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pytresumeProducing:scCsdS(N((R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt stopProducing=scCs|jj|ƒdS(N(t transporttwrite(RR((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR@s(RRR'R%R&R(R)R(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR#4s    tServercBseZdZeZRS(s8 A simple server factory using L{EchoProtocol}. (RRt__doc__R#R(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR,EstTestableThrottlingFactorycBs eZdZd„Zd„ZRS(sH L{policies.ThrottlingFactory} using a L{task.Clock} for tests. cOs#tjj|||Ž||_dS(s˜ @param clock: object providing a callLater method that can be used for tests. @type clock: C{task.Clock} or alike. N(R tThrottlingFactoryRtclock(RR0targstkwargs((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRRscCs|jj||ƒS(s0 Forward to the testable clock. (R0t callLater(Rtperiodtfunc((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR3\s(RRR-RR3(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR.Ms tTestableTimeoutFactorycBs eZdZd„Zd„ZRS(sE L{policies.TimeoutFactory} using a L{task.Clock} for tests. cOs#tjj|||Ž||_dS(s˜ @param clock: object providing a callLater method that can be used for tests. @type clock: C{task.Clock} or alike. N(R tTimeoutFactoryRR0(RR0R1R2((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRiscCs|jj||ƒS(s0 Forward to the testable clock. (R0R3(RR4R5((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR3ss(RRR-RR3(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR6ds tWrapperTestCasecBs eZdZd„Zd„ZRS(s> Tests for L{WrappingFactory} and L{ProtocolWrapper}. cCsPtƒ}tj|ƒ}|jtjdddƒƒ}|j|jj|ƒdS(sf Make sure protocol.factory is the wrapped factory, not the wrapping factory. tTCPs 127.0.0.1i#N( R,R tWrappingFactoryR"R t IPv4AddresstassertIdenticaltwrappedProtocoltfactory(RtftwfR ((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_protocolFactoryAttributes csdtfd„ƒY‰dd‡fd†ƒY}ttjƒtjƒ}tjtjdƒ|ƒ}|j|ƒƒ|j ˆj |j ƒƒdS(s² The transport wrapper passed to the wrapped protocol's C{makeConnection} provides the same interfaces as are provided by the original transport. tIStubTransportcBseZRS((RR(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRBst StubTransportcseZeˆƒRS((RRR((RB(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRC“sN(( RRR tProtocolWrapperRtProtocolR:tNonetmakeConnectiont assertTruet providedByR*(RRCtprototwrapper((RBs>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_transportInterfacesŠs  (RRR-RARL(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR8{s R:cBseZd„Zd„ZRS(cCs|S(N((tsR?R ((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt¨scCs$tjj|ƒ|jjdƒdS(N(R R:t startFactorytdeferredRRF(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyROªs(RRRRO(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR:§s tThrottlingTestCasecBs)eZdZd„Zd„Zd„ZRS(s2 Tests for L{policies.ThrottlingFactory}. c smtƒ}gtdƒD]}tƒ^q\‰‰‰‰tj|dƒ‰tˆƒ}tjƒ|_t j d|ddƒ‰ˆj ƒj ‰‡‡‡‡fd†}‡‡‡‡‡fd†}‡fd†}‡‡fd †}‡‡fd †}‡‡‡fd †} |jj |ƒ|jj |ƒ|jj |ƒ|jj |ƒ|jj |ƒ|jj | ƒ|jS( sQ Full test using a custom server limiting number of connections. iiit interfaces 127.0.0.1csXtjdˆtˆƒƒˆjj‡‡fd†ƒˆjj‡‡fd†ƒˆjS(Ns 127.0.0.1cstjdˆtˆƒƒS(Ns 127.0.0.1(Rt connectTCPR(tr(tc2tn(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRNÆscstjdˆtˆƒƒS(Ns 127.0.0.1(RRSR(RT(tc3RV(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRNÈs(RRSRRt addCallbackR(tresults(RVRURWtc1(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt _connect123Ãs   cs“ˆjgˆˆˆfD]}|j^qdddgƒˆjgˆˆˆfD]}|j^qNdddgƒˆjtˆjjƒƒdƒ|S(Niii(t assertEqualsRRtlent protocolstkeys(RYtc(ttServerRRWRURZ(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt _check123Ës88csˆjjƒˆjS(N(R*tloseConnectionR(RY(RZ(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt_lose1Ñs cs tjdˆtˆƒƒˆjS(Ns 127.0.0.1(RRSRR(RY(tc4RV(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt _connect4Öscs*ˆjˆjdƒˆjˆjdƒ|S(Nii(R\RR(RY(RRe(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt_check4ÚscsLx!ˆˆfD]}|jjƒq WtjtjˆjƒˆjˆjgƒS(N(R*RcR t DeferredListt maybeDeferredt stopListeningR(RYR`(R RURe(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt_cleanupßs (R,trangeR R R/R:R RRPRt listenTCPtgetHosttportRX( Rtservertit wrapTServerR[RbRdRfRgRk((RaRRWRURZReRVR s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt test_limitµs( . cCs3tƒ}ttjƒ|ddƒ}|jtjdddƒƒ}tƒ}||_|j |ƒ|j |_ |j dƒ|j dƒ|j |jƒdƒ|j |jd ƒ|j|j jƒ|jjd ƒ|j |jdƒ|j|j jƒ|jjd ƒ|j |jdƒ|j|j jƒd S( se Check the writeLimit parameter: write data, and check for the pause status. t writeLimiti R9s 127.0.0.1it 0123456789t abcdefghijt0123456789abcdefghijigÍÌÌÌÌÌð?N(R,R.R tClockR"R R;RRRGR=tproducerRR\tvaluetwrittenThisSecondt assertFalseR%R0tadvanceRH(RRpRaRottr((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_writeLimitðs$       cCsÍtƒ}ttjƒ|ddƒ}|jtjdddƒƒ}tƒ}||_|j |ƒ|j dƒ|j dƒ|j |j ƒdƒ|j |j d ƒ|jjd ƒ|j |j dƒ|j |jd ƒ|jjd ƒ|j |j dƒ|j |jd ƒ|jƒ|j dƒ|j dƒ|j |j ƒdƒ|j |j d ƒ|jjd ƒ|j |j dƒ|j |jd ƒ|jjd ƒ|j |j dƒ|j |jd ƒd S(sb Check the readLimit parameter: read data and check for the pause status. t readLimiti R9s 127.0.0.1iRuRvRwigÍÌÌÌÌÌð?R%t producingN(R,R.R RxR"R R;RRRGRR\RztreadThisSecondR0R}t producerStatetclear(RRpRaRoR~((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_readLimits6         (RRR-RsRR…(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRQ°s ; tTimeoutTestCasecBs2eZdZd„Zd„Zd„Zd„ZRS(s/ Tests for L{policies.TimeoutFactory}. cCs’tjƒ|_tjƒ}t|_t|j|dƒ|_|jjt j dddƒƒ|_ t ƒ|_ |j |j _|j j|j ƒdS(sq Create a testable, deterministic clock, and a set of server factory/protocol/transport. iR9s 127.0.0.1i90N(R RxR0Rt ServerFactoryR R6R>R"R R;RJRR*RG(RtwrappedFactory((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pytsetUp=s    cCse|jjdddddgƒ|j|jjjƒ|jjddgƒ|j|jjjƒdS(s¶ Make sure that when a TimeoutFactory accepts a connection, it will time out that connection if no data is read or written within the timeout period. ggà?gð?gš™™™™™Ù?gš™™™™™É?N(R0tpumptfailIfRJR=Rt failUnless(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt test_timeoutMscCsä|jjdddgƒ|j|jjjƒ|jjdƒ|jjdddgƒ|j|jjjƒ|jjdgdƒ|jjdddgƒ|j|jjjƒ|jjddgƒ|j|jjjƒdS( s’ Make sure that writing data to a transport from a protocol constructed by a TimeoutFactory resets the timeout countdown. ggà?gð?sbytes bytes bytestbytesig@N( R0RŠR‹RJR=RR+t writeSequenceRŒ(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_sendAvoidsTimeout\scCs¡|jjdddgƒ|j|jjjƒ|jjdƒ|jjdddgƒ|j|jjjƒ|jjdddgƒ|j|jjjƒdS(sR Make sure that receiving data also resets the timeout countdown. ggð?gà?sbytes bytes bytesN(R0RŠR‹RJR=RRRŒ(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_receiveAvoidsTimeoutys(RRR-R‰RRR‘(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR†8s    t TimeoutTestercBsSeZdZdZeZd„Zd„Zd„Zdd„Z d„Z d„Z RS( s” A testable protocol with timeout facility. @ivar timedOut: set to C{True} if a timeout has been detected. @type timedOut: C{bool} icCs ||_dS(sF Initialize the protocol with a C{task.Clock} object. N(R0(RR0((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR™scCs|j|jƒdS(s3 Upon connection, set the timeout. N(t setTimeoutttimeOut(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR scCs!|jƒtjj||ƒdS(s, Reset the timeout on data. N(t resetTimeoutRRER(RR((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR§s cCs|jdƒdS(sD On connection lost, cancel all timeout operations. N(R“RF(RR((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR¯scCs t|_dS(sX Flags the timedOut variable to indicate the timeout of the connection. N(R$ttimedOut(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttimeoutConnection¶scOs|jj||||ŽS(sD Override callLater to use the deterministic clock. (R0R3(RttimeoutR5R1R2((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR3½sN( RRR-R”R'R–RRRRFRR—R3(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR’s     t TestTimeoutcBsMeZdZd„Zd„Zd„Zd„Zd„Zd„Zd„Z RS(s- Tests for L{policies.TimeoutMixin}. cCs%tjƒ|_t|jƒ|_dS(sY Create a testable, deterministic clock and a C{TimeoutTester} instance. N(R RxR0R’RJ(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR‰ÊscCs0|jjdƒ|jt|jjƒdƒdS(sg Test that the callLater of the clock is used instead of C{reactor.callLater}. i iN(RJR“R\R]R0tcalls(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_overriddenCallLaterÒscCso|jjtƒƒ|jjddddgƒ|j|jjƒ|jjddgƒ|j|jjƒdS(sq Check that the protocol does timeout at the time specified by its C{timeOut} attribute. igà?gð?N(RJRGRR0RŠR‹R–RŒ(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRÛs cCs®|jjtƒƒ|jjddddgƒ|j|jjƒ|jjdƒ|jjddddgƒ|j|jjƒ|jjddgƒ|j|jjƒdS(sV Check that receiving data is delaying the timeout of the connection. igà?gð?s hello thereN( RJRGRR0RŠR‹R–RRŒ(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_noTimeoutéscCs›d|j_|jjtƒƒ|jjdƒ|j|jjdƒ|jjddgƒ|j |jj ƒ|jjddgƒ|j |jj ƒdS(sy Check that setting a new value for timeout cancel the previous value and install a new timeout. iigÍÌÌÌÌÌì?gš™™™™™É?N( RFRJR”RGRR“R\R0RŠR‹R–RŒ(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_resetTimeoutøs cCsxd|j_|jjtƒƒ|jjdƒ|j|jjdƒ|jjddddgƒ|j |jj ƒdS(sO Setting the timeout to C{None} cancel any timeout operations. iiN( RJR”RGRR“RFR\R0RŠR‹R–(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_cancelTimeout s  cCsŠd|j_|j|jjdƒdƒ|j|jjdƒdƒ|j|jjdƒdƒ|j|jjdƒ|jjdƒdS(sM setTimeout should return the value of the previous timeout. ii iN(RJR”R\R“RF(R((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt test_returns  ( RRR-R‰R›RRœRRžRŸ(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR™Ås     t$LimitTotalConnectionsFactoryTestCasecBs eZdZd„Zd„ZRS(s/Tests for policies.LimitTotalConnectionsFactorycCs³tjƒ}tj|_|jd|jƒ|jdƒ}|jd|jƒ|jdƒ}|jd|jƒ|jdƒ|jd|jƒ|jdƒ|jd|jƒdS(Niii( R tLimitTotalConnectionsFactoryRREt assertEqualtconnectionCountR"RFR(RR>tp1tp2((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttestConnectionCounting)s    cs6tjƒ‰tjˆ_dˆ_ˆjdƒ}|jd|ƒ|jdˆj ƒ|jdˆjdƒƒ|jdˆj ƒdtjf‡fd†ƒY}|ˆ_ t ˆ_ ˆjdƒ}|j dƒ|jtˆj ƒ|jdˆj ƒ|jdƒ|jdˆj ƒ|jdƒ|jdˆj ƒdS(NitOverflowProtocolcseZ‡fd†ZRS(cs tˆ_dS(N(R$t overflowed(R(R>(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyROs(RRR((R>(s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR§Nsii(R R¡RREtconnectionLimitR"RFtassertNotEqualR¢R£toverflowProtocolR'R¨RGR$R(RR R§top((R>s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttestConnectionLimiting=s&        (RRR-R¦R­(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR 's tWriteSequenceEchoProtocolcBseZd„ZRS(cCs?|jdƒdkr+|jj|gƒntj||ƒdS(Nsvector!iÿÿÿÿ(tfindR*RR#R(RRŽ((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRds(RRR(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR®cstTestLoggingFactorycBseZdZd„ZRS(cCs+|jdkstd‚tƒ|_|jS(Nsopen() called too many times(topenFileRFtAssertionErrorR(Rtname((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pytopenls N(RRRFR±R´(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyR°jstLoggingFactoryTestCasecBs eZdZd„Zd„ZRS(s6 Tests for L{policies.TrafficLoggingFactory}. cCsYtƒ}t|_tƒ}t|dƒ}|jd ƒ}||_|j|ƒ|jjƒ}|j d|kd|fƒ|j |j ƒƒ|j dƒ|jjƒ}|j d|ƒ|j d|ƒ|j|j ƒdƒ|jƒ|j d ƒ|jjƒ}|j d |ƒ|j|j ƒd ƒ|jƒ|jjƒ}|j d |ƒd S(sQ Check the output produced by L{policies.TrafficLoggingFactory}. ttests1.2.3.4i.t*s* not found in %rshere are some bytessC 1: 'here are some bytes'sS 1: 'here are some bytes's"prepare for vector! to the extremes,SV 1: ['prepare for vector! to the extreme']tConnectionDoneN(s1.2.3.4i.(R,R®RRR°R"RGR±tgetvalueRŒR‹RzRtassertInR\R„Rc(RRˆttR?R tv((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyttest_thingsGetLoggedxs.         cCs•tƒ}t|dƒ}|j|jdƒ|jd ƒ|j|jdƒd|_|jd ƒ|j|jdƒ|jƒ|j|jdƒdS( sG Test counter management with the resetCounter method. R¶is1.2.3.4i.ii/iN(s1.2.3.4i.(s1.2.3.4i/(R,R°R¢t_counterR"RFR±t resetCounter(RRˆR?((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyt test_counter›s     (RRR-R½RÀ(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyRµss #N(+R-tzope.interfaceRRRRt twisted.trialRttwisted.test.proto_helpersRRttwisted.internetRRR R R ttwisted.protocolsR RER t ClientFactoryRR#R‡R,R/R.R7R6tTestCaseR8R:RQR†t TimeoutMixinR’R™R R®tTrafficLoggingFactoryR°Rµ(((s>/usr/lib/python2.7/dist-packages/twisted/test/test_policies.pyts.( , ˆW6b<