Python unittest and multithreading

Question

I am using python unittest and would like to write a test that runs multiple threads and waits for them to complete. unittest execute a function that has several unittest statements. If any of the statements fails, I want the test to fail. This does not seem to be the case.

EDIT: Minimal runnable example (python3)

 import unittest import threading class MyTests(unittest.TestCase): def test_sample(self): t = threading.Thread(target=lambda: self.fail()) t.start() t.join() if __name__ == '__main__': unittest.main() 

and output:

 sh-4.3$ python main.py -v test_sample (__main__.MyTests) ... Exception in thread Thread-1: Traceback (most recent call last): File "/usr/lib64/python2.7/threading.py", line 813, in __bootstrap_inner self.run() File "/usr/lib64/python2.7/threading.py", line 766, in run self.__target(*self.__args, **self.__kwargs) File "main.py", line 7, in <lambda> t = threading.Thread(target=lambda: self.fail()) File "/usr/lib64/python2.7/unittest/case.py", line 450, in fail raise self.failureException(msg) AssertionError: None ok ---------------------------------------------------------------------- Ran 1 test in 0.002s OK 

Answer 1

Python unittest are passed using exceptions, so you must ensure that exceptions end in the main thread. Thus, for a stream, which means you need to run .join() , as this will throw an exception from the stream into the main stream:

  t = threading.Thread(target=lambda: self.assertTrue(False)) t.start() t.join() 

Also, make sure you don't have try/except blocks that can throw an exception before unittest can register them.

Edit : self.fail() does not really get passed when called from the stream, even if .join() present. Not sure about that.

Answer 2

In my main thread, I find that subprocesses fail by checking their exit code (nonzero).

 proc.join() self.assertEqual(proc.exitcode, 0, 'Sub-process failed, check output for stack trace')