在使用unittest与 multiprocessing时,避免重复运行测试

编程语言 2026-07-11

我们最近把Python 3.9升级到了3.14,似乎多进程的默认启动方法从 'fork' 改为了 'spawn',这导致一些测试失败,仿佛在无限地创建更多的进程池(Pools)。我有要测试的模块(thing.py):

print("IN thing.py")
import os, time

def foo(x):
    time.sleep(0.1)
    return (x, os.getpid())

def dothing():
    from multiprocessing import Pool
    with Pool(processes=4) as pool:
        x = pool.map(foo, range(10))
        a, b = zip(*x)
        print(a, b)

我也有带测试的模块(test_thing.py):

print("IN test_thing.py")
import unittest

import thing

class TestThing(unittest.TestCase):
    def testThing(self):
        thing.dothing()

还有一个运行脚本(runtest.py),因为种种原因需要它:

print("IN runtest.py")
from importlib import import_module
import unittest
import multiprocessing

print(f"{__name__=} {multiprocessing.current_process().name=}")
print(f"{multiprocessing.current_process().name=}")
print(f"{multiprocessing.parent_process()=}")

modToTest = import_module('test_thing')
suite = unittest.defaultTestLoader.loadTestsFromModule(modToTest)
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)

在运行测试(python3 runtest.py)时,对Pool的调用会导致整个测试重新执行,我收到一个RuntimeError。用于判断是否在子进程中的常用方法似乎并不起作用。原始调用和Pool调用都具有相同的 __name__、进程名,且没有报告父进程。输出看起来像:

IN runtest.py  
\__name_\_='\__main_\_' multiprocessing.current_process().name='MainProcess'  
multiprocessing.current_process().name='MainProcess'  
multiprocessing.parent_process()=None  
IN test_thing.py  
IN thing.py  
testThing (test_thing.TestThing.testThing) ... IN runtest.py  
\__name_\_='\__mp_main_\_' multiprocessing.current_process().name='MainProcess'  
multiprocessing.current_process().name='MainProcess'  
multiprocessing.parent_process()=None  
IN test_thing.py  
IN thing.py  
testThing (test_thing.TestThing.testThing) ... ERROR

======================================================================
ERROR: testThing (test_thing.TestThing.testThing)
----------------------------------------------------------------------
...
RuntimeError: 
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

我可以强制让multiprocessing以 'fork' 启动Pool,这似乎能解决问题,但相关代码在测试环境之外运行良好,因此我正尝试在仍然使用 'spawn' 的前提下,只修复测试部分。

解决方案

正如第一条注释所指出的,虽然进程名相同,但 __name__ 不同,因此我可以在主脚本中据此进行筛选,尽管到了测试层就无法筛选。

runtest.py 中的所有代码放在一个 if __name__ == '__main__': 的检查里,问题就解决了。

站内所有文章版权归属LeftHeroAI导航站,无授权禁止任何主体转载、抄袭、复制内容,亦不得私自架设镜像站点。一经侵权,本站将通过法律途径追责。

相关文章