Python 单元测试和多线程

Question

我正在使用 python 的 unittest 并且想编写一个测试来启动几个线程并等待它们完成。线程执行一个函数，该函数具有一些 unittest 断言。如果任何断言失败，我希望测试失败。情况似乎并非如此。

编辑： 最小可运行示例（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()

输出是：

 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 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

使用 concurrent.futures.ThreadPoolExecutor 或 https://docs.python.org/3/library/threading.html#threading.excepthook 来收集线程中抛出的异常

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()

在 pytest 上，这些是为您收集的： https ://docs.pytest.org/en/latest/how-to/failures.html?highlight=unraisable#warning-about-unraisable-exceptions-and-unhandled-thread-exceptions

原文由 Thomas Grainger 发布，翻译遵循 CC BY-SA 4.0 许可协议