接口自动化测试框架(pytest+allure+aiohttp+ 用例自动生成)

近期准备优先做接口测试的覆盖,为此需要开发一个测试框架,经过思考,这次依然想做点儿不一样的东西。

  • 接口测试是比较讲究效率的,测试人员会希望很快能得到结果反馈,然而接口的数量一般都很多,而且会越来越多,所以提高执行效率很有必要
  • 接口测试的用例其实也可以用来兼做简单的压力测试,而压力测试需要并发
  • 接口测试的用例有很多重复的东西,测试人员应该只需要关注接口测试的设计,这些重复劳动最好自动化来做
  • pytest和allure太好用了,新框架要集成它们
  • 接口测试的用例应该尽量简洁,最好用yaml,这样数据能直接映射为请求数据,写起用例来跟做填空题一样,便于向没有自动化经验的成员推广 加上我对Python的协程很感兴趣,也学了一段时间,一直希望学以致用,所以http请求我决定用aiohttp来实现。 但是pytest是不支持事件循环的,如果想把它们结合还需要一番功夫。于是继续思考,思考的结果是其实我可以把整个事情分为两部分。 第一部分,读取yaml测试用例,http请求测试接口,收集测试数据。 第二部分,根据测试数据,动态生成pytest认可的测试用例,然后执行,生成测试报告。 这样一来,两者就能完美结合了,也完美符合我所做的设想。想法既定,接着 就是实现了。

第一部分(整个过程都要求是异步非阻塞的)

读取yaml测试用例

一份简单的用例模板我是这样设计的,这样的好处是,参数名和aiohttp.ClientSession().request(method,url,**kwargs)是直接对应上的,我可以不费力气的直接传给请求方法,避免各种转换,简洁优雅,表达力又强。

args:- post- /xxx/add
kwargs:-caseName: 新增xxxdata:name: ${gen_uid(10)}
validator:-json:successed: True

异步读取文件可以使用aiofiles这个第三方库,yaml_load是一个协程,可以保证主进程读取yaml测试用例时不被阻塞,通过await yaml_load()便能获取测试用例的数据

async def yaml_load(dir='', file=''):"""异步读取yaml文件,并转义其中的特殊值:param file::return:"""if dir:file = os.path.join(dir, file)async with aiofiles.open(file, 'r', encoding='utf-8', errors='ignore') as f:data = await f.read()data = yaml.load(data)# 匹配函数调用形式的语法pattern_function = re.compile(r'^\${([A-Za-z_]+\w*\(.*\))}$')pattern_function2 = re.compile(r'^\${(.*)}$')# 匹配取默认值的语法pattern_function3 = re.compile(r'^\$\((.*)\)$')def my_iter(data):"""递归测试用例,根据不同数据类型做相应处理,将模板语法转化为正常值:param data::return:"""if isinstance(data, (list, tuple)):for index, _data in enumerate(data):data[index] = my_iter(_data) or _dataelif isinstance(data, dict):for k, v in data.items():data[k] = my_iter(v) or velif isinstance(data, (str, bytes)):m = pattern_function.match(data)if not m:m = pattern_function2.match(data)if m:return eval(m.group(1))if not m:m = pattern_function3.match(data)if m:K, k = m.group(1).split(':')return bxmat.default_values.get(K).get(k)return datamy_iter(data)return BXMDict(data)

可以看到,测试用例还支持一定的模板语法,如${function}$(a:b)等,这能在很大程度上拓展测试人员用例编写的能力

http请求测试接口

http请求可以直接用aiohttp.ClientSession().request(method,url,**kwargs),http也是一个协程,可以保证网络请求时不被阻塞,通过await http()便可以拿到接口测试数据

