o ¼>h=.ã@s,dZddlZddlmZddlmZddlmZmZddl m Z ddl m Z ddl mZdZzdd lmZdd lmZWn eyGd ZYnwdd lmZdd lmZGdd„deƒZGdd„deƒZGdd„deƒZGdd„deƒZGdd„deƒZGdd„deƒZGdd„deƒZGdd„deƒZdS)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)ÚPortcs,eZdZdZ‡fdd„Z‡fdd„Z‡ZS)ÚWarningCheckerTestCasezg A test case that will make sure that no warnings are left unchecked at the end of a test run. cs<tƒ ¡tj dd¡ ¡dkrt ¡r| ¡dSdSdS)NÚCIÚÚtrue) ÚsuperÚsetUpÚosÚenvironÚgetÚlowerrÚ isWindowsÚ flushWarnings©Úself©Ú __class__©ú„/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/twisted/internet/test/test_posixbase.pyr s  ÿzWarningCheckerTestCase.setUpc s z)tƒ ¡W| ¡}tj dd¡ ¡dkrt ¡rdS|  t |ƒdd|›¡dS| ¡}tj dd¡ ¡dkrCt ¡rCYdS|  t |ƒdd|›¡w)Nrrrrz'Warnings found at the end of the test: ) rÚtearDownrrrrrrrÚ assertEqualÚlen)rÚwarningsrrrr +s ÿ÷ ÿzWarningCheckerTestCase.tearDown)Ú__name__Ú __module__Ú __qualname__Ú__doc__rr Ú __classcell__rrrrrs  rc@s4eZdZdd„Zdd„Zdd„Zdd„Zd d „Zd S) ÚTrivialReactorcCsi|_i|_t |¡dS©N)Ú_readersÚ_writersrÚ__init__rrrrr->szTrivialReactor.__init__cCód|j|<dS©NT©r+©rÚreaderrrrÚ addReaderCózTrivialReactor.addReadercCó |j|=dSr*r0r1rrrÚ removeReaderFó zTrivialReactor.removeReadercCr.r/©r,©rÚwriterrrrÚ addWriterIr4zTrivialReactor.addWritercCr5r*r8r9rrrÚ removeWriterLr7zTrivialReactor.removeWriterN)r$r%r&r-r3r6r;r<rrrrr)=s  r)c@s0eZdZdZdd„Zdd„Zdd„Zdd „Zd S) ÚPosixReactorBaseTestsz( Tests for L{PosixReactorBase}. cCs2| |jt¡| |j|j¡| |j|j¡dSr*)ÚassertIsInstanceÚwakerrÚassertInÚ_internalReadersr+©rr rrrÚ _checkWakerUsz!PosixReactorBaseTests._checkWakercCstƒ}| |¡dS)z When L{PosixReactorBase} is instantiated, it creates a waker and adds it to its internal readers set. N)r)rCrBrrrÚtest_wakerIsInternalReaderZsz0PosixReactorBaseTests.test_wakerIsInternalReadercCs\tƒ}tƒ}|j |¡| |¡| |j|j¡| |¡|  ||j¡|  ||j¡dS)z‹ Any L{IReadDescriptor}s in L{PosixReactorBase._internalReaders} are left alone by L{PosixReactorBase._removeAll}. N) r)ÚobjectrAÚaddr3Ú _removeAllr+r,rCr@)rr ÚextrarrrÚ"test_removeAllSkipsInternalReadersbs   z8PosixReactorBaseTests.test_removeAllSkipsInternalReaderscCsjtƒ}tƒ}tƒ}| |¡| |¡| |j|j¡}| t|ƒ||h¡|  ||j¡|  ||j¡dS)z† L{PosixReactorBase._removeAll} returns a list of removed L{IReadDescriptor} and L{IWriteDescriptor} objects. N) r)rEr3r;rGr+r,r!ÚsetÚ assertNotIn)rr r2r:ÚremovedrrrÚ'test_removeAllReturnsRemovedDescriptorsps  z=PosixReactorBaseTests.test_removeAllReturnsRemovedDescriptorsN)r$r%r&r'rCrDrIrMrrrrr=Ps  r=c@s&eZdZdZeeeƒs dZdd„ZdS)Ú TCPPortTestsz1 Tests for L{twisted.internet.tcp.Port}. zNon-posixbase reactorcCs,tdtƒƒ}d|_dd„|_| | ¡t¡S)zƒ L{Port.stopListening} returns a L{Deferred} which errbacks if L{Port.connectionLost} raises an exception. i90TcSsddS)Nérr)ÚreasonrrrÚóz8TCPPortTests.test_connectionLostFailed..)r rÚ connectedÚconnectionLostÚ assertFailureÚ stopListeningÚZeroDivisionError)rÚportrrrÚtest_connectionLostFailedˆs  z&TCPPortTests.test_connectionLostFailedN) r$r%r&r'Ú isinstancer rÚskiprYrrrrrN€s   rNc@sTeZdZdZdd„Zdeddfdd„Zdeddfd d „Zd d „Zd d„Z dd„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. cCst |¡tƒ|_d|_dS)Néd)rr-rÚiterationTimeoutÚnowrrrrr-šs  zTimeoutReportReactor.__init__r2ÚreturnNcCódS)z— Ignore the reader. This is necessary because the waker will be added. However, we won't actually monitor it for any events. Nrr1rrrr3ŸózTimeoutReportReactor.addReadercCra)z# See L{addReader}. Nrr1rrrr6¥rbz!TimeoutReportReactor.removeReadercCsgS)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. rrrrrÚ removeAllªszTimeoutReportReactor.removeAllcCs|jS)zx Override the real clock with a deterministic one that can be easily controlled in a unit test. )r_rrrrÚseconds²szTimeoutReportReactor.secondscCs&|j}|durd|_| |¡dSdSr*)r^Úcallback)rÚtimeoutÚdrrrÚ doIteration¹s þz TimeoutReportReactor.doIteration) r$r%r&r'r-rr3r6rcrdrhrrrrr\“s r\c@sPeZdZdZdd„Zdd„Zdd„Zdd „Zd d „Zd d „Z dd„Z dd„Z dS)ÚIterationTimeoutTestsz Tests for the timeout argument L{PosixReactorBase.run} calls L{PosixReactorBase.doIteration} with in the presence of various delayed calls. cs6g}ˆj |j¡ˆj ‡fdd„¡ˆ ¡|dS)Ncsˆ ¡Sr*)Ústop)Úignoredr rrrQÊrRz>IterationTimeoutTests._checkIterationTimeout..r)r^Ú addCallbackÚappendÚrun©rr rfrr rÚ_checkIterationTimeoutÇs z,IterationTimeoutTests._checkIterationTimeoutcCstƒ}| |¡}| |¡dS)zl If there are no delayed calls, C{doIteration} is called with a timeout of L{None}. N)r\rpÚ assertIsNonerorrrÚ test_noCallsÎs z"IterationTimeoutTests.test_noCallscCs0tƒ}| ddd„¡| |¡}| |d¡dS)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. r]cSódSr*rrrrrrQÞrbz8IterationTimeoutTests.test_delayedCall..N©r\Ú callLaterrpr!rorrrÚtest_delayedCall×s z&IterationTimeoutTests.test_delayedCallcCs>tƒ}| ddd„¡|jd7_| |¡}| |d¡dS)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. r]cSrsr*rrrrrrQérbz7IterationTimeoutTests.test_timePasses..ééKN)r\rur_rpr!rorrrÚtest_timePassesâs  z%IterationTimeoutTests.test_timePassescCsPtƒ}| ddd„¡| ddd„¡| ddd„¡| |¡}| |d¡dS) 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. é2cSrsr*rrrrrrQõrbzAIterationTimeoutTests.test_multipleDelayedCalls..é cSrsr*rrrrrrQörbr]cSrsr*rrrrrrQ÷rbNrtrorrrÚtest_multipleDelayedCallsîs  z/IterationTimeoutTests.test_multipleDelayedCallscCsHtƒ}| ddd„¡}|jd7_| d¡| |¡}| |d¡dS)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. rzcSrsr*rrrrrrQrbz=IterationTimeoutTests.test_resetDelayedCall..rwéN)r\rur_Úresetrpr!©rr ÚcallrfrrrÚtest_resetDelayedCallûs   z+IterationTimeoutTests.test_resetDelayedCallcCsHtƒ}| ddd„¡}|jd7_| d¡| |¡}| |d¡dS)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. rzcSrsr*rrrrrrQrbz=IterationTimeoutTests.test_delayDelayedCall..r{éé<N)r\rur_Údelayrpr!rrrrÚtest_delayDelayedCalls   z+IterationTimeoutTests.test_delayDelayedCallcCs6tƒ}| ddd„¡}| ¡| |¡}| |¡dS)zp If the only delayed call is canceled, L{None} is the timeout passed to C{doIteration}. rzcSrsr*rrrrrrQrbz>IterationTimeoutTests.test_cancelDelayedCall..N)r\ruÚcancelrprqrrrrÚtest_cancelDelayedCalls  z,IterationTimeoutTests.test_cancelDelayedCallN) r$r%r&r'rprrrvryr|rr…r‡rrrrriÀs     ric@s,eZdZdZedur eZdd„Zdd„ZdS)ÚConnectedDatagramPortTestsz/ Test connected datagram UNIX sockets. Ncs.‡fdd„}t dtƒ¡}||_| d¡dS)z L{ConnectedDatagramPort} does not call the deprecated C{loseConnection} in L{ConnectedDatagramPort.connectionFailed}. csˆ d¡dS)z€ Dummy C{loseConnection} method. C{loseConnection} is deprecated and should not get called. z7loseConnection is deprecated and should not get called.N)ÚfailrrrrÚloseConnection0sz`ConnectedDatagramPortTests.test_connectionFailedDoesntCallLoseConnection..loseConnectionNÚgoodbye)r ÚConnectedDatagramPortr rŠÚconnectionFailed)rrŠrXrrrÚ-test_connectionFailedDoesntCallLoseConnection*s zHConnectedDatagramPortTests.test_connectionFailedDoesntCallLoseConnectioncs@dˆ_‡fdd„}t dtƒ¡}||_| d¡ˆ ˆj¡dS)zÁ L{ConnectedDatagramPort} calls L{ConnectedDatagramPort.stopListening} instead of the deprecated C{loseConnection} in L{ConnectedDatagramPort.connectionFailed}. Fcs dˆ_dS)z8 Dummy C{stopListening} method. TN)ÚcalledrrrrrVCs zYConnectedDatagramPortTests.test_connectionFailedCallsStopListening..stopListeningNr‹)rr rŒr rVrÚ assertTrue)rrVrXrrrÚ'test_connectionFailedCallsStopListening;s   zBConnectedDatagramPortTests.test_connectionFailedCallsStopListening)r$r%r&r'Ú skipSocketsr[rŽr‘rrrrrˆ"s  rˆc@seZdZdd„ZdS)Ú WakerTestscCsHtƒ}| ¡}| t|ƒd|¡| d¡| ¡}| t|ƒd|¡dS)zH No warnings are generated when constructing the waker. rN)rrr!r"rT)rr?r#rrrÚ test_noWakerConstructionWarningsPs  z+WakerTests.test_noWakerConstructionWarningsN)r$r%r&r”rrrrr“Os r“) r'rÚtwisted.internet.deferrÚtwisted.internet.interfacesrÚtwisted.internet.posixbaserrÚtwisted.internet.protocolrÚtwisted.python.runtimerÚtwisted.trial.unittestrr’Útwisted.internetr Útwisted.test.test_unixr Ú ImportErrorr Útwisted.internet.tcpr rr)r=rNr\rirˆr“rrrrÚs2       ÿ  "0-b-