如何用pytest-xdist让单Worker运行一组参数化Fixture及对应测试（不拆分文件）

针对你遇到的session级参数化Fixture在xdist并行时重复执行、官方锁方案阻塞的问题，可以通过在测试收集阶段按Fixture参数给Worker分配任务的方式解决，每个Worker只负责特定Fixture参数对应的所有测试用例，确保Fixture只执行一次，同时不阻塞其他Worker。

实现步骤 #

在项目根目录创建conftest.py，添加以下代码：

import pytest

def pytest_collection_modifyitems(config, items):
    # 非xdist模式下不做处理
    if not hasattr(config, 'workerinput'):
        return
    
    # 获取当前Worker的编号（从workerid提取，比如'gw0'对应0）
    worker_id = config.workerinput['workerid']
    worker_num = int(worker_id[2:])
    total_workers = config.workerinput['num_workers']
    
    # 与原代码中session Fixture的参数列表保持一致
    fixture_params = [1, 2, 3, 4, 5]
    total_params = len(fixture_params)
    
    # 计算当前Worker负责的参数范围，均分参数，处理余数
    params_per_worker = total_params // total_workers
    remainder = total_params % total_workers
    
    start_idx = worker_num * params_per_worker + min(worker_num, remainder)
    end_idx = start_idx + params_per_worker + (1 if worker_num < remainder else 0)
    assigned_params = fixture_params[start_idx:end_idx]
    
    # 筛选当前Worker需要执行的用例：只保留依赖目标Fixture参数的测试
    filtered_items = []
    for item in items:
        # 从测试用例的callspec中提取heavy_computation的参数值
        fixture_param = item.callspec.params.get('heavy_computation')
        if fixture_param in assigned_params:
            filtered_items.append(item)
    
    # 替换为筛选后的用例列表
    items[:] = filtered_items

原理说明 #

1. 钩子介入测试收集：通过pytest_collection_modifyitems钩子，在pytest收集完所有测试用例后，对用例列表进行修改。
2. Worker参数分配：利用xdist提供的config.workerinput获取当前Worker的ID和总Worker数，将Fixture参数均分给各个Worker，确保每个参数只被一个Worker处理。
3. 用例筛选：每个测试用例的callspec中包含了它所依赖的Fixture参数值，通过这个值筛选出当前Worker需要执行的用例，保证同一Fixture参数的所有测试都在同一个Worker中运行。

使用方法 #

运行测试时指定Worker数量（建议等于Fixture参数数量，最大化并行效率）：

pytest -n 5

如果Worker数量少于Fixture参数数量（比如2个），代码会自动将参数均分，每个Worker负责一部分参数的测试，同样能避免Fixture重复执行。

效果验证 #

每个Worker会独立执行分配给自己的Fixture参数及其对应的所有测试用例：

• 比如gw0只会执行heavy_computation参数为1的Fixture，以及对应的A-E五个测试
• 各Worker之间互不干扰，不会重复执行Fixture，也不会出现阻塞

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