Jmeter+Python实现异步接口测试

lsekfe

一.测试场景
　　压测“用户登陆”的场景，在用户登陆后，会调用大量的接口，为了 测试该场景的负载。
　　二.采用过的方案
　　方案1.直接使用Jmeter接口压测，但Jmeter在一个线程中的调用多个接口是同步的，故压测出来的结果不准确。
　　方案2.Jmeter用多个线程组压测，但这种方案压测出来是并行，并不符合场景，因为浏览器虽然是异步调用，但本质上也是一个线程执行。
　　三.最终方案
　　Jmeter结合Python3.x的asyncio和aiohttp结合。
　　因为Python的asyncio可模拟浏览器的异步调用方式。
　　四.用例设计
　　1步[User]:使用CSV Data Config读取用户名和密码
　　2步[加密密码]: 使用BeanShell调用加密的Jar为密码加密
　　3步[Login Eclinical Page]:登陆系统
　　4步[Login Design]:选择登陆子系统

　　五.核心实现
　　Jmeter部分
　　使用到的组件：Transaction Controller、HttpSampler、BeanShell Sampler、Response Assertion、Transaction Controller。
　　有两部分组成：
　　1.HttpSampler
　　这部分用户名和密码登录校验接口。

　　2.BeanShell Sampler
　　说明：为什么使用BeanShell Sampler 而不是JSR，因为JSR不支持Python 3.x。

　　关键代码说明：
　　1：传入Python的参数

#访问的服务器地址 String host = "200.200.101.97"; #执行的Python文件 String pyPath = "D:\\\sites\\\eClinical4.0_testing\\\PT\\\py_script\\\asyncio_http.py"; #需要异步调用的接口，配置在一个json文件中 String apis = "D:\\\sites\\\eClinical4.0_testing\\\PT\\\py_script\\\admin_login_api.json";

　　2：启动Python进程
　　8-10行。
　　3：等待Python 进程返回，并把结果保存在变量” ${name}_admin_login”中
　　11-23行。
　　Response Assertion
　　通过字符串判断Python的返回，如果所有接口返回正常，则变量” ${name}_admin_login”包含API OK。

　　Python部分
　　1.使用的库

import json import aiohttp import sys import asyncio

　　2.根据需要访问的接口, 生成异步任务，并调用

async def async_run(host,apis,headers):         results = []         coroutines = []         for info in apis.values():             if info.get("method") == "GET":                 coroutines.append(asyncio.create_task(get("{0}/{1}".format(host,info.get("url")),headers,info.get("params"),results)))             elif info.get("method") == "POST":                 coroutines.append(asyncio.create_task(post("{0}/{1}".format(host,info.get("url")),headers,info.get("params"),results)))             else:pass         await asyncio.gather(*coroutines)         for ret in results:             if type(ret) == Exception: raise ret

　　最后2行，如果返回的Exception ,则抛出异常。

　　3.封装Post, Get 请求，大同小异，故只贴出Get请求

async def get(url,headers,params,results):     async with aiohttp.ClientSession() as session:         async with session.get(url,headers = headers,json = params) as rsp:             r = await rsp.text()             ret = json.loads(r)             if ret.get("procCode") != 200:                 e = Exception()                 e.args = ("{0} {1}".format(ret.get("procCode"),url),ret.get("exception"))                 results.append(e)             else:                 results.append(ret)

　　如果接口返回不正正确!=200, 则生成一个Exception实例，保存在Results,如果返回正常，则把返回结果直接保存在results。
　　4.Main函数

host = "http://{0}".format(sys.argv[1])     authorization = sys.argv[2]     apis_path = sys.argv[3]     with open(apis_path, "r") as f:         apis = json.load(f)     headers = {         "Authorization":authorization,         "Accept":"application/json, text/plain, */*",         "Content-Type":"application/json"     }     try:         asyncio.run(async_run(host,apis,headers))            print(json.dumps(dict(procCode="API OK")))     except Exception as e:         if len(e.args) == 1: print(json.dumps(dict(procCode=e.args[0])))         else: print(json.dumps(dict(procCode=e.args[0],exception=e.args[1])))

　　根据jmeter传入的参数来配置执行参数和读取json文件。
　　如果无错误，则通过print输出到jmeter。
　　如果有错误，错误参数有1个，为网络错误，通过print输出。
　　错误参数有2个，为接口返回，通过print输出。