async def http(domain, *args, **kwargs):"""http请求处理器:param domain: 服务地址:param args::param kwargs::return:"""method, api = argsarguments = kwargs.get('data') or kwargs.get('params') or kwargs.get('json') or {}# kwargs中加入tokenkwargs.setdefault('headers', {}).update({'token': bxmat.token})# 拼接服务地址和apiurl = ''.join([domain, api])async with ClientSession() as session:async with session.request(method, url, **kwargs) as response:res = await response_handler(response)return {'response': res,'url': url,'arguments': arguments}

收集测试数据

协程的并发真的很快,这里为了避免服务响应不过来导致熔断,可以引入asyncio.Semaphore(num)来控制并发

async def entrace(test_cases, loop, semaphore=None):"""http执行入口:param test_cases::param semaphore::return:"""res = BXMDict()# 在CookieJar的update_cookies方法中,如果unsafe=False并且访问的是IP地址,客户端是不会更新cookie信息# 这就导致session不能正确处理登录态的问题# 所以这里使用的cookie_jar参数使用手动生成的CookieJar对象,并将其unsafe设置为Trueasync with ClientSession(loop=loop, cookie_jar=CookieJar(unsafe=True), headers={'token': bxmat.token}) as session:await advertise_cms_login(session)if semaphore:async with semaphore:for test_case in test_cases:data = await one(session, case_name=test_case)res.setdefault(data.pop('case_dir'), BXMList()).append(data)else:for test_case in test_cases:data = await one(session, case_name=test_case)res.setdefault(data.pop('case_dir'), BXMList()).append(data)return resasync def one(session, case_dir='', case_name=''):"""一份测试用例执行的全过程,包括读取.yml测试用例,执行http请求,返回请求结果所有操作都是异步非阻塞的:param session: session会话:param case_dir: 用例目录:param case_name: 用例名称:return:"""project_name = case_name.split(os.sep)[1]domain = bxmat.url.get(project_name)test_data = await yaml_load(dir=case_dir, file=case_name)result = BXMDict({'case_dir': os.path.dirname(case_name),'api': test_data.args[1].replace('/', '_'),})if isinstance(test_data.kwargs, list):for index, each_data in enumerate(test_data.kwargs):step_name = each_data.pop('caseName')r = await http(session, domain, *test_data.args, **each_data)r.update({'case_name': step_name})result.setdefault('responses', BXMList()).append({'response': r,'validator': test_data.validator[index]})else:step_name = test_data.kwargs.pop('caseName')r = await http(session, domain, *test_data.args, **test_data.kwargs)r.update({'case_name': step_name})result.setdefault('responses', BXMList()).append({'response': r,'validator': test_data.validator})return result

事件循环负责执行协程并返回结果,在最后的结果收集中,我用测试用例目录来对结果进行了分类,这为接下来的自动生成pytest认可的测试用例打下了良好的基础

def main(test_cases):"""事件循环主函数,负责所有接口请求的执行:param test_cases::return:"""loop = asyncio.get_event_loop()semaphore = asyncio.Semaphore(bxmat.semaphore)# 需要处理的任务# tasks = [asyncio.ensure_future(one(case_name=test_case, semaphore=semaphore)) for test_case in test_cases]task = loop.create_task(entrace(test_cases, loop, semaphore))# 将协程注册到事件循环,并启动事件循环try:# loop.run_until_complete(asyncio.gather(*tasks))loop.run_until_complete(task)finally:loop.close()return task.result()

第二部分

动态生成pytest认可的测试用例

首先说明下pytest的运行机制,pytest首先会在当前目录下找conftest.py文件,如果找到了,则先运行它,然后根据命令行参数去指定的目录下找test开头或结尾的.py文件,如果找到了,如果找到了,再分析fixture,如果有session或module类型的,并且参数autotest=True或标记了pytest.mark.usefixtures(a...),则先运行它们;再去依次找类、方法等,规则类似。大概就是这样一个过程。
可以看出,pytest测试运行起来的关键是,必须有至少一个被pytest发现机制认可的testxx.py文件,文件中有TestxxClass类,类中至少有一个def testxx(self)方法。
现在并没有任何pytest认可的测试文件,所以我的想法是先创建一个引导型的测试文件,它负责让pytest动起来。可以用pytest.skip()让其中的测试方法跳过。然后我们的目标是在pytest动起来之后,怎么动态生成用例,然后发现这些用例,执行这些用例,生成测试报告,一气呵成。

