超全整理，Pytest自动化测试框架-多进程(pytest-xdist)运行总结...

---

前言

平常我们功能测试用例非常多时，比如有1千条用例，假设每个用例执行需要1分钟，如果单个测试人员执行需要1000分钟才能跑完

当项目非常紧急时，会需要协调多个测试资源来把任务分成两部分，于是执行时间缩短一半，如果有10个小伙伴，那么执行时间就会变成十分之一，大大节省了测试时间

为了节省项目测试时间，10个测试同时并行测试，这就是一种分布式场景

分布式执行用例的原则：
用例之间是独立的，没有依赖关系，完全可以独立运行；
用例执行没有顺序要求，随机顺序都能正常执行；
每个用例都能重复运行，运行结果不会影响其他用例；

背景：

我们日常的工作当中进行自动化测试编写的测试用例会非常多，测试用例一个一个的执行所需要花费的时间会很长。

想象一下如果开发改动一块代码，我们需要回归一下，这时候执行一下自动化用例需要花费一小时或者好几个小时的时间，这是我们无法容忍的。

为了解决这个问题，我们采用pytest的插件pytest-xdist来进行多进程的并发执行测试用例，大大的缩短测试用例的执行时间，提高效率。

并发运行测试用例：

1、安装pytest-xdist

pip install pytest-xdist
1

2、多进程并发执行测试用例：不支持多线程

pytest test_add.py -n NUM    # NUM表示并发的进程数
1

参数配置：

-n=* ：*代表进程数

说明：
①多cpu并行执行用例，直接加个-n参数即可，后面num参数就是并行数量，比如num设置为3
②-n auto ： 自动侦测系统里的CPU数目
③-n num ： 指定运行测试的处理器进程数

3、举例

项目结构如下：

代码：

# file_name: test_a.py
import pytest
import time


def test_a_01():
    print("----------------->>> test_a_01")
    time.sleep(1)
    assert 1


def test_a_02():
    print("----------------->>> test_a_02")
    time.sleep(1)
    assert 1


def test_a_03():
    print("----------------->>> test_a_03")
    time.sleep(1)
    assert 1


def test_a_04():
    print("----------------->>> test_a_04")
    time.sleep(1)
    assert 1


if __name__ == '__main__':
    pytest.main(["-s", "test_a.py"])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# file_name: test_b.py
import pytest
import time


def test_b_01():
    print("----------------->>> test_b_01")
    time.sleep(1)
    assert 1


def test_b_02():
    print("----------------->>> test_b_02")
    time.sleep(1)
    assert 1


def test_b_03():
    print("----------------->>> test_b_03")
    time.sleep(1)
    assert 1


def test_b_04():
    print("----------------->>> test_b_04")
    time.sleep(1)
    assert 1


if __name__ == '__main__':
    pytest.main(["-s", "test_b.py"])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

①正常运行以上代码，耗时：8.09s

②设置并行运行数量为4，耗时：3.48s，大大的缩短了测试用例的执行时间。

pytest-xdist分布式测试的原理

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

在xdist中，主是master，从是workers。

大致原理：

1、xdist会产生一个或多个workers，workers都通过master来控制。

2、每个worker负责执行完整的测试用例集，然后按照master的要求运行测试，而master机不执行测试任务。

pytest-xdist分布式测试的流程

第一步：创建worker

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

第二步：收集测试项用例

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

注意：分布式测试（pytest-xdist）方式执行测试时不会输出测试用例中的print内容，因为主机并不执行测试用例，pycharm相当于一个master。

第三步：master检测workers收集到的测试用例集

master接收到所有worker收集的测试用例集之后，master会进行一些完整性检查，以确保所有worker都收集到一样的测试用例集（包括顺序）。

如果检查通过，会将测试用例的ids列表转换成简单的索引列表，每个索引对应一个测试用例的在原来测试集中的位置。

这个方案可行的原因是：所有的节点都保存着相同的测试用例集。

并且使用这种方式可以节省带宽，因为master只需要告知workers需要执行的测试用例对应的索引，而不用告知完整的测试用例信息。

第四步：测试用例分发

–dist-mode选项

each：master将完整的测试索引列表分发到每个worker。

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

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

第五步：测试用例的执行

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

第六步：测试用例再分发（–dist-mode=load）

当workers开始/结束执行时，会把测试结果返回给master，这样其他pytest hook比如： pytest_runtest_protocol就可以正常执行。

master在worker执行完一个测试后，基于测试执行时长以及每个work剩余测试用例综合决定是否向这个worker发送更多的测试用例

第七步：测试结束

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

pytest实现多线程运行测试用例（pytest-parallel）

安装

pip install pytest-parallel
1

常用参数配置：

① --workers=n ：多进程运行需要加此参数， n是进程数。默认为1
② --tests-per-worker=n ：多线程需要添加此参数，n是线程数

如果两个参数都配置了，就是进程并行；每个进程最多n个线程，总线程数：进程数*线程数

注意：

①在windows上进程数永远为1。
②需要使用 if name == “main” :，在dos中运行会报错（即在命令行窗口运行测试用例会报错）

示例：

pytest test.py --workers 3 ：3个进程运行
pytest test.py --tests-per-worker 4：4个线程运行
pytest test.py --workers 2 --tests-per-worker 4 ：2个进程并行，且每个进程最多4个线程运行，即总共最多8个线程运行。

import pytest


def test_03():
    print('测试用例3操作')


def test_04():
    print('测试用例4操作')


if __name__ == "__main__":
    pytest.main(["-s", "test_b.py", '--workers=2', '--tests-per-worker=4'])
1
2
3
4
5
6
7
8
9
10
11
12
13

pytest-parallel与pytest-xdist对比说明：

① pytest-parallel 比 pytst-xdist 相对好用，功能支持多。
② pytst-xdist 不支持多线程；
③pytest-parallel 支持python3.6及以上版本，所以如果想做多进程并发在linux或者mac上做，在Windows上不起作用（Workers=1），如果做多线程linux/mac/windows平台都支持，进程数为workers的值。

一、Python编程入门到精通

二、接口自动化项目实战

三、Web自动化项目实战

四、App自动化项目实战

五、一线大厂简历

六、测试开发DevOps体系

七、常用自动化测试工具

八、JMeter性能测试

九、总结（尾部小惊喜）

无论遇到多少困难和挫折，都不能放弃自己的梦想。要相信自己有无限的潜力和能力，勇敢地追求自己的梦想，不断超越自我，最终就能够取得成功。

成功并不是一种目的，而是一种过程。勇敢地追求梦想，不断地超越自我，坚持不懈地努力下去，最终就能够实现自己的价值。

每个人都有自己独特的价值，要勇敢地追求自己的梦想，坚持不懈地努力下去，不断提升自己的能力，最终就能够实现自己的价值。