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

问题描述

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

I am using the unittest framework to automate integration tests of multi-threaded python code, external hardware and embedded C. Despite my blatant abuse of a unittesting framework for integration testing, it works really well. Except for one problem: I need the test to fail if an exception is raised from any of the spawned threads. Is this possible with the unittest framework?

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

A simple but non-workable solution would be to either a) refactor the code to avoid multi-threading or b) test each thread separately. I cannot do that because the code interacts asynchronously with the external hardware. I have also considered implementing some kind of message passing to forward the exceptions to the main unittest thread. This would require significant testing-related changes to the code being tested, and I want to avoid that.

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

Time for an example. Can I modify the test script below to fail on the exception raised in my_thread without modifying the x.ExceptionRaiser class?

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 看起来像是一个解决方案.它是一个全局钩子，每次抛出未捕获的异常时都会调用它.

At first, sys.excepthook looked like a solution. It is a global hook which is called every time an uncaught exception is thrown.

很遗憾，这不起作用.为什么?threading 将您的 run 函数包装在代码中，该代码打印您在屏幕上看到的可爱的回溯(注意它总是告诉您 Exception in thread {Name of your thread here}?就是这样做的).

Unfortunately, this does not work. Why? well threading wraps your run function in code which prints the lovely tracebacks you see on screen (noticed how it always tells you Exception in thread {Name of your thread here}? this is how it's done).

从 Python 3.8 开始，您可以重写一个函数来完成这项工作:threading.excepthook

Starting with Python 3.8, there is a function which you can override to make this work: threading.excepthook

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

... threading.excepthook() can be overridden to control how uncaught exceptions raised by Thread.run() are handled

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

So what do we do? Replace this function with our logic, and voilà:

对于 python >= 3.8

For 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('
'.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 结节有一个未记录的导入，它执行以下操作:

For older versions of Python, this is a bit more complicated. Long story short, it appears that the threading nodule has an undocumented import which does something along the lines of:

threading._format_exc = traceback.format_exc

并不奇怪，这个函数只有在线程的run函数抛出异常时才会被调用.

Not very surprisingly, this function is only called when an exception is thrown from a thread's run function.

所以对于 python <= 3.7

So for python <= 3.7

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 的上下文管理器会检查其他线程中抛出的任何异常，并适当地引发异常.

You still need to join yourself, but upon exit, the with-statement's context manager will check for any exception thrown in other threads, and will raise an exception appropriately.

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

THE CODE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED

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

This is an undocumented, sort-of-horrible hack. I tested it on linux and windows, and it seems to work. Use it at your own risk.

这篇关于使 Python 单元测试因任何线程的异常而失败的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持html5模板网！