Moving to

Recently I discovered static site generator and I think it’s a better platform for a tech blog. From now on, my tech blog will be hosted at

Use Python’s sys.settrace() for fun and for profit

The itch to scratch

Everyone in the software industry knows Kent Beck, the pioneers of extreme programming and test-driven development and the co-author of JUnit. One of his lesser known project was JUnitMax, which aims to reduce the time developers have to wait while tests are running. One of the ideas behind that is that when code changes, only the test cases that exercise the code need to be run, instead of running the entire suite. The idea makes a lot of sense to me, but at the time, I (and the development shop I was in) wasn’t practising enough TDD, so unit test time wasn’t a big problem for me back then.

Fast-forward a few years, now as the project in my current company gets bigger, the time it takes to run tests is slowly becoming an impeding factor of my productivity. I remembered JUnitMax and say to myself, wouldn’t it be neat if something like JUnitMax were available? As the name suggests, JUnitMax is for Java while my project is in Python. Java, being a statically-typed language, has the blessings of statical analysis, which means a tool like JUnitMax can figure out which test cases cover which lines of code simply by type analysis. Python, however, being a dynamic language, doesn’t have this ability.

A few days ago, while I was running unit tests with coverage, it dawned on me that if the coverage tool knows which lines of the source code is covered by unit tests, couldn’t the same technique be used to figure out which lines are covered by which test cases?

So, I started looking into coveragepy‘s source code, and watching its author Ned Batchelder‘s excellent PyCon2011 video on sys.settrace. I wanted to build a proof-of-concept tool that integrates with the de-facto Python unit-test tool nose, that, when run, gathers the information about which lines in the files in the source folder are covered by which test cases, and hence nostrils is born.

Here comes `sys.settrace()`

Python’s motto is “batteries included”. This is manifested in many Python’s stanndard library modules, such as ast (source code parsing) and dis (bytecode disassembly). One of which is the ability to make the Python interpreter call an external function whenever a line of code is being executed. You can do a lot of fun stuff with it, for example, uses this to build code coverage data; pdb uses it to insert breakpoints into a running application and change the way a Python program is executed.

How can it be used?

For nostrils, we need to write a nose plugin that installs the trace function when a test is encountered. The trace function records the line numbers and the current test case name. After all tests are run, we have our map.

A simple use case

To start, we need a simple use case:

# this is the code-under-test
def add(x, y):
    z = x + y
    return z

def subtract(x, y):
    z = x - y
    return z


# test cases

import worker

def test_add():
    assert 1 == worker.add(1, 0)

def test_add___negative():
    assert 0 == worker.add(-1, 1)

def test_subtract():
    assert 0 == worker.subtract(0, 0)

class TestFoo(object):

    def test_add(self):
        assert 5 == worker.add(5, 0)

As you can see, we have 4 tests and 2 methods-under-test. Our goal is that when running `nosetests –with-nostrils` (–with-nostrils is the switch to turn on the nostrils plugin), we get the following mappings:

def add(x, y):
  z = x + y # test_add, test_add_negative, TestFoo.test_add
  return z  # test_add, test_add_negative, TestFoo.test_add

def subtract(x, y):
  z = x - y # test_subtract
  return z  # test_subtract

Nose plugin

I won’t go into the details about how to create a plugin for nose. You can read it here, and you can take a look at my sample setup here. In a nutshell, every plugin has a name, and when nose is supplied with –with-plugin_name, your plugin is activated. Nose provides a test lifecycle “hooks” that plugins can implement. For example, startTest is called when a test case is discovered and adapted into a nose TestCase. addSuccess is called when a test case succeeded. finalize is called when all tests are finished.

Here’s how my plugin looks like:

class Nostrils(Plugin):
    name = 'nostrils'

    def addError(self, test, err, *args):

    def addFailure(self, test, err, *args):

    def addSkip(self, test, err):

    def addSuccess(self, test, err):

    def startTest(self, test):
        self._current_test = test

    def finalize(self, result):

    def _install_tracefn(self):
        self._orig_tracefn = sys.gettrace()
        sys.settrace(self._trace) # See below

    def _restore_tracefn(self):

The idea is that we install the trace function when test starts, and restore the trace function back to what it was. We also keeps track of what’s the current test in self._current_test.

