基于JSON的接口测试框架

梦里梦外; 2023-10-09 10:10 15阅读 0赞

> 更多学习资料请加QQ群: 822601020获取

**实现效果**  
![单接口/断言][1240]

![多接口/Session依赖/中间变量][Session]

**需求场景:**

1.  公司微服务接口使用数字签名的方式, 使用Postman调试接口每次都需要修改源码将验签临时关闭, 但是关闭后,其他微服务不能正常调用该服务
2.  从ELK中抓取到的接口参数, 需要Json格式化, 我通常使用Sublime Text + PrettyJson插件, 使用Postman调试, 有些表单类接口, 还需要把每个参数及值复制出来, 特别麻烦.

**实现原理:**

*  Sublime text3可以自定义编译系统
 *  Python方便的json数据处理
 *  使用Python实现签名算法
 *  requests发送接口请求

**实现目标:**

*  采用json文件数据驱动, 支持一个json包含多个接口
 *  支持GET/POST/DELETE/...方法
 *  支持表单类/Json类请求
 *  支持headers, cookies, params, data, files, timeout
 *  支持session依赖, 中间变量

**数据文件格式:**

*  Get请求

{
      "url": "http://127.0.0.1:5000/api/user/logout/?name=张三"
    }

或

{
      "url": "http://127.0.0.1:5000/api/user/logout/",
      "params": {"name": "张三"}
    }

*  Post表单请求:

{
      "url": "http://127.0.0.1:5000/api/user/login/",
      "data": {"name": "张三","password": "123456"}
    }

*  Json格式请求:

{
      "url": "http://127.0.0.1:5000/api/user/reg/",
      "headers": {"Content-Type": "application/json"},
      "data": { "name": "张991", "password": "123456"}
    }

或

{
      "url": "http://127.0.0.1:5000/api/user/reg/",
      "type": "json",
      "data": { "name": "张991","password": "123456"}
    }

*  断言格式(ASSERT):

{
      "url": "http://127.0.0.1:5000/api/user/login/",
      "data": {"name": "张三","password": "123456"},
      "ASSERT": ["response.status_code == 200", "'成功' in response.text"]
    }

*  需要签名(SIGN):

{   "url": "http://127.0.0.1:5000/api/user/delUser/",
        "type": "json",
        "data": {"name": "张三"},
        "SIGN": true
    }

*  多接口,session依赖, 及中间变量(STORE)

[
        {
            "url": "http://127.0.0.1:5000/api/user/getToken/",
            "method": "get",
            "params": {"appid": "136425"},
            "STORE": {"token": "response.text.split('=')[1]"}
        },
        {   "url": "http://127.0.0.1:5000/api/user/updateUser/?token=%(token)s",
            "headers": {"Content-Type": "application/json"},
            "data": {"name": "张三","password": "123456"}
        }
    ]

**工具实现:**

sign.py

import hashlib
    
    appid = user
    appsecret = NTA3ZTU2ZWM5ZmVkYTVmMDBkMDM3YTBi
    
    def md5(str):
        m = hashlib.md5()
        m.update(str.encode('utf8'))
        return m.hexdigest()  #返回摘要，作为十六进制数据字符串值
    
    
    def makeSign(params):
        if not isinstance(params, dict):
            print("参数格式不正确，必须为字典格式")
            return None
        if 'sign' in params:
            params.pop('sign')
        sign = ''
        for key in sorted(params.keys()):
            sign = sign + key + '=' + str(params[key]) + '&'
        sign = md5(sign + 'appsecret=' + appsecret)
        params['sign'] = sign
        return params

post\_json.py

