使用 pytest-xdist 跨主节点和工作节点访问共享资源
Accessing shared resource across master and worker nodes using pytest-xdist
我正在尝试在主节点和工作节点之间共享数据库中的随机条目列表(我对共享资源的定义),并使用 pytest-xdist 并行化测试。我的代码遵循以下结构:
## in conftest.py
def get_db_entries():
connect to db
run a query with group by and random
returns one random entry per group as a list of dictionaries.
我采纳了 https://hackebrot.github.io/pytest-tricks/shared_directory_xdist/ 提供的建议以在主节点和工作节点之间共享数据库条目:
# in conftest.py
def pytest_configure(config):
if is_master(config):
# share db_entries across master and worker nodes.
config.db_samples = get_db_entries()
def pytest_configure_node(node):
"""xdist hook"""
node.slaveinput['db_entries'] = node.config.db_samples
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')
@pytest.fixture
def db_samples(request):
"""Returns a unique and temporary directory which can be shared by
master or worker nodes in xdist runs.
"""
if is_master(request.config):
return request.config.db_samples
else:
return request.config.slaveinput['db_entries']
我可以使用上述方法在主服务器和工作服务器之间共享这些数据库条目,如下所示:
# in test-db.py
def test_db(db_samples):
for sample in samples:
# test each sample.
到目前为止,还没有测试用例的并行化。我只是在主节点和工作节点之间共享相同的数据库条目。
我的问题:如何参数化共享资源状态(数据库条目)以便我可以使用 pytest-xdist
并行执行这些测试?
我想按照以下方式做一些事情:
# in conftest.py
samples = db_samples() # will FAIL coz fixture can't be invoked directly.
@pytest.fixture(params=samples)
def get_sample(request):
return request.param
# in test-db.py
def test_db(get_sample):
# test one sample.
感谢@hoefling 的建议,我能够使用 pytest-xdist
钩子和 pytest_generate_tests
获得可行的解决方案。
# in conftest.py
def pytest_configure(config):
if is_master(config):
# share db_entries across master and worker nodes.
config.db_samples = get_db_entries()
def pytest_configure_node(node):
"""xdist hook"""
node.slaveinput['db_entries'] = node.config.db_samples
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 pytest_generate_tests(metafunc):
if 'sample' in metafunc.fixturenames:
if is_master(metafunc.config):
samples = metafunc.config.db_samples
else:
samples = metafunc.config.slaveinput['db_entries']
metafunc.parametrize("sample", samples)
@pytest.fixture
def sample(request):
return request.param
# in test-db.py
def test_db(sample):
# test one sample.
@hoefling 建议使用 pytest_generate_tests
钩子是缺失的部分。使用此挂钩对测试夹具进行参数化有助于解决此问题。
我正在尝试在主节点和工作节点之间共享数据库中的随机条目列表(我对共享资源的定义),并使用 pytest-xdist 并行化测试。我的代码遵循以下结构:
## in conftest.py
def get_db_entries():
connect to db
run a query with group by and random
returns one random entry per group as a list of dictionaries.
我采纳了 https://hackebrot.github.io/pytest-tricks/shared_directory_xdist/ 提供的建议以在主节点和工作节点之间共享数据库条目:
# in conftest.py
def pytest_configure(config):
if is_master(config):
# share db_entries across master and worker nodes.
config.db_samples = get_db_entries()
def pytest_configure_node(node):
"""xdist hook"""
node.slaveinput['db_entries'] = node.config.db_samples
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')
@pytest.fixture
def db_samples(request):
"""Returns a unique and temporary directory which can be shared by
master or worker nodes in xdist runs.
"""
if is_master(request.config):
return request.config.db_samples
else:
return request.config.slaveinput['db_entries']
我可以使用上述方法在主服务器和工作服务器之间共享这些数据库条目,如下所示:
# in test-db.py
def test_db(db_samples):
for sample in samples:
# test each sample.
到目前为止,还没有测试用例的并行化。我只是在主节点和工作节点之间共享相同的数据库条目。
我的问题:如何参数化共享资源状态(数据库条目)以便我可以使用 pytest-xdist
并行执行这些测试?
我想按照以下方式做一些事情:
# in conftest.py
samples = db_samples() # will FAIL coz fixture can't be invoked directly.
@pytest.fixture(params=samples)
def get_sample(request):
return request.param
# in test-db.py
def test_db(get_sample):
# test one sample.
感谢@hoefling 的建议,我能够使用 pytest-xdist
钩子和 pytest_generate_tests
获得可行的解决方案。
# in conftest.py
def pytest_configure(config):
if is_master(config):
# share db_entries across master and worker nodes.
config.db_samples = get_db_entries()
def pytest_configure_node(node):
"""xdist hook"""
node.slaveinput['db_entries'] = node.config.db_samples
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 pytest_generate_tests(metafunc):
if 'sample' in metafunc.fixturenames:
if is_master(metafunc.config):
samples = metafunc.config.db_samples
else:
samples = metafunc.config.slaveinput['db_entries']
metafunc.parametrize("sample", samples)
@pytest.fixture
def sample(request):
return request.param
# in test-db.py
def test_db(sample):
# test one sample.
@hoefling 建议使用 pytest_generate_tests
钩子是缺失的部分。使用此挂钩对测试夹具进行参数化有助于解决此问题。