如何为Django并行测试的每个Worker生成独立日志文件？

问题原因 #

你当前的配置中，setup_test_environment方法是在主进程中执行的，并行测试的Worker进程通过fork继承了主进程已初始化的日志配置（包括已打开的日志文件句柄），因此所有Worker都会写入同一个主进程创建的日志文件，导致日志内容交错。

解决方案 #

需要让每个Worker进程在启动时重新初始化日志配置，而非继承主进程的配置。可以通过修改自定义测试Runner，在创建Worker池时指定初始化函数，让每个Worker启动后立即配置独立的日志文件。

步骤1：调整日志配置函数 #

更新configure_logging，确保Worker进程能清除继承的日志Handler，创建自己的独立日志文件：

import logging.config
import os

def configure_logging() -> None:
    """为每个进程生成独立日志文件的自定义配置"""
    process_id = os.getpid()
    log_file_path = f"runner-{process_id}.log"

    # 清除主进程继承的所有日志Handler，避免复用同一文件句柄
    root_logger = logging.getLogger()
    for handler in list(root_logger.handlers):
        root_logger.removeHandler(handler)
        handler.close()

    LOGGING = {
        "version": 1,
        "disable_existing_loggers": False,
        "handlers": {
            "file": {
                "level": "DEBUG",
                "class": "logging.FileHandler",
                "filename": log_file_path,
                "encoding": "utf-8"
            },
        },
        # 配置根日志器，确保所有日志都写入独立文件
        "root": {
            "handlers": ["file"],
            "level": "DEBUG",
        },
        # 保留你原有的django.log日志器配置（如果需要）
        "loggers": {
            "django.log": {
                "handlers": ["file"],
                "level": "DEBUG",
                "propagate": True,
            },
        },
    }

    logging.config.dictConfig(LOGGING)

步骤2：修改自定义测试Runner #

重写run_tests方法，为Worker池添加初始化函数，让每个Worker启动时执行日志配置：

from django.test.runner import DiscoverRunner
from multiprocessing import Pool
from myapp.tests.logging_config import configure_logging

class ParallelTestRunner(DiscoverRunner):
    """为每个并行测试Worker生成独立日志的自定义Runner"""
    def run_tests(self, test_labels, extra_tests=None, **kwargs):
        # 仅在并行模式下生效
        if self.parallel > 1:
            # 保存Pool原有的默认初始化函数
            original_initializer = Pool.__init__.__defaults__[0]

            def worker_initializer(*args):
                # 先执行Django原有的Worker初始化逻辑
                if original_initializer is not None:
                    original_initializer(*args)
                # 配置当前Worker的独立日志
                configure_logging()

            # 临时替换Pool的默认初始化函数
            Pool.__init__.__defaults__ = (worker_initializer,) + Pool.__init__.__defaults__[1:]

        try:
            # 执行测试
            return super().run_tests(test_labels, extra_tests, **kwargs)
        finally:
            # 恢复Pool原有的初始化函数，避免影响其他场景的使用
            if self.parallel > 1:
                Pool.__init__.__defaults__ = (original_initializer,) + Pool.__init__.__defaults__[1:]

步骤3：保持settings.py配置不变 #

继续使用你原有的配置：

TEST_RUNNER = "myapp.tests.runners.ParallelTestRunner"

验证效果 #

运行manage.py test --parallel后，你会看到每个Worker进程对应生成一个runner-<进程ID>.log文件，日志内容不再交错，每个Worker的日志独立存储。

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