Python接口自动化（初版）
原创
框架设计
pytest+requests+allure
目录结构
utils ——工具类
data ——用例文件存放目录
conf ——配置类
scripts ——测试用例驱动
TestCase ——测试用例
start.py ——执行脚本
pytest.ini ——pytest配置文件添加配置文件
配置文件总是项目中必不可少的部分！
将固定不变的信息集中在固定的文件中
settings.py
项目中都应该有一个文件对整体的配置进行管理，我也在这个python项目中设置了此文件。
在项目conf目录创建settings.py文件，所有的配置信息写在这个文件里面。
import os
import datetime
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 关于Excel配置表中测试用例默认配置
FILE_NAME = "本地接口测试用例.xlsx"
FILES_PATH = os.path.join(BASE_DIR, "data", FILE_NAME)
#----初始化template cookies dict
COOKIES_DICT = {}
# -----allure报告相关
ALLURE_COMMAND = "allure generate {from_json_path} -o {save_to_path} --clean".format(
from_json_path=os.path.join(BASE_DIR, "report", "json_result"),
save_to_path=os.path.join(BASE_DIR, "report", "allure_result")
)
# ---日志相关
# 日志级别
LOG_LEVEL = "debug"
LOG_STREAM_LEVEL = "debug" # 屏幕输出流
LOG_FILE_LEVEL = "info" # 文件输出流
# 日志文件命名
LOG_FILE_NAME = os.path.join(BASE_DIR, "logs", datetime.datetime.now().strftime("%Y-%m-%d") + ".log")
# 邮件相关
# 第三方SMTP服务
MAIL_HOST = "smtp.qq.com" # 设置服务器
# 邮件相关
# 第三方SMTP服务
MAIL_HOST = "smtp.qq.com" # 设置服务器
MAIL_USER = "test@qq.com" # 用户名
MAIL_TOKEN = "hhhhhhhhm" # 授权码
# 设置收件人和发件人
SENDER = "test@qq.com"
RECEIVERS = ["test@qq.com", "test@163.com"] # 接收邮箱可以设置你的qq或者其它邮箱
# 邮件主题
THEME = "接口测试报告"
# 正文内容
SEND_CONTENT = "详情看附件"
# 附件的file name
SEND_FILE_NAME = "allure_report.zip"excel工具类
用来读取Excel中的用例数据。
我们在utils目录中新建一个文件
import xlrd
from conf import settings
class ExcelOperate:
def __init__(self, file_path, sheet_by_index=0):
self.book = xlrd.open_workbook(file_path)
self.sheet = self.book.sheet_by_index(sheet_by_index)
def get_excel(self):
"""
获取Excel表中接口测试用例
:return: 构造好的测试用例数据,格式：[{'case_num': '', 'title ': '', 'url': '',}]
dict()
"""
title = self.sheet.row_values(0) #获取excel中的标题栏，
print(type(title))
case_suite=[dict(zip(title, self.sheet.row_values(row))) for row in range(1, self.sheet.nrows)] #将标题和遍历取出的值一一对应，形成多个字典，放在用例列表中。
return case_suite用python的xlrd 模块对Excel文件进行了读取。再按照制定格式输出用例。
核心驱动类
用于读取用例中所有和请求相关的数据，并且处理依赖和写入cookies
"""
处理请求
ExcelHandler模块中读取测试用例,对用例字段进去处理，然后发送请求获取响应结果
"""
import json
import re
import requests
from conf import settings
from jsonpath_rw import parse # 第三方插件，需要pip install ,作用json数据提取
# jsonpath_rw官网：https://github.com/kennknowles/python-jsonpath-rw
from utils.ExcelHandler import ExcelOperate
from utils.LogHandler import logger
class RequestsOperate:
def __init__(self, current_case, all_excel_data_list):
"""
:param current_case: 用例列表中的一条条单独测试用例(字典形式),用来给request构造发送请求
:param all_excel_data_list: 全部用例列表
"""
self.current_case = current_case
self.all_excel_data_list = all_excel_data_list
def get_response_msg(self):
"""
发送请求并且获取结果
:return:
"""
return self._send_msg()
def _send_msg(self):
"""发请求"""
logger().info("正在向{0}发送请求,{1}".format(self.current_case.get("url"), self.current_case))
response = requests.request(
method=self.current_case.get("method"),
url=self.current_case.get("url"),
data=self._check_request_data(),
json=self._check_request_json(),
params=self._check_request_params(),
cookies=self._check_request_cookies(),
headers=self._check_request_headers(),
)
self._write_cookies(response)
return json.loads(self.current_case.get("except")), response.json()
def _check_request_headers(self):
"""
校验请求头，做携带cookies 和 数据依赖的问题
{
'user':'${}$',
'testfan-id':'ca447223-876e-46ba-9e45-f775335dfcbe'
}
:return:
"""
headers = self.current_case.get("headers")
if headers:
return self._operate_re_msg(headers)
else:
return {}
def _write_cookies(self, response):
""" 监测响应结果中是否含有cookies，有就保存起来"""
for item in self.all_excel_data_list:
if item.get("case_num") == self.current_case.get("case_num"):
item["temporary_response_cookies"] = response.cookies.get_dict()
if response.headers.get("Content-Type").lower() == "application/json":
item["temporary_response_json"] = response.json()
# 如果去请求头中取数据都存一份
item["temporary_request_headers"] = self.current_case.get("headers")
item["temporary_request_data"] = self.current_case.get("data")
item["temporary_request_json"] = self.current_case.get("json")
item["temporary_request_params"] = self.current_case.get("params")
# 所有去响应头中取值的都存一份
item["temporary_response_headers"] = response.headers
print(response.cookies.get_dict())
def _check_request_data(self):
"""处理请求的data参数，检查是否有依赖"""
data = self.current_case.get("data")
if data:
return self._operate_re_msg(data)
else:
return {}
def _check_request_json(self):
"""处理请求的data参数，检查是否有依赖"""
data = self.current_case.get("json")
if data:
return json.loads(data)
else:
return {}
def _check_request_params(self):
"""处理get请求的参数，检查是否有依赖"""
params = self.current_case.get("params")
if params:
return json.loads(params)
else:
return {}
def _check_request_cookies(self):
"""处理请求中的cookies"""
cookies_case_num = self.current_case.get("cookies")
if cookies_case_num: # 当前接口需要cookies
for item in self.all_excel_data_list:
if item.get("case_num") == cookies_case_num:
return item.get("temporary_response_cookies", {})
else:
return {}
def _operate_re_msg(self, parameter):
"""
正则校验,数据依赖的字段
:param parameter: 各种参数，如：data,header,params
:return:
"""
# 使用re 将提取依赖字段 {"testfan-token":"${neeo_001>response_json>data}$"}
if isinstance(parameter, dict):
json.dumps(parameter)
pattern = re.compile("\${(.*?)}\$") # 定义规则
rule_list = pattern.findall(parameter) # 按照规则匹配
if rule_list: # 该参数有数据依赖要处理
for rule in rule_list:
case_num, params, json_path = rule.split(">")
for line in self.all_excel_data_list:
print(line)
if line.get("case_num") == case_num:
temp_data = line.get("temporary_{0}".format(params))
if isinstance(temp_data, str):
temp_data = json.loads(temp_data)
match_list = parse(json_path).find(temp_data)
print(match_list)
if match_list:
match_data = [v.value for v in match_list][0]
# 将提取出来的值替换到原来规则,
parameter = re.sub(pattern=pattern, repl=match_data, string=parameter, count=1)
return json.loads(parameter)
else:
if isinstance(parameter, str):
parameter = json.loads(parameter)
return parameter
记录操作日志
日志，大家应该都很熟悉这个名词，就是记录代码中的动作。
在utils目录中新建loggerHandler.py文件。
这个文件就是我们用来在自动化测试过程中记录一些操作步骤的。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import logging
from conf import settings
class LoggerHandler:
""" 日志操作 """
_logger_level = {
'debug': logging.DEBUG,
'info': logging.INFO,
'warning': logging.WARNING,
'error': logging.ERROR,
'critical': logging.CRITICAL
}
def __init__(self, log_name, file_name, logger_level, stream_level='info', file_level='warning'):
self.log_name = log_name
self.file_name = file_name
self.logger_level = self._logger_level.get(logger_level, 'debug')
self.stream_level = self._logger_level.get(stream_level, 'info')
self.file_level = self._logger_level.get(file_level, 'warning')
# 创建日志对象
self.logger = logging.getLogger(self.log_name)
# 设置日志级别
self.logger.setLevel(self.logger_level)
if not self.logger.handlers:
# 设置日志输出流
f_stream = logging.StreamHandler()
f_file = logging.FileHandler(self.file_name)
# 设置输出流级别
f_stream.setLevel(self.stream_level)
f_file.setLevel(self.file_level)
# 设置日志输出格式
formatter = logging.Formatter(
"%(asctime)s %(name)s %(levelname)s %(message)s"
)
f_stream.setFormatter(formatter)
f_file.setFormatter(formatter)
self.logger.addHandler(f_stream)
self.logger.addHandler(f_file)
# print(1111111, self.logger.handlers)
@property
def get_logger(self):
return self.logger
def logger(log_name='接口测试'):
return LoggerHandler(
log_name=log_name,
logger_level=settings.LOG_LEVEL,
file_name=settings.LOG_FILE_NAME,
stream_level=settings.LOG_STREAM_LEVEL,
file_level=settings.LOG_FILE_LEVEL,
).get_logger
if __name__ == '__main__':
logger().debug('aaaa')
logger().info('aaaa')
logger().warning('aaaa')
logger().error('aaaa')
在终端中运行该文件，就看到命令行打印出了：
2022-03-12 18:49:59,371 接口测试 DEBUG aaaa
2022-03-12 18:49:59,371 接口测试 INFO aaaa
2022-03-12 18:49:59,371 接口测试 WARNING aaaa
2022-03-12 18:49:59,371 接口测试 ERROR aaaa然后在项目logs目录下生成了当日的日志文件。
allure报告类
用来定制allure的输出，处理打包发送报告邮件的需求
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
allure报告相关
"""
import os
import zipfile # 打包
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
import subprocess # 执行cmd命令
from conf import settings
from utils.LogHandler import logger
class AllureOperate:
def get_allure_report(self):
"""生成报告"""
# os.system(ALLURE_COMMAND) 运行cmd命令，但是不安全
logger().info("正在生成测试报告......")
subprocess.call(settings.ALLURE_COMMAND, shell=True)
logger().info("生成测试报告成功......")
def check_zip(self):
"""打包"""
try:
logger().info("正在打包测试报告......")
BASE_DIR = os.path.join(settings.BASE_DIR, "report")
start_zip_dir = os.path.join(BASE_DIR, "allure_result") # 要压缩文件夹的根路径
zip_file_name = 'allure_report.zip' # 为压缩后的文件起个名字
zip_file_path = os.path.join(BASE_DIR, zip_file_name)
f = zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED)
for dir_path, dir_name, file_names in os.walk(start_zip_dir):
# 要是不replace，就从根目录开始复制
file_path = dir_path.replace(start_zip_dir, '')
# 实现当前文件夹以及包含的所有文件
file_path = file_path and file_path + os.sep or ''
for file_name in file_names:
f.write(os.path.join(dir_path, file_name), file_path + file_name)
f.close()
logger().info("打包测试报告完成......")
except Exception as er:
logger().error("打包测试报告失败:{0}".format(er))
def send_mail(self):
"""发送邮件"""
# 第三方SMTP服务
mail_host = settings.MAIL_HOST # 设置服务器
mail_user = settings.MAIL_USER # 用户名
mail_pass = settings.MAIL_TOKEN # 口令
# 设置收件人和发件人
sender = settings.SENDER
receivers = settings.RECEIVERS # 接收邮箱可以设置你的qq或者其它邮箱
# 创建一个带附件的实例对象
message = MIMEMultipart()
# 邮箱主题、收件人、发件人
subject = settings.THEME # 邮件主题
message["Subject"] = Header(subject, "utf-8")
message["From"] = Header("{0}".format(sender), "utf-8")
message["To"] = Header("{0}".format(";".join(receivers)), "utf-8")
# 邮件正文内容
send_content = settings.SEND_CONTENT
content_obj = MIMEText(send_content, "plain", "utf-8") # 第一个参数为邮件内容
message.attach(content_obj)
# 构造附件
att = MIMEText(_text=self._get_zip_file(), _subtype="base64", _charset="utf-8")
att["Content-Type"] = "application/octet-stream"
att["Content-Disposition"] = 'attachment;filename="{}"'.format(settings.SEND_FILE_NAME) # 邮件附件中显示什么名字
message.attach(att)
try:
smtp_obj = smtplib.SMTP()
smtp_obj.connect(mail_host, 25) # 25 为 SMTP端口号
smtp_obj.login(mail_user, mail_pass)
smtp_obj.send_message(sender, receivers, message.as_string())
smtp_obj.quit()
logger().info("邮件发送成功")
print("邮箱发送成功")
except smtplib.SMTPException as er:
logger().error("email end error:{0}".format(er))
def _get_zip_file(self):
"""获取zip文件内容"""
with open(file=os.path.join(settings.BASE_DIR, "report", "allure_report.zip"), mode="rb") as f:
return f.read()
if __name__ == '__main__':
AllureOperate().get_allure_report()执行用例
以上我们已经编写完成了整个框架和测试用例。
我们新增一个模板用例方法，并利用@pytest.mark.parametrize来进行参数化
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest
import allure
from deepdiff import DeepDiff
from utils.AssertResult import AssertUtil
from utils.ExcelHandler import ExcelOperate
from utils.RequestsHandler import RequestsOperate
from utils.LogHandler import logger
from conf import settings
excel_data_list = ExcelOperate(settings.FILES_PATH, 3).get_excel()
class TestCase:
@pytest.mark.parametrize("item", excel_data_list)
def test_case(self, item):
logger().info("正在进行断言...")
except_data, result = RequestsOperate(current_case=item, all_excel_data_list=excel_data_list).get_response_msg()
allure.dynamic.title(item.get("title"))
allure.dynamic.description(
"<b style='color:red'>请求的url:</b>{0}<hr />"
"<b style='color:red'>预期值:</b>{1}<hr />"
"<b style='color:red'>实际执行结果:</b>{2}<hr />".format(item["url"], item["except"], result)
)
assert AssertUtil().assert_in_body(except_data,result)
logger().info("完成断言,{0}-{1}".format(except_data, result))接下来就是执行命令脚本：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import shutil
import pytest
from conf.settings import BASE_DIR
from utils.AllureHandler import AllureOperate
if __name__ == '__main__':
dir_path = os.path.join(BASE_DIR, "report", "json_result")
if os.path.isdir(dir_path):
# 执行用例前，首先清除上一次测试生成的json文件
shutil.rmtree(dir_path)
pytest.main() # 执行测试用例
# 生成allure测试报告
allure_obj = AllureOperate()
allure_obj.get_allure_report()
# 压缩文件
allure_obj.check_zip()
# 执行完毕后发送邮件
# allure_obj.send_mail()z
else:
os.makedirs(dir_path)
项目的report目录中生成了一个report.html文件。
这就是生成的测试报告文件。
执行结果：
用例：
我们采用的是Excel的驱动，用例的字段为case_num，title，url，method，params，data，json，cookies，headers，except
case_num代表用例编号，一定要唯一，作为接口依赖的用例定位。
title表明用例名称，在报告中会对应体现
url为请求地址，暂时需要写入host，后续会优化成只需地址，host由配置文件控制
method为请求方式，如果为get，有参数就在params填写。如果为post，有参数就在data或者json处填写
cookies处填写用例id，对应的用例就是获取cookies的来源
headers可以自己写入，也可以通过格式关联
except表示预期结果，只要返回值包含，就会认为用例保存成功，注意，布尔True or False需要小写
然后我们讲讲用例依赖，基本上只有headers，和参数会用到依赖。
我们依赖固定的写入格式为：
{"userName":"${neeo_002>request_params>userName}$","X-Auth-Token":"${neeo_001>response_headers>X-Auth-Token}$"}
如上，指的就是从用例1中取请求参数中的userName，以及用例3中的返回头中的X-Auth-Token
运行
安装依赖
pip install -r requirements.txt执行主文件
- 在项目根目录执行
start.py文件即可运行项目
集成jekins
之前文章也讲过，一样的路子来就好了
代码地址：
https://gitee.com/czhtest/p_autotest_api.git
待优化内容：
测试环境切换
数据库断言
多种预期结果
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录