testing threads with py.test

Posted on May 12, 2008
Filed Under python |

i’m a big fan of automated unit tests. my preferred testing utility is py.test. for most things it works pretty well and its super easy to use.

threads are tricky to deal with but especially while trying to test them if you have a lot going on at the same time. by default py.test will not catch exceptions which occur in threads which, unless you’re careful to check for them, could lead to inaccurate test results.

i wrote the following class to help with thread testing:

class ThreadMonitor(object):
   """Helper class for catching exceptions generated in threads.

   Usage:

      mon = ThreadMonitor() 

      th = threading.Thread(target=mon.wrap(myFunction))
      th.start()

      th.join()

      mon.check() # raises any exception generated in the thread

   Any raised exception will include a traceback from the original
   thread, not the function calling mon.check()

   Works for multiple threads
   """
   def __init__(self):
      self.queue = Queue.Queue()

   def wrap(self, function):
      def threadMonitorWrapper(*args, **kw):
         try:
            ret = function(*args, **kw)
         except Exception, e:
            self.queue.put(sys.exc_info())
            raise

         return ret

      return threadMonitorWrapper

   def check(self):
      try:
         item = self.queue.get(block=False)
      except Queue.Empty:
         return

      klass, value, tb = item

      raise klass, value, tb # note the last parameter - traceback

this doesn’t fill all of one’s py.test thread testing needs, but it helps for certain cases.

the final line with the “raise” keyword raises the exception caught in the thread but the traceback argument causes py.test to show the original context of the exception. so with the “–pdb” option to py.test you’ll automatically drop into the context of the first exception on a failure in a monitored thread.

Comments

Leave a Reply