如何测试子进程中调用的方法?
结论
普通的 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 的调用状态跨进程同步。
备注:内容仅供参考。