Trace function

Now let’s have a look at the trace function:

class Nostrils(Plugin):
  # ...
  def _trace(self, frame, event, arg):
    if event == 'line':
    return self._trace

  def _trace_down(self, frame):
    while frame is not None:
      if frame.f_code == test.__call__.func_code:

      frame = frame.f_back

A trace function should take 3 parameters:

  • frame: the current frame object
  • event: what type of event that triggered the trace function? See here
  • &asterisk;args: any additional arguments

Here, I’m only interested in the ‘line’ event, which is triggered when a new line of code is being executed. When this happens, we invoke _trace_down, which walks the frame stack by recursing on frame.f_back. When it’s None, we’re at the bottom of the stack. Because we’re tracing the execution of tests, we can probably stop traversing when the code object of the frame is the entry point of the test case (if frame.f_code == test.__call__.func_code). This way, we save ourselves some unnecessary traversals.

Data Collection

There’s are few things we need to collect: filename, line number of the code being executed and the test case name that covers the code.

class Nostrils(Plugin):
  def __init__(self):
    super(Nostrils, self).__init__()
    self._data = defaultdict(
      lambda : defaultdict(
        lambda : set([])

  def _collect(self, frame):
    filename, lineno = frame.f_code.co_filename, frame.f_lineno
    self._data[filename][lineno].add("%s:%s.%s" % self._current_test.address())

The data structure we use here is a dictionary of dictionary. At the top level, the keys are filenames, and the values are dictionaries of with keys the line numbers and the values the set of test case names. The data structure looks like this:

      1 : set(['', '']),
      2 : set(['', '']),
      3 : set([''])

There we have it! We have a prototype of what could become a PyUnitMax 😉

Potential Problems

  • Scale: Now I’m only running nostrils on trivial code base. Profiling and optimization is needed if nostrils were to be used in real-world cases.
  • Multi-threading: No consideration was given to multi-threading at this stage.

Collaborators welcome!

I have since refactored the code, revised the data structure and published it on github. Please provide me with feedbacks and suggestions.

Realtime notification delivery using rabbitmq, Tornado and websocket

Our company has “hack-off” days once a while, where we developers get to choose whatever we would like to work on and present it to the entire company by the end of the day. I have been hearing this websocket buzz for a while now and would like to build something interesting with it.


Websocket is a persistent bi-directional connection between the browser and the server. With websocket, web browser can post message to the server, but what’s more interesting is that the server is able to push messages to the client (browser). This breaks away from the traditional web application request/response model. Traditionally, the client makes the request and waits for the server to give an answer. AJAX is revolutionary, but essentially, it’s still the same model: the client asks the server whether there’s anything interesting, but not the other way around. With websocket, the server suddenly becomes more involved and able to deliver more engaged user experience.

Our company provides web application for online invoicing. The web application allows users to create clients, create invoices, send them to clients, and so on. Each one of these are “events” which gets sent to RabbitMQ. We then have a plethora of RabbitMQ consumers that read messages off the queue and do interesting stuff with them.

Proof of concept

For this hack-off, my goal is to write a RabbitMQ consumer that reads the messages off the message queue, and deliver (notify) them to the front-end using websocket.

I’ve heard good things about Tornado. Having read their docs on websocket request handler, I felt it’s straightforward enough for me, so I chose Tornado as my backend.


One problem arises, though: The tornado server will run as a regular server, waiting for incoming websocket connections. The RabbitMQ consumer also needs to be in the same process event loop, waiting for incoming messages from the message queue. I looked into a few solutions such as sparkplug and stormed-amqp, neither seem to be a good hit here. Finally, I stumbled on Pika. It comes with a Tornado event loop adapter, which allows rabbitmq consumer and websocket handlers to run inside the same event loop. Perfect.

The entry point looks like this:

application = tornado.web.Application([
    (r'/ws', handlers.MyWebSocketHandler),

def main():

    io_loop = tornado.ioloop.IOLoop.instance()

    # PikaClient is our rabbitmq consumer
    pc = client.PikaClient(io_loop)
    application.pc = pc


class MyWebSocketHandler(tornado.websocket.WebSocketHandler):

    def open(self, *args, **kwargs):"WebSocket opened")

    def on_close(self):"WebSocket closed")

That was straightforward. However, I’m faced with the problem of how to make the amqp consumer notify websocket handlers when we receive a message from the message queue. We cannot get the handler instances from the tornado application object. Note, each websocket connection has a corresponding “MyWebSocketHandler“ instance. The instances are not available from the application object. Maybe there’s a way to get them by other means, but I’m not familiar with the tornado API enough to know that.

However, from the handler, we do get the “application“ object, and because we attached pika_client (our amqp consumer) to the application, we have access to it inside our socket handler. Hey, how about registering the handler with the client when the websocket is connected, and let the client “notify” the handler when events are received? Hey, isn’t that the observer pattern?

Here’s the code:

class MyWebSocketHandler(websocket.WebSocketHandler):

    def open(self, *args, **kwargs):
        self.application.pc.add_event_listener(self)"WebSocket opened")

    def on_close(self):"WebSocket closed")

