51Testing软件测试论坛

标题: Jmeter与Python结合测试异步接口的场景 [打印本页]
作者: lsekfe    时间: 2021-4-27 13:41
标题: Jmeter与Python结合测试异步接口的场景
一.  测试场景
压测“用户登陆”的场景，在用户登陆后，会调用大量的接口，为了 测试该场景的负载
二. 采用过的方案
方案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]:  选择登陆子系统
[attach]132631[/attach]
五.核心实现Jmeter部分
使用到的组件：Transaction Controller, HttpSampler,  BeanShell Sampler, Response Assertion
Transaction Controller
有两部分组成
1. HttpSampler
这部分用户名和密码登录校验接口
[attach]132632[/attach]

2. BeanShell Sampler
说明：为什么使用BeanShell Sampler 而不是JSR，因为JSR不支持Python 3.x
[attach]132633[/attach]

关键代码说明：
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
[attach]132634[/attach]

Python部分
1.使用的库
  1. <font size="3">import json
  2. import aiohttp
  3. import sys
  4. import asyncio</font>
复制代码

2.根据需要访问的接口, 生成异步任务，并调用
  1. <font size="3">async def async_run(host,apis,headers):
  2.         results = []
  3.         coroutines = []
  4.         for info in apis.values():
  5.             if info.get("method") == "GET":</font>
复制代码
  1. <font size="3"> coroutines.append(asyncio.create_task(get("{0}/{1}".format(host,info.get("url")),headers,info.get("params"),results)))
  2.             elif info.get("method") == "POST":
  3.                 coroutines.append(asyncio.create_task(post("{0}/{1}".format(host,info.get("url")),headers,info.get("params"),results)))
  4.             else:pass
  5.         await asyncio.gather(*coroutines)
  6.         for ret in results:
  7.             if type(ret) == Exception: raise ret</font>
复制代码
最后2行，如果返回的Exception ,则抛出异常
3.封装Post, Get 请求，大同小异，故只贴出Get请求
  1. <font size="3">async def get(url,headers,params,results):
  2.     async with aiohttp.ClientSession() as session:
  3.         async with session.get(url,headers = headers,json = params) as rsp:
  4.             r = await rsp.text()
  5.             ret = json.loads(r)
  6.             if ret.get("procCode") != 200:
  7.                 e = Exception()
  8.                 e.args = ("{0} {1}".format(ret.get("procCode"),url),ret.get("exception"))
  9.                 results.append(e)
  10.             else:
  11.                 results.append(ret)</font>
复制代码
如果接口返回不正正确!=200, 则生成一个Exception实例，保存在Results,如果返回正常，则把返回结果直接保存在results

4.Main函数
  1. <font size="3">  host = "http://{0}".format(sys.argv[1])
  2.     authorization = sys.argv[2]
  3.     apis_path = sys.argv[3]
  4.     with open(apis_path, "r") as f:
  5.         apis = json.load(f)
  6.    
  7.     headers = {
  8.         "Authorization":authorization,
  9.         "Accept":"application/json, text/plain, */*",
  10.         "Content-Type":"application/json"
  11.     }
  12.     try:
  13.         asyncio.run(async_run(host,apis,headers))   
  14.         print(json.dumps(dict(procCode="API OK")))
  15.     except Exception as e:
  16.         if len(e.args) == 1: print(json.dumps(dict(procCode=e.args[0])))
  17.         else: print(json.dumps(dict(procCode=e.args[0],exception=e.args[1])))</font>
复制代码
根据jmeter传入的参数来配置执行参数和读取json文件
如果无错误，则通过print输出到jmeter
如果有错误，错误参数有1个，为网络错误，通过print输出
                     错误参数有2个，为接口返回，通过print输出











欢迎光临 51Testing软件测试论坛 (http://bbs.51testing.com/) Powered by Discuz! X3.2