在使用unittest与 multiprocessing时,避免重复运行测试
我们最近把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导航站,无授权禁止任何主体转载、抄袭、复制内容,亦不得私自架设镜像站点。一经侵权,本站将通过法律途径追责。