自定义一个并发执行pytest测试用例的插件

前提知识点

`pytest_addoption()` 函数

pytest_addoption() 函数是 Pytest 中用于添加命令行选项和自定义参数的钩子函数。通过在此函数中定义相应的命令行选项和参数，可以让用户在运行 Pytest 时方便地控制测试的执行方式和范围，使测试框架更加灵活和易用。

具体来说，pytest_addoption() 函数主要有两个作用：

定义命令行选项参数

可以通过该函数向 Pytest 中添加自定义的命令行选项参数，例如 --myoption、--verbose 等。这些参数可以用于控制测试的输出、调试信息、日志级别等不同方面，从而更好地理解测试结果并快速定位问题。

通过 pytest_addoption() 函数定义的命令行选项参数可以帮助用户灵活地控制 Pytest 框架的行为。例如，我们可以通过向函数中传递如下代码来定义一个名为 --myoption 的命令行选项参数：

def pytest_addoption(parser):
    parser.addoption("--myoption", action="store", default="default_value", help="my option: type=str,default='default_value'")

在上面的代码中，我们使用 parser.addoption() 方法向 Pytest 中添加了一个名为 --myoption 的命令行选项参数，并指定了该选项参数的类型、默认值和帮助文本等信息。通过这个选项，用户可以在运行 Pytest 时指定不同的参数值来控制测试的行为。

例如，在终端中执行如下命令：

pytest --myoption=value test_sample.py

这样就可以将 --myoption 参数设置为 value，从而控制测试的行为。

解析自定义参数

除了标准的命令行选项参数外，pytest_addoption() 函数还可以解析自定义参数，并将其传递给测试用例。例如，我们可以定义一个名为 --myoption 的参数，并将其值传递给测试用例，以便在测试中使用该值。这样就可以根据具体需求配置不同的测试环境、数据或行为，从而实现高度的灵活性和可扩展性。

例如，在上面定义 --myoption 命令行选项参数时，我们指定了 action="store"，这意味着该参数会被解析并存储在 Pytest 的 config 对象中。因此，在测试用例中可以通过如下代码来获取该参数的值：

def test_something(config):
    myoption_value = config.getoption("--myoption")
    # do something with myoption_value

在上面的示例中，我们使用 Pytest 的 config 对象获取了 --myoption 参数的值，并将其传递给测试用例。这样，我们就可以在测试用例中根据需要使用该参数的值，以实现更高效的测试流程。

总之，pytest_addoption() 函数是 Pytest 中非常实用的钩子函数，它可以帮助用户灵活控制 Pytest 框架的行为，并实现自定义参数的解析。在测试工作中，使用该函数可以提高测试效率、降低测试成本，并增强测试框架的灵活性和可扩展性。

参数解释

name：名称，在命令行中接受该选项后面的值，即代码中的 --value;
action:对命令行中名称后面的值执行相关操作,store代表存储；
type：注册的自定义到pytest配置对象中的命令行参数对应的参数值应该转换为的类型；
dest：存储值的名称，后续取值需要用到；
default：当命令行不传入参数时，默认值；
help：注明相关信息。
运行时，在命令行窗口输入pytest -s --value Bang test-statistics.py即可将Bang存储在environment中。

parser.getgroup(): 用于创建选项组（Option Group），将多个相关的选项参数分成一组，并进行统一管理。例如：

group = parser.getgroup("my options")
group.addoption("--myoption", action="store", default="default_value", help="my option: type=str,default='default_value'")

上述代码的含义是创建一个名为 "my options" 的选项组，将 --myoption 参数添加到该组中，并指定该参数的类型、默认值和帮助信息等。

触发时机: conftest文件加载完之后执行，在测试运行开始时调用一次。

`item.ihook.pytest_runtest_protocol()`

item.ihook.pytest_runtest_protocol(item=item, nextitem=None) 是 Pytest 测试用例对象（pytest.Item）中的一个方法，用于调用 pytest_runtest_protocol() 钩子以执行测试用例。它会在 item.run() 方法中被调用，用于执行当前测试用例之前和之后的一些准备工作和清理工作，并收集测试结果。这个方法有两个参数：

