TA的每日心情 | 擦汗 前天 09:02 |
---|
签到天数: 1042 天 连续签到: 4 天 [LV.10]测试总司令
|
一.测试场景
压测“用户登陆”的场景,在用户登陆后,会调用大量的接口,为了 测试该场景的负载。
二.采用过的方案
方案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输出。
|
|