接口测试 Http 接口测试框架 (更新重试机制、清理测试数据、response 字段校验)

Heyniu · August 15, 2016 · Last by coolfish replied at January 18, 2017 · 3274 hits

相关链接

关于这个框架

设计初衷:

解决我们项目的接口测试痛点。从之前的1-2小时测试时间压缩到现在的1分钟以内,效率提升,效果显著

对于读者:

  • 完全符合你们项目
    • 庆幸的是入手即用,方便快捷
    • 需要注意的是,不要做伸手党,可以了解这个框架后,再去针对性的优化,让它更符合你们项目
  • 部分符合你们项目
    • 提取出部分内容,加入到你们项目中
  • 完全不符合你们项目
    • 提供一种思路,虽然可能没什么用,但是能了解到别人对这件事是怎么思考的

关于框架代码结构:

  • 非程序出身,能写成这样我自己还是比较满意的
  • 结构设计可能有些问题,但是不影响使用

    • 如发现设计不合理之处,欢迎指正
    • 先出成果再作优化
  • 此框架服务于测试流程、效率,是一个工具

  • 至少目前认为手工+自动化才是最符合我们项目

本次更新

  • 新增重试机制
  • 清理测试数据
  • response字段校验
  • 稳定性提升
重试机制

主要用于重试连接超时接口、response响应码非200的接口、其他异常情况

  • 待全部接口遍历后,读取写入本地的数据,取出已遍历的接口名
  • 读取遍历前的全部接口名,与上一步的数据求diff

    • 可以是fiddler录制的接口数据
    • 可以是上一次接口回归后写入本地的数据
  • diff接口再从遍历前的接口中取出相关数据,加入重试队列

  • 重试上述步骤,直至diff不存在或重试次数耗尽

代码片段
def retry11(app_type, retry=3):
"""
重试机制,默认3次
:param app_type: 0 >> A; 1 >> B; 2 >> C; 3 >> D
:param retry: 重试次数
:return:
"""

r1 = Retry(retry)
if len(r1.get_diff()) > 0:
print('发现diff接口,重试机制启动...')
r1.retry1(app_type)


class Retry(object):
def __init__(self, retry):
self.retry = retry
self.after_normal_sessions_path = '%s%s%s' % (
utils.GlobalList.SESSIONS_PATH, "\\Sessions\\", utils.GlobalList.HOST)

def __get_normal_after_sessions(self):
"""
获取遍历后正常(接口通过)的接口列表
:return:
"""

return utils.FileUtil.get_file_list(self.after_normal_sessions_path)

def __get_not_normal_after_sessions(self):
"""
获取需要人工验证的接口VerifyRequest
:return:
"""

return self.__get_check_after_sessions('VerifyRequest')

def __get_crash_after_sessions(self):
"""
获取程序异常接口ProgramCrash
:return:
"""

return self.__get_check_after_sessions('ProgramCrash')

def __get_unexpected_after_sessions(self):
"""
获取非预期接口Unexpected
:return:
"""

return self.__get_check_after_sessions('Unexpected')

def __get_field_change_after_sessions(self):
"""
获取字段改变接口FieldChange
:return:
"""

return self.__get_check_after_sessions('FieldChange')

def __get_check_after_sessions(self, sessions_type):
"""
获取需要检查的接口列表,已去重
:param sessions_type:
:return:
"""

path = '%s%s%s%s' % (self.after_normal_sessions_path, '\\Check\\', sessions_type, '.txt')
try:
l = open(path, encoding='utf-8').readlines()
sessions1 = ('%s%s' % (i.replace('\n', '')[::-1].split('/', 1)[0][::-1], '.txt') for i in l if
i.startswith('Request url: '))
return list(set(sessions1))
except FileNotFoundError:
return ()

def get_diff(self):
"""
获取diff接口
diff 接口包含类型
1.response 响应码 非200的接口
2.请求超时的接口
3.其他未知情况的接口(如 代码异常导致)
:return:
"""

