o >h9@s~dZddlZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z dd lmZmZmZdd lmZdd lmZdd lmZmZmZdd lmZddlmZddlmZm Z zddl!Z"Wn e#yodZ!Ynwe"Z!e dZ$e eeGdddZ%Gddde Z&ddZ'GdddZ(Gddde(e Z)Gddde(e Z*GdddeZ+ee! dGd d!d!e Z,dS)"z% Tests for L{twisted.internet.base}. N)Queue)Callable)skipIf) implementer) ParamSpec) FirstOneWins) DelayedCall ReactorBaseThreadedResolver)Deferred)DNSLookupError)IReactorThreads IReactorTimeIResolverSimple)Clock) ThreadPool)SkipTestTestCase_Pc@seZdZdZddZdeeefdejdej ddfd d Z d d Z d dZ ddZ defddZdeeefdejdej ddfddZddZdS) FakeReactorzl A fake reactor implementation which just supports enough reactor APIs for L{ThreadedResolver}. cs>t_jj_t_jfdd_t_dS)NcsjSN) _threadpoolselfr/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/twisted/internet/test/test_base.py1sz&FakeReactor.__init__..) r_clock callLaterrrstart getThreadPoolr _threadCallsrrrr__init__+s    zFakeReactor.__init__callableargskwargsreturnNcOs|j|||fdSr)r!putrr#r$r%rrrcallFromThread5szFakeReactor.callFromThreadcCs"|j\}}}||i|dSr)r!get)rfr$r%rrr_runThreadCalls:szFakeReactor._runThreadCallscCs|jdSr)rstoprrrr_stop>szFakeReactor._stopcCdSrrrrrrgetDelayedCallsAzFakeReactor.getDelayedCallscCr/rrrrrrsecondsEr1zFakeReactor.secondscOr/rrr(rrr callInThreadIszFakeReactor.callInThreadcCr/rr)rsizerrrsuggestThreadPoolSizeOr1z!FakeReactor.suggestThreadPoolSize)__name__ __module__ __qualname____doc__r"rrobjectr$r%r)r,r.r0floatr2r3r5rrrrr$s4    rc@0eZdZdZddZddZddZdd Zd S) ThreadedResolverTestsz( Tests for L{ThreadedResolver}. csdd}d}t}||jgg}fdd}|td|t|}|||f}||j| | |g| |g|j |d| |j j gdS) z 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__}. z 10.0.0.17zfoo.bar.example.comcs|Sr)appendnameiplookedUprrfakeGetHostByNameis z=ThreadedResolverTests.test_success..fakeGetHostByName gethostbynameN)r addCleanupr.patchsocketr getHostByName addCallbackr?r, assertEqualradvancecalls)rrAtimeoutreactor resolvedTorEresolverdrrBr test_successYs"  z"ThreadedResolverTests.test_successcCsd}t}||jdd}|td|g}t|}|d|f}||t| |j | | t |d|j|d| |jjgdS)z L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires a L{Failure} if the call to L{socket.gethostbyname} raises an exception. r>cSstd)NzENOBUFS (this is a funny joke))OSErrorr@rrrrEz=ThreadedResolverTests.test_failure..fakeGetHostByNamerF some.namerGN)rrHr.rIrJr rK assertFailurer rLr?r,rMlenrrNrOrrPrQrE failedWithrSrTrrr test_failure|s   z"ThreadedResolverTests.test_failurecsd}t}||jtfdd}|td|g}t|}|d|f}||t | |j |j |d||g|j d|t|dtddS) z If L{socket.gethostbyname} does not complete before the specified timeout elapsed, the L{Deferred} returned by L{ThreadedResolver.getHostByName} fails with L{DNSLookupError}. csr)r*r@resultrrrErWz=ThreadedResolverTests.test_timeout..fakeGetHostByNamerFrXrGzThe I/O was errorfulN)rrHr.rrIrJr rKrYr rLr?rrNrMrZr'IOErrorr[rr_r test_timeouts       z"ThreadedResolverTests.test_timeoutcsgttGfddd}Gdddt}|}|}||tt}|j|d|j|d|j|d|j|d|j|d|t d |t t t t gd |dd |dd |d d |dd|dddS)zm L{ThreadedResolver.getHostByName} is passed L{str}, encoded using IDNA if required. cseZdZdfdd ZdS)zAThreadedResolverTests.test_resolverGivenStr..FakeResolverrcs|tSr)r?r )rrAtimeoutsrOrrrKs zOThreadedResolverTests.test_resolverGivenStr..FakeResolver.getHostByNameN)r)r6r7r8rKrrdrr FakeResolversrec@seZdZddZdS)zFThreadedResolverTests.test_resolverGivenStr..JustEnoughReactorcSr/rrrrrr installWakerzSThreadedResolverTests.test_resolverGivenStr..JustEnoughReactor.installWakerN)r6r7r8rfrrrrJustEnoughReactors rhzexample.exampleuvääntynyt.exampleu рф.examplezxn----7sbb4ac0ad0be6cf.xn--p1airrGzxn--vntynyt-5waa.examplezxn--p1ai.exampleN)rrr installResolverrr nameResolverresolveHostNamerMrZlistmaptypestr)rrerhfakerQrecrrdrtest_resolverGivenStrs(  z+ThreadedResolverTests.test_resolverGivenStrN)r6r7r8r9rUr]rbrvrrrrr=Ts # !r=cCdS)z8 Function used by L{DelayedCallTests.test_str}. Nrrrrrnothingrxc@s`eZdZdZddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ dS)DelayedCallMixin L{DelayedCall} cCs dd}t|dddi||dS)z Get a L{DelayedCall} instance at a given C{time}. @param time: The absolute time at which the returned L{DelayedCall} will be scheduled. cSr/rr)callrrrnooprgz0DelayedCallMixin._getDelayedCallAt..noopcSr/rrrrrrrryz4DelayedCallMixin._getDelayedCallAt..rN)r)rtimer}rrr_getDelayedCallAtsz"DelayedCallMixin._getDelayedCallAtcCs|d|_|d|_dS)zb Create two L{DelayedCall} instanced scheduled to run at different times. rrGN)rzeroonerrrrsetUps zDelayedCallMixin.setUpc Cs:tdtdddidddd}|t|dt|fdS) The string representation of a L{DelayedCall} instance, as returned by L{str}, includes the unsigned id of the instance, as well as its state, the function to be called, and the function arguments. rkAriNcSrwNg?rrrrrrryz+DelayedCallMixin.test_str..z?)rrxrMrsidrdcrrrtest_strszDelayedCallMixin.test_strc Cs4tdtdddidddd}|t|t|dS)z The string representation of a L{DelayedCall} instance, as returned by {repr}, is identical to that returned by L{str}. )r NcSrw)Ng?rrrrrrryz,DelayedCallMixin.test_repr..)rrxrMrsreprrrrr test_repr szDelayedCallMixin.test_reprcCsJ|j|j}}|||k|||k|||k|||kdS)z 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}. Nrr assertTrue assertFalserrrrrrtest_lt zDelayedCallMixin.test_ltcCsJ|j|j}}|||k|||k|||k|||kdS)z 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}. Nrrrrrtest_le zDelayedCallMixin.test_lecCsJ|j|j}}|||k|||k|||k|||kdS)z 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}. Nrrrrrtest_gt*rzDelayedCallMixin.test_gtcCsJ|j|j}}|||k|||k|||k|||kdS)z 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}. Nrrrrrtest_ge5rzDelayedCallMixin.test_gecCs:||j|jk||j|jk||j|jkdS)zD A L{DelayedCall} instance is only equal to itself. N)rrrrrrrrtest_eqAzDelayedCallMixin.test_eqcCs:||j|jk||j|jk||j|jkdS)zM A L{DelayedCall} instance is not equal to any other object. N)rrrrrrrrtest_neKrzDelayedCallMixin.test_neN)r6r7r8r9rrrrrrrrrrrrrrrzs      rzc@(eZdZdZddZddZddZdS) DelayedCallNoDebugTestsr{cC|tddt|dS)z! Turn debug off. debugFNrIrrzrrrrrr[zDelayedCallNoDebugTests.setUpc C>tdtdddidddd}dt|}|t||dS) rrrrriNcSrwrrrrrrrhryz2DelayedCallNoDebugTests.test_str..zA)rrxformatrrMrs)rrexpectedrrrrbs  z DelayedCallNoDebugTests.test_strc C6tddddittdd}d|_|dt|dS) z If L{DelayedCall.debug} changes from C{0} to C{1} between L{DelayeCall.__init__} and L{DelayedCall.__repr__} then L{DelayedCall.__repr__} returns a string that does not include the creator stack. rkcSr/rrrrrrrvryz.rcSrwNrjrrrrrrvryrGtraceback at creationN)rrxr assertNotInrrrrrtest_switchToDebugoz*DelayedCallNoDebugTests.test_switchToDebugN)r6r7r8r9rrrrrrrrVs  rc@r) DelayedCallDebugTestsr{cCr)z Turn debug on. rTNrrrrrrrzDelayedCallDebugTests.setUpc Cr) rrrrriNcSrwrrrrrrrryz0DelayedCallDebugTests.test_str..z\.rcSrwrrrrrrrryrrN)rrxrassertInrrrrrtest_switchFromDebugrz*DelayedCallDebugTests.test_switchFromDebugN)r6r7r8r9rrrrrrrr{s  rc@seZdZdZddZdS)TestSpySignalCapturingReactorza Subclass of ReactorBase to capture signals delivered to the reactor for inspection. cCrw)z* Required method, unused. Nrrrrrrfryz*TestSpySignalCapturingReactor.installWakerN)r6r7r8r9rfrrrrrs rzsignal module not availablec@r<) ReactorBaseSignalTestszE Tests to exercise ReactorBase's signal exit reporting path. cCst}|d|jdS)zI The default value of the _exitSignal attribute is None. N)rassertIs _exitSignalrrQrrrtest_exitSignalDefaultsToNonesz4ReactorBaseSignalTests.test_exitSignalDefaultsToNonecC(t}|tjd|tj|jdS)zn ReactorBase's SIGINT handler saves the value of SIGINT to the _exitSignal attribute. N)rsigIntsignalSIGINT assertEqualsrrrrrtest_captureSIGINTz)ReactorBaseSignalTests.test_captureSIGINTcCr)zp ReactorBase's SIGTERM handler saves the value of SIGTERM to the _exitSignal attribute. N)rsigTermrSIGTERMrrrrrrtest_captureSIGTERMrz*ReactorBaseSignalTests.test_captureSIGTERMcCs:ttds tdt}|tjd|tj|jdS)zr ReactorBase's SIGBREAK handler saves the value of SIGBREAK to the _exitSignal attribute. SIGBREAKz$signal module does not have SIGBREAKN)hasattrrrrsigBreakrrrrrrrtest_captureSIGBREAKs z+ReactorBaseSignalTests.test_captureSIGBREAKN)r6r7r8r9rrrrrrrrrs  r)-r9rJqueuertypingrunittestrzope.interfacertyping_extensionsrtwisted.internet._resolverrtwisted.internet.baserr r twisted.internet.deferr twisted.internet.errorr twisted.internet.interfacesr rrtwisted.internet.taskrtwisted.python.threadpoolrtwisted.trial.unittestrrr_signal ImportErrorrrr=rxrzrrrrrrrrsB            / r%&