pytest默认不支持直接测试多进程代码，因其单进程上下文无法捕获子进程异常、exit code和覆盖率，易导致死锁、静默失败或雪崩启动；需用Queue/Pipe显式传递结果与异常，并设join timeout。
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()，就会递归启动新进程 —— 导致雪崩。
必须确保所有进程启动逻辑只在显式入口中执行：
立即学习“Python免费学习笔记（深入）”；
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()
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
同时，在测试前显式设置启动方式（尤其 CI 环境需统一）：
import multiprocessing
<p>def test_multiprocessing_with_spawn():
multiprocessing.set_start_method("spawn", force=True) # force=True 避免 RuntimeError</p><h1>... 启动子进程并验证</h1><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 < 0:
raise ValueError("negative not allowed")
queue.put(("result", x ** 2))
except Exception as e:
queue.put(("error", (str(e), traceback.format_exc())))</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]
最麻烦的从来不是怎么写进程，而是怎么让它们“说清楚自己干了什么、哪里错了”。别指望子进程自动上报，得靠队列、管道、信号或日志文件主动交出证据。
