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

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

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

还要确保您没有任何可能会在注册异常try/except之前耗尽异常的块。unittest

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