使用pytest - 丸趣TV

行业资讯
服务器
云计算
使用 pytest-xdist 分布式插件如何保证 scope=session 的 fixture 在多进程运行情况下仍然能只运行一次

使用 pytest-xdist 分布式插件如何保证 scope=session 的 fixture 在多进程运行情况下仍然能只运行一次
发布时间：2021-12-04 09:13:25
来源：丸趣 TV
阅读：261
作者：柒染
栏目：云计算

这篇文章给大家介绍使用 pytest-xdist 分布式插件如何保证 scope=session 的 fixture 在多进程运行情况下仍然能只运行一次，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。

背景

使用 pytest-xdist 分布式插件可以加快运行，充分利用机器多核 CPU 的优势

将常用功能放到 fixture，可以提高复用性和维护性

做接口自动化测试的时候，通常我们会将登录接口放到 fixture 里面，并且 scope 会设置为 session，让他全局只运行一次

但是当使用 pytest-xdist 的时候，scope=session 的 fixture 无法保证只运行一次，官方也通报了这一问题

官方描述

pytest-xdist 的设计使每个工作进程将执行自己的测试集合并执行所有测试子集，这意味着在不同的测试过程中，要求高级范围的 fixture（如：session）将会被多次执行，这超出了预期，在某些情况下可能是不希望的

尽管 pytest-xdist 没有内置支持来确保 scope=session 的 fixture 仅执行一次，但是可以通过使用锁定文件进行进程间通信来实现

官方解决办法（直接套用就行）

import jsonimport pytestfrom filelock import FileLock

@pytest.fixture(scope= session)def session_data(tmp_path_factory, worker_id):if worker_id ==  master :# not executing in with multiple workers, just produce the data and let# pytest s fixture caching do its jobreturn produce_expensive_data()# get the temp directory shared by all workersroot_tmp_dir = tmp_path_factory.getbasetemp().parent
 fn = root_tmp_dir /  data.json with FileLock(str(fn) +  .lock ):if fn.is_file():
 data = json.loads(fn.read_text())else:
 data = produce_expensive_data()
 fn.write_text(json.dumps(data))return data

若某个 scope = session 的 fixture 需要确保只运行一次的话，可以用上面的方法，直接套用，然后改需要改的部分即可（这个后面详细讲解）

官方原话：这项技术可能并非在每种情况下都适用，但对于许多情况下，它应该是一个起点，在这种情况下，对于 scope = session 的 fixture 只执行一次很重要

后续栗子的代码项目结构

xdist+fixture(文件夹)
│ tmp(存放  allure  数据文件夹)
│ conftest.py
│ test_1.py
│ test_2.py
│ test_3.py
│ __init__.py │

test_1.py 代码

import osdef test_1(test):print(os  环境变量 ,os.environ[ token])print(test1  测试用例 , test)

test_2.py 代码

import osdef test_2(test):print(os  环境变量 ,os.environ[ token])print(test2  测试用例 , test)

test_3.py 代码

import osdef test_3(test):print(os  环境变量 ,os.environ[ token])print(test3  测试用例 , test)

未解决情况下的栗子 conftest.py 代码

import osimport pytestfrom random import random

@pytest.fixture(scope= session)def test():
 token = str(random())print(fixture: 请求登录接口，获取 token , token)
 os.environ[token] = tokenreturn token

运行命令

pytest -n 3 --alluredir=tmp

运行结果

scope=session 的 fixture 很明显执行了三次，三个进程下的三个测试用例得到的数据不一样，明显不会是我们想要的结果

使用官方解决方法的栗子 rt

#!/usr/bin/env python# -*- coding: utf-8 -*- __title__ = 
__Time__ = 2021/4/27 11:28
__Author__ =  小菠萝测试笔记
__Blog__ = https://www.cnblogs.com/poloyy/ import jsonimport osimport pytestfrom random import randomfrom filelock import FileLock
@pytest.fixture(scope= session)def test(tmp_path_factory, worker_id):#  如果是单机运行   则运行这里的代码块【不可删除、修改】if worker_id ==  master :【自定义代码块】  这里就写你要本身应该要做的操作，比如：登录请求、新增数据、清空数据库历史数据等等 token = str(random())print(fixture: 请求登录接口，获取 token , token)
 os.environ[token] = token#  如果测试用例有需要，可以返回对应的数据，比如  tokenreturn 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  的代码块一样就行 token = str(random())print(fixture: 请求登录接口，获取 token , token)# 【不可删除、修改】 fn.write_text(json.dumps(token))print(f 首次执行，token  是 {token}  )#  最好将后续需要保留的数据存在某个地方，比如这里是  os  的环境变量 os.environ[token] = tokenreturn token

运行命令