如何Mock含非简单对象的logging.config.dictConfig配置字典？

问题描述 #

测试setup_logging函数时，直接使用assert_called_once_with(expected_conf_dict)会失败，原因是io.StringIO未实现__eq__方法——即使两个StringIO实例功能一致，也会被判定为不相等：

>>> s1=StringIO()
>>> s2=StringIO()
>>> assert s1==s2
AssertionError

下面提供三种可行的解决方案：

方案1：提取实际参数，逐字段针对性校验 #

通过mock对象的call_args获取传入dictConfig的真实配置，对普通字段直接比较相等性，对stream字段仅校验类型是否为StringIO。

修改后的测试代码：

from io import StringIO
from unittest import mock
from application.utils import setup_logging

@mock.patch("logging.config.dictConfig")
def test_setup_logging(mock_logging_config):
    setup_logging()
    
    # 获取实际传入的配置字典
    actual_config = mock_logging_config.call_args[0][0]
    
    # 定义预期配置（剔除stream的具体实例）
    expected_config = {
        "version": 1,
        "disable_existing_loggers": False,
        "incremental": False,
        "formatters": {
            "standard": {
                "format": "[%(threadName)s] %(asctime)s %(levelname)s %(name)s: %(message)s",
                "datefmt": "%Y-%m-%d %H:%M:%S",
            }
        },
        "handlers": {
            "console": {
                "level": "INFO",
                "formatter": "standard",
                "class": "logging.StreamHandler",
            }
        },
        "root": {"handlers": ["console"], "level": "INFO"},
        "loggers": {
            "py4j.java_gateway": {"level": "ERROR"},
            "botocore": {"level": "ERROR"},
            "boto3": {"level": "ERROR"},
            "s3transfer": {"level": "ERROR"},
            "urllib3": {"level": "ERROR"},
        },
    }
    
    # 复制实际配置并移除stream字段，与预期配置比较
    actual_config_copy = actual_config.copy()
    actual_console_handler = actual_config_copy["handlers"]["console"].copy()
    stream_instance = actual_console_handler.pop("stream")
    actual_config_copy["handlers"]["console"] = actual_console_handler
    
    assert actual_config_copy == expected_config
    
    # 单独验证stream的类型
    assert isinstance(stream_instance, StringIO)

方案2：Mock StringIO类，让实例支持相等比较 #

通过unittest.mock.patch替换原StringIO类，给mock实例添加__eq__方法，使其可以被比较。

修改后的测试代码：

from io import StringIO
from unittest import mock
from application.utils import setup_logging

@mock.patch("logging.config.dictConfig")
@mock.patch("application.utils.StringIO")
def test_setup_logging(mock_string_io, mock_logging_config):
    # 让mock的StringIO实例支持相等判断
    mock_stream = mock_string_io.return_value
    mock_stream.__eq__ = lambda self, other: isinstance(other, (mock.Mock, StringIO))
    
    expected_conf_dict = {
        "version": 1,
        "disable_existing_loggers": False,
        "incremental": False,
        "formatters": {
            "standard": {
                "format": "[%(threadName)s] %(asctime)s %(levelname)s %(name)s: %(message)s",
                "datefmt": "%Y-%m-%d %H:%M:%S",
            }
        },
        "handlers": {
            "console": {
                "level": "INFO",
                "formatter": "standard",
                "class": "logging.StreamHandler",
                "stream": mock_stream,
            }
        },
        "root": {"handlers": ["console"], "level": "INFO"},
        "loggers": {
            "py4j.java_gateway": {"level": "ERROR"},
            "botocore": {"level": "ERROR"},
            "boto3": {"level": "ERROR"},
            "s3transfer": {"level": "ERROR"},
            "urllib3": {"level": "ERROR"},
        },
    }

    setup_logging()
    mock_logging_config.assert_called_once_with(expected_conf_dict)

方案3：重构原代码，解耦stream依赖 #

将StringIO的创建逻辑抽离，允许函数接收自定义的stream参数，测试时传入可比较的mock对象即可。

修改后的原代码（application/utils.py）

import logging
from io import StringIO

def setup_logging(conf=None, stream=None):
    if conf is None:
        # 优先使用传入的stream，否则创建新的StringIO
        log_stream = stream if stream is not None else StringIO()
        config = {
            "version": 1,
            "disable_existing_loggers": False,
            "incremental": False,
            "formatters": {
                "standard": {
                    "format": "[%(threadName)s] %(asctime)s %(levelname)s %(name)s: %(message)s",
                    "datefmt": "%Y-%m-%d %H:%M:%S",
                }
            },
            "handlers": {
                "console": {
                    "level": "INFO",
                    "formatter": "standard",
                    "class": "logging.StreamHandler",
                    "stream": log_stream,
                }
            },
            "root": {"handlers": ["console"], "level": "INFO"},
            "loggers": {
                "py4j.java_gateway": {"level": "ERROR"},
                "botocore": {"level": "ERROR"},
                "boto3": {"level": "ERROR"},
                "s3transfer": {"level": "ERROR"},
                "urllib3": {"level": "ERROR"},
            },
        }
        logging.config.dictConfig(config)
    else:
        # 基于配置文件读取的相关代码...

对应的测试代码

from unittest import mock
from application.utils import setup_logging

@mock.patch("logging.config.dictConfig")
def test_setup_logging(mock_logging_config):
    # 创建可比较的mock stream
    mock_stream = mock.Mock()
    
    expected_conf_dict = {
        "version": 1,
        "disable_existing_loggers": False,
        "incremental": False,
        "formatters": {
            "standard": {
                "format": "[%(threadName)s] %(asctime)s %(levelname)s %(name)s: %(message)s",
                "datefmt": "%Y-%m-%d %H:%M:%S",
            }
        },
        "handlers": {
            "console": {
                "level": "INFO",
                "formatter": "standard",
                "class": "logging.StreamHandler",
                "stream": mock_stream,
            }
        },
        "root": {"handlers": ["console"], "level": "INFO"},
        "loggers": {
            "py4j.java_gateway": {"level": "ERROR"},
            "botocore": {"level": "ERROR"},
            "boto3": {"level": "ERROR"},
            "s3transfer": {"level": "ERROR"},
            "urllib3": {"level": "ERROR"},
        },
    }

    setup_logging(stream=mock_stream)
    mock_logging_config.assert_called_once_with(expected_conf_dict)

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