TL;DR: fork and os._exit(0) the fixture callable
Getting py.test to test Twisted apps is supported to some extent, albeit somewhat briefly documented; also there’s a py.test plugin to help with testing Twisted apps.
Tests usually require fixtures to be set up. Let’s assume your tests require running something in a separate process, for example a server such as MySQL. What to do? Ok, you can use subprocess.Popen, or Twisted spawnProcess to spin up the database. Note that you should probably not use multiprocessing: It uses its own loop for which there is no support in Twisted.
But what if it’s Python code you want to run? Yes, you can put it into a module and run the module using the above methods. However, if you want to use a Python callable defined in your test module you’re out of luck: neither subprocess.Popen and spawnProcess: nor can run a Python callable in a subprocess.
In that case, you need os.fork. Simply run the callable in the child, and depending on your use case, either wait for it to complete in parent, or kill it at the end of the test. However there’s one gotcha, at least when using py.test: since you’re forking a running test, py.test will now report two tests running and completing. The solution is to exit the child abnormally; simple sys.exit() will raise an exception, but doing os._exit(0) does not.
Here’s example code that spins up a simple test HTTP server for one request, and checks that content fetched by HTTP client matches that served by the HTTP server:
import os
from httpserver import BaseHTTPServer
import pytest
import treq
from turq import TurqHandler
def serve_request(host, port, rulecode):
TurqHandler.rules = parse_rules(rulecode)
server = BaseHTTPServer.HTTPServer((host,port), TurqHandler)
server.handle_request() # nothing more needed fro this one test
@pytest.inlineCallbacks
def test_something():
pid = os.fork()
# set up fixture in child
if pid == 0:
serve_request("127.0.0.1", 8080, "path('*').text('Hello')")
os._exit(0)
# proceed in parent (test), wait a bit first for the server fixture to come up
time.sleep(0.5)
# make a request
r = treq.get("http://127.0.0.1:8080")
# kill server in child if we cannot connect
try:
response = yield r
except Exception as exc:
os.kill(pid, signal.SIGKILL)
raise
responsetext = yield treq.content(response)
assert responsetext == "Hello"
I don’t know whether the same technique works with nose and/or Twisted Trial – let me know if you find out!
