Python multi-threaded unittesting

Question

I use a multi-threaded design (had no choice), but most of my code resides in a single thread where all events in it are managed via a queue. In this fashion most of my code behaves as if it is single threaded, and I don't have to worry about locks, semaphores and what not.

Alas I've come to the point where I need to unittest my code (please don't lash for not TDDing in the first place), and I'm at a loss - how do you test something in another thread?

For instance, say I have the following class:

class MyClass():
    def __init__(self):
        self.a=0
        # register event to self.on_event

    def on_some_event(self, b):
        self.a += b

    def get(self):
        return self.a

and I want to test:

import unittest
from queued_thread import ThreadedQueueHandler

class TestMyClass(unittest.TestCase):
    def setUp(self):
        # create the queued thread and assign the queue to self.queue

    def test_MyClass(self):
        mc = MyClass()
        self.queue.put({'event_name':'some_event', 'val':1})
        self.queue.put({'event_name':'some_event', 'val':2})
        self.queue.put({'event_name':'some_event', 'val':3})
        self.assertEqual(mc.get(),6)

if __name__ == '__main__':
    unittest.main()

MyClass.get() works fine for anything inside the queued thread, but it will be called asynchronously in the main thread by the test, thus the result may not be correct!

Jonathan Livni · Accepted Answer · 2011-07-08 08:30:31Z

10

If your design assumes everything must go through the queue, then don't fight it - make everything go through it!

Add an on_call event to your queued event handler, and register to it the following function:

def on_call(self, callback):
    callback()

then modify your test to:

def test_MyClass(self):
    def threaded_test():
        self.assertEqual(mc.get(),6)

    mc = MyClass()
    self.queue.put(1)
    self.queue.put(2)
    self.queue.put(3)
    self.queue.put({'event_name':'call','val':threaded_test})

edited Jul 8, 2011 at 8:30

answered Jun 26, 2011 at 17:20

Jonathan Livni

108k112 gold badges278 silver badges367 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Jared Smith Over a year ago

This might be the most practical (i.e. not just a dupe target) self-answered question I've seen on SO.

Senthil Kumaran · Accepted Answer · 2011-06-26 17:26:09Z

2

You can have a look at test_threading.py in the stdlib tests which does something similar to what you are trying to do. The basic idea is to protect a thread execution with mutex and semaphore so that the execution is complete before the test condition is asserted.

answered Jun 26, 2011 at 17:26

Senthil Kumaran

57.3k15 gold badges99 silver badges139 bronze badges

2 Comments

Jonathan Livni Over a year ago

Any chance you know of a tutorial\doc for this?

Senthil Kumaran Over a year ago

The code I linked itself is easier to follow and you may take some help with threading documentation at python.org docs.

Collectives™ on Stack Overflow

Python multi-threaded unittesting

2 Answers 2

1 Comment

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related