Richard Jones' Log

Mon, 30 Jan 2012
PyPI is an OpenID provider

PyPI is now an OpenID provider.

To use this OpenID provider, enter into any form that expects an OpenID*. Should the service not support OpenID 2, you will have to enter instead (using your PyPI username.) Log into PyPI and visit your details page if you'd like to cut-n-paste the URL.

We follow the emerging approach that you have to sign into PyPI before signing into the actual services. This is intended to prevent phishing, as otherwise the relying party may fake PyPI's login page and collect your PyPI password (which they can still do if you fall for it.) It also avoids "nested" logins (i.e. where you need to log into PyPI with an OpenID while trying to login elsewhere with the PyPI id.)

If you find any problems with this service, please report them to the PyPI bug tracker.

*: of course for sites that extend PyPI this can be simplified to a simple button saying "link to my PyPI account".

category: Python | permanent link
Fri, 06 Jan 2012
Tips for Testing Twisted

This post contains some tips for testing Twisted applications. Note that I'm not using trial, the official Twisted test runner, for a variety of reasons.

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:


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:

observer = log.PythonLoggingObserver()

and then patch nose.twistedtools as follows:

--- original/nose/
+++ installed/nose/
@@ -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
         from twisted.internet import reactor
@@ -135,6 +142,8 @@
             def errback(failure):
                 # Retrieve and save full exception info
+                    if issubclass(failure.type, StandardError):
+                        print failure
@@ -156,6 +165,9 @@
                 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')
def step(context):
    # wait until we're all done
    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)

More tips as I happen to think of them or discover them :-)

category: Python | permanent link