# test_bootstrap.py
import pytestclass TestStarter(object):def test_start(self):pytest.skip('此为测试启动方法, 不执行')

我想到的是通过fixture,因为fixture有setup的能力,这样我通过定义一个scope为session的fixture,然后在TestStarter上面标记use,就可以在导入TestStarter之前预先处理一些事情,那么我把生成用例的操作放在这个fixture里就能完成目标了。

# test_bootstrap.py
import pytest@pytest.mark.usefixtures('te', 'test_cases')
class TestStarter(object):def test_start(self):pytest.skip('此为测试启动方法, 不执行')

pytest有个--rootdir参数,该fixture的核心目的就是,通过--rootdir获取到目标目录,找出里面的.yml测试文件,运行后获得测试数据,然后为每个目录创建一份testxx.py的测试文件,文件内容就是content变量的内容,然后把这些参数再传给pytest.main()方法执行测试用例的测试,也就是在pytest内部再运行了一个pytest!最后把生成的测试文件删除。注意该fixture要定义在conftest.py里面,因为pytest对于conftest中定义的内容有自发现能力,不需要额外导入。

# conftest.py
@pytest.fixture(scope='session')
def test_cases(request):"""测试用例生成处理:param request::return:"""var = request.config.getoption("--rootdir")test_file = request.config.getoption("--tf")env = request.config.getoption("--te")cases = []if test_file:cases = [test_file]else:if os.path.isdir(var):for root, dirs, files in os.walk(var):if re.match(r'\w+', root):if files:cases.extend([os.path.join(root, file) for file in files if file.endswith('yml')])data = main(cases)content = """
import allurefrom conftest import CaseMetaClass@allure.feature('{}接口测试({}项目)')
class Test{}API(object, metaclass=CaseMetaClass):test_cases_data = {}
"""test_cases_files = []if os.path.isdir(var):for root, dirs, files in os.walk(var):if not ('.' in root or '__' in root):if files:case_name = os.path.basename(root)project_name = os.path.basename(os.path.dirname(root))test_case_file = os.path.join(root, 'test_{}.py'.format(case_name))with open(test_case_file, 'w', encoding='utf-8') as fw:fw.write(content.format(case_name, project_name, case_name.title(), data.get(root)))test_cases_files.append(test_case_file)if test_file:temp = os.path.dirname(test_file)py_file = os.path.join(temp, 'test_{}.py'.format(os.path.basename(temp)))else:py_file = varpytest.main(['-v',py_file,'--alluredir','report','--te',env,'--capture','no','--disable-warnings',])for file in test_cases_files:os.remove(file)return test_cases_files

可以看到,测试文件中有一个TestxxAPI的类,它只有一个test_cases_data属性,并没有testxx方法,所以还不是被pytest认可的测试用例,根本运行不起来。那么它是怎么解决这个问题的呢?答案就是CaseMetaClass

function_express = """
def {}(self, response, validata):with allure.step(response.pop('case_name')):validator(response,validata)"""class CaseMetaClass(type):"""根据接口调用的结果自动生成测试用例"""def __new__(cls, name, bases, attrs):test_cases_data = attrs.pop('test_cases_data')for each in test_cases_data:api = each.pop('api')function_name = 'test' + apitest_data = [tuple(x.values()) for x in each.get('responses')]function = gen_function(function_express.format(function_name),namespace={'validator': validator, 'allure': allure})# 集成allurestory_function = allure.story('{}'.format(api.replace('_', '/')))(function)attrs[function_name] = pytest.mark.parametrize('response,validata', test_data)(story_function)return super().__new__(cls, name, bases, attrs)

