分布式测试插件pytest-xdist-摩杜云开发者社区

一、问题说明

在自动化测试的时候存在如下问题：

功能测试用例非常多时，比如有1000用例，假设,每个用例执行需要1分钟，如果单个测试人员执行需要1000分钟才能跑完，当项目非常紧急时，会需要协调多个测试资源来把任务分成两部分，于是执行时间缩短一半，如果有10个小伙伴，那么执行时间就会变成十分之一，大大节省了测试时间，为了节省项目测试时间，10个测试同时并行测试，这就是一种分布式场景
同样道理，当我们自动化测试用例非常多的时候，一条条按顺序执行会非常慢，pytest-xdist的作用就是为了让自动化测试用例可以分布式执行，从而节省自动化测试时间
pytest-xdist是属于进程级别的并发

二、分布式执行用例的设计原则

要使用pytest完成自动化测试，那么用例需要满足一下的要求：

用例之间是独立的，用例之间没有依赖关系，用例可以完全独立运行【独立运行】
用例执行没有顺序，随机顺序都能正常执行【随机执行】
每个用例都能重复运行，运行结果不会影响其他用例【不影响其他用例】

三、插件安装

在pytest中需要安装xdist

pip3 install pytest-xdist -i http://pypi.douban.com/simple/ --trusted-host pypi.douban.com

pytest-xdist通过测试执行模式对于pytest进行了扩展：

测试运行并行化执行：如果有多个CPU或主机，则可以将它们用于组合的测试运行。可以加快开发速度或使用远程计算机的特殊资源。
--looponfail：在子进程中重复运行测试。每次运行之后，pytest都会等到项目中的文件更改后再运行之前失败的测试。重复此过程，直到所有测试通过，然后再次执行完整运行。
跨平台覆盖：可以指定不同的Python解释程序或不同的平台，并在所有这些平台上并行运行测试。

四、案例说明

4.1.准备测试用例

准备测试的用例代码，这是运行代码的包结构，如下：

分布式测试插件pytest-xdist_执行时间

最外层的conftest.py代码如下：

@pytest.fixture(scope="session")
def login():
    print("*****************进行登录操作，返回账号和密码*****************")
    name = "Augus"
    pwd = "123312"
    yield name, pwd
    print("*****************退出登录操作*****************")

准备test_case01代码如下：

import time

class Test_pass01:

    def test_01(self):
        print('********用例01********')
        time.sleep(1)
        assert 1 == 1

    def test_02(self):
        print('********用例02********')
        time.sleep(1)
        assert 2 == 2

    def test_03(self):
        print('********用例03********')
        time.sleep(1)
        assert 2 == 2

准备test_case02代码如下：

import time

class Test_pass02:

    def test_04(self):
        print('********用例01********')
        time.sleep(1)
        assert 1 == 1

    def test_05(self):
        print('********用例02********')
        time.sleep(1)
        assert 2 == 2

    def test_06(self):
        print('********用例03********')
        time.sleep(1)
        assert 2 == 2

准备test_case03代码如下：

import time

class Test_pass03:

    def test_07(self):
        print('********用例01********')
        time.sleep(1)
        assert 1 == 1

    def test_08(self):
        print('********用例02********')
        time.sleep(1)
        assert 2 == 2

    def test_09(self):
        print('********用例03********')
        time.sleep(1)
        assert 2 == 2

4.2.执行用例

4.2.1.不使用分布式测试的命令和所需执行时间

执行命令：

pytest -sv .\test_case\

不使用分布式执行如下：

分布式测试插件pytest-xdist_测试用例_02

可以看到，执行一条用例大概1s（因为每个用例都加了 sleep(1) ），一共9条用例，总共运行9.19s；那么如果有1000条用例，执行时间就真的是1000s

4.2.2.使用分布式测试的命令和所需执行时间

命令如下：

pytest -sv -n auto .\test_case\

说明：

可以看到，最终运行时间只需要3.63s，我的电脑是真6核12线程
-n auto：可以自动检测到系统的CPU核数；从测试结果来看，检测到的是逻辑处理器的数量，即6核心
使用auto等于利用了所有CPU来跑用例，此时CPU占用率会特别高

