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

Question

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

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

是时候举个例子了。我可以修改下面的测试脚本以在 my_thread 中引发的异常上失败修改 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()

Answer 1

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

很遗憾，这不起作用。为什么？好吧，threading 将您的 run 函数包装在代码中，该函数会打印您在屏幕上看到的可爱回溯（注意它总是如何告诉您 Exception in thread {Name of your thread here}？这就是它的完成方式）。

从 Python 3.8 开始，您可以重写一个函数来完成这项工作：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，这有点复杂。 长话短说，threading 结节似乎有一个未记录的导入，它的作用类似于：

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)

用法：

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

您仍然需要自己join，但退出时，with-statement 的上下文管理器将检查其他线程中抛出的任何异常，并适当地引发异常。

---

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

这是一个无证的、可怕的黑客攻击。我在linux和windows上测试过，似乎可以。使用它需要您自担风险。

Answer 2

我自己也遇到过这个问题，我能想出的唯一解决方案是将 Thread 子类化以包含一个属性，以判断它是否在没有未捕获异常的情况下终止：

from threading import Thread

class ErrThread(Thread):
    """                                                                                                                                                                                               
    A subclass of Thread that will log store exceptions if the thread does                                                                                                                            
    not exit normally                                                                                                                                                                                 
    """
    def run(self):
        try:
            Thread.run(self)
        except Exception as self.err:
            pass
        else:
            self.err = None


class TaskQueue(object):
    """                                                                                                                                                                                               
    A utility class to run ErrThread objects in parallel and raises and exception                                                                                                                     
    in the event that *any* of them fail.                                                                                                                                                             
    """

    def __init__(self, *tasks):

        self.threads = []

        for t in tasks:
            try:
                self.threads.append(ErrThread(**t)) ## passing in a dict of target and args
            except TypeError:
                self.threads.append(ErrThread(target=t))

    def run(self):

        for t in self.threads:
            t.start()
        for t in self.threads:
            t.join()
            if t.err:
                raise Exception('Thread %s failed with error: %s' % (t.name, t.err))

Answer 3

我已经使用上面接受的答案有一段时间了，但是从 Python 3.8 开始，该解决方案不再起作用，因为 threading 模块不再具有此 _format_exc 导入。

另一方面，threading 模块现在有一个很好的方法来注册自定义，除了 Python 3.8 中的钩子，所以这里有一个简单的解决方案来运行单元测试，它断言在线程内部引发了一些异常：

def test_in_thread():
    import threading

    exceptions_caught_in_threads = {}

    def custom_excepthook(args):
        thread_name = args.thread.name
        exceptions_caught_in_threads[thread_name] = {
            'thread': args.thread,
            'exception': {
                'type': args.exc_type,
                'value': args.exc_value,
                'traceback': args.exc_traceback
            }
        }

    # Registering our custom excepthook to catch the exception in the threads
    threading.excepthook = custom_excepthook

    # dummy function that raises an exception
    def my_function():
        raise Exception('My Exception')

    # running the funciton in a thread
    thread_1 = threading.Thread(name='thread_1', target=my_function, args=())

    thread_1.start()
    thread_1.join()

    assert 'thread_1' in exceptions_caught_in_threads  # there was an exception in thread 1
    assert exceptions_caught_in_threads['thread_1']['exception']['type'] == Exception
    assert str(exceptions_caught_in_threads['thread_1']['exception']['value']) == 'My Exception'