01声明
在介绍pytest-xdist时,不讲任何原理,需要看原理的请移至官方:pytest-xdist · PyPI
当我们自动化测试用例非常多的时候, 一条条按顺序执行会非常慢,pytest-xdist的出现就是为了让自动化测试用例可以分布式执行,从而节省自动化测试时间,pytest-xdist是属于进程级别的并发。
pytest-xdist插件在测试过程中可以使我们的测试用例一起并行测试,运行情况是根据你运行环境存在多个CPU,运行过程中可以进行组合测试运行,从而缩短我们的测试时间。
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)
deflogin(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
-
示例只需要执行一次login(因为它是只需要执行一次来定义配置选项,等等)
-
当第一次请求这个fixture时,则会利用FileLock仅产生一次fixture数据
-
当其他进程再次请求这个fixture时,则会从文件中读取数据
最后感谢每一个认真阅读我文章的人,礼尚往来总是要有的,虽然不是什么很值钱的东西,如果你用得到的话可以直接拿走:【文末自行领取】
这些资料,对于【软件测试】的朋友来说应该是最全面最完整的备战仓库,这个仓库也陪伴上万个测试工程师们走过最艰难的路程,希望也能帮助到你!