ó [³XMc@sHdZddlmZddlmZddlmZddlmZm Z ddl m Z dZ yddlmZWnek rdZ nXdd lmZdd lmZdd lmZd efd „ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZdS(s> Tests for L{twisted.internet.posixbase} and supporting code. iÿÿÿÿ(tset(tTestCase(tDeferred(tPosixReactorBaset_Waker(t ServerFactory(tunixs)Platform does not support AF_UNIX sockets(tPort(treactor(t ClientPrototTrivialReactorcBs5eZd„Zd„Zd„Zd„Zd„ZRS(cCs#i|_i|_tj|ƒdS(N(t_readerst_writersRt__init__(tself((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyR s  cCst|j|¿s(RRRt connectedtconnectionLostt assertFailuret stopListeningtZeroDivisionError(Rtport((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_connectionLostFailed¸s  (RRR>t isinstanceRRtskipRH(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyR?°s tTimeoutReportReactorcBs;eZdZd„Zd„Zd„Zd„Zd„ZRS(sà A reactor which is just barely runnable and which cannot monitor any readers or writers, and which fires a L{Deferred} with the timeout passed to its C{doIteration} method as soon as that method is invoked. cCs&tj|ƒtƒ|_d|_dS(Nid(RR RtiterationTimeouttnow(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyR Ês  cCsdS(s— Ignore the reader. This is necessary because the waker will be added. However, we won't actually monitor it for any events. N((RR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRÐscCsgS(s° There are no readers or writers, so there is nothing to remove. This will be called when the reactor stops, though, so it must be implemented. ((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyt removeAll×scCs|jS(sx Override the real clock with a deterministic one that can be easily controlled in a unit test. (RM(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pytsecondsàscCs2|j}|dk r.d|_|j|ƒndS(N(RLtNonetcallback(Rttimeouttd((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyt doIterationès   (RRR>R RRNRORT(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRKÄs    tIterationTimeoutTestscBsVeZdZd„Zd„Zd„Zd„Zd„Zd„Zd„Z d„Z RS( s Tests for the timeout argument L{PosixReactorBase.run} calls L{PosixReactorBase.doIteration} with in the presence of various delayed calls. csDg}ˆjj|jƒˆjj‡fd†ƒˆjƒ|dS(Ncs ˆjƒS(N(tstop(tignored(R(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRAùsi(RLt addCallbacktappendtrun(RRRR((RsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyt_checkIterationTimeoutös  cCs,tƒ}|j|ƒ}|j|dƒdS(sl If there are no delayed calls, C{doIteration} is called with a timeout of C{None}. N(RKR[R/RP(RRRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyt test_noCallsþs cCs?tƒ}|jdd„ƒ|j|ƒ}|j|dƒdS(sÁ If there is a delayed call, C{doIteration} is called with a timeout which is the difference between the current time and the time at which that call is to run. idcSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRAsN(RKt callLaterR[R/(RRRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_delayedCalls cCsNtƒ}|jdd„ƒ|jd7_|j|ƒ}|j|dƒdS(s¯ If a delayed call is scheduled and then some time passes, the timeout passed to C{doIteration} is reduced by the amount of time which passed. idcSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRAsiiKN(RKR]RMR[R/(RRRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_timePassess  cCsetƒ}|jdd„ƒ|jdd„ƒ|jdd„ƒ|j|ƒ}|j|dƒdS(sÜ If there are several delayed calls, C{doIteration} is called with a timeout which is the difference between the current time and the time at which the earlier of the two calls is to run. i2cSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRA(si cSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRA)sidcSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRA*sN(RKR]R[R/(RRRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_multipleDelayedCalls!s  cCs]tƒ}|jdd„ƒ}|jd7_|jdƒ|j|ƒ}|j|dƒdS(sÀ If a delayed call is reset, the timeout passed to C{doIteration} is based on the interval between the time when reset is called and the new delay of the call. i2cSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRA6siiN(RKR]RMtresetR[R/(RRtcallRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_resetDelayedCall/s   cCs]tƒ}|jdd„ƒ}|jd7_|jdƒ|j|ƒ}|j|dƒdS(sñ If a delayed call is re-delayed, the timeout passed to C{doIteration} is based on the remaining time before the call would have been made and the additional amount of time passed to the delay method. i2cSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRAEsi ii<N(RKR]RMtdelayR[R/(RRRbRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_delayDelayedCall=s   cCsKtƒ}|jdd„ƒ}|jƒ|j|ƒ}|j|dƒdS(sp If the only delayed call is canceled, C{None} is the timeout passed to C{doIteration}. i2cSsdS(N(RP(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRARsN(RKR]tcancelR[R/RP(RRRbRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyttest_cancelDelayedCallLs   ( RRR>R[R\R^R_R`RcReRg(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRUðs    tConnectedDatagramPortTestCasecBs5eZdZedk r!eZnd„Zd„ZRS(s/ Test connected datagram UNIX sockets. cs>‡fd†}tjdtƒƒ}||_|jdƒdS(s L{ConnectedDatagramPort} does not call the deprecated C{loseConnection} in L{ConnectedDatagramPort.connectionFailed}. csˆjdƒdS(s€ Dummy C{loseConnection} method. C{loseConnection} is deprecated and should not get called. s7loseConnection is deprecated and should not get called.N(tfail((R(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pytloseConnectionfstgoodbyeN(RtConnectedDatagramPortRPR RjtconnectionFailed(RRjRG((RsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyt-test_connectionFailedDoesntCallLoseConnectionas csZtˆ_‡fd†}tjdtƒƒ}||_|jdƒˆjˆjt ƒdS(sÁ L{ConnectedDatagramPort} calls L{ConnectedDatagramPort.stopListening} instead of the deprecated C{loseConnection} in L{ConnectedDatagramPort.connectionFailed}. cs tˆ_dS(s8 Dummy C{stopListening} method. N(Rtcalled((R(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyREzsRkN( tFalseRoRRlRPR RERmR/R(RRERG((RsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyt'test_connectionFailedCallsStopListeningrs    N(RRR>t skipSocketsRPRJRnRq(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyRhYs    N(R>ttwisted.python.compatRttwisted.trial.unittestRttwisted.internet.deferRttwisted.internet.posixbaseRRttwisted.internet.protocolRRPRrR+Rt ImportErrorttwisted.internet.tcpRRttwisted.test.test_unixR R RR?RKRURh(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_posixbase.pyts&  ,i