如何使用Python pytest测试多进程（Multiprocessing）代码？-Python教程-PHP中文网

pytest默认不支持直接测试多进程代码，因其单进程上下文无法捕获子进程异常、exit code和覆盖率，易导致死锁、静默失败或雪崩启动；需用Queue/Pipe显式传递结果与异常，并设join timeout。

如何使用python pytest测试多进程（multiprocessing）代码？

pytest 默认不支持直接测试多进程代码，因为 pytest 的测试运行器本身依赖单进程上下文，而 multiprocessing 会 fork 或 spawn 新进程，导致测试断言、异常捕获、覆盖率收集失效，甚至出现挂起或静默失败。

为什么 pytest 默认跑 multiprocessing 测试会卡住或没报错？

根本原因是：子进程无法将异常回传到主测试进程，pytest 也收不到子进程的 exit code；如果子进程抛出未捕获异常，它会直接退出，但主进程可能还在等 join()，造成死锁。常见现象包括：

测试长时间无响应（尤其用了 process.join() 但没设 timeout）
assert 在子进程中失败，但测试显示 PASSED
使用 pytest-cov 时，子进程的代码完全不计入覆盖率
在 Windows 上用 spawn 启动方式时，子进程反复导入测试模块，可能触发重复执行或 NameError

用 multiprocessing.set_start_method("spawn") + if name == "main" 防止 Windows 崩溃

Windows 和 macOS 默认用 spawn，Linux 默认用 fork。但 pytest 运行时不是通过 python -m pytest 启动的常规脚本入口，所以子进程重新导入模块时，若模块顶层有 multiprocessing.Process(...).start()，就会递归启动新进程 —— 导致雪崩。

必须确保所有进程启动逻辑只在显式入口中执行：

def worker(x):
    return x * x
<p>if <strong>name</strong> == "<strong>main</strong>":</p><h1>只有直接运行此文件时才启动进程</h1><pre class="brush:php;toolbar:false;"># pytest 运行时不执行这里
p = multiprocessing.Process(target=worker, args=(4,))
p.start()
p.join()

同时，在测试前显式设置启动方式（尤其 CI 环境需统一）：

python-script-generator

快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。

下载

import multiprocessing
<p>def test_multiprocessing_with_spawn():
multiprocessing.set_start_method("spawn", force=True)  # force=True 避免 RuntimeError</p><h1>... 启动子进程并验证</h1><p></p>

用 pytest-xdist 跑并行测试？别混淆概念

pytest-xdist 是让多个 pytest 实例在不同进程中并行运行测试用例，但它不解决「单个测试函数内使用 multiprocessing」的问题。两者目标相反：

pytest-xdist：把测试分发到多个 Python 解释器进程（每个进程跑一部分 test_ 函数）
你的需求：在一个 test_ 函数里启动多个 multiprocessing.Process 或 multiprocessing.Pool

混用反而容易出问题 —— 比如 xdist 的 worker 进程再 fork 子进程，可能触发 pickle 错误或资源竞争。真要测多进程逻辑，关掉 xdist（加 -n0）更稳妥。

真正可行的测试策略：mock + 显式错误传播 + timeout

不追求“真实并发”，而是验证逻辑正确性与异常路径。关键点：

用 multiprocessing.Queue 或 multiprocessing.Pipe 把子进程结果/异常传回来，主进程做 assert
永远给 join(timeout=3)，避免无限等待
捕获子进程中的异常并序列化（比如用 cloudpickle 或手动包装成 dict）
对 Pool 类场景，改用 concurrent.futures.ProcessPoolExecutor，它原生支持 as_completed 和异常传播

示例（安全传递异常）：

import multiprocessing
import traceback
<p>def safe_worker(queue, x):
try:
if x </p><p>def test_worker_error_propagation():
q = multiprocessing.Queue()
p = multiprocessing.Process(target=safe_worker, args=(q, -1))
p.start()
p.join(timeout=2)
assert p.exitcode != 0 or not q.empty()  # 至少得有反馈
kind, payload = q.get_nowait()
assert kind == "error"
assert "negative not allowed" in payload[0]
</p>

最麻烦的从来不是怎么写进程，而是怎么让它们“说清楚自己干了什么、哪里错了”。别指望子进程自动上报，得靠队列、管道、信号或日志文件主动交出证据。