before_sessions = utils.GlobalList.BEFORE_SESSIONS
after_sessions = []
normal_after_sessions = self.__get_normal_after_sessions()
not_normal_after_sessions = self.__get_not_normal_after_sessions()
unexpected_after_sessions = self.__get_unexpected_after_sessions()
field_change_after_sessions = self.__get_field_change_after_sessions()
crash_after_sessions = self.__get_crash_after_sessions()
if str(type(normal_after_sessions)) != "<class 'NoneType'>":
after_sessions.extend(normal_after_sessions)
if str(type(not_normal_after_sessions)) != "<class 'NoneType'>":
after_sessions.extend(not_normal_after_sessions)
if str(type(crash_after_sessions)) != "<class 'NoneType'>":
after_sessions.extend(crash_after_sessions)
if str(type(unexpected_after_sessions)) != "<class 'NoneType'>":
after_sessions.extend(unexpected_after_sessions)
if str(type(field_change_after_sessions)) != "<class 'NoneType'>":
after_sessions.extend(field_change_after_sessions)
return list(set(before_sessions).difference(set(after_sessions)))

def __get_diff_sessions(self):
"""
从本地磁盘读取diff接口sessions(request url; request header; ...)
pass: 一个接口多条session
:return:
"""

diff = self.get_diff()
for d in diff:
print('diff sessions: %s' % (d, ))
total_session = sessions.ReadSessions.ReadSessions().get_single_session(d)
if len(total_session) == 0:
print('发现录制异常接口:' + d)
print('执行移除操作,移除重试队列')
# 移除录制异常的接口
os.remove('%s%s%s%s%s' % (utils.GlobalList.SESSIONS_PATH, "\\Api\\", utils.GlobalList.HOST, "\\", d))
# 全局变量遍历前的全部接口也需要移除
utils.GlobalList.BEFORE_SESSIONS.remove(d)
else:
yield sessions.ReadSessions.ReadSessions().get_single_session(d)

def __will_request_sessions(self):
"""
将要重跑的sessions,把多个接口的多个session合并为一个列表
:return:
"""

s = self.__get_diff_sessions()
for i in s:
for j in i:
yield j

def __request_sessions(self, app_type):
"""
请求接口
:return:
"""

s = self.__will_request_sessions()
base.Request.thread_pool(app_type, s)

def retry1(self, app_type):
"""
重试的接口类型
1.response 响应码 非200的接口
2.请求超时的接口
3.其他未知情况的接口(如 代码异常导致)
注意:已经知道失败的接口不会再重试,目前是这样考虑的
:return:
"""

temp = self.retry
while self.retry > 0:
self.retry -= 1
# 请求接口
self.__request_sessions(app_type)
print('第%d次尝试请求diff...' % (temp - self.retry, ))
# 再次求差异化文件,还有diff继续,否则停止
if len(self.get_diff()) > 0 and self.retry > 0:
print('发现diff存在,继续尝试请求...')
continue
else:
break
print('diff请求完成...')
清理测试数据

目的:在于避免接口回放创建的数据(如发朋友圈)对线上数据的影响

思路:通过一个接口对(如:发朋友圈与删除朋友圈),即创建数据与删除数据完成操作

  • 配置文件填写创建数据接口名以及response body json中创建数据成功的字段
  • 配置文件填写删除数据的接口名以及request body中传入删除数据的字段
  • 移除删除接口不加入执行接口列表
  • 执行接口请求,请求完毕后读取写入的创建接口数据文件,提取出字段
  • 执行接口后提取出删除接口的url request body response body存入list
  • 接口执行完毕时遍历创建数据的list,调用相应的删除接口删除数据
代码片段
def clear_up(app_type):
"""
执行清理创建的数据
:param app_type:
:return:
"""

print("清理创建的接口数据...")
DelaySessions().request_sessions(app_type)


class DelaySessions(object):
def __init__(self):
self.create_session_path = '%s%s%s%s' % (utils.GlobalList.SESSIONS_PATH, "\\Sessions\\", utils.GlobalList.HOST, "\\")
self.delete_session_path = '%s%s%s%s' % (utils.GlobalList.SESSIONS_PATH, "\\Api\\", utils.GlobalList.HOST, "\\")
self.create_sessions_parameter_value = self.__get_all_session_create_parameter()

def __get_single_session_create_parameter(self, session_name):
"""
获取单个创建数据接口response body json中创建数据成功的字段
:return:
"""

total_session = []
parameter = []
file_path = '%s%s%s' % (self.create_session_path, session_name, '.txt')
if os.path.exists(file_path):
total_session = sessions.ReadSessions.ReadSessions().get_single_session_full_path(
'%s%s%s' % (self.create_session_path, session_name, '.txt'))
req = re.compile(r'"%s":[0-9]+' % (utils.GlobalList.CREATE_DICT[session_name], ))
for i in total_session:
parameter.append(re.findall(req, i[-1])[0].split(":")[-1])
return parameter

