Spawn a simple service fixture listening to a port¶

Let’s create a test that spawns a dummy HTTP server that listens to port 8080:

>>> import socket

>>> from testtools import TestCase
>>> from txfixtures import Reactor, Service

>>> HTTP_SERVER = "python3 -m http.server 8080".split(" ")

>>> class HTTPServerTest(TestCase):
...
...     def setUp(self):
...         super().setUp()
...         self.useFixture(Reactor())
...
...         # Create a service fixture that will spawn the HTTP server
...         # and wait for it to listen to port 8080.
...         self.service = Service(HTTP_SERVER)
...         self.service.expectPort(8080)
...
...         self.useFixture(self.service)
...
...     def test_connect(self):
...         connection = socket.socket()
...         connection.connect(("127.0.0.1", 8080))
...         self.assertEqual(connection.getsockname()[0], "127.0.0.1")

>>> test = HTTPServerTest(methodName="test_connect")
>>> test.run().wasSuccessful()
True

Forward standard output to the Python logging system¶

Let’s spawn a simple HTTP server and have its standard output forwarded to the Python logging system:

>>> import requests

>>> from fixtures import FakeLogger

>>> TWIST_COMMAND = "twistd -n web".split(" ")

# This format string will be used to build a regular expression to parse
# each output line of the service, and map it to a Python LogRecord. A
# sample output line from the twistd web command looks like:
#
#   2016-11-17T22:18:36+0000 [-] Site starting on 8080
#
>>> TWIST_FORMAT = "{Y}-{m}-{d}T{H}:{M}:{S}\+0000 \[{name}\] {message}"

# This output string will be used as a "marker" indicating that the service
# has initialized, and should shortly start listening to the expected port (if
# one was given). The fixture.setUp() method will intercept this marker and
# then wait for the service to actually open the port.
>>> TWIST_OUTPUT = "Site starting on 8080"

>>> class TwistedWebTest(TestCase):
...
...     def setUp(self):
...         super().setUp()
...         self.logger = self.useFixture(FakeLogger())
...         self.useFixture(Reactor())
...         self.service = Service(TWIST_COMMAND)
...         self.service.setOutputFormat(TWIST_FORMAT)
...         self.service.expectOutput(TWIST_OUTPUT)
...         self.service.expectPort(8080)
...         self.useFixture(self.service)
...
...     def test_request(self):
...         response = requests.get("http://localhost:8080")
...         self.assertEqual(200, response.status_code)
...         self.assertIn('"GET / HTTP/1.1" 200', self.logger.output)
...
>>> test = TwistedWebTest(methodName="test_request")
>>> test.run().wasSuccessful()
True

txfixtures.reactor¶

class txfixtures.reactor.Reactor(reactor=None, timeout=5)[source]¶

A fixture to run the Twisted reactor in a separate thread.

This fixture will spawn a new thread in the test process and run the Twisted reactor in it. Test code can then invoke asynchronous APIs by using blockingCallFromThread().

Parameters:	reactor – The Twisted reactor to run. timeout – Raise an exception if the reactor or the thread is it runs in doesn’t start (at setUp time) or doesn’t stop (at cleanUp time) in this amount of seconds.
Variables:	thread – The Thread that the reactor runs in.

call(timeout, f, *a, **kw)[source]¶

Convenience around blockingCallFromThread, with timeout support.

The function f will be invoked in the reactor’s thread with the given arguments and keyword arguments. If f returns a Deferred, the calling code will block until it has fired.

Returns:	The value returned by f or the value fired by the Deferred it returned. If f traces back or the Deferred it returned errbacks, the relevant exception will be propagated to the caller of this method.
Raises:	CallFromThreadTimeout – If timeout seconds have elapsed and the Deferred returned by f hasn’t fired yet.

reset()[source]¶

Make sure that the reactor is still running.

If the reactor and its thread have died, this method will try to recover them by creating a new thread and starting the reactor again.

txfixtures.service¶

class txfixtures.service.Service(command, reactor=None, timeout=15, env=None)[source]¶

Spawn, control and monitor a background service.

allocatePort()[source]¶

Allocate an unused port.

This method can be used by subclasses to allocate a random ports for the service they spawn.

There is a small race condition here (between the time we allocate the port, and the time it actually gets used), but for the purposes for which this method gets used it isn’t a problem in practice.

class txfixtures.service.ServiceOutputParser(service, logger=None, pattern=None)[source]¶

Parse the standard output stream of a service and forward it to the Python logging system.

The stream is assumed to be a UTF-8 sequence of lines each delimited by a (configurable) delimiter character.

Each received line is tested against the RegEx pattern provided in the constructor. If a match is found, a LogRecord is built using the information from the groups of the match object, otherwise default values will be used.

The record is then passed to the Logger provided in the constructor.

Match objects that result from the RegEx pattern are supposed to provide groups named after the substitutions below.

Parameters:	service – A string identifying the service whose output is being parsed. It will be attached as ‘service’ attribute to all log records emitted.

delimiter = '\n'¶

The delimiter character identifying the end of a line.

lineLengthExceeded(line)[source]¶

Simply truncate the line.

lineReceived(line)[source]¶

Foward the received line to the Python logging system.

substitutions = {'M': '(?P<M>\\d{2})', 'S': '(?P<S>\\d{2})', 'd': '(?P<d>\\d{2})', 'Y': '(?P<Y>\\d{4})', 'H': '(?P<H>\\d{2})', 'message': '(?P<message>.+)', 'm': '(?P<m>\\d{2})', 'levelname': '(?P<levelname>[a-zA-Z]+)', 'msecs': '(?P<msecs>\\d{3})', 'name': '(?P<name>.+)'}¶

Substitutions for commonly used groups in line match patterns. For example, this allows you to use “{Y}-{m}-{S}” as pattern snippet, as

whenLineContains(text, callback)[source]¶

Fire the given callback when a line contains the given text.

The callback will be fired only once when and if a match is found.

class txfixtures.service.ServiceProtocol(reactor=None, parser=None, timeout=15)[source]¶

Start and stop a background service process.

This ProcessProtocol manages the start up and termination phases of a background service process. The process is considered ‘running’ when it has stayed up for at least 0.1 seconds (or any other non default value which minUptime is set too), and optionally when it has emitted a certain string and/or it has started listening to a certain port.

expectedOutput = None¶

Optional text that we expect the process to emit in standard output before we consider it ready.

expectedPort = None¶

Optional port number that we expect the service process to listen, before we consider it ready.

minUptime = 0.1¶

The service process must stay up at least this amount of seconds, before it’s considered running. This allows to catch common issues like the service process executable not being in PATH or not being executable.

ready = None¶

Deferred that will fire when the service is considered ready, i.e. it has stayed up for at least minUptime seconds, has produced the expected output (if any), and is listening to the expected port (if any). Upon cancellation, any waiting activity will be stopped.

terminated = None¶

Deferred that will fire when the service has fully terminated, i.e. it has exited and we parent process have read any outstanding data in the pipes and have closed them.

timeout = None¶

Maximum amount of seconds to wait for the service to be ready. After that, the ‘ready’ deferred will errback with a TimeoutError.

txfixtures.phantomjs¶