asyncio加速pytest异步测试速度无提升的优化方案咨询

你当前的异步测试完全没有发挥并发能力，核心问题是在异步函数中使用了同步阻塞的requests库：
requests所有请求都是阻塞式IO，哪怕包裹在async def定义的函数中、加上await调用，也不会让出事件循环执行权，每次请求都会堵死整个异步调度流程，所以你的循环实际还是串行执行100次请求，耗时自然和串行一致。

方案1：替换为异步HTTP库 + 并发调度（推荐，性能最好） #

使用异步HTTP库aiohttp替代同步的requests，再通过asyncio.gather一次性并发发起所有请求，耗时会接近单次请求的耗时：

import asyncio
import aiohttp
import pytest

@pytest.mark.asyncio
async def test_async():
    # 复用aiohttp会话，减少连接开销
    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session:
        # 一次性创建所有请求任务
        tasks = [get_files_async(session, file_id) for file_id in range(100)]
        # 并发等待所有任务完成
        res = await asyncio.gather(*tasks)
    assert res == [200 for _ in range(100)]


async def get_files_async(session: aiohttp.ClientSession, file_id):
    # cert参数可以通过ClientSession的参数配置，或者直接传到get方法
    async with session.get(url, cert=(cert, key)) as resp:
        return resp.status

方案2：线程池并发运行同步请求（不改HTTP库的兼容方案） #

如果你不想更换requests库，可以用线程池把同步请求放到后台线程运行，避免阻塞事件循环，同样能实现并发效果：

import asyncio
import requests
import pytest

@pytest.mark.asyncio
async def test_async():
    # 创建所有请求任务，用to_thread把同步requests调用放到独立线程运行
    tasks = [asyncio.to_thread(get_files_sync, file_id) for file_id in range(100)]
    res = await asyncio.gather(*tasks)
    assert res == [200 for _ in range(100)]

# 不需要async修饰，就是原有的同步请求逻辑
def get_files_sync(file_id):
    kwargs = {"verify": False, "cert": (cert, key)}
    resp = requests.request('GET', url, **kwargs)
    return resp.status_code

方案3：单元测试场景下Mock接口（最快，适合不需要真实调用的场景） #

如果这个测试是单元测试，不需要真实调用远端接口，可以直接Mock掉请求逻辑的返回值，完全跳过IO等待，耗时可以降到毫秒级。

内容的提问来源于stack exchange，提问作者Jeffrey S.