当前位置：首页 > article >正文

高效执行自动化用例：分布式执行工具pytest-xdist实战

article 2025/2/21 4:07:53

在自动化测试领域，测试用例的执行速度往往决定了项目迭代的效率。你是否也曾因为测试用例数量庞大而苦恼，手动串行执行导致周期过长？别担心，pytest-xdist 这款分布式执行工具能让你轻松实现多进程并发执行，真正让测试效率翻倍！

那么，如何利用 pytest-xdist 进行分布式并发测试？它的核心原理是什么？又该如何配置和优化，才能让自动化用例高速、稳定地执行？

在介绍pytest-xdist时，不讲任何原理，需要看原理的请移至官方：https://pypi.org/project/pytest-xdist/

当我们自动化测试用例非常多的时候，一条条按顺序执行会非常慢，pytest-xdist的出现就是为了让自动化测试用例可以分布式执行，从而节省自动化测试时间，pytest-xdist是属于进程级别的并发。

pytest-xdist插件在测试过程中可以使我们的测试用例一起并行测试，运行情况是根据你运行环境存在多个CPU，运行过程中可以进行组合测试运行，从而缩短我们的测试时间。

核心原理与优势

多进程并发执行：pytest-xdist 通过启动多个进程来并行执行测试用例，每个进程运行一部分测试，从而大幅缩短整体测试时间。
负载均衡调度：它支持不同的分布策略（如 loadscope、loadfile），能根据测试用例的执行时间智能分配任务，确保各进程负载均衡。

1、pytest-xdist插件安装

只需要在终端中运行如下命令：

pip install pytest-xdist

2、pytest-xdist执行用例的条件

需满足以下条件：

每一条用例必须是独立的。用例之间没有依赖关系，用例可以完全独立运行【独立运行】
每一条用例没有特定的执行顺序，就是每条用例都要遵循随机执行【随机执行】
每条用例的测试结果不能影响到其他的测试用例。【不影响其他用例】

pytest-sdist 与 pytest-parallel的小区别

pytest-xdist在win中调用的是多进程，而parallel在win中调用的是多线程，win中只有单进程
pytest-xdist调用时，会使session执行多次，算是一个win的bug，但是官方给出的解决方案是，使用filelock锁
使用parallel时，在win中调用多线程，但是allure报告不支持多线程，allure是集成报告，根据session来集成，所以就是出现一个线程一个报告

3、pytest-xdist使用方法

pytest -n x

n：表示使用并行参数
x：表示需要启动多少个分布式，也即使用CPU的个数

n auto：可以自动检测到系统的CPU核数；从测试结果来看，检测到的是逻辑处理器的数量
使用auto等于利用了所有CPU来跑用例，此时CPU占用率会特别高

说明：建议最多使用1/2的CPU个数来进行执行，消耗资源太多，导致电脑太卡

接下来看实例：

在终端中分别输入并执行：pytest -vs、pytest -vs -n 2

从结果可以看出，不是用分布式运行测试用例，总共用时10.71秒

那么接下来我们使用分布式进行运行用例，看看他运行的时间：

可以看出使用分布式运行用例，同时有2个线程进行执行用例，时间为6.08秒，缩短了很多的时间。

4、pytest-xdist自定义执行模式

按照同一个作用域方法来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行:

--dist=loadscope #每个worker按类执行
示例：pytest -v -n 3 --dist=loadscope test_demo.py

按照同一个文件名来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行：

--dist=loadfile #每个worker按文件执行
示例：pytest -v -n 3 --dist = loadfile test_xdist.py test_xdist_02.py test_xdist_03.py

将每个用例，分别发给所有的执行器worker，相当于开了几个执行器worker，同一个用例就执行几遍：

--dist=each
示例：pytest -v -n 3 --dist = each test_xdist.p

将待运行的用例随机发给可用的执行器worker，用例执行顺序随机的，目前默认采用这种方式：

--dist=load 和 --dist==no
示例：pytest -v -n 3 --dist = load test_xdist.py

5、如何让scope=session的fixture在test session中仅仅执行一次

pytest-xdist是让每个worker进程执行属于自己的测试用例集下的所有测试用例，这意味着在不同进程中，不同的测试用例可能会调用同一个scope范围级别较高（例如session）的fixture，该fixture则会被执行多次，这不符scope=session的预期。

虽然pytest-xdist没有内置的支持来确保会话范围的夹具仅执行一次，但是可以通过使用锁定文件进行进程间通信来实现。

import pytest

from filelock import FileLock



@pytest.fixture(scope="session",autouse=True)

def login(tmp_path_factory, worker_id):

    # 如果是单机运行 则运行这里的代码块【不可删除、修改】

    if worker_id == "master":

        """

        【自定义代码块】

        这里就写你要本身应该要做的操作，比如：登录请求、新增数据、清空数据库历史数据等等

        """

        uuid_value = uuid.uuid1()

        token = uuid_value.hex

        print("fixture:请求登录接口，获取token", token)

        os.environ['token'] = token

        # 如果测试用例有需要，可以返回对应的数据，比如 token

        return token

    # 如果是分布式运行

    # 获取所有子节点共享的临时目录，无需修改【不可删除、修改】

    root_tmp_dir = tmp_path_factory.getbasetemp().parent

    # 【不可删除、修改】

    fn = root_tmp_dir / "data.json"

    # 【不可删除、修改】

    with FileLock(str(fn) + ".lock"):

        # 【不可删除、修改】

        if fn.is_file():

          # 缓存文件中读取数据，像登录操作的话就是 token 【不可删除、修改】

          token = json.loads(fn.read_text())

          print(f"读取缓存文件，token 是{token} ")

        else:

            """

            【自定义代码块】

             跟上面 if 的代码块一样就行

            """

            uuid_value = uuid.uuid1()

            token = uuid_value.hex

            print("fixture:请求登录接口，获取token", token)

            # 【不可删除、修改】

            fn.write_text(json.dumps(token))

            print(f"首次执行，token 是{token} ")

          # 最好将后续需要保留的数据存在某个地方，比如这里是 os 的环境变量

        os.environ['token'] = token

    return token