CaseMetaClass是一个元类,它读取test_cases_data属性的内容,然后动态生成方法对象,每一个接口都是单独一个方法,在相继被allure的细粒度测试报告功能和pytest提供的参数化测试功能装饰后,把该方法对象赋值给test+api的类属性,也就是说,TestxxAPI在生成之后便有了若干testxx的方法,此时内部再运行起pytest,pytest也就能发现这些用例并执行了。

def gen_function(function_express, namespace={}):"""动态生成函数对象, 函数作用域默认设置为builtins.__dict__,并合并namespace的变量:param function_express: 函数表达式,示例 'def foobar(): return "foobar"':return:"""builtins.__dict__.update(namespace)module_code = compile(function_express, '', 'exec')function_code = [c for c in module_code.co_consts if isinstance(c, types.CodeType)][0]return types.FunctionType(function_code, builtins.__dict__)

在生成方法对象时要注意namespace的问题,最好默认传builtins.__dict__,然后自定义的方法通过namespace参数传进去。

后续(yml测试文件自动生成)

至此,框架的核心功能已经完成了,经过几个项目的实践,效果完全超过预期,写起用例来不要太爽,运行起来不要太快,测试报告也整的明明白白漂漂亮亮的,但我发现还是有些累,为什么呢?
我目前做接口测试的流程是,如果项目集成了swagger,通过swagger去获取接口信息,根据这些接口信息来手工起项目创建用例。这个过程很重复很繁琐,因为我们的用例模板已经大致固定了,其实用例之间就是一些参数比如目录、用例名称、method等等的区别,那么这个过程我觉得完全可以自动化。
因为swagger有个网页啊,我可以去提取关键信息来自动创建.yml测试文件,就像搭起架子一样,待项目架子生成后,我再去设计用例填传参就可以了。
于是我试着去解析请求swagger首页得到的HTML,然后失望的是并没有实际数据,后来猜想应该是用了ajax,打开浏览器控制台的时,我发现了api-docs的请求,一看果然是json数据,那么问题就简单了,网页分析都不用了。

import re
import os
import sysfrom requests import Sessiontemplate ="""
args:- {method}- {api}
kwargs:-caseName: {caseName}{data_or_params}:{data}
validator:-json:successed: True
"""def auto_gen_cases(swagger_url, project_name):"""根据swagger返回的json数据自动生成yml测试用例模板:param swagger_url::param project_name::return:"""res = Session().request('get', swagger_url).json()data = res.get('paths')workspace = os.getcwd()project_ = os.path.join(workspace, project_name)if not os.path.exists(project_):os.mkdir(project_)for k, v in data.items():pa_res = re.split(r'[/]+', k)dir, *file = pa_res[1:]if file:file = ''.join([x.title() for x in file])else:file = dirfile += '.yml'dirs = os.path.join(project_, dir)if not os.path.exists(dirs):os.mkdir(dirs)os.chdir(dirs)if len(v) > 1:v = {'post': v.get('post')}for _k, _v in v.items():method = _kapi = kcaseName = _v.get('description')data_or_params = 'params' if method == 'get' else 'data'parameters = _v.get('parameters')data_s = ''try:for each in parameters:data_s += each.get('name')data_s += ': \n'data_s += ' ' * 8except TypeError:data_s += '{}'file_ = os.path.join(dirs, file)with open(file_, 'w', encoding='utf-8') as fw:fw.write(template.format(method=method,api=api,caseName=caseName,data_or_params=data_or_params,data=data_s))os.chdir(project_)

现在要开始一个项目的接口测试覆盖,只要该项目集成了swagger,就能秒生成项目架子,测试人员只需要专心设计接口测试用例即可,我觉得对于测试团队的推广使用是很有意义的,也更方便了我这样的懒人。

【整整200集】超超超详细的Python接口自动化测试进阶教程合集,真实模拟企业项目实战

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/97091.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【ARM CoreLink 系列 5 -- CI-700 控制器介绍 】

