Richard Jones' Log: Tips for Testing Twisted

Twisted Deferreds in Testing

You could use trial, but if you need to use another testing tool (nose, or behave) then you'll need to do a little more work.

To let your deferreds run (and make sure they do finish) wrap your test functions in the "@deferred" decorator from nose.twistedtools:

@deferred(1)

This will run the reactor for 1 second before declaring it hung.

If you want to see deferreds that are not behaving you will need to enable logging of Twisted events if they're not already enabled:

defer.setDebugging(1)
observer = log.PythonLoggingObserver()
observer.start()

and then patch nose.twistedtools as follows:

--- original/nose/twistedtools.py
+++ installed/nose/twistedtools.py
@@ -39,12 +39,19 @@

 _twisted_thread = None

+_deferreds = []
+
 def threaded_reactor():
     """
     Start the Twisted reactor in a separate thread, if not already done.
     Returns the reactor.
     The thread will automatically be destroyed when all the tests are done.
     """
+    from twisted.internet import defer
+    def __init__(self):
+        _deferreds.append(self)
+    defer.DebugInfo.__init__ = __init__
+
     global _twisted_thread
     try:
         from twisted.internet import reactor
@@ -135,6 +142,8 @@
             def errback(failure):
                 # Retrieve and save full exception info
                 try:
+                    if issubclass(failure.type, StandardError):
+                        print failure
                     failure.raiseException()
                 except:
                     q.put(sys.exc_info())
@@ -156,6 +165,9 @@
             try:
                 error = q.get(timeout=timeout)
             except Empty:
+                for d in _deferreds:
+                    if hasattr(d, 'creator') and not hasattr(d, 'invoker'):
+                        print d._getDebugTracebacks()
                 raise TimeExpired("timeout expired before end of test (%f s.)"
                                   % timeout)
             # Re-raise all exceptions

(I tried a couple of times to figure out how trial does the above but failed miserably)

This will result in a display of all the deferreds still pending when the deferred decorator timeout fires. Note that the deferred decorator introduces a deferred that will not have terminated.

One outstanding problem I've not solved is that inlineCallbacks aren't displayed usefully in the failure debug output - you get told that it is some inline callback that is still hanging, but not where it was created.

Twisted Web and Selenium

Testing Twisted web using Selenium is made more complicated by the need to allow the Twisted reactor to run at the same time as the selenium webdriver is trying to poke at it.

In other applications I've tested like this I just run up the server being tested in a separate thread but I can't do that in the case of Twisted.

To make this work any selenium operation that will cause a Twisted reactor event must be deferred to a thread, hence the liberal sprinkling of deferToThread throughout this code. Some sections do not touch the reactor (for example, filling out a form once the form page has been loaded doesn't), so it can be straight selenium calls.

Note also that the pages are generated in a server dynamically using deferreds so I use a "done" tag marker at the end to indicate to the tests that the page has been fully generated. Selenium's timeout on element access allows me to cause the test to fail if the done marker does not appear.

For example this step function directly from a behave feature implementation:

@then('the message should be sent')
@deferred(1)
@inlineCallbacks
def step(context):
    # wait until we're all done
    context.browser.implicitly_wait(1)
    yield threads.deferToThread(lambda: get_element(context.browser, id="done"))

    # there should be a result container
    results = yield threads.deferToThread(lambda: find_elements(context.browser, id="result"))
    result = results[-1]
    status = yield threads.deferToThread(lambda: get_element(result, id="status"))

    # the status could be a number of values depending on synchronisiation of
    # the tests / runner
    assert status.text in 'created new accepted done'.split()

    message = context.db.messages.values()[0]
    if message.status != message.STATUS_DONE:
        yield esme._wait_until_done(context)