ó [³XMc@sôdZddlZddlmZddlmZddlmZddlmZddl m Z m Z ddl m Z dd lmZmZdd lmZdd lmZd efd „ƒYZdefd„ƒYZdefd„ƒYZdS(s% Tests for L{twisted.internet.base}. iÿÿÿÿN(tQueue(t implements(t ThreadPool(t setIDFunction(t IReactorTimetIReactorThreads(tDNSLookupError(tThreadedResolvert DelayedCall(tClock(tTestCaset FakeReactorcBs?eZdZeeeƒd„Zd„Zd„Zd„Z RS(sl A fake reactor implementation which just supports enough reactor APIs for L{ThreadedResolver}. csVtƒˆ_ˆjjˆ_tƒˆ_ˆjjƒ‡fd†ˆ_tƒˆ_dS(NcsˆjS(N(t _threadpool((tself(sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt#s( R t_clockt callLaterRR tstartt getThreadPoolRt _threadCalls(R ((R sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt__init__s    cOs|jj|||fƒdS(N(Rtput(R tftargstkwargs((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pytcallFromThread(scCs)|jjƒ\}}}|||ŽdS(N(Rtget(R RRR((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt_runThreadCalls,scCs|jjƒdS(N(R tstop(R ((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt_stop1s( t__name__t __module__t__doc__RRRRRRR(((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyR s    tThreadedResolverTestscBs)eZdZd„Zd„Zd„ZRS(s( Tests for L{ThreadedResolver}. csëd‰d}d}tƒ}|j|jƒg‰g}‡‡fd†}|jtd|ƒt|ƒ}|j||fƒ}|j|jƒ|j ƒ|j ˆ|gƒ|j |ˆgƒ|j j |dƒ|j |j j gƒdS(sí L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires with the value returned by the call to L{socket.gethostbyname} in the threadpool of the reactor passed to L{ThreadedResolver.__init__}. s 10.0.0.17sfoo.bar.example.comicsˆj|ƒˆS(N(tappend(tname(tiptlookedUp(sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pytfakeGetHostByNameIs t gethostbynameiN(R t addCleanupRtpatchtsocketRt getHostByNamet addCallbackR"Rt assertEqualRtadvancetcalls(R R#ttimeouttreactort resolvedToR&tresolvertd((R$R%sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt test_success:s"   cCsÐd}tƒ}|j|jƒd„}|jtd|ƒg}t|ƒ}|jd|fƒ}|j|tƒ|j |j ƒ|j ƒ|j t |ƒdƒ|jj|dƒ|j |jjgƒdS(s¦ L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires a L{Failure} if the call to L{socket.gethostbyname} raises an exception. icSstdƒ‚dS(NsENOBUFS (this is a funny joke)(tIOError(R#((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyR&gsR's some.nameiN(R R(RR)R*RR+t assertFailureRR,R"RR-tlenRR.R/(R R0R1R&t failedWithR3R4((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt test_failure]s    csòd}tƒ}|j|jƒtƒ‰‡fd†}|jtd|ƒg}t|ƒ}|jd|fƒ}|j|t ƒ|j |j ƒ|j j |dƒ|j|gƒ|j j dƒ|jt|ƒdƒˆjtdƒƒdS(sÐ If L{socket.gethostbyname} does not complete before the specified timeout elapsed, the L{Deferred} returned by L{ThreadedResolver.getHostByBame} fails with L{DNSLookupError}. i csˆjƒ‚dS(N(R(R#(tresult(sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyR&‡sR's some.nameisThe I/O was errorfulN(R R(RRR)R*RR+R7RR,R"RR.R-R8RR6(R R0R1R&R9R3R4((R;sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt test_timeout{s    (RRR R5R:R<(((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyR!6s # tDelayedCallTestscBs_eZdZd„Zd„Zd„Zd„Zd„Zd„Zd„Z d„Z d „Z RS( s# Tests for L{DelayedCall}. cCs(d„}t|d„di||dƒS(s· Get a L{DelayedCall} instance at a given C{time}. @param time: The absolute time at which the returned L{DelayedCall} will be scheduled. cSsdS(N((tcall((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pytnoop¨scSsdS(N(tNone(((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyRªs(N(RR@(R ttimeR?((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyt_getDelayedCallAt¡s cCs(|jdƒ|_|jdƒ|_dS(sb Create two L{DelayedCall} instanced scheduled to run at different times. iiN(RBtzerotone(R ((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pytsetUp­scs}d„}td|d idd6d d d„ƒ}id|6‰‡fd†}|jtt|ƒƒ|jt|ƒd ƒd S( sç The string representation of a L{DelayedCall} instance, as returned by C{str}, includes the unsigned id of the instance, as well as its state, the function to be called, and the function arguments. cSsdS(N((((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pytnothing¼si iitAcSsdS(Ngø?((((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyR¾siÈcs1y ˆ|SWnttfk r,t|ƒSXdS(N(t TypeErrortKeyErrortid(tobj(tids(sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pytfakeIDÀs s?N(i(RR@R(Rt assertEqualststr(R RFtdcRM((RLsC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_str¶s (  cCsc|j|j}}|j||kƒ|j||kƒ|j||kƒ|j||kƒdS(s’ For two instances of L{DelayedCall} C{a} and C{b}, C{a < b} is true if and only if C{a} is scheduled to run before C{b}. N(RCRDt assertTruet assertFalse(R RCRD((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_ltËs cCsc|j|j}}|j||kƒ|j||kƒ|j||kƒ|j||kƒdS(s· For two instances of L{DelayedCall} C{a} and C{b}, C{a <= b} is true if and only if C{a} is scheduled to run before C{b} or at the same time as C{b}. N(RCRDRRRS(R RCRD((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_le×s cCsc|j|j}}|j||kƒ|j||kƒ|j||kƒ|j||kƒdS(s‘ For two instances of L{DelayedCall} C{a} and C{b}, C{a > b} is true if and only if C{a} is scheduled to run after C{b}. N(RCRDRRRS(R RCRD((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_gtäs cCsc|j|j}}|j||kƒ|j||kƒ|j||kƒ|j||kƒdS(sµ For two instances of L{DelayedCall} C{a} and C{b}, C{a > b} is true if and only if C{a} is scheduled to run after C{b} or at the same time as C{b}. N(RCRDRRRS(R RCRD((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_geðs cCsO|j|j|jkƒ|j|j|jkƒ|j|j|jkƒdS(sD A L{DelayedCall} instance is only equal to itself. N(RSRCRDRR(R ((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_eqýscCsO|j|j|jkƒ|j|j|jkƒ|j|j|jkƒdS(sM A L{DelayedCall} instance is not equal to any other object. N(RRRCRDRS(R ((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyttest_nes( RRR RBRERQRTRURVRWRXRY(((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyR=s  (R R*Rtzope.interfaceRttwisted.python.threadpoolRttwisted.python.utilRttwisted.internet.interfacesRRttwisted.internet.errorRttwisted.internet.baseRRttwisted.internet.taskR ttwisted.trial.unittestR tobjectR R!R=(((sC/usr/lib/python2.7/dist-packages/twisted/internet/test/test_base.pyts  g