item: pytest.Item 对象，表示当前要执行的测试用例。
nextitem: pytest.Item 对象，表示即将执行的下一个测试用例。

重写pytest_runtestloop() 方法

首先原 pytest_runtestloop 钩子实现对会话 ( ) 中收集的所有项目执行 runtest 协议session.items，除非收集失败或collectonly设置了 pytest 选项。如果在任何时候pytest.exit()调用，循环将立即终止。如果在任何点session.shouldfail或session.shouldstop设置，循环在当前项目的运行测试协议完成后终止。

在conftest.py中编写

# pip install pytest-xdist
from typing import List

import pytest
import gevent
from gevent import monkey

monkey.patch_all()


# conftest.py是pytest的配置文件，文件名称是固定的


def pytest_addoption(parser):
    """
    用于添加命令行选项参数
    名称是固定的
    :param parser: Parser对象
    :return:
    """
    # 1.添加参数分组
    group = parser.getgroup('dev16', 'manual pytest plugin')

    # 2.添加参数和帮助信息
    group.addoption('--task', type=str, default='module',
                    help='Specify the type of task to be executed concurrently, you can fill in case or model, '
                         'the default is case')
    group.addoption('--concurrent', type='int', default=1,
                    help='Specify the number of concurrent tasks, the default is 1')


# 根据不同的任务类型来创建协程并发执行任务
def pytest_runtestloop(session: pytest.Session) -> bool:
    """
    用于控制测试用例的执行
    名称是固定的
    :param session: 包含了当前会话的所有信息
    :return:
    """
    # 获取选项参数信息
    task_level = session.config.getoption('--task')
    print(f'任务类型为：{task_level}')

    concurrent = session.config.getoption('--concurrent')
    print(f'并发用户数：{concurrent}')

    if task_level == 'case':
        run_case(session)
    elif task_level == 'module':
        run_module(session)
    else:
        raise Exception('The task type is wrong, it can only be case or module')
    return True


def run_case(session: pytest.Session):
    concurrent = session.config.getoption('--concurrent')
    gs = [gevent.spawn(run_task, session.items) for _ in range(concurrent)]
    # 等待所有的协程执行结束
    gevent.joinall(gs)


def run_module(session: pytest.Session):
    # concurrent = session.config.getoption('--concurrent')
    # 定义一个字典用于保存不同模块的用例
    # 把模块作为key，把同一个模块下的测试用例列表作为value
    # {'test_demo01': [item, item], 'test_demo02': [item, item]}
    module_dict = {}
    for item in session.items:
        module = item.module
        # if module not in module_dict:
        #     module_dict[module] = []
        # else:
        #     module_dict[module].append(item)
        module_dict.setdefault(module, []).append(item)
    print(module_dict)
    gs = []
    for module_name, items in module_dict.items():
        gs.append(gevent.spawn(run_task, items))
    # 等待所有的协程执行结束
    gevent.joinall(gs)


def run_task(items: List):
    while len(items) != 0:
        item = items.pop()
        result = item.ihook.pytest_runtest_protocol(item=item, nextitem=None)

然后再运行入口文件run.py中

import pytest

if __name__ == '__main__':
    pytest.main(['-s', '--task=case', '--concurrent=3'])
    # pytest.main(['-s', '--task=module'])

这段代码是一个自定义的 Pytest 插件，主要实现了根据命令行选项参数来控制测试任务的类型和并发执行任务的数量。下面详细解析这段代码的逻辑思路：

添加命令行选项参数

通过 pytest_addoption 函数来对 pytest 的命令行选项参数进行添加，其中 parser 参数是用于解析命令行参数的解析器，我们可以通过 parser.getgroup 方法来创建一个参数分组，并通过 group.addoption 方法添加具体的选项参数。

本代码中添加的选项参数有两个，分别是 --task 和 --concurrent，前者用于指定任务类型（可以是模块级别或用例级别），后者用于指定并发执行任务的数量。

控制测试用例执行

通过 pytest_runtestloop 函数来控制测试用例的执行流程。该函数在 pytest 启动时会被调用，其中 session 参数包含了当前会话的所有信息。在该函数内部，首先获取选项参数的值，然后判断任务类型，如果是用例级别，则调用 run_case 函数来并发执行测试用例任务；如果是模块级别，则调用 run_module 函数来并发执行测试模块任务；如果任务类型不符合要求，则抛出异常。