文章目录 1.1 什么是 CI-700?1.1.1 关于 CI-7001.1.2 CI-700 特点1.2 全局配置参数1.2.1 寻址能力1.3 组件和配置1.3.1 CI-700 互联的结构1.3.2 Crosspoint(XP)1.3.3 外部接口1.4 组件(Components)1.1 什么是 CI-700? CI-700是一种AMBA 5 CHI互连,具有可定制的网状拓扑结构…

<HarmonyOS第一课>ArkTS开发语言介绍——闯关习题及答案

判断题 1.循环渲染ForEach可以从数据源中迭代获取数据,并为每个数组项创建相应的组件。( 对 ) 2.Link变量不能在组件内部进行初始化。( 对 ) 单选题 1.用哪一种装饰器修饰的struct表示该结构体具有组件化能力&#…

Maven 下载安装配置

Maven 下载安装配置 下载 maven maven 官网:https://maven.apache.org/ maven 下载页面:https://maven.apache.org/download.cgi 安装 maven 将下载的apache-maven.zip文件解压到安装目录 将加压后的apache-maven目录改名为maven maven 配置环…

Kafka 高可用

正文 一、高可用的由来 1.1 为何需要Replication 在Kafka在0.8以前的版本中,是没有Replication的,一旦某一个Broker宕机,则其上所有的Partition数据都不可被消费,这与Kafka数据持久性及Delivery Guarantee的设计目标相悖。同时Pr…

threejs 透明贴图,模型透明,白边

