上篇帖子组合测试生成参数在接口测试中的探索篇

说明

• 比如登陆接口的 3 个参数可以生产 13 组参数组合，那么就要对这个接口进行 13 次测试，这里有两种请求方式：

  • 多线程发请求
  • 协程发请求

• 新增对各个接口的每组参数进行压力测试

代码分析

简单的封装

• api.ymal

---
title: XXXX接口测试
host: rap.taobao.org
port: 80
protocol: http://
header: {"Accept":"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8","User-Agent":"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36"}
api_list:
- id: 1001
  name: 登陆
  method: post
  stress: 10 # 对每组参数进行压测次数
  url: /mockjs/11463/login
  hope: {"appStatus":{"errorCode":0,"message":"操作成功"}, "content":{"token":"dfafdafdafsa","user_id":30,"nickname":"18576759587"}}
  params:
  - "user_name:user_nam:empty:true:type:bool, user_name:right:1851111111:type:str, user_name:error:18522222222:type:str, user_name:format-error:18522222:type:str"
  - "pwd:pwd:empty:true:type:bool, pwd:right:111111:type:str, pwd:error:222222:type:str"
  login_key: user_id
  is_first: 1
- id: 1002
  method: get
  name: 得到个人信息
  url: /mockjs/11463/getUserInfo
  hope: {"appStatus":{"errorCode":0,"message":"操作成功"}, "content":{"sex":0,"user_id":30,"test":"11111"}}
  stress: 10
  params:
  - "user_id:user_id:empty:true:type:bool, user_id:right:1851111111:type:str, user_id:error:18522222222:type:str, user_id:format-error:18522222:type:str"
  - "token:token:empty:true:type:bool, token:right:111111:type:str, token:error:222222:type:str"
  login_key: user_id

• http 请求

class ConfigHttp():
    def __init__(self, dict_http):
        '''
        http请求的封装，传入dict
        :param dict_http:
        '''
        self.dict_http = dict_http
    def get(self, dict_get, param):
        result = {}
        url = self.dict_http["protocol"] + self.dict_http["host"] + ":" + str(self.dict_http["port"]) + dict_get["url"]
        # print(url)
        r = requests.get(url, headers=self.dict_http["header"], params=param)
        r.encoding = 'UTF-8'
        if r.status_code == 200:
            result = json.loads(r.text)
        result["status_code"] = r.status_code
        return result
    def post(self, dict_post, param):
        url = self.dict_http["protocol"] + self.dict_http["host"] + ':' + str(self.dict_http["port"]) + dict_post["url"]
        r = requests.post(url, files=None, data=param)
        result = {}
        if r.status_code == 200:
            result = json.loads(r.text)
        result["status_code"] = r.status_code
        return result

• 执行 pict 组合参数，读取组合参数

def pict_param(**kwargs):
    '''
    pict处理请求的参数
    :param kwargs:
    params: 请求的参数列表，类型为list
    pict_param: 用配对处理好的参数存储的
    :return:
    '''

    for item in kwargs["params"]:
        operateFile.write_txt(kwargs["pict_params"], item)
    os.popen("pict " + kwargs["pict_params"] + ">"+kwargs["pict_params_result"])

def read_pict_param(pict_params):
    '''
    读取本地e生成好了的接口请求参数
    :param pict_params:  本地路径
    :return: list
    '''
    result = operateFile.read_txt(pict_params)
    l_result = []
    if result:
        for info in range(len(result)):
            for item in range(len(result[info])):
                t_result = result[info][item].split(",")
                d_t = {}
                for i in t_result:
                    temp = i.split(":")
                    t = {}
                    t[temp[1]] = temp[2]
                    t[temp[3]] = temp[4]
                    d_t[temp[0]] = t
                l_result.append(d_t)
    return l_result

• 线程封装

import threading
class Thread(threading.Thread):
  def __init__(self, func):
      threading.Thread.__init__(self)
      self.func = func
  def run(self):
      self.func

• 协程封装

from gevent import Greenlet
from gevent import monkey; monkey.patch_all()
class requestGevent(Greenlet):
    def __init__(self, func):
        Greenlet.__init__(self)
        self.func = func
    def _run(self):
        # gevent.sleep(self.n)
        self.func

• 读取 api.yaml

def get_config(api_yaml):
    '''
    得到api.yaml中的设置的接口信息
    :param yaml:
    :return:
    '''
    http_config = {} # http信息的记录
    api_config = [] # api的记录记录
    get_api_list = operateYaml.getYam(api_yaml)
    for key in get_api_list:
        if type(get_api_list[key]) != list:
            http_config[key] = get_api_list[key]
        else:
            api_config = get_api_list[key]
    return http_config, api_config

入口

def myRequest(**kwargs):
    if kwargs["param_req"]["method"] == get_managers.HTTP_POST:
        for item in range(kwargs["param_req"]["stress"]): # 压力测试
            really_result = kwargs["http_config"].post(dict_post=kwargs["param_req"], param=kwargs["param_result"])
            print("压力请求开始post："+str(item))
            if really_result.get("status_code") == 200:
                print("post请求成功鸟")
                hope = kwargs["param_req"]["hope"] # 得到期望值
    elif kwargs["param_req"]["method"] == get_managers.HTTP_GET:
        for item in range(kwargs["param_req"]["stress"]):  # 压力测试
            print("压力请求开始get：" + str(item))
            really_result = kwargs["http_config"].get(dict_get=kwargs["param_req"], param=kwargs["param_result"])
            if really_result.get("status_code") == 200:
                print("get请求成功鸟")
        else:
            print("请求方法暂时不支持")



def multi_thread_request(**kwargs):
    '''
     多线程发请求
    :param kwargs:
    :return:
    '''
  kwargs["http_config"] = kwargs["http_config"] # http的配置信息，如host,port，header等
    for i in kwargs["api_config"]: # 读取各个接口的配置，api.ymal
        # 生成参数
        pict_param(params=i["params"], pict_params=get_managers.PICT_PARAMS,
                   pict_params_result=get_managers.PICT_PARAMS_RESULT)
        # 读取参数
        get_param = read_pict_param(get_managers.PICT_PARAMS_RESULT)
        count = len(get_param) #根据不同的分组参数，启动多个协程
        threads = []
        req = {}
        for key in i:
            if key != "params": #过滤请求参数，参数上面已经处理好了
                req[key] = i[key]
        for k in range(0, count):
            kwargs["param_result"] = get_param[k] #接口中不同的参数组合，是dict类型
            kwargs["param_req"] = req # 每次请求除组合参数之外的参数，如逾期只，请求的url,method等
            kwargs["login_response_param"] = {} # 记录结果，后面拓展
            threads.append(httpThread.Thread(myRequest(**kwargs)))
        for k in range(0, count):
            threads[k].start()
        for k in range(0, count):
            threads[k].join()

        # 协程是这样调用
        # for k in range(0, count):
        #     kwargs["param_result"] = get_param[k]
        #     kwargs["param_req"] = req
        #     kwargs["login_response_param"] = {}
        #     green_let.append(gevents.requestGevent(myRequest(**kwargs)))
        # for k in range(0, count):
        #     green_let[k].start()
        # for k in range(0, count):
        #     green_let[k].join()


if __name__ == "__main__":
    get_api_config = get_config(PATH("../api.ymal"))
    http_conf = hc.ConfigHttp(dict_http=get_api_config[0]) # http请求的设置
    multi_thread_request(http_config=http_conf, api_config=get_api_config[1])

执行结果

post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
post接口请求成功鸟
....
....
get请求成功鸟
get请求成功鸟
get请求成功鸟