一. 测试场景 压测“用户登陆”的场景,在用户登陆后,会调用大量的接口,为了 测试该场景的负载 二. 采用过的方案方案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.使用的库 - <font size="3">import json
- import aiohttp
- import sys
- import asyncio</font>
复制代码
2.根据需要访问的接口, 生成异步任务,并调用 - <font size="3">async def async_run(host,apis,headers):
- results = []
- coroutines = []
- for info in apis.values():
- if info.get("method") == "GET":</font>
复制代码- <font size="3"> 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</font>
复制代码最后2行,如果返回的Exception ,则抛出异常 3.封装Post, Get 请求,大同小异,故只贴出Get请求 - <font size="3">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)</font>
复制代码如果接口返回不正正确!=200, 则生成一个Exception实例,保存在Results,如果返回正常,则把返回结果直接保存在results
4.Main函数 - <font size="3"> 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])))</font>
复制代码根据jmeter传入的参数来配置执行参数和读取json文件 如果无错误,则通过print输出到jmeter 如果有错误,错误参数有1个,为网络错误,通过print输出 错误参数有2个,为接口返回,通过print输出
|