使用 pytest-xdist 跨主节点和工作节点访问共享资源答案

【问题标题】：Accessing shared resource across master and worker nodes using pytest-xdist使用 pytest-xdist 跨主节点和工作节点访问共享资源
【发布时间】：2019-08-19 07:00:20
【问题描述】：

我正在尝试跨主节点和工作节点共享数据库中的随机条目列表（我对共享资源的定义），并使用 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']

我可以使用上述方法在 master 和 worker 之间共享这些 db 条目，如下所示：

# 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.

【问题讨论】：

我认为这与xdist 没有太大关系，而是因为您不能将夹具作为参数化参数。当夹具尚未初始化时，参数化发生在测试集合上。请参阅著名的issue #349 了解当前的讨论状态。
然而，IIRC 参数化应该发生在 xdist 场景中的 master 上，所以你应该能够实现一个自定义的 pytest_generate_tests 钩子，你可以通过 metafunc.config.db_samples 获取样本，例如用它们参数化sample 测试参数。
谢谢@hoefling。你的建议奏效了。如果我知道如何将您的评论作为答案，我会这样做。所以，我添加了一个完整的工作示例作为答案。
别担心，很高兴我能帮上忙！

标签： python pytest xdist

【解决方案1】：

感谢@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 钩子是缺失的部分。使用这个钩子参数化测试夹具有助于解决这个问题。

【讨论】：