用nose测试python多处理池代码

Question

我试图用鼻子编写测试,使用多处理计算得出的东西.

我有这个目录结构:

code/
    tests/
        tests.py

tests.py看起来像这样:

import multiprocessing as mp


def f(i):
    return i ** 2


pool = mp.Pool()
out = pool.map(f, range(10))


def test_pool():
    """Really simple test that relies on the output of pool.map.
    The actual tests are much more complicated, but this is all
    that is needed to produce the problem."""
    ref_out = map(f, range(10))
    assert out == ref_out

if __name__ == '__main__':
    test_pool()

从code目录运行,python tests/tests.py 传递.

nosetests tests/tests.py 未能完成.它启动,但永远不会通过呼叫pool.map,只是挂起.

为什么这是最简单的解决方案？

Answer 1

pool.map这个问题与所谓的“全球层面”的事实有关。通常您希望避免这种情况，因为即使您的文件只是导入，这些语句也会被执行。

Nose 必须导入您的模块才能找到您的测试并稍后执行它们，因此我相信导入机制启动时会出现问题（我没有花时间尝试找出此行为的确切原因）

您应该将初始化代码移至测试装置；鼻子用装饰器支撑固定装置with_setup。pool这是一种可能性（可能是保留and作为全局变量时最简单的更改out）：

import multiprocessing as mp
from nose import with_setup

pool = None
out  = None

def f(i):
    return i ** 2

def setup_func():
    global pool
    global out
    pool = mp.Pool()
    out  = pool.map(f, range(10))

@with_setup(setup_func)
def test_pool():
    """Really simple test that relies on the output of pool.map.
    The actual tests are much more complicated, but this is all
    that is needed to produce the problem."""
    global out
    ref_out = map(f, range(10))
    assert out == ref_out

if __name__ == '__main__':
    test_pool()

执行：

$ nosetests tests/tests.py
.
----------------------------------------------------------------------
Ran 1 test in 0.011s

OK