import json
    import os
    import requests
    from sign import makeSign
    
    timeout = 10
    
    
    def postJson(path, timeout=60):
        try:
            with open(path, encoding='utf-8') as f:
                apis = json.load(f)
        except IOError as e:
            print(e)
    
        except json.decoder.JSONDecodeError:
            print("json文件格式有误")
            
        if isinstance(apis, dict):
            apis=[apis]
    
        session = requests.session()
        for api in apis:
    
            # 处理全局变量
            str_api = json.dumps(api)
            if '%' in str_api:
                api = json.loads(str_api % globals())
    
            # 获取接口相关项
            url = api.get('url')
            method = api.get('method')
            type = api.get('type')
            headers = api.get('headers')
            cookies = api.get('cookies')
            params = api.get('params')
            data = api.get('data')
            files = api.get('files')
    
            # 特定动作(系统标签)
            _sign = api.get('SIGN')  # 对数据进行签名
            _store = api.get('STORE') # 存储变量
            _assert = api.get('ASSERT') # 结果断言
            
            # 处理签名
            if _sign:
                data = makeSign(data)
    
            # 如果json请求，转换一下数据
            if type and type == 'json' or headers and 'json' in json.dumps(headers):
                data = json.dumps(data)
    
            # 根据method发送不同请求
            if not method or method.lower() == 'post': # 有data字段默认使用post方法
                response = session.post(url=url, headers=headers, cookies=cookies, params=params, data=data, files=files, timeout=timeout)  
            elif not data or method.lower() == 'get': # 没有data字段默认采用get方法
                  response = session.get(url=url, headers=headers, cookies=cookies, params=params, timeout=timeout)
            elif method.lower() == 'delete':
                response = session.delete(url=url, headers=headers, cookies=cookies, params=params, data=data, files=files, timeout=timeout)
            elif method.lower() == 'put':
                response = session.put(url=url, headers=headers, cookies=cookies, params=params, data=data, files=files, timeout=timeout)
            elif method.lower() == 'patch':
                response = session.patch(url=url, headers=headers, cookies=cookies, params=params, data=data, files=files, timeout=timeout)
            elif method.lower() == 'head':
                response = session.head(url=url, headers=headers, cookies=cookies, params=params, data=data, files=files, timeout=timeout)
            else:
                print("不支持当前方法")
            
            # 存储中间结果
            if _store:
                for key in _store:
                    globals()[key]=eval(_store[key])
            
            # 处理响应
            try:
                response_text = json.dumps(response.json(), ensure_ascii=False, indent=2)
            except json.decoder.JSONDecodeError:  # only python3
                try:
                    response_text = response.text
                except UnicodeEncodeError:
                    # print(response.content.decode("utf-8","ignore").replace('\xa9', ''))
                    response_text = response.content
            finally:
                pass
    
            # 处理断言
            status = "PASS"
            if _assert:
                assert_results = []
                for item in _assert:
                    try:
                        assert eval(item)
                        assert_results.append("PASS: <%s>" % item)
                    except AssertionError:
                        assert_results.append("FAIL: <%s>" % item)
                        status = "FAIL"
                    except Exception as e:
                        assert_results.append("ERROR: <%s>\n%s" % (item, repr(e)))
                        status = "ERROR"
            
            # 打印结果
            print("="*80)
            print("请求:")
            print("Url: %s\nHeaders: %s\nData: %s" % (url, headers, data if isinstance(data, str) else json.dumps(data)))
            print("-"*80)
            print("响应:")
            print(response_text)
            if _assert:
                print("-"*80)
                print("断言:")
                for assert_result in assert_results:
                    print(assert_result)
    
    def main():
        if len(sys.argv) != 2:
            print("缺少参数：json文件")
        else:
            postJson(sys.argv[1])
    
    main()

**集成到Sublime**

1.  Sublime Text3 -> Tools -> Build System -> New Build System  
    复制以下代码, 修改Python为你的Python3环境, 脚本路径为你的post\_json.py路径, 保存为PostJson.sublimet-build

*  Windows 系统

{
        "cmd": ["python3","C:/PostJson/post_json.py","$file"],
        "file_regex": "^[ ] *File \"(...*?)\", line ([0-9]*)",
        "selector": "source.json",
        "encoding":"cp936",
    }

*  Linux & Mac

{
        "cmd": ["/Library/Frameworks/Python.framework/Versions/3.6/bin/python3","-u","/Users/apple/Applications/PostJson/post_json.py","$file"],
        "file_regex": "^[ ] *File \"(...*?)\", line ([0-9]*)",
        "selector": "source.json",
        "env": {"PYTHONIOENCODING": "utf8"}, 
    }

**使用方法**

> 建议安装PrettyJson插件

1.  在Sublime text3中, 新建json文件,并保存为\*.json

{
      "url": "http://127.0.0.1:5000/api/user/login/",
      "data": {"name": "张三","password": "123456"}
    }

1.  按Ctrl + B运行

**项目路径**

> Github: [PostJson][]

补充了以下功能:

*  用例discover, 自动发现和批量执行
 *  生成报告
 *  命令行参数
    
     *  verbose: 显示级别: 1. 只显示用例结果 2. 只显示响应文本(默认) 3. 显示请求,响应及断言信息
     *  host: 指定服务器地址
     *  env: 指定环境
     *  timeout: 指定timeout
     *  html: 指定输出html报告路径
     *  log: 指定输出log路径
     *  collect-only: 只列出所有用例

**ToDo**

*  完善测试报告
 *  支持数据库对比
 *  支持执行Sql, Shell, Cmd命令等
 *  使用协程, 进行异步并发
 *  支持分布式执行
 *  支持定时执行
 *  进行持续集成

转载于:https://www.cnblogs.com/superhin/p/11455230.html

[1240]: https://upload-images.jianshu.io/upload_images/7575721-7579065783875ce6.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240
[Session]: https://upload-images.jianshu.io/upload_images/7575721-1ef3850c1da9fc21.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240
[PostJson]: https://github.com/hanzhichao/post_json.git