如何在pytest-xdist测试时为每行stdout添加xdist网关编号？

当然可以！要让每条日志都带上对应的xdist worker编号，你只需要通过pytest的配置钩子结合logging的过滤器来实现全局的日志上下文注入，具体步骤如下：

1. 在conftest.py中添加配置代码 #

在你的测试项目根目录下创建（或修改）conftest.py，加入以下代码：

import logging
import pytest
from pytest_xdist import get_xdist_worker_id

def pytest_configure(config):
    # 自定义日志过滤器，给每条日志记录注入worker_id属性
    class WorkerIDFilter(logging.Filter):
        def filter(self, record):
            # 获取当前worker编号，主进程返回"main"
            record.worker_id = get_xdist_worker_id(config) or "main"
            return True

    # 给所有日志处理器添加这个过滤器
    root_logger = logging.getLogger()
    for handler in root_logger.handlers:
        handler.addFilter(WorkerIDFilter())

    # 修改日志格式，把worker_id加入到输出中
    for handler in root_logger.handlers:
        if isinstance(handler, logging.StreamHandler):
            # 这里可以根据你原来的日志格式调整，只需要加入%(worker_id)s即可
            original_format = handler.formatter._fmt if handler.formatter else "%(levelname)s: %(message)s"
            new_format = f"[%(worker_id)s] {original_format}"
            handler.setFormatter(logging.Formatter(new_format))

2. 效果说明 #

• 当你用pytest -n auto（或指定worker数量，比如-n 2）运行测试时，每条stdout日志都会自动带上对应的worker编号，比如：

  [gw0] INFO: 处理用户数据
  [gw1] DEBUG: 连接数据库成功

• 如果是主进程产生的日志（比如初始化代码），会显示[main]标识。

自定义格式提示 #

如果你已经有自己的日志格式字符串，只需要在其中插入%(worker_id)s占位符即可，不用完全替换格式。比如原来的格式是%(asctime)s %(levelname)s %(message)s，修改后可以是%(asctime)s [%(worker_id)s] %(levelname)s %(message)s。

这个方案的好处是全局生效，不需要修改任何测试用例里的日志调用代码，所有通过logging模块输出的日志都会自动带上worker标识，完美解决你分不清日志归属的问题！

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