Python Mock侧边效应配置：side_effect与return_value怎么用

在编写 Python 单元测试时，unittest.mock 模块通过模拟外部依赖来保证被测逻辑的独立性。其中控制 Mock 实例响应的核心手段主要是两个属性：return_value 负责设定静态、固定的单次或重复调用返回值；而 side_effect 则负责处理“非单一固定返回值”的各类动态行为，包括抛出指定异常、按调用次序逐个产出序列元素，以及依据传入参数动态执行自定义逻辑。当两个属性同时存在时，side_effect 具有更高的生效优先级并直接遮蔽 return_value。

一、核心机制：return_value 与 side_effect 的运行分发与优先级

单元测试中最常碰到的困惑之一，是为同一个 Mock 实例同时配置了返回值与侧边效应后，测试结果往往只遵循 side_effect 的输出。这是由于 unittest.mock 内部的分发机制所决定的：一旦检测到 side_effect 属性不为 None，Mock 对象就会优先将调用转交给该属性指向的对象处理。

只有当 side_effect 的计算结果显式返回了内部哨兵对象 unittest.mock.DEFAULT 时，调用控制流才会向下回退并读取 return_value。如果需要撤销侧边效应恢复普通的静态返回值，必须显式将 mock_obj.side_effect = None 进行状态重置。

二、side_effect 的三种核心配置形态与实操示例

为了满足不同的测试断言目标，side_effect 支持接收三种完全不同的数据结构，每种结构对应一套清晰的执行逻辑。

1. 赋值为异常类型或异常实例：测试容灾与错误处理

当被测函数包含 try...except 结构，或者需要确保底层网络超时、权限不足时系统能准确向上抛出业务自定义异常时，可以直接将异常类或具体的异常实例赋给 side_effect。

from unittest.mock import Mock

# 模拟网络客户端
client = Mock()

# 方式 A：直接传入异常类，调用时自动实例化抛出
client.get.side_effect = TimeoutError

# 方式 B：传入已经初始化好的异常实例（携带具体错误信息）
# client.get.side_effect = TimeoutError("HTTP连接池获取超时，重试次数超限")

try:
    client.get("https://api.example.com/data")
except TimeoutError as err:
    print(f"捕获到预期的网络异常: {err}")

2. 赋值为可迭代对象：测试多次调用间的时序状态流转

在涉及轮询任务状态（例如排队中、执行中、最终完成）或接口自动重试（例如前两次失败、第三次成功）的代码路径中，静态的 return_value 无法表达“随着调用次数增加返回不同内容”的需求。此时可以将一个列表、元组或生成器传入 side_effect。

每次调用该 Mock 实例，Mock 内部会自动执行一次 next() 并返回当前迭代项。当列表内所有项被消耗完毕后，如果被测代码再次发起调用，Mock 会严格抛出 StopIteration 异常。

from unittest.mock import Mock

mock_fetch_status = Mock()

# 配置时序返回值：前两次返回处理中，第三次返回成功
mock_fetch_status.side_effect = ["PROCESSING", "PROCESSING", "SUCCESS"]

print(mock_fetch_status())  # 输出: PROCESSING
print(mock_fetch_status())  # 输出: PROCESSING
print(mock_fetch_status())  # 输出: SUCCESS

# 注意：若继续调用第四次，将引发 StopIteration 异常
# print(mock_fetch_status()) -> StopIteration

如果希望在序列消耗过程中穿插异常抛出，可以将异常实例与正常值混排在同一个列表中：

# 第一次调用抛出网络错误，第二次调用返回正常数据 200
mock_fetch_status.side_effect = [ConnectionResetError("对端重置连接"), {"code": 200, "data": "ok"}]

3. 赋值为自定义函数（Callable）：测试基于参数的动态分支与校验

当返回的数据依赖于调用时传入的具体实参，或者需要在被调用时同步修改测试夹具内的某些状态时，可以编写一个独立的函数或 lambda 表达式传给 side_effect。该函数会完整接收调用 Mock 时传来的所有位置参数与关键字参数。

from unittest.mock import Mock, DEFAULT

def mock_database_query(user_id, with_profile=False):
    if user_id <= 0:
        raise ValueError("无效的用户 ID")
    if user_id == 101:
        return {"id": 101, "name": "Alice", "role": "admin"}
    # 当入参为其他值时，返回 DEFAULT 触发回退机制
    return DEFAULT

db_mock = Mock()
db_mock.return_value = {"id": 0, "name": "Anonymous", "role": "guest"}
db_mock.side_effect = mock_database_query

