tf=.dZddlZddlmZddlmZddlmZmZddl m Z ddl m Z ddl mZdZ dd lmZdd lmZn #e$rd ZYnwxYwdd lmZdd lmZGddeZGddeZGddeZGddeZGddeZGddeZGddeZGddeZdS)z> Tests for L{twisted.internet.posixbase} and supporting code. N)Deferred)IReadDescriptor)PosixReactorBase_Waker) ServerFactory)platform)TestCase)unix) ClientProtoz)Platform does not support AF_UNIX sockets)reactor)Portc,eZdZdZfdZfdZxZS)WarningCheckerTestCasezg A test case that will make sure that no warnings are left unchecked at the end of a test run. c ttjdddkr)t jr|dSdSdS)NCItrue) supersetUposenvirongetlowerr isWindows flushWarnings)self __class__s c/var/www/surfInsights/venv3-11/lib/python3.11/site-packages/twisted/internet/test/test_posixbase.pyrzWarningCheckerTestCase.setUp sq   :>>$ # # ) ) + +v 5 5(:L:N:N 5     6 5 5 5cp t|}tjdddkrtjrdS| t|dd|dS#|}tjdddkrtjrYdS| t|dd|wxYw)Nrrrrz'Warnings found at the end of the test: ) rtearDownrrrrrrr assertEquallen)rwarningsrs rr!zWarningCheckerTestCase.tearDown+s-  GG     ))++Hz~~dB''--//699h>P>R>R9   H q"WX"W"W     ))++Hz~~dB''--//699h>P>R>R9   H q"WX"W"W    s B++AD5 (D5)__name__ __module__ __qualname____doc__rr! __classcell__)rs@rrrs[ ! ! ! ! !rrc,eZdZdZdZdZdZdZdS)TrivialReactorcJi|_i|_tj|dSN)_readers_writersr__init__rs rr0zTrivialReactor.__init__>s'  !$'''''rcd|j|<dSNTr.rreaders r addReaderzTrivialReactor.addReaderC $ frc|j|=dSr-r4r5s r removeReaderzTrivialReactor.removeReaderF M& ! ! !rcd|j|<dSr3r/rwriters r addWriterzTrivialReactor.addWriterIr8rc|j|=dSr-r=r>s r removeWriterzTrivialReactor.removeWriterLr;rN)r%r&r'r0r7r:r@rBrrr+r+=s_((( %%%"""%%%"""""rr+c*eZdZdZdZdZdZdZdS)PosixReactorBaseTestsz( Tests for L{PosixReactorBase}. c||jt||j|j||j|jdSr-)assertIsInstancewakerrassertIn_internalReadersr.rr s r _checkWakerz!PosixReactorBaseTests._checkWakerUsR gmV444 gmW%=>>> gmW%566666rcLt}||dS)z When L{PosixReactorBase} is instantiated, it creates a waker and adds it to its internal readers set. N)r+rLrKs rtest_wakerIsInternalReaderz0PosixReactorBaseTests.test_wakerIsInternalReaderZs) !"" !!!!!rc^t}t}|j||j||j|j|j||| ||j| ||jdS)z Any L{IReadDescriptor}s in L{PosixReactorBase._internalReaders} are left alone by L{PosixReactorBase._removeAll}. N) r+objectrJaddr7 _removeAllr.r/rLrI)rr extras r"test_removeAllSkipsInternalReadersz8PosixReactorBaseTests.test_removeAllSkipsInternalReadersbs !"" $$U+++%   7+W-=>>> !!! eW5666 eW-.....rct}t}t}|j||j||j|j|j}|t|||h| ||j| ||jdS)z L{PosixReactorBase._removeAll} returns a list of removed L{IReadDescriptor} and L{IWriteDescriptor} objects. N) r+rPr7r@rRr.r/r"set assertNotIn)rr r6r?removeds r'test_removeAllReturnsRemovedDescriptorsz=PosixReactorBaseTests.test_removeAllReturnsRemovedDescriptorsps !""&!!!&!!!$'$W%5w7GHH W'7888 !1222 !122222rN)r%r&r'r(rLrNrTrYrCrrrErEPsZ777 """ / / / 3 3 3 3 3rrEc4eZdZdZeeesdZdZdS) TCPPortTestsz1 Tests for L{twisted.internet.tcp.Port}. zNon-posixbase reactorctdt}d|_d|_||t S)z L{Port.stopListening} returns a L{Deferred} which errbacks if L{Port.connectionLost} raises an exception. i90Tc ddzS)NrrC)reasons rz8TCPPortTests.test_connectionLostFailed..s Q!Vr)r r connectedconnectionLost assertFailure stopListeningZeroDivisionError)rports rtest_connectionLostFailedz&TCPPortTests.test_connectionLostFailedsL E=??++33!!$"4"4"6"68IJJJrN) r%r&r'r( isinstancer rskiprgrCrrr[r[sN :g/ 0 0'&KKKKKrr[cJeZdZdZdZdeddfdZdeddfdZdZd Z d Z dS) TimeoutReportReactorz 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. cbtj|t|_d|_dS)Nd)rr0riterationTimeoutnowr1s rr0zTimeoutReportReactor.__init__s*!$''' ( rr6returnNcdS)z Ignore the reader. This is necessary because the waker will be added. However, we won't actually monitor it for any events. NrCr5s rr7zTimeoutReportReactor.addReaderrcdS)z# See L{addReader}. NrCr5s rr:z!TimeoutReportReactor.removeReaderrrrcgS)z 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. rCr1s r removeAllzTimeoutReportReactor.removeAlls  rc|jS)zx Override the real clock with a deterministic one that can be easily controlled in a unit test. )ror1s rsecondszTimeoutReportReactor.secondss xrcT|j}|d|_||dSdSr-)rncallback)rtimeoutds r doIterationz TimeoutReportReactor.doIterations6  ! =$(D ! JJw      =r) r%r&r'r(r0rr7r:rurwr|rCrrrkrks    D     ? t         rrkcBeZdZdZdZdZdZdZdZdZ dZ d Z d S) IterationTimeoutTestsz Tests for the timeout argument L{PosixReactorBase.run} calls L{PosixReactorBase.doIteration} with in the presence of various delayed calls. cg}j|jjfdj|dS)Nc"jSr-)stop)ignoredr s rr`z>IterationTimeoutTests._checkIterationTimeout..s\W\^^rr)rn addCallbackappendrunrr rzs ` r_checkIterationTimeoutz,IterationTimeoutTests._checkIterationTimeoutsY ,,W^<<< ,,-K-K-K-KLLL qzrcvt}||}||dS)zl If there are no delayed calls, C{doIteration} is called with a timeout of L{None}. N)rkr assertIsNoners r test_noCallsz"IterationTimeoutTests.test_noCallss< '((--g66 '"""""rct}|jdd||}||ddS)z 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. rmcdSr-rCrCrrr`z8IterationTimeoutTests.test_delayedCall..trNrk callLaterrr"rs rtest_delayedCallz&IterationTimeoutTests.test_delayedCallsU '((#||,,,--g66 #&&&&&rct}|jdd|xjdz c_||}||ddS)z 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. rmcdSr-rCrCrrr`z7IterationTimeoutTests.test_timePasses..rrKN)rkrrorr"rs rtest_timePassesz%IterationTimeoutTests.test_timePassesse '((#||,,, r --g66 "%%%%%rct}|jdd|jdd|jdd||}||ddS)z 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. 2cdSr-rCrCrrr`zAIterationTimeoutTests.test_multipleDelayedCalls..dr cdSr-rCrCrrr`zAIterationTimeoutTests.test_multipleDelayedCalls..rrrmcdSr-rCrCrrr`zAIterationTimeoutTests.test_multipleDelayedCalls..rrNrrs rtest_multipleDelayedCallsz/IterationTimeoutTests.test_multipleDelayedCallss '(("ll+++"ll+++#||,,,--g66 "%%%%%rct}|jdd}|xjdz c_|d||}||ddS)z 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. rcdSr-rCrCrrr`z=IterationTimeoutTests.test_resetDelayedCall..TrrN)rkrroresetrr"rr callrzs rtest_resetDelayedCallz+IterationTimeoutTests.test_resetDelayedCallss '(( w \\22 r  2--g66 "%%%%%rct}|jdd}|xjdz c_|d||}||ddS)z 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. rcdSr-rCrCrrr`z=IterationTimeoutTests.test_delayDelayedCall..rrr<N)rkrrodelayrr"rs rtest_delayDelayedCallz+IterationTimeoutTests.test_delayDelayedCallss'(( w \\22 r  2--g66 "%%%%%rct}|jdd}|||}||dS)zp If the only delayed call is canceled, L{None} is the timeout passed to C{doIteration}. rcdSr-rCrCrrr`z>IterationTimeoutTests.test_cancelDelayedCall..rrN)rkrcancelrrrs rtest_cancelDelayedCallz,IterationTimeoutTests.test_cancelDelayedCalls_ '(( w \\22 --g66 '"""""rN) r%r&r'r(rrrrrrrrrCrrr~r~s ### ' ' ' & & & & & & & & & & & & # # # # #rr~c&eZdZdZeeZdZdZdS)ConnectedDatagramPortTestsz/ Test connected datagram UNIX sockets. Ncfd}tjdt}||_|ddS)z L{ConnectedDatagramPort} does not call the deprecated C{loseConnection} in L{ConnectedDatagramPort.connectionFailed}. c2ddS)z Dummy C{loseConnection} method. C{loseConnection} is deprecated and should not get called. z7loseConnection is deprecated and should not get called.N)failr1srloseConnectionz`ConnectedDatagramPortTests.test_connectionFailedDoesntCallLoseConnection..loseConnection0s IIO P P P P PrNgoodbye)r ConnectedDatagramPortr rconnectionFailed)rrrfs` r-test_connectionFailedDoesntCallLoseConnectionzHConnectedDatagramPortTests.test_connectionFailedDoesntCallLoseConnection*s[  Q Q Q Q Q)$ >>, i(((((rcd_fd}tjdt}||_|djdS)z L{ConnectedDatagramPort} calls L{ConnectedDatagramPort.stopListening} instead of the deprecated C{loseConnection} in L{ConnectedDatagramPort.connectionFailed}. Fcd_dS)z8 Dummy C{stopListening} method. TN)calledr1srrdzYConnectedDatagramPortTests.test_connectionFailedCallsStopListening..stopListeningCsDKKKrNr)rr rr rdr assertTrue)rrdrfs` r'test_connectionFailedCallsStopListeningzBConnectedDatagramPortTests.test_connectionFailedCallsStopListening;ss        )$ >>* i(((  $$$$$r)r%r&r'r( skipSocketsrirrrCrrrr"sI)))"%%%%%rrceZdZdZdS) WakerTestsc,t}|}|t|d||d|}|t|d|dS)zH No warnings are generated when constructing the waker. rN)rrr"r#rb)rrHr$s r test_noWakerConstructionWarningsz+WakerTests.test_noWakerConstructionWarningsPs%%'' X8444 T"""%%'' X844444rN)r%r&r'rrCrrrrOs# 5 5 5 5 5rr) r(rtwisted.internet.deferrtwisted.internet.interfacesrtwisted.internet.posixbaserrtwisted.internet.protocolrtwisted.python.runtimertwisted.trial.unittestr rtwisted.internetr twisted.test.test_unixr ImportErrorr twisted.internet.tcpr rr+rEr[rkr~rrrCrrrs} ++++++777777????????333333++++++++++++ >%%%%%%2222222>>>=KKK>%$$$$$%%%%%%XD"""""%"""&-3-3-3-3-32-3-3-3`KKKKK)KKK&* * * * * +* * * Z_#_#_#_#_#2_#_#_#D*%*%*%*%*%!7*%*%*%Z 5 5 5 5 5' 5 5 5 5 5s =AA