使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('\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结节具有未记录的导入，可以执行以下操作:

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

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单元测试因任何线程的异常而失败

相关推荐