如何在不同的进程中调用带有固定装置的测试

Question

我正在为基于 pytorch 的库编写测试，许多测试需要通过在不同进程上运行相同的函数体并相互通信（例如使用 torch.distributed.all_reduce）来执行 torch.distributed。

我目前有一个装饰器，它包装了这样的测试：

import inspect

@wrap()
def foo():
    assert true == false

def wrap():
    def fixer(function)
        def replacement():
            torch.distributed.mp.spawn(function, nprocs=2)

        setattr(inspect.getmodule(function), "test_" + function.__name__, replacement)
    return fixer

但这不适用于 pytest 固定装置，我希望能够使用固定装置。我尝试声明我自己的

pytest_runtest_call

/

pytest_runtest_protocol

但测试项目不可腌制，所以我回到了第一个方向。有没有一些简单的方法可以最大限度地减少每个测试的样板？

当前需要固定装置的测试在包含实际测试主体的另一个函数上调用

mp.spawn

。

Answer 1

我想问题不在于 pytorch 或不同的进程，而只是丢失函数签名和参数传递

wrap

def wrap():
    def fixer(function):
        paramertes = inspect.signature(function).parameters
        @wraps(function)  # this will make pytest pass fixtures to the wrapped function  
        def replacement(*args, **kwargs):
            assert args == tuple()
            args = tuple(kwargs[p] for p in paramertes)  # converting named parameters to positional parameters to pass to `spawn`
            torch.distributed.mp.spawn(function, args=args, nprocs=2)

        setattr(inspect.getmodule(function), "test_" + function.__name__, replacement)
    return fixer


@fixture()
def d():
    return 1


@wrap()
def foo(d):
    assert d == 1