如何在py.test环境中实现非阻塞sleep？

不需要修改pytest源码，以下几种简单方案可以解决你的阻塞问题：

方案1：线程+事件标记替代阻塞sleep #

核心思路是把硬件测试序列放到后台线程执行，用threading.Event标记测试完成，主线程通过非阻塞的方式等待，避免长时间time.sleep卡住整个进程。

修改fixture和测试函数的示例：

import pytest
import threading
from time import sleep
from hardware_library import hardware_fixture

@pytest.fixture
def hardware_thread_fixture():
    test_done = threading.Event()
    hardware = hardware_fixture()

    def run_test():
        hardware.start_the_testing_sequence()
        # 这里是硬件测试的实际执行逻辑，完成后标记事件
        # 如果硬件库支持回调，可在回调里直接调用test_done.set()
        sleep(120)
        test_done.set()

    # 启动后台守护线程
    test_thread = threading.Thread(target=run_test)
    test_thread.daemon = True
    test_thread.start()
    
    yield hardware, test_done
    
    # 测试结束后的清理逻辑
    if not test_done.is_set():
        hardware.stop_testing()

def hardware_test(hardware_thread_fixture):
    hardware, test_done = hardware_thread_fixture
    # 非阻塞等待：每次等待1秒，期间可做状态检查
    while not test_done.wait(timeout=1):
        print("硬件测试执行中...")
        # 可选：实时读取硬件状态、打印日志等
    # 验证测试结果
    assert hardware.get_test_result() == "pass"

方案2：用线程池管理异步任务 #

借助concurrent.futures.ThreadPoolExecutor可以更简洁地管理后台线程，避免手动维护线程生命周期：

import pytest
from concurrent.futures import ThreadPoolExecutor
from time import sleep
from hardware_library import hardware_fixture

@pytest.fixture
def hardware_executor():
    executor = ThreadPoolExecutor(max_workers=1)
    yield executor
    executor.shutdown(wait=False)

def hardware_test(hardware_executor):
    hardware = hardware_fixture()
    # 提交硬件测试任务到线程池
    test_future = hardware_executor.submit(hardware.start_the_testing_sequence)
    
    # 非阻塞等待任务完成
    while not test_future.done():
        print("测试进行中...")
        sleep(1)
    
    # 获取并验证测试结果
    test_result = test_future.result()
    assert test_result == "success"

方案3：异步测试（适配异步硬件库） #

如果你的硬件库提供异步API，配合pytest-asyncio插件可以实现完全非阻塞的测试流程：

1. 先安装插件：pip install pytest-asyncio
2. 编写异步测试代码：

import pytest
import asyncio
from hardware_library import async_hardware_fixture  # 假设硬件库有异步版本

@pytest.fixture
async def async_hardware():
    hardware = async_hardware_fixture()
    await hardware.initialize()
    yield hardware
    await hardware.shutdown()

@pytest.mark.asyncio
async def test_hardware_async(async_hardware):
    # 创建异步测试任务
    test_task = asyncio.create_task(async_hardware.start_the_testing_sequence())
    
    # 非阻塞等待，期间可执行其他异步操作
    while not test_task.done():
        print("异步测试执行中...")
        await asyncio.sleep(1)
    
    test_result = await test_task
    assert test_result == "pass"

关键注意点 #

• time.sleep()会阻塞当前线程的所有操作，主线程调用时会导致整个测试进程卡住，必须用非阻塞等待方式替代。
• 如果硬件测试序列支持回调触发完成信号，优先用回调设置事件/标记，不要依赖固定时长的sleep，这样测试逻辑更准确。

内容的提问来源于stack exchange，提问作者kehrazy