从任何线程的异常使Python unittest失败

Question

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

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

是时候了.我是否可以在不修改x.ExceptionRaiser类的情况下修改my_thread中引发的异常,从而修改下面的测试脚本？

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

\n

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

\n

从 Python 3.8 开始，您可以重写一个函数来实现此功能：threading.excepthook

\n

\n

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

\n

\n

那么我们该怎么办？用我们的逻辑替换这个函数，然后voil\xc3\xa0：

\n

对于 python >= 3.8

\n

import traceback\nimport threading \nimport os\n\n\nclass GlobalExceptionWatcher(object):\n    def _store_excepthook(self, args):\n        \'\'\'\n        Uses as an exception handlers which stores any uncaught exceptions.\n        \'\'\'\n        self.__org_hook(args)\n        formated_exc = traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback)\n        self._exceptions.append(\'\\n\'.join(formated_exc))\n        return formated_exc\n\n    def __enter__(self):\n        \'\'\'\n        Register us to the hook.\n        \'\'\'\n        self._exceptions = []\n        self.__org_hook = threading.excepthook\n        threading.excepthook = self._store_excepthook\n\n    def __exit__(self, type, value, traceback):\n        \'\'\'\n        Remove us from the hook, assure no exception were thrown.\n        \'\'\'\n        threading.excepthook = self.__org_hook\n        if len(self._exceptions) != 0:\n            tracebacks = os.linesep.join(self._exceptions)\n            raise Exception(f\'Exceptions in other threads: {tracebacks}\')\n

\n

对于较旧版本的 Python，这有点复杂。\n长话短说，该threading结节似乎有一个未记录的导入，它执行以下操作：

\n

threading._format_exc = traceback.format_exc\n

\n

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

\n

所以对于 python <= 3.7

\n

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

\n

用法：

\n

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

\n

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

\n\n

\n

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

\n

\n

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

\n