Spawn, control and monitor test services

The Service fixture can be used to spawn a background service process (for instance a web application), and leave it running for the duration of the test suite (see testresources-integration).

It supports real-time streaming of the service standard output to Python’s logging system.

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