Pytest分布式测试插件之pytest-xdist的详细使用
1前言
当项目非常紧急时，会需要协调多个测试资源来把任务分成两部分，于是执行时间缩短一半，如果有10个小伙伴，那么执行时间就会变成十分之一，大大节省了测试时间
为了节省项目测试时间，10个测试同时并行测试，这就是一种分布式场景
pytest-xdist是属于进程级别的并发
2分布式执行用例的设计原则（重中之重的重点）
用例之间是独立的，用例之间没有依赖关系，用例可以完全独立运行【独立运行】
用例执行没有顺序，随机顺序都能正常执行【随机执行】
每个用例都能重复运行，运行结果不会影响其他用例【不影响其他用例】
3插件安装
| pip3 install pytest-xdist -i http://pypi.douban.com/simple/ --trusted-host pypi.douban.com |
4pytest-xdist通过一些独特的测试执行模式扩展了pytest
测试运行并行化：如果有多个CPU或主机，则可以将它们用于组合的测试运行。 这样可以加快开发速度或使用远程计算机的特殊资源。
--looponfail：在子进程中重复运行测试。 每次运行之后，pytest都会等到项目中的文件更改后再运行之前失败的测试。 重复此过程，直到所有测试通过，然后再次执行完整运行。
5快速入门
这是运行代码的包结构
14xdist是项目文件夹名称 │ conftest.py test_1.py │ __init__.py │ ├─test_51job │ │ conftest.py │ test_case1.py │ │ __init__.py │ ├─test_toutiao │ test_case2.py │ ├─test_weibo │ │ conftest.py │ test_case3.py │ │ __init__.py │ |
具体代码
最外层的conftest.py
# 外层conftest.py @pytest.fixture(scope="session")def login(): print("====登录功能，返回账号，token===") name = "testyy" token = "npoi213bn4" yield name, token print("====退出登录！！！====") |
最外层的test_1.py
import pytest @pytest.mark.parametrize("n", list(range(5)))def test_get_info(login, n): sleep(1) name, token = login print("***基础用例：获取用户个人信息***", n) print(f"用户名:{name}, token:{token}") |
test_51job包下的conftest.py
import pytest @pytest.fixture(scope="module")def open_51(login): name, token = login print(f"###用户 {name} 打开51job网站###") |
test_51job包下的test_case1.py
from time import sleep import pytest @pytest.mark.parametrize("n", list(range(5)))def test_case2_01(open_51, n): sleep(1) print("51job，列出所有职位用例", n) @pytest.mark.parametrize("n", list(range(5)))def test_case2_02(open_51, n): sleep(1) print("51job，找出所有python岗位", n) |
test_toutiao包下的test_case2.py
from time import sleep import pytest @pytest.mark.parametrize("n", list(range(5)))def test_no_fixture(login, n): sleep(1) print("==没有__init__测试用例，我进入头条了==", login) |
test_weibo包下的conftest.py
| import pytest @pytest.fixture(scope="function")def open_weibo(login): name, token = login print(f"&&& 用户 {name} 返回微博首页 &&&") |
test_weibo包下的test_case3.py
from time import sleep import pytest @pytest.mark.parametrize("n", list(range(5)))class TestWeibo: def test_case1_01(self, open_weibo, n): sleep(1) print("查看微博热搜", n) def test_case1_02(self, open_weibo, n): sleep(1) print("查看微博范冰冰", n) |
不使用分布式测试的命令和所需执行时间
| pytest -s |
可以看到，执行一条用例大概1s（因为每个用例都加了 sleep(1) ），一共30条用例，总共运行30s；那么如果有1000条用例，执行时间就真的是1000s
使用分布式测试的命令和所需执行时间
| pytest -s -n auto |
知识点
可以看到，最终运行时间只需要6s，我的电脑是真6核，假12核
-n auto：可以自动检测到系统的CPU核数；从测试结果来看，检测到的是逻辑处理器的数量，即假12核
使用auto等于利用了所有CPU来跑用例，此时CPU占用率会特别高
可以指定需要多少个CPU来跑用例
| pytest -s -n 2 |
pytest-xdist是可以和pytest-html很好的相结合
pytest -s -n auto --html=report.html --self-contained-html
6pytest-xdist的分布式测试原理
前言
xdist的分布式类似于一主多从的结构，master机负责下发命令，控制slave机；slave机根据master机的命令执行特定测试任务
在xdist中，主是master，从是workers
大致原理
1.xdist会产生一个或多个workers，workers都通过master来控制
2.每个worker负责执行完整的测试用例集，然后按照master的要求运行测试
0pytest-xdist的分布式测试流程
1第一步：创建worker
1.master会在总测试会话（test session）开始前产生一个或多个worker
2.master和worker之间是通过execnet和网关来通信的
3.实际编译执行测试代码的worker可能是本地机器也可能是远程机器
2第二步：收集测试项用例
1.每个worker类似一个迷你型的pytest执行器
2.worker会执行一个完整的test collection过程【收集所有测试用例的过程】
3.然后把测试用例的ids返回给master
4.master是不会执行任何测试用例集的
注意
所以为什么上面通过分布式测试的结果截图是没有输出用例的print内容，因为主机并不执行测试用例，pycharm相当于一个master
3第三步：master检车workers收集到的测试用例集
1.master接收到所有worker收集的测试用例集之后，master会进行一些完整性检查，以确保所有worker都收集到一样的测试用例集（包括顺序）
2.如果检查通过，会将测试用例的ids列表转换成简单的索引列表，每个索引对应一个测试用例的在原来测试集中的位置
3.这个方案可行的原因是：所有的节点都保存着相同的测试用例集
4.并且使用这种方式可以节省带宽，因为master只需要告知workers需要执行的测试用例对应的索引，而不用告知完整的测试用例信息
4第四步：测试用例分发
--dist-mode选项
each：master将完整的测试索引列表分发到每个worker
load：master将大约25%的测试用例以轮询的方式分发到各个worker，剩余的测试用例则会等待workers执行完测试用例以后再分发
注意
可以使用 pytest_xdist_make_scheduler 这个hook来实现自定义测试分发逻辑。
5第五步：测试用例的执行
1.workers 重写了 pytest_runtestloop ：pytest的默认实现是循环执行所有在test session这个对象里面收集到的测试用例
2.但是在xdist里, workers实际上是等待master为其发送需要执行的测试用例
3.当worker收到测试任务, 就顺序执行 pytest_runtest_protocol
4.值得注意的一个细节是：workers 必须始终保持至少一个测试用例在的任务队列里, 以兼容 pytest_runtest_protocol(item, nextitem) hook的参数要求，为了将 nextitem传给hook
5.worker会在执行最后一个测试项前等待master的更多指令
6.如果它收到了更多测试项, 那么就可以安全的执行 pytest_runtest_protocol , 因为这时nextitem参数已经可以确定
7.如果它收到一个 "shutdown"信号, 那么就将 nextitem 参数设为 None, 然后执行 pytest_runtest_protocol
6第六步：测试用例再分发（--dist-mode=load）
当workers开始/结束执行时，会把测试结果返回给master，这样其他pytest hook比如： pytest_runtest_protocol 和 pytest_runtest_protocol 就可以正常执行
master在worker执行完一个测试后，基于测试执行时长以及每个work剩余测试用例综合决定是否向这个worker发送更多的测试用例
7第七步：测试结束
当master没有更多执行测试任务时，它会发送一个“shutdown”信号给所有worker
当worker将剩余测试用例执行完后退出进程
master等待所有worker全部退出
然此时仍需要处理诸如 pytest_runtest_logreport 等事件.
TAG: