如何使用py.test创建thread fixture并实现测试结束后自动终止的后台线程

我明白你想要实现的效果——在测试中启动持续运行的后台线程，当测试结束时自动让这些线程停下来，避免它们一直占用资源。你已经尝试了用KillableThread的fixture，但也清楚直接“杀死”线程不是最佳实践，同时又不想手动处理threading.Event。下面我给你两种方案，分别对应你的需求和更优雅的最佳实践：

方案一：优化你的KillableThread Fixture（满足你不想用Event的需求） #

首先，我们补全KillableThread的完整实现（用ctypes实现线程终止逻辑），再封装成pytest fixture，确保测试结束后自动清理存活的线程：

import threading
import ctypes
import pytest

class KillableThread(threading.Thread):
    def kill(self):
        if not self.is_alive():
            return
        # 通过系统API强制终止线程
        thread_id = self.ident
        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, ctypes.py_object(SystemExit))
        if res > 1:
            # 清理多余的异常标识
            ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
            raise RuntimeError("线程终止失败")

@pytest.fixture
def threads():
    active_threads = []
    
    def run_background(target):
        thread = KillableThread(target=target)
        thread.start()
        active_threads.append(thread)
    
    yield run_background
    
    # 测试结束后终止所有存活的后台线程
    for thread in active_threads:
        if thread.is_alive():
            thread.kill()

你的测试代码可以直接按预期使用：

def test_with_background_thread(threads):
    def background_thread():
        while True:
            print("running in background")
    
    threads.run_background(background_thread)
    
    # 这里写你的测试逻辑
    assert True  # 替换成实际断言
    
    # 测试结束后，fixture会自动终止后台线程

⚠️ 注意：这种强制终止线程的方式存在风险——如果线程正在持有锁、打开文件或操作其他资源，强行终止可能导致资源泄漏或程序状态异常，仅推荐在测试场景下临时使用。

方案二：用threading.Event实现优雅退出（推荐） #

虽然你提到不想用threading.Event，但我们可以把它封装在fixture内部，让测试代码完全不用关心Event的细节，同时实现线程的优雅退出：

import threading
import pytest

@pytest.fixture
def threads():
    active_threads = []
    stop_events = []
    
    def run_background(target):
        stop_event = threading.Event()
        # 包装目标函数，让它在stop_event触发时退出循环
        def wrapped_target():
            while not stop_event.is_set():
                target()
        thread = threading.Thread(target=wrapped_target)
        thread.start()
        active_threads.append(thread)
        stop_events.append(stop_event)
    
    yield run_background
    
    # 测试结束后，触发所有终止信号并等待线程退出
    for event in stop_events:
        event.set()
    for thread in active_threads:
        thread.join()

测试代码的用法和之前完全一致，不需要任何额外修改：

def test_with_background_thread(threads):
    def background_thread():
        print("running in background")
    
    threads.run_background(background_thread)
    
    # 执行你的测试逻辑
    assert True
    
    # 测试结束后，fixture会自动触发终止信号，等待线程优雅退出

这种方案的优势在于线程是优雅退出的，不会有资源泄漏的风险，而且测试代码依然保持简洁——所有Event的处理逻辑都被fixture封装起来了，完全符合你的使用需求。

备注：内容来源于stack exchange，提问作者kehrazy