使 Python 单元测试因任何线程的异常而失败

我正在使用单元测试框架来自动化多线程 python 代码、外部硬件和嵌入式 C 的集成测试。尽管我公然滥用单元测试框架进行集成测试，但它的效果非常好。除了一个问题：如果任何生成的线程引发异常，我需要测试失败。这可以通过单元测试框架实现吗？

一个简单但不可行的解决方案是 a) 重构代码以避免多线程或 b) 单独测试每个线程。我不能这样做，因为代码与外部硬件异步交互。我还考虑过实现某种消息传递以将异常转发到主单元测试线程。这将需要对正在测试的代码进行与测试相关的重大更改，我想避免这种情况。

是时候举个例子了。我可以修改下面的测试脚本以在 my_thread 中引发的异常上失败吗without修改 x.ExceptionRaiser 类？

import unittest
import x

class Test(unittest.TestCase):
    def test_x(self):
        my_thread = x.ExceptionRaiser()
        # Test case should fail when thread is started and raises
        # an exception.
        my_thread.start()
        my_thread.join()

if __name__ == '__main__':
    unittest.main()

首先，sys.excepthook看起来像是一个解决方案。它是一个全局钩子，每次抛出未捕获的异常时都会调用它。

不幸的是，这不起作用。为什么？出色地threading包裹你的run代码中的函数打印您在屏幕上看到的可爱的回溯（注意到它总是告诉您Exception in thread {Name of your thread here}？这就是它的完成方式）。

从 Python 3.8 开始，您可以重写一个函数来实现此功能：线程.excepthook https://docs.python.org/3/library/threading.html#threading.excepthook

...可以重写 threading.excepthook() 以控制如何处理 Thread.run() 引发的未捕获异常

那么我们该怎么办？用我们的逻辑替换这个函数，并且voilà:

对于Python >= 3.8

import traceback
import threading 
import os


class GlobalExceptionWatcher(object):
    def _store_excepthook(self, args):
        '''
        Uses as an exception handlers which stores any uncaught exceptions.
        '''
        self.__org_hook(args)
        formated_exc = traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback)
        self._exceptions.append('\n'.join(formated_exc))
        return formated_exc

    def __enter__(self):
        '''
        Register us to the hook.
        '''
        self._exceptions = []
        self.__org_hook = threading.excepthook
        threading.excepthook = self._store_excepthook

    def __exit__(self, type, value, traceback):
        '''
        Remove us from the hook, assure no exception were thrown.
        '''
        threading.excepthook = self.__org_hook
        if len(self._exceptions) != 0:
            tracebacks = os.linesep.join(self._exceptions)
            raise Exception(f'Exceptions in other threads: {tracebacks}')

对于旧版本的 Python，这有点复杂。 长话短说，看来threadingnodule 有一个未记录的导入，它执行以下操作：

threading._format_exc = traceback.format_exc

毫不奇怪，只有当线程抛出异常时才会调用该函数run功能。

所以对于 python

import threading 
import os

class GlobalExceptionWatcher(object):
    def _store_excepthook(self):
        '''
        Uses as an exception handlers which stores any uncaught exceptions.
        '''
        formated_exc = self.__org_hook()
        self._exceptions.append(formated_exc)
        return formated_exc
        
    def __enter__(self):
        '''
        Register us to the hook.
        '''
        self._exceptions = []
        self.__org_hook = threading._format_exc
        threading._format_exc = self._store_excepthook
        
    def __exit__(self, type, value, traceback):
        '''
        Remove us from the hook, assure no exception were thrown.
        '''
        threading._format_exc = self.__org_hook
        if len(self._exceptions) != 0:
            tracebacks = os.linesep.join(self._exceptions)
            raise Exception('Exceptions in other threads: %s' % tracebacks)

Usage:

my_thread = x.ExceptionRaiser()
# will fail when thread is started and raises an exception.
with GlobalExceptionWatcher():
    my_thread.start()
    my_thread.join()
            

你还需要join你自己，但在退出时，with语句的上下文管理器将检查其他线程中抛出的任何异常，并适当地引发异常。

本代码“按原样”提供，不提供任何形式的保证， 明示或暗示

这是一个没有记录的、有点可怕的黑客行为。我在linux和windows上测试了它，似乎可以工作。需要您自担风险使用它。