Now, our PikaClient object need to support add_event_listener() and remove_event_listener() methods.

class PikaClient(object):

    def __init__(self, io_loop):'PikaClient: __init__')
        self.io_loop = io_loop

        self.connected = False
        self.connecting = False
        self.connection = None = None

        self.event_listeners = set([])

    def connect(self):
        if self.connecting:
  'PikaClient: Already connecting to RabbitMQ')
            return'PikaClient: Connecting to RabbitMQ')
        self.connecting = True

        cred = pika.PlainCredentials('guest', 'guest')
        param = pika.ConnectionParameters(

        self.connection = TornadoConnection(param,

    def on_connected(self, connection):'PikaClient: connected to RabbitMQ')
        self.connected = True
        self.connection = connection

    def on_channel_open(self, channel):'PikaClient: Channel open, Declaring exchange') = channel
        # declare exchanges, which in turn, declare
        # queues, and bind exchange to queues

    def on_closed(self, connection):'PikaClient: rabbit connection closed')

    def on_message(self, channel, method, header, body):'PikaClient: message received: %s' % body)

    def notify_listeners(self, event_obj):
        # here we assume the message the sourcing app
        # post to the message queue is in JSON format
        event_json = json.dumps(event_obj)

        for listener in self.event_listeners:
  'PikaClient: notified %s' % repr(listener))

    def add_event_listener(self, listener):
        self.event_listeners.add(listener)'PikaClient: listener %s added' % repr(listener))

    def remove_event_listener(self, listener):
  'PikaClient: listener %s removed' % repr(listener))
        except KeyError:

I left out the queue setup code here for brevity. `on_message` callback is called when the consumer gets a message from the queue. The client, in turn, notifies all registered websocket handlers. Obviously, in real applications, you may want to do some kind of credentials and filtering, so the right message get to the right receiver. Then we simply call handler.write_message(), so the message gets relayed to the front-end’s websocket.onmessage callback.

