Go to the latest version of this document.

# ``twisted`.`internet`.`testing`.`MemoryReactor``

class documentation

`class MemoryReactor: (source)`

Known subclasses: `twisted.internet.testing.MemoryReactorClock`

Constructor: `MemoryReactor()`

Implements interfaces: `twisted.internet.interfaces.IReactorCore`, `twisted.internet.interfaces.IReactorFDSet`, `twisted.internet.interfaces.IReactorPluggableNameResolver`, `twisted.internet.interfaces.IReactorSocket`, `twisted.internet.interfaces.IReactorSSL`, `twisted.internet.interfaces.IReactorTCP`, `twisted.internet.interfaces.IReactorUNIX`

View In Hierarchy

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.

Method | `__init__` | Initialize the tracking lists.
Method | `addReader` | Fake `IReactorFDSet.addReader` which adds the reader to a local set.
Method | `addSystemEventTrigger` | Fake `IReactorCore.run`. Keep track of trigger by appending it to self.triggers[phase][eventType].
Method | `addWriter` | Fake `IReactorFDSet.addWriter` which adds the writer to a local set.
Method | `adoptDatagramPort` | Fake `IReactorSocket.adoptDatagramPort`, that logs the call and returns a fake `IListeningPort`.
Method | `adoptStreamConnection` | Record the given stream connection in `adoptedStreamConnections`.
Method | `adoptStreamPort` | Fake `IReactorSocket.adoptStreamPort`, that logs the call and returns an `IListeningPort`.
Method | `callWhenRunning` | Fake `IReactorCore.callWhenRunning`. Keeps a list of invocations to make in `self.whenRunningHooks`.
Method | `connectSSL` | Fake `IReactorSSL.connectSSL`, that logs the call and returns an `IConnector`.
Method | `connectTCP` | Fake `IReactorTCP.connectTCP`, that logs the call and returns an `IConnector`.
Method | `connectUNIX` | Fake `IReactorUNIX.connectUNIX`, that logs the call and returns an `IConnector`.
Method | `crash` | Fake `IReactorCore.crash`. Sets `self.running` to `None`, because that feels crashy. Sets `self.hasCrashed` to `True`.
Method | `fireSystemEvent` | Not implemented; raises `NotImplementedError`.
Method | `getReaders` | Fake `IReactorFDSet.getReaders` which returns a list of readers from the local set.
Method | `getWriters` | Fake `IReactorFDSet.getWriters` which returns a list of writers from the local set.
Method | `install` | Fake install callable to emulate reactor module installation.
Method | `installNameResolver` | Implement `IReactorPluggableNameResolver`.
Method | `iterate` | Not implemented; raises `NotImplementedError`.
Method | `listenSSL` | Fake `IReactorSSL.listenSSL`, that logs the call and returns an `IListeningPort`.
Method | `listenTCP` | Fake `IReactorTCP.listenTCP`, that logs the call and returns an `IListeningPort`.
Method | `listenUNIX` | Fake `IReactorUNIX.listenUNIX`, that logs the call and returns an `IListeningPort`.
Method | `removeAll` | Fake `IReactorFDSet.removeAll` which removed all readers and writers from the local sets.
Method | `removeReader` | Fake `IReactorFDSet.removeReader` which removes the reader from a local set.
Method | `removeSystemEventTrigger` | Not implemented; raises `NotImplementedError`.
Method | `removeWriter` | Fake `IReactorFDSet.removeWriter` which removes the writer from a local set.
Method | `resolve` | Not implemented; raises `NotImplementedError`.
Method | `run` | Fake `IReactorCore.run`. Sets `self.running` to `True`, runs all of the hooks passed to `self.callWhenRunning`, then calls `self.stop` to simulate a request to stop the reactor. Sets `self.hasRun` to `True`.
Method | `stop` | Fake `IReactorCore.run`. Sets `self.running` to `False`. Sets `self.hasStopped` to `True`.
Instance Variable | `adoptedPorts` | Keeps track of server listen attempts (ie, calls to `adoptStreamPort`).
Instance Variable | `adoptedStreamConnections` | Keeps track of stream-oriented connections added using `adoptStreamConnection`.
Instance Variable | `connectors` | Undocumented
Instance Variable | `hasCrashed` | Keeps track of whether this reactor has crashed.
Instance Variable | `hasInstalled` | Keeps track of whether this reactor has been installed.
Instance Variable | `hasRun` | Undocumented
Instance Variable | `hasStopped` | Keeps track of whether this reactor has been stopped.
Instance Variable | `nameResolver` | Read-only attribute; the resolver installed with `installResolver`. An `IHostnameResolver`.
Instance Variable | `readers` | Undocumented
Instance Variable | `running` | Keeps track of whether this reactor is running.
Instance Variable | `sslClients` | Keeps track of connection attempts (ie, calls to `connectSSL`).
Instance Variable | `sslServers` | Keeps track of server listen attempts (ie, calls to `listenSSL`).
Instance Variable | `tcpClients` | Keeps track of connection attempts (ie, calls to `connectTCP`).
Instance Variable | `tcpServers` | Keeps track of server listen attempts (ie, calls to `listenTCP`).
Instance Variable | `triggers` | Keeps track of hooks registered with `addSystemEventTrigger`.
Instance Variable | `unixClients` | Keeps track of connection attempts (ie, calls to `connectUNIX`).
Instance Variable | `unixServers` | Keeps track of server listen attempts (ie, calls to `listenUNIX`).
Instance Variable | `whenRunningHooks` | Keeps track of hooks registered with `callWhenRunning`.
Instance Variable | `writers` | Undocumented

