Go to the latest version of this document.

# ``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.
Class | `SynchronousResolver` | A very simple `IHostnameResolver` that immediately, synchronously resolves all host names to a single static address (TCPv4, 127.0.0.1) while preserving any requested port number.
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.
Class | `_ProtocolConnectionMadeHaver` | Explicit stipulation of the implicit requirement of `AccumulatingProtocol`'s factory.
Class | `_SynchronousResolution` | No class docstring; 0/1 class variable, 1/1 method documented
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:`IReactorTime` | The reactor to schedule the checks on.
protocols:A `list` of `IProtocol`s. | The protocols to wait for disconnecting.

def _benchmarkWithReactor(test_target: `Callable[[], 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')