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

相关链接

关于这个框架

设计初衷:

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

框架地址

发表评论

电子邮件地址不会被公开。 必填项已用*标注