如何仅为pytest-asyncio异步测试参数化不同任务工厂？

核心问题分析 #

你之前的思路方向正确，但fixture未生效大概率是因为没有正确将fixture依赖注入到测试函数的依赖列表中，或是fixture的定义/参数化逻辑存在疏漏。

分步解决方案 #

1. 定义参数化的task factory fixture #

在conftest.py中先实现参数化的fixture，返回两种不同的task工厂：

import pytest
import asyncio

@pytest.fixture(params=["default", "custom"])
def task_factory(request):
    if request.param == "default":
        # 默认的asyncio task创建工厂
        return asyncio.create_task
    else:
        # 自定义task工厂示例（可根据需求修改逻辑）
        def custom_task_creator(coro):
            task = asyncio.create_task(coro)
            task.set_name("custom_test_task")
            return task
        return custom_task_creator

2. 动态为异步测试绑定fixture依赖 #

在conftest.py中添加pytest_collection_modifyitems钩子，自动识别所有异步测试函数，并将task_factory fixture注入到它们的依赖列表：

import inspect

def pytest_collection_modifyitems(items):
    for item in items:
        # 适配pytest-asyncio的auto模式，判断是否为异步测试函数
        if inspect.iscoroutinefunction(item.function):
            # 避免重复绑定fixture
            if "task_factory" not in item.fixturenames:
                item.fixturenames.append("task_factory")

3. 编写异步测试（无需手动标记） #

异步测试函数直接接收task_factory参数即可，pytest会自动为每个参数化的task工厂运行一次测试：

async def test_async_api_call(task_factory):
    async def mock_api():
        return {"code": 200}
    
    task = task_factory(mock_api())
    result = await task
    assert result["code"] == 200

async def test_async_data_process(task_factory):
    async def process_data():
        return sorted([3,1,2])
    
    task = task_factory(process_data())
    assert await task == [1,2,3]

4. 同步测试不受影响 #

同步测试不会被钩子识别，因此不会绑定task_factory fixture，只会执行一次：

def test_sync_calculation():
    assert 5 * 4 == 20

def test_sync_string_check():
    assert "test".capitalize() == "Test"

关键注意事项 #

• 确保inspect.iscoroutinefunction判断逻辑准确：pytest-asyncio的auto模式下，async函数会被直接识别为测试，无需额外标记，这个判断能精准筛选异步测试。
• 避免重复绑定fixture：钩子中加入存在性判断，防止因重复添加依赖导致的异常。
• 自定义task工厂可按需扩展：比如添加超时控制、日志埋点等逻辑，只需修改fixture中对应分支的实现即可。

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