def __get_all_session_create_parameter(self):
"""
获取所有创建数据接口response body json中创建数据成功的字段
:return:
"""

create_parameter = {}
for i in utils.GlobalList.CREATE_DICT.keys():
create_parameter[i] = self.__get_single_session_create_parameter(i)
return create_parameter

def __get_single_session_delete_parameter(self, session_name):
"""
替换单个删除数据接口request body中传入删除数据的字段值i
:return:
"""

total_session = []
delete_session_name = utils.GlobalList.DELETE_DICT[session_name]
file_path = '%s%s%s' % (self.delete_session_path, session_name, '.txt')
if os.path.exists(file_path):
total_session = sessions.ReadSessions.ReadSessions().get_single_session_full_path(file_path)
req = re.compile(r'%s=[0-9]+' % (delete_session_name, ))
for i in total_session:
if len(i) == 4:
temp = re.findall(req, i[1])[0].split('=')[-1]
for j in utils.GlobalList.MAPPING_DICT.keys():
if j == session_name:
# 匹配对应的创建数据接口
create_session_name = utils.GlobalList.MAPPING_DICT[session_name]
# 取第一个值,用完删除
value = self.create_sessions_parameter_value[create_session_name][0]
i = str(i).replace(str(temp), value) # 替换value
l = list(self.create_sessions_parameter_value[create_session_name])
l.remove(value)
self.create_sessions_parameter_value[create_session_name] = l
return eval(i)

def __get_all_session_delete_parameter(self):
"""
替换全部删除数据接口request body中传入删除数据的字段值,并返回一个即将请求接口的list
:return:
"""

# 根据创建数据接口找到对应的删除接口,并拿出创建数据接口值的长度,多长就调用多少次对应删除接口
for i in self.create_sessions_parameter_value.keys():
for j in utils.GlobalList.MAPPING_DICT.keys():
if utils.GlobalList.MAPPING_DICT[j] == i:
for k in range(1, len(self.create_sessions_parameter_value[i]) + 1):
# 此处可优化 目前会读取多次文件
yield self.__get_single_session_delete_parameter(j)

def request_sessions(self, app_type):
"""
请求删除接口
:return:
"""

s = self.__get_all_session_delete_parameter()
base.Request.thread_pool(app_type, s)
print("接口数据清理完成!")
response字段校验

目的:校验是否缺失字段、校验字段类型是否改变、校验结果是否预期结果

思路:遍历response body json生成 <字段|字段类型的item> 存入list

  • 提取本地接口response body json存入list
  • 提取请求接口后的response body json存入list
  • 求diff list

pass:

  • 2个list可能存在字段相差太大的情况

    • 如一个接口本身返回了数据与无数据返回的情况
  • 解决方案

    • 求2个list的相似度
    • 100%相似则2个list长度相等用于验证字段类型改变
    • 相似度介于80%与100%之间(开区间)则判断是否缺失字段
    • 其他相似度则由于数据影响,没有太大比较意义,暂不考虑

收获:

字段类型改变(一般会导致客户端崩溃,当然客户端容错机制也没做到位)

Diff: ['CircleUserName|int', 'CircleUserName|str']
"CircleId":6420,"CircleUserName":0,"PostDate":""
"CircleId":6152,"CircleUserName":"虾米","PostDate":"2016年07月22日"

再细化的response字段校验

  • 目前的字段校验是抽象的,适用全部接口

  • 后面单个接口自动化时再针对单个接口的更细的字段校验

代码片段

遍历json取出字段及其类型

def decode_json(self, json_data):
"""
解析json并返回对应的key|value
:param json_data: json数据源
:return: 返回json各字段以及字段值
"""

try:
data = json.loads(json_data)
except Exception as e:
print(e)
print("JSON format error")
return []
self.__iterate_json(data)
return self.json_list


def __iterate_json(self, json_data, i=0):
"""
遍历json
:param i: 遍历深度
:param json_data: json数据源
:return: 返回json各字段以及字段值
"""