分布式测试插件pytest-xdist_执行时间_03

执行结果如下：

分布式测试插件pytest-xdist_执行时间_04

观察到同样是执行9条用例，6个核心分布式测试，时间变成了3.63s，明显的减少了

4.2.3.指定需要多少个CPU来跑用例

通过参数-n，指定所需要的核心数：

pytest -sv -n 2 .\test_case\

执行如下：

分布式测试插件pytest-xdist_用例_05

之前6个核心执行需要3.63s，这次调整成2个核心执行，如下需要6.14s，时间略有增长：

分布式测试插件pytest-xdist_用例_06

4.3.4.pytest-xdist结合pytest-html生成测试报告

pytest-xdist是可以和pytest-html相结合生成测试报告，如下：

pytest -sv -n auto --html=report.html --self-contained-html

生成测试报告如下：

分布式测试插件pytest-xdist_用例_07

五、pytest-xdist执行顺序

pytest-xdist默认是无序执行的，可以通过 --dist 参数来控制顺序

5.1.--dist的执为loadscope

--dist=loadscope

将按照同一个模块module下的函数和同一个测试类class下的方法来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行，目前无法自定义分组，按类class分组优先于按模块module分组，执行如下：

分布式测试插件pytest-xdist_测试用例_08

5.2.--dist的执为loadfile

--dist=loadfile 参数指定了使用 loadfile 分发模式，它会按照文件加载的顺序执行测试用例。

--dist=loadfile

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

分布式测试插件pytest-xdist_执行时间_09

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

6.1.问题说明

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

需要注意的是当使用 pytest-xdist 的 -n 参数进行并发执行时，由于测试用例在不同的进程中执行，因此无法获取用例和fixtrue中的print中的执行信息，为了方便测试，修改conftest.py中的代码，利用selenium打开和关闭浏览器进行测试，如下：

from selenium import webdriver
import pytest

@pytest.fixture(scope="session",autouse=True)
def login():
    print("*****************打开浏览器*****************")
    driver = webdriver.Chrome()
    driver.implicitly_wait(10)
    driver.maximize_window()
    yield driver
    print("*****************关闭浏览器*****************")
    driver.quit()

启动执行所有的测试用例，

pytest -sv -n 2 --dist=loadscope .\test_case\

由于指定了2个核心，代码会在两个核心执行，会发现浏览器启动进行了两次，这样有多少个核心，就需要

6.2.如何解决上面的问题？

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

案例说明：

下面的示例只需要执行一次打开和关闭浏览器的操作（因为它是只需要执行一次来定义配置选项，等等）
当第一次请求这个fixture时，则会利用FileLock仅产生一次fixture数据
当其他进程再次请求这个fixture时，则会从文件中读取数据

代码实现如下：

我们使用 filelock 库来获取文件锁。如果锁文件 session.lock 存在，则说明其他进程已经获取了锁，当前进程直接返回。在获取锁之前，先创建一个空的锁文件。然后，使用 FileLock 上下文管理器在获得锁后执行 session fixture 中的浏览器操作。最后，在完成 session fixture 执行后，删除锁文件。
请确保已安装 filelock 库

pip install filelock

这样，无论使用几个并发进程运行测试，都能够确保 session fixture 只运行一次。

修改conftest.py代码如下：

import os
import pytest
from selenium import webdriver
from filelock import FileLock

@pytest.fixture(scope='session', autouse=True)
def browser(request):
    lock_file = "session.lock"

    # 如果锁文件存在，则说明其他进程已经获取了锁，当前进程直接返回
    if os.path.isfile(lock_file):
        return

    # 在获取锁之前，创建一个空的锁文件
    with open(lock_file, 'w') as f:
        pass

    with FileLock(lock_file):
        print("*****************打开浏览器*****************")
        driver = webdriver.Chrome()
        driver.implicitly_wait(10)
        driver.maximize_window()

        def close_browser():
            print("*****************关闭浏览器*****************")
            driver.quit()
        
        # 在pytest中想要做teardown的处理，除了使用带有yield的fixture函数，还可以直接添加终结器addfinalizer
        request.addfinalizer(close_browser)

    # 在完成 `session` fixture 执行后，删除锁文件
    if os.path.exists(lock_file):
        os.remove(lock_file)

    return driver