并发执行用例级别任务

run_case 函数用于并发执行用例级别的测试任务。首先获取并发执行任务数量的值，然后定义 gs 列表来存放协程对象。接下来，使用列表推导式创建多个协程对象，并将其添加到 gs 列表中。最后，调用 gevent.joinall 方法等待所有协程执行完毕。

并发执行模块级别任务

run_module 函数用于并发执行模块级别的测试任务。首先定义一个空字典 module_dict 用于存放不同模块下的测试用例。然后遍历所有的测试用例（通过 session.items 获取），并将其所属模块作为 key，将同一个模块下的测试用例列表作为 value 存储到 module_dict 中。接下来，使用列表推导式创建多个协程对象，并将其添加到 gs 列表中。最后，调用 gevent.joinall 方法等待所有协程执行完毕。

执行测试用例任务

run_task 函数用于执行单个测试用例任务。该函数接收一个测试用例列表参数 items，在函数内部使用一个 while 循环来不断取出 items 末尾的元素（即测试用例），并调用 item.ihook.pytest_runtest_protocol 方法来执行测试用例。执行完当前测试用例后继续取出下一个测试用例，直到 items 列表为空停止循环。

以上就是这段代码的主要逻辑思路。该代码实现了根据命令行选项参数来控制测试任务的类型和并发执行任务的数量，可以有效提升测试效率。同时，使用协程并发执行任务，可以更加高效地利用系统资源。

`session` 对象

在 Pytest 中，session 是一个全局的对象，代表着测试会话，在整个测试过程中只有一个。它包含了所有的测试信息，比如命令行参数、插件配置、测试用例等等。session 对象是在 Pytest 启动时创建的，并且在整个测试过程中始终存在。

`session.config`

session.config 是 session 对象中的一个属性，它是一个字典，包含了当前测试会话的所有配置信息，可以通过它来获取命令行选项参数等配置信息。在这段代码中，我们使用 session.config.getoption 方法来获取命令行选项参数的值，其实质是从 session.config 字典中查找对应的键值对。所以，session.config.getoption 中的 session 参数就是当前测试会话的 session 对象。

`session.items`

在 Pytest 中，session.items 是一个属性，它返回一个生成器对象，其中包含了当前测试会话中所有的测试项。测试项可以是测试模块、测试用例、测试类等，每个测试项都是通过 pytest_collection_modifyitems 钩子函数进行收集和修改的。

知识点补充

module_dict.setdefault(module, []).append(item)

module_dict 是一个字典对象，用来按模块名存储测试用例。在 run_module 函数中，我们使用 session.items 来遍历所有的测试项，将同一个模块下的测试用例存储到同一个列表中，并将该列表作为值，以模块名为键保存到 module_dict 中。具体地，setdefault 方法可以取出字典中给定键 module 对应的值，如果没有找到，就创建一个空列表并将其作为默认值返回再添加新项，然后将当前测试项 item 添加到这个列表中。

所以，module_dict.setdefault(module, []).append(item) 这一行代码的意思是：如果 module 在 module_dict 中不存在，则创建一个空列表，并将其作为值关联到该键上；接着，将当前测试项 item 添加到该列表中。如果 module 已经存在于 module_dict 中，直接取出对应的列表，并将当前测试项 item 添加到该列表中。这样就实现了按模块名存储测试用例的功能。

优化：支持传递不同的并发类型参数，通过线程池、进程池或协程并发执行用例

from typing import List
import pytest
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import gevent
from gevent import monkey
from multiprocessing import Manager
monkey.patch_all()


def pytest_addoption(parser):
    """
    用于添加命令行选项参数
    名称是固定的
    :param parser: Parser对象
    :return:
    """
    # 1.添加参数分组
    group = parser.getgroup('dev16', 'manual pytest plugin')

    # 2.添加参数和帮助信息
    group.addoption('--task', type=str, default='module',
                    help='Specify the type of task to be executed concurrently, you can fill in case or module, '
                         'the default is module')

    group.addoption('--concurrent', type=int, default=1,
                    help='Specify the number of concurrent tasks, the default is 1')

    group.addoption('--concurrent_type', type=str, default='coroutine',
                    help='Specify the concurrent execution mode, you can choose process, thread or coroutine. '
                         'The default is coroutine')


