快速搞定API自动化

赵云龙龙

发布于 2020-11-05 20:21:28

6990

发布于 2020-11-05 20:21:28

文章被收录于专栏：python爱好部落python爱好部落

在自动化测试中，个人觉得API是比较好测的了。首先它比较稳定，不像UI总是变化。只要对接口比较熟悉，就能通过传递不同的参数，或者业务组合，就能验证不同的场景了。

看过别人写的不少的框架，有用excel的，有用httprunner的。我这里用数据处理的方法来做API自动化测试。

首先是要写case, 这里用代理来录制。

先用mitmdump来抓取包，无论是自动化跑，或者手工操作UI，就能得到所有的接口。

import mitmproxy.http
import json
import csv
import os


class Follower():

    def response(self, flow: mitmproxy.http.HTTPFlow):
        # url = flow.request.url
        # with open("/Users/anderson/Downloads/log.txt", "a+", encoding="utf-8") as fp:
        #     fp.write(str(url) + '\n')

        if '/api/' in flow.request.path:
            request_name = (flow.request.path).split("/")[-1]
            host = flow.request.host
            method = flow.request.method
            request_body = flow.request.get_text()
            request_path = flow.request.path

            response_code = flow.response.status_code
            response_body = json.loads(flow.response.text)

            api_content = [request_name, host, response_code, method, request_path, request_body, response_body]

            csvfile = open("/Users/anderson/Downloads/apirecode.csv", "a+")
            csvread = csv.writer(csvfile)
            csvread.writerow(api_content)


addons = [
    Follower()
]


def run():
    print("将会启动mitmproxy服务")
    os.system("mitmdump -q -s trackhost.py -p 8889")  # -q 屏蔽 mitmdump 默认的控制台日志，只显示自己脚本中的 -s 入口脚本文件


if __name__ == '__main__':
    run()

得到文件如下：

这样就可以组合接口，设计测试测试用例了，将参数改好。这里设置不同的测试环境，对应不同的账号来获取token, 替换掉。然后就可以批量发了。

from sys import argv
import pandas as pd
import requests
import jsonpath
import json
import re
import arrow
from jsonschema import validate
env = "stgcn"
# print(env)
appversion = "2.1.9"


date = arrow.now().format('MM-DD HH-mm-ss')


def get_hosts(env):
    envs = {
        "uatus": "https://uat.cn",
        "uatcn": "https://uat-us.cn",
        "qaus": "https://qa.com",
        "qacn": "https://qa.com.cn",
        "stgus": "https://stg.com",
        "stgcn": "https://stg.com.cn"
    }[env]
    return envs


def get_accounts(env):
    accounts = {
        "uatus": "test97894",
        "uatcn": "test25646",
        "qaus": "test181612",
        "qacn": "test15043",
        "stgus": "test14708",
        "stgcn": "test6294"
    }[env]
    return accounts


login_json_data = {
    'serviceRequest': {
        'appVersion': appversion,
        'userName': get_accounts(env),
        'password': 'wodemingzihaochang',
        'platform': 'pc',
        'productId': 400099
    }
}
login_url = "{}/services/api/mobile/service/login"

r = requests.session()
r.verify = False
response = r.post(url=login_url.format(get_hosts(env)), json=login_json_data)
print(response.json())

token = jsonpath.jsonpath(response.json(), '$..token')[0]
session = jsonpath.jsonpath(response.json(), '$..sessionId')[0]


#
csvfile = pd.read_csv("/Users/anderson/Downloads/api.csv")
final = csvfile.iloc[:, :6]

templat=pd.read_csv("/Users/anderson/Downloads/apitemplates.csv")

regex1 = r'"token":"(.*?)"'
regex2 = r'"sessionId":"(.*?)"'

my_result = []
for ix, row in final.iterrows():

    row[5] = "" if isinstance(row[5], float) else row[5]

    if row[3] == 'POST':

        if row[5] == "":
            response = r.post(url=get_hosts(env) + row[4])

        else:

            if "?" not in row[4]:

                result = re.sub(regex2, '"sessionId":"{}"'.format(session), row[5])
                result = re.sub(regex1, '"token":"{}"'.format(token), result)

                response = r.post(url=get_hosts(env) + row[4], json=json.loads(result))


            else:
                response = r.post(url=get_hosts(env) + row[4], data=row[5])

    if row[3] == 'GET':

        if row[5] == "":
            response = r.get(url=get_hosts(env) + row[4])

        else:
            response = r.get(url=get_hosts(env) + row[4], data=row[5])


    my_result.append([row[4], response.status_code, response.text])


print(my_result)

final_result=pd.DataFrame(my_result,columns=["path","statuscode","response"])
error_df = final_result[(final_result["statuscode"] != 200)]
final_result.to_csv("/Users/anderson/Downloads/api_result.csv")
error_df.to_csv("/Users/anderson/Downloads/{}_api_{}_error.csv".format(env,date))

很快就能跑完。这样就能得到结果：

同时也能得到出错的结果，出错的结果：

这样的结果还是比较简单，只是验证了返回的状态码，如果返回的结果中有错误，就难发现。

用jsonshema来对比一下吧在线转换工具：

https://www.liquid-technologies.com/online-json-to-schema-converter

生成schema的包：

https://pypi.org/project/genson/

生成:

from genson import SchemaBuilder
def to_json_schema(target):

  print(target)

  builder = SchemaBuilder()
  builder.add_schema({"type": "object", "properties": {}})


  builder.add_object(json.loads(target.replace("\'","\"")))

  builder.to_schema()

  #
  print(builder.to_json(indent=2))
  return builder.to_json(indent=2)



csvfiles =pd.read_csv("/Users/anderson/Downloads/apitemplates.csv")

csvfiles.iloc[:,2]=csvfiles.iloc[:,2].apply(to_json_schema)

csvfiles.to_csv("/Users/anderson/Downloads/api_schema.csv")

对比的话，比较简单

from jsonschema import validate, draft7_format_checker
from jsonschema.exceptions import SchemaError, ValidationError

result=validate(instance=loginresponse, schema=loginschema)
try:
    validate(instance=loginresponse, schema=loginschema, format_checker=draft7_format_checker)
except SchemaError as e:
    print("验证模式schema出错：\n出错位置：{}\n提示信息：{}".format(" --> ".join([i for i in e.path]), e.message))
except ValidationError as e:
    print("json数据不符合schema规定：\n出错字段：{}\n提示信息：{}".format(" --> ".join([i for i in e.path]), e.message))
else:
    print("验证成功！")

然后字段对比就可以加入进去了

    test_result=""
    if row[0] in templat.iloc[:,0]:
      test=templat[templat.iloc[:,0]==row[0]]
      diff_result=validate(instance=response.json(), schema=test[3])

      if test_result is None:
        test_result="pass"
      else:
        test_result = "fail"


    my_result.append([row[4], response.status_code, response.text,test_result])

print(my_result)

final_result=pd.DataFrame(my_result,columns=["path","statuscode","response","schemadiff"])

对比出错的结果就可以筛出来了。