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     

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

Answer 2

您的测试没有失败，原因与此代码将打印“无异常”相同

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

当 unittest 运行您的测试函数时，它会通过查看代码执行是否导致某些异常来确定通过/失败。如果异常发生在线程内部，那么主线程中仍然没有异常。

如果你认为你必须通过在线程中运行某些东西来获得通过/失败结果，你可以这样做。但这确实不是 unittest 的设计方式，并且可能有一种更简单的方法来完成您想要完成的工作。

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

Answer 3

Python unittest 断言通过异常进行通信，因此您必须确保异常最终出现在主线程中。所以对于一个线程，这意味着你必须运行.join()，因为这会将异常从线程抛出到主线程：

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

还要确保在unittest 注册它们之前没有任何try/except 块可能会吃掉异常。

编辑：self.fail() 在从线程调用时确实没有通信，即使 .join() 存在。不知道这是怎么回事。