twisted.internet.testing

module documentation
(source)

Assorted functionality which is commonly useful when writing unit tests.

Class AccumulatingProtocol AccumulatingProtocol is an IProtocol implementation which collects the data delivered to it and can fire a Deferred when it is connected or disconnected.
Class EventLoggingObserver ILogObserver That stores its events in a list for later inspection. This class is similar to LimitedHistoryLogObserver save that the internal buffer is public and intended for external inspection. The observer implements the sequence protocol to ease iteration of the events.
Class FakeDatagramTransport Undocumented
Class LineSendingProtocol Undocumented
Class MemoryReactor A fake reactor to be used in tests. This reactor doesn't actually do much that's useful yet. It accepts TCP connection setup attempts, but they will never succeed.
Class MemoryReactorClock Undocumented
Class NonStreamingProducer A pull producer which writes 10 times only.
Class RaisingMemoryReactor A fake reactor to be used in tests. It accepts TCP connection setup attempts, but they will fail.
Class StringIOWithoutClosing A BytesIO that can't be closed.
Class StringTransport A transport implementation which buffers data in memory and keeps track of its other state without providing any behavior.
Class StringTransportWithDisconnection A StringTransport which on disconnection will trigger the connection lost on the attached protocol.
Function waitUntilAllDisconnected Take a list of disconnecting protocols, callback a Deferred when they're all done.
Class _FakeConnector A fake IConnector that allows us to inspect if it has been told to stop connecting.
Class _FakePort A fake IListeningPort to be used in tests.
Function _benchmarkWithReactor Decorator for running a benchmark tests that loops the reactor.
Function _runReactor (re)Start a reactor that might have been previously started.
Function _stopReactor Stop the reactor and allow it to be re-started later.
Constant _P Undocumented
Type Variable _T Undocumented
def waitUntilAllDisconnected(reactor, protocols): (source)

Take a list of disconnecting protocols, callback a Deferred when they're all done.

This is a hack to make some older tests less flaky, as ITransport.loseConnection is not atomic on all reactors (for example, the CoreFoundation, which sometimes takes a reactor turn for CFSocket to realise). New tests should either not use real sockets in testing, or take the advice in https://jml.io/pages/how-to-disconnect-in-twisted-really.html to heart.

Parameters
reactor:IReactorTimeThe reactor to schedule the checks on.
protocols:A list of IProtocols.The protocols to wait for disconnecting.
def _benchmarkWithReactor(test_target: Callable[[], Union[Coroutine[Deferred[Any], Any, _T], Generator[Deferred[Any], Any, _T], Deferred[_T]]]) -> Callable[[Any], None]: (source)

Decorator for running a benchmark tests that loops the reactor.

This is designed to decorate test method executed using pytest and pytest-benchmark.

def _runReactor(callback: Callable[[], Deferred[_T]]): (source)

(re)Start a reactor that might have been previously started.

def _stopReactor(reactor): (source)

Stop the reactor and allow it to be re-started later.

_P = (source)

Undocumented

Value
ParamSpec('_P')
_T = (source)

Undocumented

Value
TypeVar('_T')