if isinstance(json_data, dict):
for k in json_data.keys():
self.json_list.append('%s|%s' % (k, str(type(json_data[k])).split("'")[1]))
if str(type(json_data[k])).startswith("<class 'list'>"):
if len((json_data[k])) and isinstance(json_data[k][0], dict):
self.__iterate_json(json_data[k][0], i=i + 1)
if isinstance(json_data[k], dict):
self.__iterate_json(json_data[k], i=i + 1)
else:
print("JSON format error")

接口流程走向

接口回归测试启动...
清理测试数据...
读取配置文件中...
读取接口数据中...
接口请求中,请等待...
http://a-b.test.c.com/api/Circle/AddCancelCollectCircle
....................................................
http://a-b.test.c.com/api/GroupActivity/UploadActivityImage
http://a-b.test.c.com/api/photo/UploadImage
RequestException url: http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
HTTPConnectionPool(host='http://a-b.test.c.com', port=80): Read timed out. (read timeout=30)
IndexError url:
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
接口请求完成!
发现diff接口,重试机制启动...
1次尝试请求diff...
diff sessions: GetDemandKnockSourceListV4.txt
diff sessions: GetSecondHouseTopic.txt
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
http://a-b.test.c.com/api/SecondHouseSource/GetSecondHouseTopic
RequestException url: http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
HTTPConnectionPool(host='http://a-b.test.c.com', port=80): Read timed out. (read timeout=30)
IndexError url:
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
发现diff存在,继续尝试请求...
2次尝试请求diff...
diff sessions: GetDemandKnockSourceListV4.txt
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
RequestException url: http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
HTTPConnectionPool(host='http://a-b.test.c.com', port=80): Read timed out. (read timeout=30)
IndexError url:
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
发现diff存在,继续尝试请求...
3次尝试请求diff...
diff sessions: GetDemandKnockSourceListV4.txt
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
RequestException url: http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
HTTPConnectionPool(host='http://a-b.test.c.com', port=80): Read timed out. (read timeout=30)
IndexError url:
http://a-b.test.c.com/api/Demand/GetDemandKnockSourceListV4
diff请求完成...
正在整理创建的数据...
清理创建的接口数据...
http://a-b.test.c.com/api/Circle/DeleteContent
http://a-b.test.c.com/api/Group/DeleteAnnouncement
http://a-b.test.c.com/api/RentDemand/DeleteRentDemandById
http://a-b.test.c.com/api/GroupFile/DeleteGroupFile
http://a-b.test.c.com/api/GroupDynamic/DeleteGroupDynamic
http://a-b.test.c.com/api/Demand/DeleteDemandById
http://a-b.test.c.com/api/GroupActivity/DeleteGroupActivity
接口数据清理完成!
测试报告准备中...
接口回归测试完成!
耗时: 125s

请求接口后写入本地的数据说明

FieldChange >> 字段改变的接口写入该文件
ProgramCrash >> 程序异常接口写入该文件
Unexpected >> 未达到预期字段校验的接口写入该文件
VerifyRequest >> 需要再次确认的接口写入该文件
GetUserInfoV2 >> 正常接口(一个接口一个文件)

关于接口回放的数据

  • 第一次的数据来自fiddler录制
  • 第二次及以后的数据来自第一次请求写入本地的正常接口文件 + fiddler继续录制需要检查的接口
  • 一般来说不要一直拿第一次fiddler录制的数据使用,目的在于测试接口对各个客户端版本的兼容情况
  • 接口回放可以跑线上运行的客户端全部版本接口(一个版本一套接口)

框架的下一步

  • 优雅的Html报告
  • 邮件通知
  • 持续集成
  • 测试数据另存,方便后续查阅

框架的更下一步

  • 接口压测
  • 接口自动化(api测试)pass:目前只是回归验证
  • 简单的GUI界面

GitHub

框架地址

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
共收到 7 条回复 时间 点赞

已经关注,有个疑问,如何给没有代码能力的测试人员做接口测试?

Heyniu #2 · August 15, 2016 作者

#1楼 @lose 后面考虑增加GUI界面,这样就不用关注代码了,但是牺牲了灵活性,增加了工作量,后面看需求吧

mac上用什么工具可以把请求自动保存到本地。

Heyniu #4 · August 16, 2016 作者

#3楼 @wanxi3 fiddler好像也有mac版本,你搜下

这个接口框架适用于app跟web吧

#5楼 @mads 适合的

7Floor has been deleted

#1楼 @lose 我觉得 不会可以学习

需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up