Pytest-xdist：如何知晓各Worker将运行的测试用例？

当用request.session.items在xdist多worker模式下获取测试用例时，每个worker拿到的都是全量测试用例，这是因为xdist的测试收集由master进程统一完成，之后master会把所有测试用例的元数据同步给每个worker，但实际执行时worker只会从master的任务队列中获取分配给自己的任务，本地的session.items并不代表当前worker要执行的内容。

要在首个测试执行前获取每个worker的任务列表，需要从xdist的任务调度机制入手，以下是可行的解决方案：

方法：自定义调度器+worker传递任务列表 #

1. 自定义xdist调度器，预分配任务 #

在conftest.py中实现自定义调度器，继承xdist默认的LoadScheduling，在首次分配时将测试用例平均（或按负载）分配给各个worker，并记录每个worker的任务列表：

from pytest_xdist.scheduler import LoadScheduling
from pytest_xdist.remote import WorkerController

class CustomLoadScheduler(LoadScheduling):
    def __init__(self, config, log):
        super().__init__(config, log)
        self.worker_tasks = {}  # 存储每个worker的任务ID列表

    def distribute(self, node):
        # 仅在首次分配时生成任务列表
        if not self.worker_tasks:
            test_ids = [item.nodeid for item in self.config.session.items]
            worker_ids = list(self.nodes.keys())
            num_workers = len(worker_ids)
            
            # 平均分配测试用例到各个worker
            for idx, worker_id in enumerate(worker_ids):
                self.worker_tasks[worker_id] = test_ids[idx::num_workers]
        
        # 将当前worker的任务加入待执行队列
        if node.id in self.worker_tasks:
            for task_id in self.worker_tasks[node.id]:
                if task_id in self.collection:
                    self.collection.remove(task_id)
                    self.pending.append((node, task_id))
        # 调用父类的分发逻辑
        super().distribute(node)

# 替换默认调度器
def pytest_xdist_make_scheduler(config, log):
    return CustomLoadScheduler(config, log)

2. Master端将任务列表传递给Worker #

在master端通过pytest_configure_node钩子，把每个worker的任务列表写入workerinput（master与worker之间的通信载体）：

def pytest_configure_node(node: WorkerController):
    # 检查调度器是否有预分配的任务列表
    if hasattr(node.sched, 'worker_tasks'):
        node.workerinput['assigned_tasks'] = node.sched.worker_tasks.get(node.id, [])

3. Worker端获取并输出任务列表 #

在worker端的会话启动钩子pytest_sessionstart中，从workerinput中读取当前worker的任务列表，这会在首个测试执行前触发：

def pytest_sessionstart(session):
    worker_id = session.config.workerinput.get('workerid')
    assigned_tasks = session.config.workerinput.get('assigned_tasks', [])
    
    if worker_id and assigned_tasks:
        print(f"\n=== Worker {worker_id} 即将执行以下测试 ===")
        for task in assigned_tasks:
            print(f"- {task}")

注意事项 #

• 该方案依赖xdist的调度器API，不同版本的xdist可能存在API差异，若出现兼容性问题，可根据xdist的源码调整调度器逻辑。
• 若需要更复杂的任务分配策略（如按模块、按标记分组），可修改CustomLoadScheduler中的任务分配逻辑，比如根据测试用例的标记或模块路径进行分组。

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