如何在Python中mock有状态API，解决unittest.mock递归报错问题

问题根因 #

你尝试2的递归问题根源是：自定义类继承Mock时，访问实例的Publish/Subscribe属性会触发Mock的__getattr__逻辑，自动创建对应子Mock实例，而子Mock默认会复用当前自定义类作为类型，因此会再次调用__init__方法，形成无限递归。

可行方案 #

方案1：状态与Mock实例解耦（推荐） #

该方案完全避开继承Mock的潜在坑点，状态管理与Mock的断言能力分离，扩展性和可维护性更高，可直接封装为独立模块复用。

可复用模块实现

from unittest.mock import create_autospec, ANY
import mqtt

class MqttMockState:
    def __init__(self):
        # 按主题存储订阅回调，支持多主题订阅场景
        self._cached_handlers = {}
        # 按主题存储预设响应列表
        self._topic_responses = {}

    def set_topic_response(self, request_topic: str, responses: list):
        """配置指定请求主题对应的返回消息列表"""
        self._topic_responses[request_topic] = responses

    def on_subscribe(self, topic: str, handler):
        """Subscribe方法的side_effect实现"""
        self._cached_handlers[topic] = handler

    def on_publish(self, topic: str, payload, dummy=None):
        """Publish方法的side_effect实现"""
        if topic in self._topic_responses and topic in self._cached_handlers:
            # 取出对应主题的第一条响应调用回调
            resp = self._topic_responses[topic].pop(0)
            self._cached_handlers[topic](resp)

def create_mocked_mqtt() -> tuple[mqtt.Mqtt, MqttMockState]:
    """创建带状态的MQTT Mock实例，返回Mock对象和状态管理实例"""
    mock_client = create_autospec(mqtt.Mqtt, spec_set=True, instance=True)
    state = MqttMockState()
    mock_client.Subscribe.side_effect = state.on_subscribe
    mock_client.Publish.side_effect = state.on_publish
    return mock_client, state

测试用例使用示例

def test_mqtt_business_flow():
    # 初始化Mock实例和状态
    mock_mqtt, mqtt_state = create_mocked_mqtt()
    # 配置指定主题的返回消息
    mqtt_state.set_topic_response("storage/request/test_fooapp_00", ["Response1"])

    # 注入Mock运行业务代码
    thing_to_test = Thing(mock_mqtt)
    thing_to_test.Do()

    # 执行调用断言
    mock_mqtt.Publish.assert_called_with("storage/request/test_fooapp_00", ANY)

方案2：修复继承Mock的实现 #

如果要保留继承Mock的写法，只需要在自定义类中指定child_mock_type为默认Mock类，避免子节点复用当前自定义类导致递归即可，修复后的代码如下：

from unittest.mock import Mock, ANY
import mqtt

class MockedMqtt(Mock):
    # 指定子Mock使用默认Mock类，切断递归链路
    child_mock_type = Mock

    def __init__(self, *args, **kwargs):
        kwargs['spec'] = mqtt.Mqtt
        super().__init__(*args, **kwargs)
        self._cached_handlers = {}
        self._topic_responses = {}

        self.Publish.side_effect = self.on_publish
        self.Subscribe.side_effect = self.on_subscribe

    def set_topic_response(self, request_topic: str, responses: list):
        self._topic_responses[request_topic] = responses

    def on_subscribe(self, topic: str, handler):
        self._cached_handlers[topic] = handler

    def on_publish(self, topic: str, payload, dummy=None):
        if topic in self._topic_responses and topic in self._cached_handlers:
            resp = self._topic_responses[topic].pop(0)
            self._cached_handlers[topic](resp)

# 测试用例写法和你之前的尝试2一致
def test_mocking2():
    mock_mqtt_client = MockedMqtt()
    mock_mqtt_client.set_topic_response("storage/request/test_fooapp_00", ["Response1"])

    thing_to_test = Thing(mock_mqtt_client)
    thing_to_test.Do()

    mock_mqtt_client.Publish.assert_called_with("storage/request/test_fooapp_00", ANY)

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