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
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.

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
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