pytest-asyncio异步测试未并发执行 运行耗时远超预期问题

我正在尝试实现一套简易的异步测试套件。按照我对异步机制的理解，下述测试用例的总运行时长应仅为2秒左右，但实际运行总耗时达到6秒。需要明确缺少的配置或写法调整，让测试真正实现异步（“同时”）运行。

复现代码 #

import logging
import pytest
import asyncio

MSG_FORMAT = "%(asctime)s.%(msecs)03d %(module)s->%(funcName)-15s |%(levelname)s| %(message)s"
MSG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
LOG_LEVEL = logging.INFO
# Create logger
logger = logging.getLogger(__name__)
logger.setLevel(LOG_LEVEL)
# Create Stream Handler
log_stream = logging.StreamHandler()
log_format = logging.Formatter(fmt=MSG_FORMAT, datefmt=MSG_DATE_FORMAT)
log_stream.setFormatter(log_format)
log_stream.setLevel(LOG_LEVEL)
logger.addHandler(log_stream)

class TestMyStuff:

    @staticmethod
    async def foo(seconds):
        await asyncio.sleep(seconds)
        return 1

    @pytest.mark.asyncio
    async def test_1(self, event_loop):
        logger.info("start")
        assert await event_loop.create_task(self.foo(2)) == 1
        logger.info("end")

    @pytest.mark.asyncio
    async def test_2(self, event_loop):
        logger.info("start")
        assert await event_loop.create_task(self.foo(2)) == 1
        logger.info("end")

    @pytest.mark.asyncio
    async def test_3(self, event_loop):
        logger.info("start")
        # assert await event_loop.run_in_executor(None, self.foo) == 1
        assert await event_loop.create_task(self.foo(2)) == 1
        logger.info("end")

环境信息 #

• pytest插件版本：asyncio-0.18.3、aiohttp-1.0.4
• 运行效果截图：

总耗时达到6秒的核心逻辑：

• pytest-asyncio默认给每个异步测试用例分配函数级独立事件循环，测试用例默认串行执行：跑完test_1全部逻辑才会启动test_2，test_2执行完才会启动test_3。单个测试内调用create_task仅能实现当前测试内部的协程并发，无法跨测试用例实现并发。
• 每个测试内部都会同步等待一个耗时2秒的asyncio.sleep，3个测试串行累加总耗时就是2*3=6秒，和预期的2秒并发耗时存在差距。

根据实际需求选择对应方案即可：

方案1：单测试内实现协程并发（推荐，无额外依赖） #

如果目标是验证异步逻辑的并发执行效果，不需要把三个等待逻辑拆成三个独立测试函数，直接在同一个测试内创建多个协程任务，通过asyncio.gather批量等待执行完成即可，示例代码：

@pytest.mark.asyncio
async def test_concurrent_foo(self):
    logger.info("all tasks start")
    # 同时创建3个耗时2秒的协程任务
    task1 = asyncio.create_task(self.foo(2))
    task2 = asyncio.create_task(self.foo(2))
    task3 = asyncio.create_task(self.foo(2))
    # 等待所有任务执行完毕
    res1, res2, res3 = await asyncio.gather(task1, task2, task3)
    assert res1 == res2 == res3 == 1
    logger.info("all tasks end")

该写法是单事件循环内协程并发的标准实现，总耗时稳定在2秒左右。另外代码里的event_loop.create_task可以直接替换为asyncio.create_task，无需显式传入事件循环参数，效果完全一致。

方案2：保留独立测试用例实现并发 #

如果必须保留三个独立的测试用例结构，同时要求并发执行，可通过pytest-xdist实现多worker并行：

1. 安装插件：

   pip install pytest-xdist
   

2. 执行测试时指定并发worker数量：

   pytest -n 3
   

该方式通过多进程实现并行，总耗时约2秒，但运行机制和单线程异步协程不同。注意你当前使用的0.18.3版本pytest-asyncio本身不支持单事件循环内跨测试用例调度，升级高版本后配置会话级事件循环+异步调度插件的方案成本较高，非特殊需求不推荐。

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