不修改Python类的测试注入及生产流水线单元测试难题求助

听起来你遇到的痛点很典型——要测试不能修改的生产代码，还得拿到它内部悄悄创建的对象。别担心，我们可以用unittest.mock的补丁机制来拦截对象的创建过程，完全不用碰生产代码。

核心思路 #

既然流水线内部创建对象的方法不返回它，那我们就拦截这个对象的类的实例化过程，在它被创建的瞬间把它“截胡”下来，存到我们测试代码能访问的地方。

具体步骤（附代码示例） #

假设你的生产流水线代码大概是这样（完全不用修改它）：

# 生产代码：production_pipeline.py
class ProductionPipeline:
    def process(self):
        # 你已经mock了这些队列操作
        self._read_from_queue()
        # 这里创建了我们要捕获的对象，但没返回
        internal_obj = BusinessObject()
        internal_obj.execute_task()
        self._write_to_queue()

    def _read_from_queue(self):
        # 实际队列读取逻辑
        pass

    def _write_to_queue(self):
        # 实际队列写入逻辑
        pass

class BusinessObject:
    def execute_task(self):
        # 业务逻辑
        pass

接下来是测试代码的实现：

方法1：捕获真实对象实例（适合需要验证对象内部状态的场景）

import unittest
from unittest.mock import patch, MagicMock
from production_pipeline import ProductionPipeline, BusinessObject

class TestProductionPipeline(unittest.TestCase):
    def test_process_creates_business_object(self):
        # 用一个列表来存捕获到的对象（列表是可变对象，能在闭包里修改）
        captured_objects = []

        def capture_business_obj(*args, **kwargs):
            # 创建真实的BusinessObject实例
            obj = BusinessObject(*args, **kwargs)
            captured_objects.append(obj)
            return obj

        # 关键：patch流水线代码中引用的BusinessObject类
        with patch('production_pipeline.BusinessObject', side_effect=capture_business_obj):
            pipeline = ProductionPipeline()
            # 继续mock你已经处理过的队列方法
            pipeline._read_from_queue = MagicMock()
            pipeline._write_to_queue = MagicMock()

            # 执行流水线
            pipeline.process()

            # 现在就能拿到内部创建的对象了！
            self.assertEqual(len(captured_objects), 1)
            internal_obj = captured_objects[0]
            # 验证对象的方法是否被正确调用
            internal_obj.execute_task.assert_called_once()
            # 如果需要，还能检查对象的属性状态
            # self.assertEqual(internal_obj.some_attribute, expected_value)

方法2：用Mock对象替代（适合只需要验证方法调用的场景）

如果不需要真实的对象实例，只想验证它的方法是否被调用，这个方法更简单：

def test_process_calls_business_object_methods(self):
    # 创建一个Mock对象，模拟BusinessObject的行为
    mock_business_obj = MagicMock(spec=BusinessObject)

    # patch后，流水线里创建的BusinessObject都会变成这个mock
    with patch('production_pipeline.BusinessObject', return_value=mock_business_obj):
        pipeline = ProductionPipeline()
        pipeline._read_from_queue = MagicMock()
        pipeline._write_to_queue = MagicMock()

        pipeline.process()

        # 直接用mock对象断言
        mock_business_obj.execute_task.assert_called_once()

关键注意事项 #

• 补丁路径要正确：必须patch生产流水线代码中实际导入BusinessObject的路径。比如如果流水线是从myapp.core import BusinessObject，那补丁路径应该是production_pipeline.BusinessObject（因为流水线里的引用是这个），而不是myapp.core.BusinessObject。
• 闭包变量的坑：如果用方法1的捕获函数，要注意用可变对象（比如列表）来存实例，因为不可变对象（比如单个变量）在闭包里无法修改。

这样就能在完全不修改生产代码的前提下，拿到流水线内部创建的对象，完成你的单元测试了！

内容的提问来源于stack exchange，提问作者nicholas.reichel