# 命中动态逻辑
print(db_mock(101))        # 输出: {'id': 101, 'name': 'Alice', 'role': 'admin'}

# 返回 DEFAULT，自动回退到 return_value
print(db_mock(999))        # 输出: {'id': 0, 'name': 'Anonymous', 'role': 'guest'}

三、配置与验证 Mock 侧边效应的四步标准流程

在实际的单元测试工程中，编写可信赖的 Mock 测试通常遵循“定位外部依赖”、“配置响应行为”、“注入执行目标”、“断言调用痕迹”这四个标准化步骤。

1. 步骤一：识别模拟桩点。观察被测方法调用链，将易发生网络抖动、数据库连接异常或耗时较长的第三方外部调用作为隔离对象。
2. 步骤二：设定 side_effect 响应内容。根据测试用例的目的（如测试重试机制、错误熔断或分支判断），挑选适合的数据形态。
3. 步骤三：注入并执行目标函数。通过 unittest.mock.patch 替换目标模块中的符号，保证被测代码在执行期间调用的是被修饰后的 Mock 实例。
4. 步骤四：断言调用痕迹与状态清理。除了断言被测函数的返回值之外，必须使用 assert_called_with 校验传参，并使用 mock.call_count 确保调用的次数完全符合预期设计。

四、实战演练：测试带有重试机制的网络请求函数

以下为一个具备指数退避或固定重试机制的业务函数 fetch_with_retry。该函数在遭遇网络错误时最多重试 3 次，若重试成功则返回响应字典，若达到最大重试次数仍失败则抛出自定义异常 ServiceUnavailableError。

# 业务代码：network_client.py
import time
import requests

class ServiceUnavailableError(Exception):
    """业务自定义服务不可用异常"""
    pass

def fetch_with_retry(url: str, max_retries: int = 3) -> dict:
    attempts = 0
    while attempts < max_retries:
        try:
            response = requests.get(url, timeout=2.0)
            if response.status_code == 200:
                return response.json()
            attempts += 1
        except (requests.ConnectionError, requests.Timeout):
            attempts += 1
            time.sleep(0.1)
            
    raise ServiceUnavailableError(f"请求 {url} 失败，已达最大重试次数 {max_retries}")

在编写对应的单元测试时，我们希望验证两个核心路径：

• 路径 1（网络瞬时故障后恢复）：前两次请求抛出 requests.ConnectionError，第三次请求返回 200 状态码，验证函数最终能够成功拿到数据，且重试了 3 次。
• 路径 2（持续网络故障）：连续 3 次全部抛出 requests.Timeout，验证函数是否如期触发 ServiceUnavailableError。

# 测试代码：test_network_client.py
import unittest
from unittest.mock import patch, Mock
import requests
from network_client import fetch_with_retry, ServiceUnavailableError

class TestFetchWithRetry(unittest.TestCase):

    @patch("network_client.time.sleep")  # 消除真实的 sleep 等待耗时
    @patch("network_client.requests.get")
    def test_retry_success_after_two_failures(self, mock_get, mock_sleep):
        # 构造成功响应对象
        success_response = Mock()
        success_response.status_code = 200
        success_response.json.return_value = {"status": "ok", "retried": True}

        # 配置 side_effect 序列：前两次抛异常，第三次返回正常响应对象
        mock_get.side_effect = [
            requests.ConnectionError("连接被拒绝"),
            requests.Timeout("读取数据超时"),
            success_response
        ]

        # 执行被测方法
        result = fetch_with_retry("https://api.internal/service", max_retries=3)

        # 断言结果正确性
        self.assertEqual(result, {"status": "ok", "retried": True})
        # 断言调用总次数为 3 次
        self.assertEqual(mock_get.call_count, 3)
        # 验证 sleep 确实被触发了 2 次
        self.assertEqual(mock_sleep.call_count, 2)

    @patch("network_client.time.sleep")
    @patch("network_client.requests.get")
    def test_retry_exhausted_raises_service_unavailable(self, mock_get, mock_sleep):
        # 配置持续超时
        mock_get.side_effect = requests.Timeout("持续超时")

        # 断言抛出业务异常
        with self.assertRaises(ServiceUnavailableError):
            fetch_with_retry("https://api.internal/service", max_retries=3)

        # 断言恰好尝试了 3 次后退出
        self.assertEqual(mock_get.call_count, 3)

if __name__ == "__main__":
    unittest.main()

五、常见避坑点与调试指南

六、选型速查与交互关系总结

为了在编写单元测试时快速决策，下表与图 4 总结了 return_value 与 side_effect 在不同维度上的关键特性差异。