如何使用pytest-xdist访问共享资源？

Question

我希望访问一个包含所有帐户凭据的列表，以便将它们提供给pytest-xdist中的每个单独线程。我怎样才能做到这一点？据我所知，pytest，对于每个已启动的线程，它是一个单独的测试会话，它们在各自的内存中分别初始化每个python模块。我有一个代码片段示例，如下所示

import sys
import pytest
import threading
import time

lock = threading.Lock()

i = 0
lis = []
lis.append(1)
lis.append(2)


class TestAAA(object):

    def test_function1(self, fixture_function_feature1):
        global i, lock
        with lock:
            print "Lock Acquired"
            time.sleep(2)
            print >> sys.stderr, 'test_: '+str(lis[i])

执行的命令：

pytest -sv -n 2 --count=5 

输出：

test_: 1
test_: 1

预期产出：

test_: 1
test_: 2

我还尝试使用https://github.com/pytest-dev/pytest/issues/1402#issuecomment-186299177中提到的共享资源

 import pytest

    def pytest_configure(config):        
        if is_master(config):
            config.shared_directory = tempdir.mktemp()

    def pytest_unconfigure(config):        
        if is_master(config):
            shutil.rmtree(config.shared_directory)


    def pytest_configure_node(self, node):
        """xdist hook"""
        node.slaveinput['shared_dir'] = node.config.shared_directory


    @pytest.fixture
    def shared_directory(request):
        if is_master(request.config):
            return request.config.shared_directory
        else:
            return request.config.slaveinput['shared_dir']


    def is_master(config):
        """True if the code running the given pytest.config object is running in a xdist master
        node or not running xdist at all.
        """
        return not hasattr(config, 'slaveinput')

    def test_shared_dir(shared_directory):
        print >> sys.stderr, 'master logs dir: '+str(shared_directory) 

如果使用pytest-xdist运行它们，则两者都会出现错误。

命令：pytest -sv test_parallel_share.py

输出：

―――――――――――――――――――――――――――――――――――――――――――――― ERROR at setup of test_shared_dir ―――――――――――――――――――――――――――――――――――――――――――――――

request = <SubRequest 'shared_directory' for <Function 'test_shared_dir'>>

    @pytest.fixture
    def shared_directory(request):
        if is_master(request.config):
>           return request.config.shared_directory
E           AttributeError: 'Config' object has no attribute 'shared_directory'

test_parallel_share.py:21: AttributeError
                                                                                                                 100% ██████████

Results (0.05s):
       1 error

命令：pytest -n 2 -sv test_parallel_share.py

输出：

scheduling tests via LoadScheduling


―――――――――――――――――――――――――――――――――――――――――――――― ERROR at setup of test_shared_dir ―――――――――――――――――――――――――――――――――――――――――――――――

request = <SubRequest 'shared_directory' for <Function 'test_shared_dir'>>

    @pytest.fixture
    def shared_directory(request):
        if is_master(request.config):
            return request.config.shared_directory
        else:
>           return request.config.slaveinput['shared_dir']
E           KeyError: 'shared_dir'

test_parallel_share.py:23: KeyError
                                                                                                                 100% ██████████

Results (0.55s):
       1 error

Answer 1

我认为有很多方法可以解决你的问题。config.option内容在主用户和工作人员之间共享。一个特殊的钩子可以定制运行时信息..。

在固定设备中测试is_master是没有意义的，接受它的价值是因为xdist不存在(没有-n)。当xdist处于运行状态时，固定装置不是在主中运行，而是在钩子中运行。只有在封装在pytest_runtestloop中的普通协议中的子进程中才会执行固定装置。

slave_input中的自定义键--您的取消引用--需要从为从服务器做准备的钩子中的主键中设置。查看pytest-xdist newhooks.py，您将发现类似于在node.workerinput中设置该键的主页。看看pytest_configure_node(self, node: WorkerController)。您应该编写其中之一，并在其中设置node.workerinput['shared_dir']，奴隶将找到要分发的公共数据。

def pytest_configure_node(self, node: WorkerController):
    node.workerinput['shared_dir'] = node.config.option.shared_dir