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

Heyniu · 2016年08月15日 · 最后由 coolfish 回复于 2017年01月18日 · 3598 次阅读

关于这个框架

设计初衷：

解决我们项目的接口测试痛点。从之前的 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

框架地址

23 个赞

如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作！

打赏支持

共收到 7 条回复时间点赞

测试小书童 #1 · 2016年08月15日

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

Heyniu #7 · 2016年08月15日 Author

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

淳罡 #6 · 2016年08月16日

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

Heyniu #5 · 2016年08月16日 Author

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

恺 #5 · 2016年11月07日

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

Heyniu #6 · 2016年11月08日 Author

#5 楼 @mads 适合的

2楼已删除

coolfish #8 · 2017年01月18日

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

需要登录后方可回复, 如果你还没有账号请点击这里注册。

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

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

相关链接

关于这个框架

本次更新

重试机制

代码片段

清理测试数据

代码片段

response 字段校验

代码片段

接口流程走向

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

关于接口回放的数据

框架的下一步

框架的更下一步

GitHub