def pytest_runtestloop(session: pytest.Session) -> bool:
    """
    用于控制测试用例的执行
    名称是固定的
    :param session: 包含了当前会话的所有信息
    :return:
    """
    # 获取选项参数信息
    task_level = session.config.getoption('--task')
    print(f'任务类型为：{task_level}')

    concurrent = session.config.getoption('--concurrent')
    print(f'并发任务数：{concurrent}')

    concurrent_type = session.config.getoption('--concurrent_type')
    print(f'并发执行方式：{concurrent_type}')

    if task_level == 'case':
        run_case(session, concurrent, concurrent_type)
    elif task_level == 'module':
        run_module(session, concurrent, concurrent_type)
    else:
        raise Exception('The task type is wrong, it can only be case or module')

    return True


def run_in_thread(tasks: List, concurrent: int) -> List:
    with ThreadPoolExecutor(max_workers=concurrent) as executor:
        results = list(executor.map(_run_task, tasks))
    return results


def run_in_process(tasks: List, concurrent: int) -> List:
    with ProcessPoolExecutor(max_workers=concurrent) as executor:
        results = list(executor.map(_run_task, tasks))
    return results


def run_in_coroutine(tasks: List, concurrent: int):
    gs = [gevent.spawn(_run_task, task) for task in tasks]
    gevent.joinall(gs)
    return


def _run_task(task):
    item = task
    result = item.ihook.pytest_runtest_protocol(item=item, nextitem=None)
    return result


def run_case(session: pytest.Session, concurrent: int, concurrent_type: str):
    # 获取要执行的任务列表
    tasks = session.items[:]

    if concurrent_type == 'thread':
        results = run_in_thread(tasks, concurrent)
    elif concurrent_type == 'process':
        results = run_in_process(tasks, concurrent)
    else:  # concurrent_type == 'coroutine'
        run_in_coroutine(tasks, concurrent)
        results = []

    print(f"执行完成，共执行测试用例：{len(results)} 条")


def run_module(session: pytest.Session, concurrent: int, concurrent_type: str):
    # 定义一个字典用于保存不同模块的用例
    # 把模块作为key，把同一个模块下的测试用例列表作为value
    # {'test_demo01': [item, item], 'test_demo02': [item, item]}
    module_dict = {}
    for item in session.items:
        module = item.module
        module_dict.setdefault(module, []).append(item)
    print(module_dict)

    tasks_list = list(module_dict.values())
    tasks_count = sum(len(tasks) for tasks in tasks_list)

    if concurrent_type == 'thread':
        results = []
        with ThreadPoolExecutor(max_workers=concurrent) as executor:
            for tasks in tasks_list:
                results += executor.map(_run_task, tasks)
    elif concurrent_type == 'process':
        results = []
        with ProcessPoolExecutor(max_workers=concurrent) as executor:
            for tasks in tasks_list:
                results += executor.map(_run_task, tasks)
    else:  # concurrent_type == 'coroutine'
        new_tasks = []
        for tasks in tasks_list:
            new_tasks += tasks
        run_in_coroutine(new_tasks, concurrent)
        results = []

    print(f"执行完成，共执行测试用例：{tasks_count} 条")
    return results


if __name__ == "__main__":
    pytest.main(['-s', '--task=case', '--concurrent=10', '--concurrent_type=process'])

# 这段代码多进程有问题还在修改，待续,[运行环境为python3.8；pytest==5.2.1]

自定义一个并发执行pytest测试用例的插件

自定义一个并发执行pytest测试用例的插件

前提知识点

pytest_addoption() 函数

定义命令行选项参数

解析自定义参数

参数解释

item.ihook.pytest_runtest_protocol()

重写pytest_runtestloop() 方法

session 对象

session.config

session.items

知识点补充

优化：支持传递不同的并发类型参数，通过线程池、进程池或协程并发执行用例

评论

`pytest_addoption()` 函数

`item.ihook.pytest_runtest_protocol()`

`session` 对象

`session.config`

`session.items`