如何在Pytest 3.0.7中动态并行运行一些测试

Question

我正在使用Pytest 3.0.7（在Python 2.7中），我需要能够在不修改测试的情况下动态地并行运行选定的测试。以下面的示例代码为例：

import time
import logging
import pytest

logger = logging.getLogger('test1')
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# File handler
logfh = logging.FileHandler('results.log')
logfh.setLevel(logging.DEBUG)
logfh.setFormatter(formatter)
logger.addHandler(logfh)
# Console handler
logch = logging.StreamHandler()
logch.setLevel(logging.DEBUG)
logch.setFormatter(formatter)
logger.addHandler(logch)

def waste_time(prefix, range_end=5, sleep_time=2):
    logger.info("WASTE TIME")
    for i in range(range_end):
        logger.debug("{}, iteration {} of {}.".format(prefix, i, range_end))
        time.sleep(sleep_time)

def test_test1():
    logger.info("Test1 is starting.")
    waste_time(prefix='TEST 1')
    assert True
    logger.info("Test1 is done.")

def test_test2():
    logger.info("Test2 is starting.")
    waste_time(prefix='TEST 2')
    assert True
    logger.info("Test2 is done.")

def test_test3():
    logger.info("Test3 is starting.")
    waste_time(prefix='TEST 3')
    assert True
    logger.info("Test3 is done.")

我需要能够动态地决定并行运行哪些测试（如果有的话）。有时它将是test1和test2，而test3将在之后连续运行。有时测试1将并行运行，然后是测试2和3，有时我将并行运行所有测试，有时无测试，等等

我不知道提前组合，我计划在一个单独的配置文件中提供它们，我将使用插件进行评估

我一直在玩pytest_generate_tests插件，并试图在我的测试中添加标记，但我还没有找到成功标记测试的方法。我一直使用pytest xdist作为并行测试运行程序

这是我的一个pytest_generate_测试实验（我现在只是想让它跳过一个测试）

def pytest_generate_tests(metafunc):
    if metafunc.definition.name == 'test_test2':
        metafunc.definition.own_markers = [_pytest.mark.Mark(name='skip', args=(), kwargs={})]
        logger.debug("Mark set on test_test2")

我们有数以千计的测试，我们想应用这一点，我不能修改测试模块。如果可以的话，我甚至不知道该怎么做。我必须让它动态工作

是的，我知道Python2.7已经被弃用了。我们将过渡到3.0，但是有很多代码和一个相当大的组织需要迁移

任何帮助都将不胜感激

Answer 1

通过使用pytest_collection_finish钩子动态设置标记，我可以跳过测试，如下所示：

import _pytest.mark

def pytest_collection_finish(session):
    for itm in session.session.items:
        if itm.name == 'test_test2':
            itm.own_markers = [_pytest.mark.Mark(name='skip', args=(), kwargs={})]

现在看看我是否能把它应用到我真正的问题上