Here’s some front-end code:

    $(document).ready(function() {
        var ws = new WebSocket('ws://localhost:8888/ws');
        ws.onmessage = function(evt){

Yes, we simply echo the message back. For the hackoff, I did parse the data, render a slightly more detailed notification message, and display the notification using jquery-toaster.


This is my first stab at websocket and the tornado web framework. I’m not an expert on either subject, so chances are there are better ways to achieve the same result.

I think websocket is a very interesting technology. It opens a wide range of possibilities for more interactive and engaging web applications. Our web application is of traditional architecture: server renders most of the page, and every request involves page loads. Having a websocket may not be very beneficial as the application doesn’t have that much of user interaction. My hackoff is more of a proof of concept. However, if the application is a one-page web app (no full page reloads), the websocket model works very well.

Scala Simple Build Tool — Not so simple after all, at least for now…

Update:I got sbt working by building directly from the master branch from their github repo. The current version is 0.7.5. The tagged 0.9.4 version is actually an older version. Anyway, tried it and kinda loved it.

This is just another late night rambling…I was trying to get a proper scala build system setup. I was using Maven scala plugin for a while, but longing for something simpler and more scalanic (is there such a word?). I was pretty happy at Cake, the Clojure build system and expected SBT to allow me to break away from using Maven to build Scala projects…boy, was I wrong…

First off, when you google ‘simple build tool’, you get a link to the SBT Google code home page. Well, nothing wrong there, except the “latest” version on Google code was 0.7.4 and it was half a year ago…Maybe it’s not that outdated, so I downloaded it, followed this instruction and setup my ~/bin/sbt script. Running it, it asked me to setup projects, and it only supported up until Scala 2.7.7…Hrm, 2.8 was out for a while now, so obviously, SBT 0.7.4 isn’t the latest. Reading their home page more carefully, they’re moving the repository to Github. Awesome! I’d pick Github over Google Code any time too.

Heading over to their Github repo, and found the latest stable version is 0.9.2. Good! So it should support Scala 2.8 now! Downloaded the zip, unzipped it, and of course it wasn’t executable. You need to build it. There’s a, so quickly I less’ed it. For step 1, it asked me to go to the setup wiki page on Google Code (!), which is the steps I did setting up 0.7.4…I guess they’re using 0.7.4 as a bootstrapping build…Anyways, I did that. Step 2 was to run `sbt update “project Launcher” proguard “project Simple Build Tool” “publish-local”`. Of course it didn’t work. It’s complained 0.7.4 version of sbt-launch can’t download Scala 2.7.7 from any of the repository…bummer! But hey, I can download Scala 2.7.7 lib from Maven! So I quickly updated pom.xml of one of my projects to use Scala 2.7.7 and did an upgrade. Now 2.7.7 is happily in my local Maven repo. Ran that command again, hooray! It started to build, and judging by the number of packages it’s building, “simple” isn’t the first adjective that comes into my mind. Anyway, it’s building at least, so even if it’s a little complicated, so be it…Except…of course it broke half way… and why?

[info] Post-analysis: 107 classes.
[info] == Precompiled 2.7.7 / compile ==
[info] Precompiled 2.8.0 / compile …
[info] == Precompiled 2.8.0 / compile ==
[info] Source analysis: 9 new/modified, 0 indirectly invalidated, 0 removed.
[info] Compiling main sources…
[warn] there were deprecation warnings; re-run with -deprecation for details
[warn] one warning found
[info] Compilation successful.
[info] Post-analysis: 108 classes.
[info] == Precompiled 2.8.0 / compile ==
java.lang.OutOfMemoryError: PermGen space
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClassCond(
at java.lang.ClassLoader.defineClass(

You’ve gotta be kidding me! I set -Xmx512M and it’s not enough? And why is it building every version of Scala *from source*?? Is there something called a…JAR?

Anyway, increased -Xmx from 512 to 1024M, ran again, wait, and same thing happened again! Out of PermGen space…urrgh…

I decided to give up, at least for the day… SBT is anything but simple, at least from my experience. I know it’s open source and people put efforts into it without compensation, so I shouldn’t be critical about it. I’ll give it a try again, and hopefully it’s worth the time investment.

Cake – the yummy Clojure build system

About 10 minutes ago I heard about cake clojure build system, and gave it a try. And 10 minutes later, it won me over! Wow, it addresses all the pain points of leiningen

Sorry for using all CAPS but I’m very excited about this improvement over leiningen — OK, it may not be the fault of leiningen that JVM cold startup time is non-trivial but hey, someone came up with an idea of having a long running JVM process in the background, so subsequent clojure tasks reuse the same JVM instance. Cake folks integrated that nicely. It takes about 10-15 seconds to boot up a JVM but subsequent cake tasks or execution of clojure code is virtually instant! Comparing to leiningen, which doesn’t take this approach and every single task (such as common ones like lein test) takes around 5 seconds. This adds up quickly and makes you less efficient. The speed improvement alone is enough for me to switch to cake.

Advanced REPL functionalities: tab completion, history
It just works. Very useful for having instant feedbacks while exploring the language and API. No more manually adding jLine to your classpath or hack around tab completion wrapper…It just works! (I know I said it already)

run clojure files directly
OK, leiningen can do this too, but through plugin. I feel this is a very handy functionality, which probably should be included in the core.

Detects your code change and automatically run your test suites! Sweet.

compatible with leiningen project definition files
Cake understand project.clj, so I don’t need to do anything for my existing leiningen projects. Change directory to the project and `cake` away 😀

Overall, it just works out of the box. No more mucking around with dev-dependencies and other chores and let you focus on what you’d love to do.