/usr/lib/python3/dist-packages/twisted/internet/test/__pycache__
Edit: /usr/lib/python3/dist-packages/twisted/internet/test/__pycache__/reactormixins.cpython-310.pyc (11808B)
o
b9 @ s d Z g dZddlZddlZddlZddlmZmZmZm Z m
Z
ddl m Z ddl
mZ ddlmZ ddlmZ dd lmZ dd
lmZ dd lmZmZ dd lm Z m Z zdd
l m Z W n e!yg dZ Y nw e Z G dd de"Z#dd Z$dddZ%G dd dZ&dS )a
Utilities for unit testing reactor implementations.
The main feature of this module is L{ReactorBuilder}, a base class for use when
writing interface/blackbox tests for reactor implementations. Test case classes
for reactor features should subclass L{ReactorBuilder} instead of
L{SynchronousTestCase}. All of the features of L{SynchronousTestCase} will be
available. Additionally, the tests will automatically be applied to all
available reactor implementations.
)TestTimeoutErrorReactorBuilderneedsRunningReactor N)DictOptionalSequenceTypeUnion) Interfacelog)_fullyQualifiedName)Failure)namedAny)platform)SkipTestSynchronousTestCase)DEFAULT_TIMEOUT_DURATIONacquireAttribute)processc @ s e Zd ZdZdS )r zm
The reactor was still running after the timeout period elapsed in
L{ReactorBuilder.runReactor}.
N)__name__
__module__ __qualname____doc__ r r E/usr/lib/python3/dist-packages/twisted/internet/test/reactormixins.pyr * s r c C s | | dS )a*
Various functions within these tests need an already-running reactor at
some point. They need to stop the reactor when the test has completed, and
that means calling reactor.stop(). However, reactor.stop() raises an
exception if the reactor isn't already running, so if the L{Deferred} that
a particular API under test returns fires synchronously (as especially an
endpoint's C{connect()} method may do, if the connect is to a local
interface address) then the test won't be able to stop the reactor being
tested and finish. So this calls C{thunk} only once C{reactor} is running.
(This is just an alias for
L{twisted.internet.interfaces.IReactorCore.callWhenRunning} on the given
reactor parameter, in order to centrally reference the above paragraph and
repeating it everywhere as a comment.)
@param reactor: the L{twisted.internet.interfaces.IReactorCore} under test
@param thunk: a 0-argument callable, which eventually finishes the test in
question, probably in a L{Deferred} callback.
N)callWhenRunning)reactorthunkr r r r 1 s r c sD |du r
ddl m} dg fdd}|| | |j| dS )a
Stop the reactor as soon as any error is logged on the given publisher.
This is beneficial for tests which will wait for a L{Deferred} to fire
before completing (by passing or failing). Certain implementation bugs may
prevent the L{Deferred} from firing with any result at all (consider a
protocol's {dataReceived} method that raises an exception: this exception
is logged but it won't ever cause a L{Deferred} to fire). In that case the
test would have to complete by timing out which is a much less desirable
outcome than completing as soon as the unexpected error is encountered.
@param case: A L{SynchronousTestCase} to use to clean up the necessary log
observer when the test is over.
@param reactor: The reactor to stop.
@param publisher: A L{LogPublisher} to watch for errors. If L{None}, the
global log publisher will be watched.
Nr r c s* r| dr d S d S d S )NisError)getpopstop)eventr runningr r stopIfError_ s z stopOnError.
.stopIfError)twisted.pythonr addObserver
addCleanupremoveObserver)caser publisherr& r r$ r stopOnErrorI s
r- c
@ s e Zd ZU dZdgZe reg d n'eg d ed e r,ed neddg e
s=ed g d
Z d
Z d
Z
eeee ed < i Zeeef ed < d
d Zdd Zdd Zdd Zdd Zd ddZe ded deee ed ee f f fdd Z d
S ) r a
L{SynchronousTestCase} mixin which provides a reactor-creation API. This
mixin defines C{setUp} and C{tearDown}, so mix it in before
L{SynchronousTestCase} or call its methods from the overridden ones in the
subclass.
@cvar skippedReactors: A dict mapping FQPN strings of reactors for
which the tests defined by this class will be skipped to strings
giving the skip message.
@cvar requiredInterfaces: A C{list} of interfaces which the reactor must
provide or these tests will be skipped. The default, L{None}, means
that no interfaces are required.
@ivar reactorFactory: A no-argument callable which returns the reactor to
use for testing.
@ivar originalHandler: The SIGCHLD handler which was installed when setUp
ran and which will be re-installed when tearDown runs.
@ivar _reactors: A list of FQPN strings giving the reactors for which
L{SynchronousTestCase}s will be created.
z,twisted.internet.selectreactor.SelectReactor)z/twisted.internet.gtk2reactor.PortableGtkReactorz,twisted.internet.gireactor.PortableGIReactorz0twisted.internet.gtk3reactor.PortableGtk3Reactorz/twisted.internet.win32eventreactor.Win32Reactorz0twisted.internet.iocpreactor.reactor.IOCPReactor)z*twisted.internet.glib2reactor.Glib2Reactorz(twisted.internet.gtk2reactor.Gtk2Reactorz$twisted.internet.gireactor.GIReactorz(twisted.internet.gtk3reactor.Gtk3Reactorz6twisted.internet.asyncioreactor.AsyncioSelectorReactorz$twisted.internet.cfreactor.CFReactorz(twisted.internet.pollreactor.PollReactorz*twisted.internet.epollreactor.EPollReactorz(twisted.internet.kqreactor.KQueueReactorNrequiredInterfacesskippedReactorsc C s" t sttjtj| _dS dS )z
Clear the SIGCHLD handler, if there is one, to ensure an environment
like the one which exists prior to a call to L{reactor.run}.
N)r isWindowssignalSIGCHLDSIG_DFLoriginalHandlerselfr r r setUp s zReactorBuilder.setUpc C s | j dur
ttj| j tdurLt }tjrNtdtjf t t d t | dkrGtjD ] }t
|tj q5t
dtjf tjsdS dS dS )z|
Restore the original SIGCHLD handler and reap processes as long as
there seem to be any remaining.
Nz1ReactorBuilder.tearDown reaping some processes %rgMbP?< z/Timeout waiting for child processes to exit: %r)r4 r1 r2 r timereapProcessHandlersr msgreapAllProcessessleeposkillSIGKILL Exception)r6 beginpidr r r tearDown s.
zReactorBuilder.tearDownc C sh | t|dddur!|jD ] }|| |d q|j | | }|D ]}| q+dS )z
Clean up any resources which may have been allocated for the given
reactor by its creation or by a test which used it.
_internalReadersN) _uninstallHandlergetattrrE removeReaderconnectionLostclear
disconnectAllgetDelayedCallscancel)r6 r readercallscr r r unbuildReactor s
z ReactorBuilder.unbuildReactorc s zddl m} ddlm} W n ty Y nw t||r&| j|u r&tdz| W n tyD t
dd | tt
w | jdurn fdd| jD }|rn| td t jd
d d |D f | | j S ) zK
Create and return a reactor using C{self.reactorFactory}.
r r ) CFReactorzuCFReactor uses APIs which manipulate global state, so it's not safe to run its own reactor-builder tests under itselfNzFailed to install reactorc s g | ] }| s|qS r )
providedBy).0requiredrR r r
s z/ReactorBuilder.buildReactor..z%s does not provide %s,c S s g | ]}t |qS r )fullyQualifiedName)rU xr r r rW & s )twisted.internetr twisted.internet.cfreactorrS ImportError
isinstancereactorFactoryr
BaseExceptionr errflushLoggedErrorsr getErrorMessager. rQ rY __class__joinr) )r6
globalReactorrS missingr rR r buildReactor sD
zReactorBuilder.buildReactorc C s t | jdtS )z
Determine how long to run the test before considering it failed.
@return: A C{int} or C{float} giving a number of seconds.
timeout)r _parentsr r5 r r r
getTimeout, s zReactorBuilder.getTimeoutc sV |du r| }g fdd} ||} r%td| d| dS )a
Run the reactor for at most the given amount of time.
@param reactor: The reactor to run.
@type timeout: C{int} or C{float}
@param timeout: The maximum amount of time, specified in seconds, to
allow the reactor to run. If the reactor is still running after
this much time has elapsed, it will be stopped and an exception
raised. If L{None}, the default test method timeout imposed by
Trial will be used. This depends on the L{IReactorTime}
implementation of C{reactor} for correct operation.
@raise TestTimeoutError: If the reactor is still running after
C{timeout} seconds.
Nc s d d S N)appendr" r r timedOutr r r" J s
z'ReactorBuilder.runReactor..stopz reactor still running after z seconds)rk callLaterrunr rM )r6 r ri r" timedOutCallr rn r
runReactor4 s zReactorBuilder.runReactorclsreturnc s i } j D ]:dd } jd | d dd}G fddd t}||_d j dd |g |_|||j< q|S )z
Create a L{SynchronousTestCase} subclass which mixes in C{cls} for each
known reactor and return a dict mapping their names to them.
.Tests_c sP e Zd Z jZ jv r j ZzeZW dS ey' e ZY dS w )z4ReactorBuilder.makeTestCaseClasses..testcaseN)
r r r r/ skipr r_ r` r rc r rt r r r testcased s
r| r ) _reactorssplitr replacer re r )rt classesshortReactorNamenamer| r r{ r makeTestCaseClassesU s
z"ReactorBuilder.makeTestCaseClassesrl ) r r r r r} r r0 extendrm isMacOSXisLinuxr_ r4 r. r r r r
__annotations__r/ r strr7 rD rQ rh rk rs classmethodr r r r r r r r h sP
2
!r rl )'r __all__r> r1 r9 typingr r r r r zope.interfacer
r' r twisted.python.deprecater
rY twisted.python.failurer twisted.python.reflectr twisted.python.runtimer twisted.trial.unittestr r twisted.trial.utilr r r[ r _processr] rA r r r- r r r r r s0