如何实现pytest延迟执行断言不阻塞测试且可更新测试结果

该需求完全可实现，不需要替换现有测试栈，基于pytest原生钩子能力搭配轻量任务调度逻辑即可落地，无需引入Celery等重中间件。

• 拆分测试用例为两个执行阶段：第一阶段运行Selenium E2E流程+前2个立即执行的断言，同时记录需要延迟校验的参数（数据ID、延迟时长、对应测试用例唯一标识）
• 用pytest内置钩子接管测试结果生成逻辑，第一阶段跑完无报错时，标记用例为「待校验」状态，不直接输出最终通过/失败结果
• 用内存队列+独立线程处理所有延迟校验任务，调度完一个测试的延迟任务后立刻启动下一个测试，全程不阻塞测试执行流程
• 所有延迟断言执行完成后，统一更新对应测试用例的结果，输出完整测试报告

该方案的核心逻辑是利用pytest的异步执行能力，把需要延迟的测试步骤标记为异步任务，pytest会挂起这个测试用例的结果，直到异步任务执行完成后再输出最终结果。具体实现是用pytest-asyncio插件，把需要延迟的断言写成async函数，用asyncio.sleep代替time.sleep，这样不会阻塞整个测试进程，其他测试用例可以同时运行，等待所有异步任务都完成后再统一生成测试报告，和上面的实现思路本质一致。

1. 全局延迟任务存储与fixture定义 #

import pytest
from datetime import datetime, timedelta
import threading
import time
import mysql.connector

# 全局存储延迟校验任务、测试临时结果
delayed_tasks = []
test_final_results = {}

@pytest.fixture(scope="function")
def register_delay_assert():
    """用于在测试用例中注册延迟断言的fixture"""
    def wrapper(delay_minutes, test_case_id, assert_func):
        # 计算断言的实际执行时间
        run_at = datetime.now() + timedelta(minutes=delay_minutes)
        delayed_tasks.append({
            "run_time": run_at,
            "test_id": test_case_id,
            "assert_logic": assert_func
        })
    return wrapper

2. 测试用例写法 #

def test_e2e_order_submit(register_delay_assert):
    # 第一部分：Selenium E2E页面操作
    driver.get("https://你的被测站点/下单页")
    # 省略点击、填写表单等操作逻辑
    order_id = driver.find_element(value="order_id").text # 拿到需要后续校验的业务ID

    # 立即执行的前2个断言
    assert driver.title == "下单成功"
    assert len(order_id) == 12, "订单ID格式错误"

    # 注册20分钟后执行的延迟断言
    register_delay_assert(
        delay_minutes=20,
        test_case_id="test_e2e_order_submit",
        assert_func=lambda: check_order_status(order_id)
    )

def check_order_status(order_id):
    """延迟执行的数据库校验逻辑"""
    conn = mysql.connector.connect(host="数据库地址", user="账号", password="密码", database="库名")
    cursor = conn.cursor()
    cursor.execute("SELECT status FROM order_table WHERE order_id = %s", (order_id,))
    status = cursor.fetchone()[0]
    assert status == "已完成", f"订单{order_id}20分钟后状态异常，实际为{status}"

3. pytest钩子配置（写在conftest.py中） #

def pytest_runtest_makereport(item, call):
    """接管测试结果生成逻辑，第一阶段跑完先标记为待校验"""
    if call.when == "call" and not call.excinfo:
        test_final_results[item.name] = "pending"
        # 暂时不返回最终结果
        return
    # 延迟断言执行完成后，更新测试用例最终结果
    if test_final_results.get(item.name) in ["passed", "failed"]:
        report = super().pytest_runtest_makereport(item, call)
        report.outcome = test_final_results[item.name]
        return report

def pytest_sessionfinish(session, exitstatus):
    """所有测试用例第一阶段跑完后，启动独立线程执行延迟任务"""
    def run_delay_tasks():
        while delayed_tasks:
            current_time = datetime.now()
            for task in delayed_tasks[:]:
                if current_time >= task["run_time"]:
                    try:
                        task["assert_logic"]()
                        test_final_results[task["test_id"]] = "passed"
                    except AssertionError as e:
                        test_final_results[task["test_id"]] = "failed"
                        session.config.test_errors[task["test_id"]] = str(e)
                    delayed_tasks.remove(task)
            time.sleep(60) # 每分钟轮询一次即可，不需要过高频率
        # 所有延迟任务跑完后，生成最终的html/控制台测试报告
        generate_final_test_report(test_final_results)
    # 启动非守护线程，等待所有延迟任务执行完再结束整个pytest进程
    threading.Thread(target=run_delay_tasks, daemon=False).start()

• 本地执行时直接运行pytest命令即可，会自动等待所有延迟任务执行完成后输出最终结果
• 接入CI流水线时，可把延迟任务、临时结果存储换成Redis，避免流水线进程异常中断丢失数据，同时设置流水线超时时间大于最大延迟时长即可
• 测试用例量级大的场景，可把延迟校验单独拆成一个CI Job，前序E2E Job跑完后触发延迟校验Job，不占用前序E2E执行的机器资源

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