Python unittest and multithreading
Solution 1
use a concurrent.futures.ThreadPoolExecutor or https://docs.python.org/3/library/threading.html#threading.excepthook to collect exceptions thrown in threads
import unittest
import threading
from concurrent import futures
class catch_threading_exception:
"""
https://docs.python.org/3/library/test.html#test.support.catch_threading_exception
Context manager catching threading.Thread exception using
threading.excepthook.
Attributes set when an exception is catched:
* exc_type
* exc_value
* exc_traceback
* thread
See threading.excepthook() documentation for these attributes.
These attributes are deleted at the context manager exit.
Usage:
with support.catch_threading_exception() as cm:
# code spawning a thread which raises an exception
...
# check the thread exception, use cm attributes:
# exc_type, exc_value, exc_traceback, thread
...
# exc_type, exc_value, exc_traceback, thread attributes of cm no longer
# exists at this point
# (to avoid reference cycles)
"""
def __init__(self):
self.exc_type = None
self.exc_value = None
self.exc_traceback = None
self.thread = None
self._old_hook = None
def _hook(self, args):
self.exc_type = args.exc_type
self.exc_value = args.exc_value
self.exc_traceback = args.exc_traceback
self.thread = args.thread
def __enter__(self):
self._old_hook = threading.excepthook
threading.excepthook = self._hook
return self
def __exit__(self, *exc_info):
threading.excepthook = self._old_hook
del self.exc_type
del self.exc_value
del self.exc_traceback
del self.thread
class MyTests(unittest.TestCase):
def test_tpe(self):
with futures.ThreadPoolExecutor() as pool:
pool.submit(self.fail).result()
def test_t_excepthook(self):
with catch_threading_exception() as cm:
t = threading.Thread(target=self.fail)
t.start()
t.join()
if cm.exc_value is not None:
raise cm.exc_value
if __name__ == '__main__':
unittest.main()
on pytest these are collected for you: https://docs.pytest.org/en/latest/how-to/failures.html?highlight=unraisable#warning-about-unraisable-exceptions-and-unhandled-thread-exceptions
Solution 2
Your test isn't failing for the same reason that this code will print "no exception"
import threading
def raise_err():
raise Exception()
try:
t = threading.Thread(target=raise_err)
t.start()
t.join()
print('no exception')
except:
print('caught exception')
When unittest runs your test function, it determines pass/fail by seeing if the code execution results in some exception. If the exception occurs inside the thread, there still is no exception in the main thread.
You could do something like this if you think you HAVE to get a pass/fail result from running something in a thread. But this is really not how unittest is designed to work, and there's probably a much easier way to do what you're trying to accomplish.
import threading
import unittest
def raise_err():
raise Exception()
def no_err():
return
class Runner():
def __init__(self):
self.threads = {}
self.thread_results = {}
def add(self, target, name):
self.threads[name] = threading.Thread(target = self.run, args = [target, name])
self.threads[name].start()
def run(self, target, name):
self.thread_results[name] = 'fail'
target()
self.thread_results[name] = 'pass'
def check_result(self, name):
self.threads[name].join()
assert(self.thread_results[name] == 'pass')
runner = Runner()
class MyTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
runner.add(raise_err, 'test_raise_err')
runner.add(no_err, 'test_no_err')
def test_raise_err(self):
runner.check_result('test_raise_err')
def test_no_err(self):
runner.check_result('test_no_err')
if __name__ == '__main__':
unittest.main()
Solution 3
Python unittest assertions are communicated by exceptions, so you have to ensure that the exceptions end up in the main thread. So for a thread that means you have to run .join(), as that will throw the exception from the thread over into the main thread:
t = threading.Thread(target=lambda: self.assertTrue(False))
t.start()
t.join()
Also make sure that you don't have any try/except blocks that might eat up the exception before the unittest can register them.
Edit: self.fail() is indeed not communicated when called from a thread, even if .join() is present. Not sure what's up with that.
Ben RR
Updated on July 09, 2022Comments
-
Ben RR almost 2 years
I am using python's
unittestand would like to write a test that starts a few threads and waits for them to finish. The threads execute a function that has someunittestassertions. If any of the assertions fail, I wish the test to, well, 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 the output is:
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 -
Ben RR over 7 yearsI tried your example. It does not work for me (python3.x) and yields the same result as my code. The test still passes.
-
Syranolic almost 4 yearsNice, but new in version 3.8
