forked from DxvLwRYF/apiAutoTest
优化请求前数据处理方法，支持用户使用jsonpath提取断言数据，支持多数据断言
This commit is contained in:
zy7y
parent
260d963fee
commit
876cc53ef4
|
|
@ -122,6 +122,9 @@ https://www.bilibili.com/video/BV1EE411B7SU?p=10
|
|||
2020/11/21 更新用例文档，合并文件对象，文件地址，优化文件上传处理方式
|
||||
|
||||
2020/11/21 config.yaml文件中新增request_headers 选项，默认header在此设置，优化test_api.py文件，整合read_file.py
|
||||
|
||||
2020/11/22 优化请求断言方法支持用户自定义提取响应自定内容，支持多数据断言，整合请求方法，优化测试启动方法，部分日志移除，修改预期结果处理
|
||||
|
||||
#### 博客园首发
|
||||
https://www.cnblogs.com/zy7y/p/13426816.html
|
||||
|
||||
|
|
|
|||
|
|
@ -7,59 +7,79 @@
|
|||
@ide: PyCharm
|
||||
@time: 2020/7/31
|
||||
"""
|
||||
from loguru import logger
|
||||
import requests
|
||||
|
||||
from tools import convert_json
|
||||
import requests
|
||||
from tools import allure_step, allure_title, logger, extractor
|
||||
from tools.data_process import DataProcess
|
||||
from tools.read_file import ReadFile
|
||||
|
||||
|
||||
class BaseRequest(object):
|
||||
def __init__(self):
|
||||
# 修改时间：2020年9月14日17:09
|
||||
# 确保，整个接口测试中，使用同一个requests.Session() 来管理cookie
|
||||
self.session = requests.Session()
|
||||
session = None
|
||||
|
||||
# 请求
|
||||
def api_send(self, method, url, parametric_key=None, data=None, file_obj=None, header=None):
|
||||
@classmethod
|
||||
def get_session(cls):
|
||||
if cls.session is None:
|
||||
cls.session = requests.Session()
|
||||
return cls.session
|
||||
|
||||
@classmethod
|
||||
def send_request(cls, case: list, env: str = 'dev') -> object:
|
||||
"""处理case数据，转换成可用数据发送请求
|
||||
:param case: 读取出来的每一行用例内容，可进行解包
|
||||
:param env: 环境名称 默认使用config.yaml server下的 dev 后面的基准地址
|
||||
return: 响应结果， 预期结果
|
||||
"""
|
||||
case_number, case_title, path, token, method, parametric_key, file_obj, data, expect = case
|
||||
# allure报告 用例标题
|
||||
allure_title(case_title)
|
||||
# 处理url、header、data、file、的前置方法
|
||||
url = ReadFile.read_config(f'$.server.{env}') + DataProcess.handle_path(path)
|
||||
allure_step('请求地址', url)
|
||||
header = DataProcess.handle_header(token)
|
||||
allure_step('请求头', header)
|
||||
data = DataProcess.handle_data(data)
|
||||
allure_step('请求参数', data)
|
||||
file = DataProcess.handler_files(file_obj)
|
||||
allure_step('上传文件', file)
|
||||
# 发送请求
|
||||
res = cls.send_api(url, method, parametric_key, header, data, file)
|
||||
allure_step('响应耗时(s)', res.elapsed.total_seconds())
|
||||
allure_step('响应内容', res.json())
|
||||
# 响应后操作
|
||||
if token == '写':
|
||||
DataProcess.have_token['Authorization'] = extractor(res.json(), ReadFile.read_config('$.expr.token'))
|
||||
allure_step('请求头中添加Token', DataProcess.have_token)
|
||||
DataProcess.save_response(case_number, res.json())
|
||||
allure_step('存储实际响应', DataProcess.response_dict)
|
||||
return res.json(), expect
|
||||
|
||||
@classmethod
|
||||
def send_api(cls, url, method, parametric_key, header=None, data=None, file=None) -> object:
|
||||
"""
|
||||
:param method: 请求方法
|
||||
:param url: 请求url
|
||||
:param parametric_key: 入参关键字， get/delete/head/options/请求使用params,
|
||||
post/put/patch请求可使用json（application/json）/data
|
||||
|
||||
:param data: 参数数据，默认等于None
|
||||
:param file_obj: 文件对象的地址， 单个文件直接放地址：/Users/zy7y/Desktop/vue.js
|
||||
多个文件格式：["/Users/zy7y/Desktop/vue.js","/Users/zy7y/Desktop/jenkins.war"]
|
||||
:param file: 文件对象
|
||||
:param header: 请求头
|
||||
:return: 返回json格式的响应
|
||||
:return: 返回res对象
|
||||
"""
|
||||
# 修改时间：2020年9月14日17:09
|
||||
session = self.session
|
||||
|
||||
files = DataProcess.handler_files(file_obj)
|
||||
session = cls.get_session()
|
||||
|
||||
if parametric_key == 'params':
|
||||
logger.info(f'{method, url, data, header}')
|
||||
res = session.request(method=method, url=url, params=data, headers=header)
|
||||
elif parametric_key == 'data':
|
||||
res = session.request(method=method, url=url, data=data, files=files, headers=header)
|
||||
res = session.request(method=method, url=url, data=data, files=file, headers=header)
|
||||
elif parametric_key == 'json':
|
||||
res = session.request(method=method, url=url, json=data, files=files, headers=header)
|
||||
res = session.request(method=method, url=url, json=data, files=file, headers=header)
|
||||
else:
|
||||
raise ValueError(
|
||||
'可选关键字为：get/delete/head/options/请求使用params, post/put/patch请求可使用json（application/json）/data')
|
||||
logger.info(f'请求方法:{method}，请求路径:{url}, 请求参数:{data}, 请求文件:{files}, 请求头:{header})')
|
||||
return res.json()
|
||||
|
||||
def api_front(self):
|
||||
"""请求api的前置处理方法"""
|
||||
pass
|
||||
|
||||
def api_position(self):
|
||||
"""请求api的后置处理方法"""
|
||||
pass
|
||||
logger.info(f'\n请求地址:{url}\n请求方法:{method}\n请求头:{header}\n请求参数:{data}\n响应数据:{res.json()}')
|
||||
return res
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -11,18 +11,15 @@ request_headers:
|
|||
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36
|
||||
|
||||
|
||||
# 实际响应jsonpath提取规则设置
|
||||
# 提取规则设置
|
||||
expr:
|
||||
# 提取token的jsonpath表达式
|
||||
token: $.data.token
|
||||
# 提取实际响应的断言数据jsonpath表达式，与excel中预期结果的数据进行比对用
|
||||
response: $.meta
|
||||
|
||||
file_path:
|
||||
case_data: ../data/case_data.xlsx
|
||||
report_data: ../report/data/
|
||||
report_generate: ../report/html/
|
||||
log_path: ../log/运行日志{time}.log
|
||||
test_case: ../data/case_data.xlsx
|
||||
report: ../report/
|
||||
log: ../log/run{time}.log
|
||||
|
||||
email:
|
||||
# 发件人邮箱
|
||||
|
|
@ -35,6 +32,4 @@ email:
|
|||
# 收件人邮箱
|
||||
addressees: ["收件人邮箱1","收件人邮箱2","收件人邮箱3"]
|
||||
title: 接口自动化测试报告(见附件)
|
||||
# 附件地址
|
||||
enclosures: ["../report/html/apiAutoTestReport.zip",]
|
||||
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -5,83 +5,43 @@
|
|||
@author: zy7y
|
||||
@file: test_api.py
|
||||
@ide: PyCharm
|
||||
@time: 2020/7/31
|
||||
@time: 2020/11/22
|
||||
@desc: 测试方法
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
|
||||
import pytest
|
||||
|
||||
from tools import logger
|
||||
from api.base_requests import BaseRequest
|
||||
from tools import *
|
||||
from tools.data_process import DataProcess
|
||||
from tools.read_file import ReadFile
|
||||
|
||||
base_url = ReadFile.read_config('$.server.dev')
|
||||
res_reg = ReadFile.read_config('$.expr.response')
|
||||
report_data = ReadFile.read_config('$.file_path.report_data')
|
||||
report_generate = ReadFile.read_config('$.file_path.report_generate')
|
||||
log_path = ReadFile.read_config('$.file_path.log_path')
|
||||
report = ReadFile.read_config('$.file_path.report')
|
||||
logfile = ReadFile.read_config('$.file_path.log')
|
||||
# 读取excel数据对象
|
||||
data_list = ReadFile.read_testcase()
|
||||
# 请求对象
|
||||
br = BaseRequest()
|
||||
logger.info(f'配置文件/excel数据/对象实例化，等前置条件处理完毕\n\n')
|
||||
cases = ReadFile.read_testcase()
|
||||
|
||||
|
||||
class TestApiAuto(object):
|
||||
# 启动方法
|
||||
def run_test(self):
|
||||
import os, shutil
|
||||
if os.path.exists('../report') and os.path.exists('../log'):
|
||||
class TestApi:
|
||||
|
||||
@classmethod
|
||||
def run(cls):
|
||||
if os.path.exists('../report'):
|
||||
shutil.rmtree(path='../report')
|
||||
shutil.rmtree(path='../log')
|
||||
# 日志存取路径
|
||||
logger.add(log_path, encoding='utf-8')
|
||||
pytest.main(args=[f'--alluredir={report_data}'])
|
||||
# 本地生成 allure 报告文件，需注意 不用pycharm等类似ide 打开会出现无数据情况
|
||||
os.system(f'allure generate {report_data} -o {report_generate} --clean')
|
||||
logger.add(logfile, enqueue=True, encoding='utf-8')
|
||||
logger.info('开始测试...')
|
||||
pytest.main(args=[f'--alluredir={report}/data'])
|
||||
os.system(f'allure generate {report}/data -o {report}/html --clean')
|
||||
logger.success('报告已生成')
|
||||
|
||||
# 直接启动allure报告（会占用一个进程，建立一个本地服务并且自动打开浏览器访问，ps 程序不会自动结束，需要自己去关闭）
|
||||
# os.system(f'allure serve {report_data}')
|
||||
logger.warning('报告已生成')
|
||||
|
||||
@pytest.mark.parametrize('case_number,case_title,path,is_token,method,parametric_key,file_obj,'
|
||||
'data,expect', data_list)
|
||||
def test_main(self, case_number, case_title, path, is_token, method, parametric_key, file_obj,
|
||||
data, expect):
|
||||
logger.debug(f'⬇️⬇️⬇️...执行用例编号:{case_number}...⬇️⬇️⬇️️')
|
||||
|
||||
allure_title(case_title)
|
||||
path = DataProcess.handle_path(path)
|
||||
allure_step('请求地址', base_url + path)
|
||||
allure_step('请求方式', method)
|
||||
header = DataProcess.handle_header(is_token)
|
||||
allure_step('请求头', header)
|
||||
data = DataProcess.handle_data(data)
|
||||
allure_step('请求数据', data)
|
||||
res = br.api_send(method=method, url=base_url + path, parametric_key=parametric_key, file_obj=file_obj,
|
||||
data=data, header=header)
|
||||
allure_step('实际响应结果', res)
|
||||
DataProcess.save_response(case_number, res)
|
||||
allure_step('响应结果写入字典', DataProcess.response_dict)
|
||||
# 写token的接口必须是要正确无误能返回token的
|
||||
if is_token == '写':
|
||||
DataProcess.have_token['Authorization'] = extractor(res, ReadFile.read_config('$.expr.token'))
|
||||
really = extractor(res, res_reg)
|
||||
allure_step('根据配置文件的提取响应规则提取实际数据', really)
|
||||
expect = convert_json(expect)
|
||||
allure_step('处理读取出来的预期结果响应', expect)
|
||||
allure_step('响应断言', (really == expect))
|
||||
|
||||
assert really == expect
|
||||
@pytest.mark.parametrize('case', cases)
|
||||
def test_main(self, case):
|
||||
response, expect = BaseRequest.send_request(case)
|
||||
# 断言操作
|
||||
DataProcess.assert_result(response, expect)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
TestApiAuto().run_test()
|
||||
|
||||
# 使用jenkins集成将不会使用到这两个方法（邮件发送/报告压缩zip）
|
||||
# from tools.zip_file import zipDir
|
||||
# from tools.send_email import send_email
|
||||
# zipDir(report_generate, report_zip)
|
||||
# send_email(email_setting)
|
||||
|
||||
|
||||
|
||||
|
||||
TestApi.run()
|
||||
|
|
@ -38,10 +38,8 @@ def rep_expr(content: str, data: dict, expr: str = '&(.*?)&') -> str:
|
|||
:param expr: 查找用的正则表达式
|
||||
return content： 替换表达式后的字符串
|
||||
"""
|
||||
logger.info(f'替换前内容{content}')
|
||||
for ctt in re.findall(expr, content):
|
||||
content = content.replace(f'&{ctt}&', str(extractor(data, ctt)))
|
||||
logger.info(f'替换后内容{content}')
|
||||
return content
|
||||
|
||||
|
||||
|
|
@ -67,7 +65,6 @@ def convert_json(dict_str: str) -> dict:
|
|||
dict_str = dict_str.replace('false', 'False')
|
||||
dict_str = eval(dict_str)
|
||||
logger.error(e)
|
||||
logger.info(f'{dict_str}, {type(dict_str)}')
|
||||
return dict_str
|
||||
|
||||
|
||||
|
|
@ -83,7 +80,3 @@ def allure_step(step: str, var: str) -> None:
|
|||
"""
|
||||
with allure.step(step):
|
||||
allure.attach(json.dumps(var, ensure_ascii=False, indent=4), step, allure.attachment_type.TEXT)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print(convert_json('["1","2"]'))
|
||||
|
|
@ -7,16 +7,13 @@
|
|||
@ide: PyCharm
|
||||
@time: 2020/11/18
|
||||
"""
|
||||
from tools import *
|
||||
from tools import logger, extractor, convert_json, rep_expr, allure_step
|
||||
from tools.read_file import ReadFile
|
||||
|
||||
|
||||
class DataProcess:
|
||||
response_dict = {}
|
||||
header = ReadFile.read_config('$.request_headers')
|
||||
|
||||
logger.error(header)
|
||||
|
||||
have_token = header.copy()
|
||||
|
||||
@classmethod
|
||||
|
|
@ -57,7 +54,7 @@ class DataProcess:
|
|||
实例- 单个文件: &file&D:
|
||||
"""
|
||||
if file_obj == '':
|
||||
return None
|
||||
return
|
||||
for k, v in convert_json(file_obj).items():
|
||||
# 多文件上传
|
||||
if isinstance(v, list):
|
||||
|
|
@ -79,5 +76,21 @@ class DataProcess:
|
|||
return
|
||||
data = rep_expr(variable, cls.response_dict)
|
||||
variable = convert_json(data)
|
||||
logger.info(f'最终的请求数据如下: {variable}')
|
||||
return variable
|
||||
|
||||
@classmethod
|
||||
def assert_result(cls, response: dict, expect_str: str):
|
||||
""" 预期结果实际结果断言方法
|
||||
:param response: 实际响应字典
|
||||
:param expect_str: 预期响应内容，从excel中读取
|
||||
return None
|
||||
"""
|
||||
expect_dict = convert_json(expect_str)
|
||||
index = 0
|
||||
for k, v in expect_dict.items():
|
||||
actual = extractor(response, k)
|
||||
index += 1
|
||||
logger.info(f'第{index}个断言,实际结果:{actual} | 预期结果:{v} \n断言结果 {actual == v}')
|
||||
allure_step(f'第{index}个断言', f'实际结果:{actual} = 预期结果:{v}')
|
||||
assert actual == v
|
||||
|
||||
|
|
|
|||
|
|
@ -10,14 +10,13 @@
|
|||
"""
|
||||
import yaml
|
||||
import xlrd
|
||||
from tools import *
|
||||
from tools import extractor
|
||||
|
||||
|
||||
class ReadFile:
|
||||
config_dict = None
|
||||
|
||||
@classmethod
|
||||
@logger.catch
|
||||
def get_config_dict(cls, config_path: str = '../config/config.yaml') -> dict:
|
||||
"""读取配置文件，并且转换成字典
|
||||
:param config_path: 配置文件地址， 默认使用当前项目目录下的config/config.yaml
|
||||
|
|
@ -30,7 +29,6 @@ class ReadFile:
|
|||
return cls.config_dict
|
||||
|
||||
@classmethod
|
||||
@logger.catch
|
||||
def read_config(cls, expr: str = '.') -> dict:
|
||||
"""默认读取config目录下的config.yaml配置文件，根据传递的expr jsonpath表达式可任意返回任何配置项
|
||||
:param expr: 提取表达式, 使用jsonpath语法,默认值提取整个读取的对象
|
||||
|
|
@ -39,14 +37,13 @@ class ReadFile:
|
|||
return extractor(cls.get_config_dict(), expr)
|
||||
|
||||
@classmethod
|
||||
@logger.catch
|
||||
def read_testcase(cls):
|
||||
"""
|
||||
读取excel格式的测试用例
|
||||
:return: data_list - pytest参数化可用的数据
|
||||
"""
|
||||
data_list = []
|
||||
book = xlrd.open_workbook(cls.read_config('$.file_path.case_data'))
|
||||
book = xlrd.open_workbook(cls.read_config('$.file_path.test_case'))
|
||||
# 读取第一个sheet页
|
||||
table = book.sheet_by_index(0)
|
||||
for norw in range(1, table.nrows):
|
||||
|
|
@ -55,11 +52,5 @@ class ReadFile:
|
|||
value = table.row_values(norw)
|
||||
value.pop(3)
|
||||
data_list.append(list(value))
|
||||
logger.info(f'{data_list}')
|
||||
return data_list
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
ReadFile.read_config()
|
||||
ReadFile.read_testcase()
|
||||
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
@time: 2020/8/3
|
||||
"""
|
||||
import yagmail
|
||||
from loguru import logger
|
||||
from tools import logger
|
||||
|
||||
|
||||
def send_email(setting):
|
||||
|
|
@ -25,7 +25,7 @@ def send_email(setting):
|
|||
"""
|
||||
yag = yagmail.SMTP(setting['user'], setting['password'], setting['host'])
|
||||
# 发送邮件
|
||||
yag.send(setting['addressees'], setting['title'], setting['contents'], setting['enclosures'])
|
||||
yag.send(setting['addressees'], setting['title'], setting['contents'])
|
||||
# 关闭服务
|
||||
yag.close()
|
||||
logger.info("邮件发送成功！")
|
||||
Loading…
Reference in New Issue