Python单元测试mock多进程方法后call_count为0的排查方法

1. 优先排查显性传参错误 #

• 对照multiprocessing.Process的定义校验入参：
  该类的构造函数签名为 Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)，第一个位置参数是预留的group字段，并非target入参。
• 现有代码写法 p1=multiprocessing.Process(self.fn2, (arg1,)) 实际的传参效果是：把self.fn2赋值给group参数，把元组(arg1,)赋值给name参数，真正的target字段全程保持默认值None。
• 这种错误下，子进程完全不会触发对fn2的任何调用，完全匹配观察到的现象：真实fn2的内部逻辑不执行、mock对象的call_count为0——不是mock失效，是根本没走到调用fn2的代码分支。
• 修正方式：初始化Process时显式指定参数名，避免位置参数错位：

  p1 = multiprocessing.Process(target=self.fn2, args=(arg1,))
  

2. 定位跨进程场景下的mock隔离问题 #

• 就算修正了上述传参错位的问题，现有测试写法依然无法正常统计mock调用次数，核心原因是多进程的内存隔离机制：
  • multiprocessing启动子进程时，传递给子进程的target、args、kwargs全量需要经过pickle序列化，从父进程内存拷贝到独立的子进程内存空间。
  • 测试代码中挂在obj1实例上的MagicMock对象，是父进程内存中的临时对象：一方面pickle无法跨进程序列化MagicMock类型的对象，另一方面子进程反序列化生成的对象和父进程的mock对象完全是独立的内存地址，不存在共享。
  • 哪怕子进程真的触发了fn2调用，调用的也是子进程空间内的独立对象，完全不会修改父进程里mock对象的call_count属性。
• 针对这类隐蔽问题，可以通过以下步骤排查：
  • 初始化完Process对象后，直接打印p1._target、p1._args两个内部属性，确认target确实绑定到了传入的mock对象，而非类上定义的原始fn2方法。
  • 由于子进程的序列化报错、运行时异常默认不会冒泡到父进程，会表现为静默失败，需要临时扩展Process类捕获异常：

    # 替换原有的Process类做调试
    class DebugProcess(multiprocessing.Process):
        def run(self):
            try:
                super().run()
            except Exception:
                import traceback
                traceback.print_exc()
    

    运行时即可打印出子进程内部的真实报错，快速定位序列化失败、参数错位等问题。

3. 适配多进程场景的单元测试方案 #

• 单元测试阶段不需要真正启动跨进程执行逻辑，应当直接mock掉multiprocessing.Process本身，将跨进程调用替换为同进程内的对象交互，才能让mock的调用计数正常生效。
• 修正后的测试代码示例：

  from unittest.mock import patch, MagicMock
  import multiprocessing
  import unittest
  
  class aTest(unittest.TestCase):
      def testfn1(self):
          # 替换a类依赖的Process类，拦截实际的跨进程启动
          with patch.object(multiprocessing, 'Process') as mock_process_cls:
              mock_process_instance = MagicMock()
              mock_process_cls.return_value = mock_process_instance
              
              obj1 = a()
              # 替换fn2为独立的mock对象
              mock_fn2 = MagicMock()
              obj1.fn2 = mock_fn2
              test_arg1 = "测试传入参数"
              
              obj1.fn1()
              
              # 校验Process初始化的参数是否符合预期
              mock_process_cls.assert_called_once_with(target=mock_fn2, args=(test_arg1,))
              # 校验子进程的start、join方法被正常调用
              mock_process_instance.start.assert_called_once()
              mock_process_instance.join.assert_called_once()
  

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