PAYATHON 2026

如何测试子进程中调用的方法?

数据小许

结论

普通的 unittest.mock.Mock 不适合用父进程中的调用记录来验证子进程行为。Process 启动后,子进程有自己的进程状态。子进程中的 mock 调用不会自动同步到父进程,因此 mock.method.assert_called() 仍可能失败。

示例中,Main 把 self.task 作为进程目标。子进程执行 self._mock.method() 时,调用记录保存在子进程自己的 mock 状态中,父进程中的 mock 无法看到。

优先选择:直接测试业务逻辑

如果要验证 task 中的业务逻辑,可以将业务方法与启动进程的代码分开,在父进程中直接调用并断言:

from multiprocessing import Process

class Main:
    def __init__(self, service):
        self._service = service

    def task(self):
        self._service.method()

    def start(self):
        process = Process(target=self.task)
        process.start()
        process.join()
from unittest.mock import Mock


def test_task_calls_method():
    service = Mock()
    main = Main(service)

    main.task()

    service.method.assert_called_once_with()

这类测试检查业务逻辑本身,不受多进程调度影响。进程的启动和退出可以单独测试,也可以检查进程退出码,确认子进程是否成功结束。

必须验证子进程行为时:使用进程间通信

如果要确认某个调用确实由子进程执行,可以让子进程通过 multiprocessing.Queue 显式报告事件。下面用一个记录调用的替身代替 Mock:

from multiprocessing import Process, Queue


class RecordingService:
    def __init__(self, calls):
        self.calls = calls

    def method(self):
        self.calls.put("method")


class Main:
    def __init__(self, service):
        self._service = service
        self.process = Process(target=self.task)
        self.process.start()
        self.process.join()

    def task(self):
        self._service.method()
def test_process_calls_method():
    calls = Queue()
    main = Main(RecordingService(calls))

    assert main.process.exitcode == 0
    assert calls.get(timeout=2) == "method"

队列可以在进程之间传递消息,因此父进程能收到子进程发出的记录。实际项目中,也可以让子进程返回结果,或通过其他合适的 IPC 机制传递需要断言的数据。使用 Queue 时,应为读取设置超时,并检查进程退出码,避免子进程异常或没有发送消息时测试一直等待。

注意 pytest 的收集方式

错误日志显示,test_Process.py 在模块导入时直接调用了 test_process()。通常应删除测试文件顶层的直接调用,让 pytest 自行收集并运行测试。这个问题可能改变测试的执行方式,但不会影响父子进程之间不共享普通 mock 调用记录这一事实。

multiprocessing 在不同操作系统和启动方式下创建子进程的细节有所不同。编写跨平台测试时,应确保传给子进程的目标和参数能在当前启动方式下使用。不要依赖 Mock 的调用状态跨进程同步。

备注:内容仅供参考。