如何用asyncio.gather()按依赖分组异步执行测试用例？

原代码存在三个核心问题：

1. 测试用例是同步函数，无法直接用await调用
2. groups中存储的是字符串形式的调用语句，不是可执行的函数对象
3. 多次调用asyncio.run虽然能保证组顺序，但写法冗余且内部逻辑错误（无法正确执行测试用例）

下面分两种场景给出修改方案：

场景1：测试用例可改为异步函数 #

如果允许将测试用例改为异步定义，直接调整如下：

第一步：修改测试用例为异步函数 #

async def test1(driver):
    # 原功能逻辑，可加入异步操作（如await asyncio.sleep()）
    pass

# 其他test2、test3等同理，全部改为async def

第二步：重构执行代码 #

import asyncio
import tests  # 导入你的测试模块

# 存储函数对象，而非字符串调用
groups = [
    [tests.test1, tests.test2, tests.test3],
    [tests.test5, tests.test6, tests.test9],
    [tests.test4, tests.test7, tests.test8],
    [tests.test0, tests.test10]
]

async def main():
    # 初始化你的driver实例
    driver = ... 
    
    # 顺序遍历每个组，组内异步执行所有测试用例
    for group in groups:
        # 为每个测试用例传入driver，生成协程对象后用gather异步执行
        await asyncio.gather(*[test(driver) for test in group])

# 启动主协程
asyncio.run(main())

场景2：测试用例必须保持同步 #

如果测试用例无法改为异步，需要用线程池包装同步函数，避免阻塞事件循环：

第一步：保留同步测试用例不变 #

def test1(driver):
    # 原功能逻辑
    pass

# 其他测试用例保持def定义

第二步：重构执行代码 #

import asyncio
import tests
from concurrent.futures import ThreadPoolExecutor

# 创建线程池，可根据测试用例数量调整max_workers
executor = ThreadPoolExecutor(max_workers=4)

# 存储函数对象
groups = [
    [tests.test1, tests.test2, tests.test3],
    [tests.test5, tests.test6, tests.test9],
    [tests.test4, tests.test7, tests.test8],
    [tests.test0, tests.test10]
]

# 包装同步函数为可await的协程
async def run_sync_test(test_func, driver):
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(executor, test_func, driver)

async def main():
    driver = ...  # 初始化driver
    
    for group in groups:
        # 组内所有同步测试用例包装后异步执行
        await asyncio.gather(*[run_sync_test(test, driver) for test in group])
    
    executor.shutdown()  # 执行完毕后关闭线程池

asyncio.run(main())

关键修改说明 #

• 替换groups中的字符串为函数对象，确保能实际调用测试用例
• 使用单个主协程main统一管理流程，避免多次创建事件循环
• 组内用asyncio.gather实现异步并行，组间通过for循环+await保证顺序执行
• 同步测试用例必须通过run_in_executor包装，否则会阻塞整个事件循环

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