如何为基于QueueHandler的多进程logging模块编写pytest单元测试

由于capsys、caplog仅捕获主进程的标准输出/日志记录，而你的haplog通过QueueHandler将子进程日志异步传递到主进程处理，需要针对性改造测试逻辑来捕获跨进程日志。以下是三种实用方案：

方案一：自定义内存型QueueListener捕获日志 #

核心思路是临时替换haplog的QueueListener，将日志记录存储到内存列表中，测试时直接断言列表内容。

1. 实现自定义MemoryQueueListener #

import threading
import logging
from logging.handlers import QueueListener

class MemoryQueueListener(QueueListener):
    def __init__(self, queue):
        self.log_records = []
        self._lock = threading.Lock()  # 保证线程安全
        # 用空Handler避免额外输出，仅收集日志记录
        super().__init__(queue, logging.NullHandler())
    
    def handle(self, record):
        with self._lock:
            self.log_records.append(record)

2. 编写pytest Fixture做环境替换 #

import pytest
import haplog  # 导入你的haplog模块

@pytest.fixture
def captured_multi_logs(monkeypatch):
    # 停掉原Listener并替换为自定义内存Listener
    original_listener = haplog.listener
    original_listener.stop()

    memory_listener = MemoryQueueListener(haplog.log_queue)
    memory_listener.start()
    monkeypatch.setattr(haplog, 'listener', memory_listener)

    yield memory_listener.log_records

    # 测试后恢复原环境
    memory_listener.stop()
    original_listener.start()
    monkeypatch.setattr(haplog, 'listener', original_listener)

3. 编写测试用例 #

from multiprocessing import Process
from haplog import get_logger

def test_child_process_log(captured_multi_logs):
    # 定义子进程任务
    def child_task():
        logger = get_logger("child_process")
        logger.info("子进程输出的INFO日志")
        logger.error("子进程输出的ERROR日志")
    
    # 启动并等待子进程结束
    p = Process(target=child_task)
    p.start()
    p.join()

    # 断言日志内容
    info_records = [r for r in captured_multi_logs if r.levelname == "INFO"]
    assert len(info_records) == 1
    assert info_records[0].message == "子进程输出的INFO日志"

    error_records = [r for r in captured_multi_logs if r.levelname == "ERROR"]
    assert len(error_records) == 1
    assert error_records[0].message == "子进程输出的ERROR日志"

方案二：临时输出到临时文件断言 #

如果不想修改haplog内部逻辑，可让haplog临时输出到pytest提供的临时文件，测试后读取文件内容断言。

测试用例示例 #

import pytest
from multiprocessing import Process
from haplog import get_logger, configure

def test_log_to_temp_file(tmp_path):
    # 配置haplog输出到临时文件
    log_file = tmp_path / "haplog_test.log"
    configure(log_file=str(log_file))  # 假设haplog有configure方法配置输出路径

    # 子进程任务
    def child_task():
        logger = get_logger("child")
        logger.warning("子进程警告日志")
    
    p = Process(target=child_task)
    p.start()
    p.join()

    # 读取并断言日志内容
    log_content = log_file.read_text()
    assert "WARNING" in log_content
    assert "子进程警告日志" in log_content

方案三：让caplog捕获主进程处理的日志 #

如果haplog的主进程Handler是基于日志器的，可将caplog的Handler添加到haplog的根日志器中，直接捕获主进程处理后的日志记录。

测试用例示例 #

from multiprocessing import Process
from haplog import get_logger

def test_caplog_capture(caplog):
    # 将caplog的Handler添加到haplog的根日志器
    root_logger = get_logger()
    root_logger.addHandler(caplog.handler)

    def child_task():
        logger = get_logger("child")
        logger.debug("子进程调试日志")
    
    p = Process(target=child_task)
    p.start()
    p.join()

    # 直接用caplog断言
    assert "子进程调试日志" in caplog.text
    assert any(record.levelname == "DEBUG" for record in caplog.records)

注意事项 #

• 确保子进程完全结束后再断言，避免日志尚未通过队列传递到主进程。
• 如果haplog未暴露queue和listener对象，需要在haplog中添加对应的获取接口（如get_queue()、get_listener()），方便测试时替换。
• 多线程场景下，内存收集日志时必须加锁，避免并发写入导致的列表数据异常。

内容的提问来源于stack exchange，提问作者張治猷