def __init__(self): (source)
overridden in `twisted.internet.testing.MemoryReactorClock`

Initialize the tracking lists.

def addReader(self, reader): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.addReader` which adds the reader to a local set.

def addSystemEventTrigger(self, phase: `str`, eventType: `str`, callable: `Callable[_P, object]`, *args: `_P.args`, **kw: `_P.kwargs`): (source)
from `twisted.internet.interfaces.IReactorCore`

Fake `IReactorCore.run`. Keep track of trigger by appending it to self.triggers[phase][eventType].

def addWriter(self, writer): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.addWriter` which adds the writer to a local set.

def adoptDatagramPort(self, fileno, addressFamily, protocol, maxPacketSize=8192): (source)
from `twisted.internet.interfaces.IReactorSocket`

Fake `IReactorSocket.adoptDatagramPort`, that logs the call and returns a fake `IListeningPort`.

See Also
`twisted.internet.interfaces.IReactorSocket.adoptDatagramPort`

def adoptStreamConnection(self, fileDescriptor, addressFamily, factory): (source)
from `twisted.internet.interfaces.IReactorSocket`

Record the given stream connection in `adoptedStreamConnections`.

See Also
`twisted.internet.interfaces.IReactorSocket.adoptStreamConnection`

def adoptStreamPort(self, fileno, addressFamily, factory): (source)
from `twisted.internet.interfaces.IReactorSocket`

Fake `IReactorSocket.adoptStreamPort`, that logs the call and returns an `IListeningPort`.

def callWhenRunning(self, callable: `Callable[_P, object]`, *args: `_P.args`, **kw: `_P.kwargs`): (source)
from `twisted.internet.interfaces.IReactorCore`

Fake `IReactorCore.callWhenRunning`. Keeps a list of invocations to make in `self.whenRunningHooks`.

If the reactor has not started, the callable will be scheduled to run when it does start. Otherwise, the callable will be invoked immediately.

def connectSSL(self, host, port, factory, contextFactory, timeout=30, bindAddress=None): (source)
from `twisted.internet.interfaces.IReactorSSL`

Fake `IReactorSSL.connectSSL`, that logs the call and returns an `IConnector`.

def connectTCP(self, host, port, factory, timeout=30, bindAddress=None): (source)
from `twisted.internet.interfaces.IReactorTCP`

Fake `IReactorTCP.connectTCP`, that logs the call and returns an `IConnector`.

def connectUNIX(self, address, factory, timeout=30, checkPID=0): (source)
from `twisted.internet.interfaces.IReactorUNIX`

Fake `IReactorUNIX.connectUNIX`, that logs the call and returns an `IConnector`.

def crash(self): (source)
from `twisted.internet.interfaces.IReactorCore`

Fake `IReactorCore.crash`. Sets `self.running` to `None`, because that feels crashy. Sets `self.hasCrashed` to `True`.

def fireSystemEvent(self, eventType): (source)
from `twisted.internet.interfaces.IReactorCore`

Not implemented; raises `NotImplementedError`.