问题 使用Threejs加载模型时,模型出现了上面的问题。模型边缘部分白边,或者模型出现透明问题。 原因 出现这种问题是模型制作时使用了透明贴图。threejs无法直接处理贴图。 解决 场景一 模型有多个贴图时(一个透贴和其他贴图&#xff0…

笔记01:随机过程——随机游动

一、伯努利随机过程 1. n次伯努利实验中(x1)发生的总次数Yn: (二项分布) 2. 伯努利实验中事件第一次发生的时间L1: (几何分布) 3. n次伯努利实验中事件第k次发生的时间Lk: (帕斯卡分…

list的模拟实现

全部代码 #pragma once namespace HQJ {template<class T>struct __list_node//节点类{T __data;__list_node<T>* __prev;__list_node<T>* __next;__list_node(const T& x T())//由于不知道要存储的数据类型&#xff0c;使用匿名对象进行初始化:__data…

HiveServer2 Service Crashes(hiveServer2 服务崩溃)

Troubleshooting Hive | 5.9.x | Cloudera Documentation 原因&#xff1a;别人用的都好好的&#xff0c;我的集群为什么会崩溃&#xff1f; 1.hive分区表太多(这里没有说具体数量。) 2.并发连接太多&#xff0c;我记的以前默认是200个连接 3.复杂的hive查询访问表的的分区…

(一)实现一个简易版IoC容器【手撸Spring】

一、前言 相信大家在看本篇文章的时候&#xff0c;对IoC应该有一个比较清晰的理解&#xff0c;我在这里再重新描述下&#xff1a;它的作用就是实现一个容器将一个个的Bean&#xff08;这里的Bean可以是一个Java的业务对象&#xff0c;也可以是一个配置对象&#xff09;统一管理…

JetPack系列:001-JetPack概要介绍

文章目录 1. 概念介绍2. 主要内容2.1 框架库2.2 UI界面库 3. 核心思想4. 内容总结 本章回是一起Talk AndroidJetpack吧专栏的第一章回&#xff0c;本章回中主要介绍JetPack的基本概念和编程思想&#xff0c;同时也会介绍它的基础知识。闲话休提&#xff0c;请我们一起Talk Andr…

趣味工具箱小程序源码

趣味工具箱小程序源码&#xff0c;支持功能去水印&#xff0c;精选壁纸&#xff0c;图片压缩&#xff0c;文字生成二维码&#xff0c;图片加水印&#xff0c;模拟来电&#xff0c;手持弹幕&#xff0c;掷骰子…等 使用小工具&#xff0c;一个小程序有几十个功能。 源码下载&am…

面试打底稿⑥ 项目一的第二部分

简历原文 抽查部分 计算运费模块板块扩展性优化&#xff0c;采用责任链模式&#xff0c;实现不同地区间寄件的运费模板扩展的优化&#xff0c;为模块解耦&#xff0c;提高了系统的扩展性 短信模块设计&#xff0c;设计了短信发送数据模板的数据化存储&#xff0c;规范了发送短…

uniapp微信小程序蓝牙连接与设备数据对接

蓝牙连接并通信方法封装大致步骤。 初始化蓝牙并搜索&#xff1b;获取并启用service服务&#xff1b;数据读取和监听设备返回数据 需要使用uniapp官方提供api&#xff1a; // 关闭蓝牙 uni.closeBluetoothAdapter({}) // 打开蓝牙 uni.openBluetoothAdapter({}) // 搜索附近…

服务器or虚拟机安装SSH和虚拟机or服务器设置远程服务权限

第一步 服务器/虚拟机安装SSH工具,这是外部SSH终端连接服务器/虚拟机的第一步! sudo apt update && sudo apt upgrade#更新apt sudo apt install openssh-server#安装SSH工具 service ssh status#查看SSh运行状态 sudo systemctl enable --now ssh#运行SSH工具第二步…

Pytorch-学习记录-1-Tensor

1. 张量 (Tensor): 数学中指的是多维数组&#xff1b; torch.Tensor data: 被封装的 Tensor dtype: 张量的数据类型 shape: 张量的形状 device: 张量所在的设备&#xff0c;GPU/CPU requires_grad: 指示是否需要计算梯度 grad: data 的梯度 grad_fn: 创建 Tensor 的 Functio…

Google Cloud dataflow streaming job简介

简单介绍 首先 gcp 的dataflow 是1个ETL 组件, 它是基于Apache beam的 Apache beam 是1个较新的开源ETL 框架。 对于我们常用的ETL tool Spring batch 有下面的区别 spring batch 更偏向batch &#xff08;后台处理&#xff09;的ETL&#xff0c; 而apache beam 同时支持bat…

[python 刷题] 3 Longest Substring Without Repeating Characters

[python 刷题] 3 Longest Substring Without Repeating Characters 题目&#xff1a; Given a string s, find the length of the longest substring without repeating characters. 这到提要求找的是最长的&#xff0c;没有重复符号的子字符串 解题思路是用双指针哈希表&…

mysql MVCC(多版本并发控制)理解

最近看MVCC相关资料&#xff0c;这边做一个记录总结&#xff0c;方便后续理解。 目录 一、MVCC相关概念 二、MVCC实现原理 1.隐藏字段 2.undo log 3.Read View 4.MVCC的整体处理流程 5. RC&#xff0c;RR级级别下的innoDB快照读有什么不同 6.总结 一、MVCC相关概念 1…

华为云云耀云服务器L实例评测|部署私有网盘 Nextcloud

华为云云耀云服务器L实例评测&#xff5c;部署私有网盘 Nextcloud 一、云耀云服务器L实例介绍1.1 云服务器介绍1.2 产品规格1.3 应用场景 二、云耀云服务器L实例配置2.1 重置密码2.2 服务器连接2.3 安全组配置 三、部署 Nextcloud3.1 Nextcloud 介绍3.2 Docker 环境搭建3.3 Nex…

LeetCode 面试题 08.06. 汉诺塔问题

文章目录 一、题目二、C# 题解 一、题目 在经典汉诺塔问题中&#xff0c;有 3 根柱子及 N 个不同大小的穿孔圆盘&#xff0c;盘子可以滑入任意一根柱子。一开始&#xff0c;所有盘子自上而下按升序依次套在第一根柱子上(即每一个盘子只能放在更大的盘子上面)。移动圆盘时受到以…