pytest的内置插件盘点32. threadexception | 对接python和pytest底层收集子线程异常

本文系《pytest源码剖析》系列内容

32. threadexception

插件路径：_pytest.threadexception

实现的 hook

调用的 hook

无

定义的 fixture

无

插件功能

创建上下文管理器catch_threading_exception

并在在以下 hook 包装器中使用：

pytest_runtest_setup

pytest_runtest_call

pytest_runtest_teardown

代码片段

def thread_exception_runtest_hook() -> Generator[None, None, None]: with catch_threading_exception() as cm: yield if cm.args: thread_name = "" if cm.args.thread is None else cm.args.thread.name msg = f"Exception in thread {thread_name}\n\n" msg += "".join( traceback.format_exception( cm.args.exc_type, cm.args.exc_value, cm.args.exc_traceback, ) ) warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))

通过上下文管理器的方式，收集例运行时的子进程异常，并发出警告

简评

threading.excepthook是 python 3.8 新增的一个钩子，在子线程遇到“ 未捕获的异常 ”时自动调用。

通过设置该 hook，便可以自助处理子线程的异常。

本插件正是让 pytest 设置了这个 python 的钩子，在遇到这些情况时发出警告

...

假设有下面这样的代码

import threading

def f(): assert False # 会引发异常的代码

t = threading.Thread(target=f) # 创建子线程，执行f函数t.start() # 启动子线程

子线程启动后，会引发异常。

在默认情况下，子线程会搜集收集异常信息，并在标准错误中进行输出

Exception in thread Thread-1 (f):Traceback (most recent call last): File "C:\Users\admin\AppData\Local\Programs\Python\Python312\Lib\threading.py", line 1052, in _bootstrap_inner self.run() File "C:\Users\admin\AppData\Local\Programs\Python\Python312\Lib\threading.py", line 989, in run self._target(*self._args, **self._kwargs) File "D:\pytest_7.2.x\aaa.py", line 5, in f assert False # 会引发异常的代码 ^^^^^^^^^^^^AssertionError程序结束Process finished with exit code 0

这会影响主线程的输出

实现这个 hook，将子线程的异常信息暂存起来

exc_list = []

def sanmu_funcs(exc): exc_list.append(exc)

threading.excepthook = sanmu_funcs

并在需要时重新处理

print('程序结束，收集到以下异常')for exc in exc_list: print(exc)

...

本插件通过上下文管理器的方式，分别在用例执行的三个阶段进行处理：

setup

call

teardown

如果发现了“子线程中未捕获的异常 ”，在测试结束后统一发出警告