def getReaders(self): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.getReaders` which returns a list of readers from the local set.

def getWriters(self): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.getWriters` which returns a list of writers from the local set.

def install(self): (source)

Fake install callable to emulate reactor module installation.

def installNameResolver(self, resolver: `IHostnameResolver`) -> `IHostnameResolver`: (source)
from `twisted.internet.interfaces.IReactorPluggableNameResolver`

Implement `IReactorPluggableNameResolver`.

def iterate(self, delay=0): (source)
from `twisted.internet.interfaces.IReactorCore`

Not implemented; raises `NotImplementedError`.

def listenSSL(self, port, factory, contextFactory, backlog=50, interface=''): (source)
from `twisted.internet.interfaces.IReactorSSL`

Fake `IReactorSSL.listenSSL`, that logs the call and returns an `IListeningPort`.

def listenTCP(self, port, factory, backlog=50, interface=''): (source)
from `twisted.internet.interfaces.IReactorTCP`

Fake `IReactorTCP.listenTCP`, that logs the call and returns an `IListeningPort`.

def listenUNIX(self, address, factory, backlog=50, mode=438, wantPID=0): (source)
from `twisted.internet.interfaces.IReactorUNIX`

Fake `IReactorUNIX.listenUNIX`, that logs the call and returns an `IListeningPort`.

def removeAll(self): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.removeAll` which removed all readers and writers from the local sets.

def removeReader(self, reader): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.removeReader` which removes the reader from a local set.

def removeSystemEventTrigger(self, triggerID): (source)
from `twisted.internet.interfaces.IReactorCore`

Not implemented; raises `NotImplementedError`.

def removeWriter(self, writer): (source)
from `twisted.internet.interfaces.IReactorFDSet`

Fake `IReactorFDSet.removeWriter` which removes the writer from a local set.

def resolve(self, name, timeout=10): (source)
from `twisted.internet.interfaces.IReactorCore`

Not implemented; raises `NotImplementedError`.

def run(self): (source)
from `twisted.internet.interfaces.IReactorCore`

Fake `IReactorCore.run`. Sets `self.running` to `True`, runs all of the hooks passed to `self.callWhenRunning`, then calls `self.stop` to simulate a request to stop the reactor. Sets `self.hasRun` to `True`.

def stop(self): (source)
from `twisted.internet.interfaces.IReactorCore`

Fake `IReactorCore.run`. Sets `self.running` to `False`. Sets `self.hasStopped` to `True`.

adoptedPorts: `list` = (source)

Keeps track of server listen attempts (ie, calls to `adoptStreamPort`).

adoptedStreamConnections: `list` = (source)

Keeps track of stream-oriented connections added using `adoptStreamConnection`.

connectors: `list` = (source)

Undocumented

hasCrashed: `bool` = (source)

Keeps track of whether this reactor has crashed.

hasInstalled: `bool` = (source)

Keeps track of whether this reactor has been installed.

hasRun: `bool` = (source)

Undocumented

hasStopped: `bool` = (source)

Keeps track of whether this reactor has been stopped.

nameResolver: `IHostnameResolver` = (source)
from `twisted.internet.interfaces.IReactorPluggableNameResolver`

Read-only attribute; the resolver installed with `installResolver`. An `IHostnameResolver`.

readers: `set` = (source)

Undocumented

running: `bool` = (source)
from `twisted.internet.interfaces.IReactorCore`

Keeps track of whether this reactor is running.

sslClients: `list` = (source)

Keeps track of connection attempts (ie, calls to `connectSSL`).

sslServers: `list` = (source)

Keeps track of server listen attempts (ie, calls to `listenSSL`).

tcpClients: `list` = (source)

Keeps track of connection attempts (ie, calls to `connectTCP`).

tcpServers: `list` = (source)

Keeps track of server listen attempts (ie, calls to `listenTCP`).

triggers: `dict` = (source)

Keeps track of hooks registered with `addSystemEventTrigger`.

unixClients: `list` = (source)

Keeps track of connection attempts (ie, calls to `connectUNIX`).

unixServers: `list` = (source)

Keeps track of server listen attempts (ie, calls to `listenUNIX`).

whenRunningHooks: `list` = (source)

Keeps track of hooks registered with `callWhenRunning`.

writers: `set` = (source)

Undocumented