执行test_case包下所有用例，如下：

pytest -sv -n 2 --dist=loadscope .\test_case\

执行后，发现我们在多个线程中执行代码，但是上面的fixture中打开个关闭浏览器只执行了一次

七、pytest-xdist分布式测试的原理和执行流程

7.1.原理说明

xdist的分布式类似于一主多从的结构，master机负责下发命令，控制slave机；slave机根据master机的命令执行特定测试任务：

在xdist中，主是master，从是workers

原理说明：

xdist会产生一个或多个workers，workers都通过master来控制
每个worker负责执行完整的测试用例集，然后按照master的要求运行测试，而master机不执行测试任务

7.2.pytest-xdist分布式测试的流程

7.2.1.步骤一：创建worker

master会在总测试会话（test session）开始前产生一个或多个worker
master和worker之间是通过execnet和网关来通信的
实际编译执行测试代码的worker可能是本地机器也可能是远程机器

7.2.2.步骤二：收集测试项用例

每个worker类似一个迷你型的pytest执行器
worker会执行一个完整的test collection过程【收集所有测试用例的过程】
然后把测试用例的ids返回给master
master是不会执行任何测试用例集的

注意

所以为什么上面通过分布式测试的结果截图是没有输出用例的print内容，因为主机并不执行测试用例，pycharm相当于一个master

7.2.3.步骤三：master检测workers收集到的测试用例集

master接收到所有worker收集的测试用例集之后，master会进行一些完整性检查，以确保所有worker都收集到一样的测试用例集（包括顺序）
如果检查通过，会将测试用例的ids列表转换成简单的索引列表，每个索引对应一个测试用例的在原来测试集中的位置
这个方案可行的原因是：所有的节点都保存着相同的测试用例集
并且使用这种方式可以节省带宽，因为master只需要告知workers需要执行的测试用例对应的索引，而不用告知完整的测试用例信息

7.2.4.步骤四：测试用例分发

--dist-mode选项

each：master将完整的测试索引列表分发到每个worker
load：master将大约25%的测试用例以轮询的方式分发到各个worker，剩余的测试用例则会等待workers执行完测试用例以后再分发

注意

可以使用 pytest_xdist_make_scheduler 这个hook来实现自定义测试分发逻辑。

7.2.5.步骤五：测试用例的执行

workers 重写了 pytest_runtestloop ：pytest的默认实现是循环执行所有在test session这个对象里面收集到的测试用例
但是在xdist里, workers实际上是等待master为其发送需要执行的测试用例
当worker收到测试任务, 就顺序执行 pytest_runtest_protocol
值得注意的一个细节是：workers 必须始终保持至少一个测试用例在的任务队列里, 以兼容 pytest_runtest_protocol(item, nextitem) hook的参数要求，为了将 nextitem传给hook
worker会在执行最后一个测试项前等待master的更多指令
如果它收到了更多测试项, 那么就可以安全的执行 pytest_runtest_protocol , 因为这时nextitem参数已经可以确定
如果它收到一个 "shutdown"信号, 那么就将 nextitem 参数设为 None, 然后执行 pytest_runtest_protocol

7.2.6.步骤六：测试用例再分发（--dist-mode=load）

当workers开始/结束执行时，会把测试结果返回给master，这样其他pytest hook比如： pytest_runtest_protocol 和 pytest_runtest_protocol 就可以正常执行
master在worker执行完一个测试后，基于测试执行时长以及每个work剩余测试用例综合决定是否向这个worker发送更多的测试用例

7.2.7.步骤七：测试结束

当master没有更多执行测试任务时，它会发送一个“shutdown”信号给所有worker
当worker将剩余测试用例执行完后退出进程
master等待所有worker全部退出
然此时仍需要处理诸如 pytest_runtest_logreport 等事件