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

Question

我想访问一个包含所有帐户凭据的列表，以便将它们提供给pytest xdist中的每个单独线程。我该如何做到这一点？据我所知，pytest-dist，对于每个启动的线程，它都是一个单独的测试会话，它们在内存中分别初始化每个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 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内容在master和worker之间共享。一个特殊的钩子可以自定义运行时信息。。。

在fixture中测试is_master是没有意义的，接受它的价值是在xdist不存在的情况下(no-n(。当xdist在播放时，固定装置不在主中运行，只有钩子。fixture仅在pytest_runtestloop中封装的正常协议中的子进程内执行。

您取消引用的slave_input中的自定义键需要从准备slave的钩子中的master设置。在pytest xdist newhooks.py中，您会发现类似于在node.workerinput中设置该键的homepytest_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