程序员文章、书籍推荐和程序员创业信息与资源分享平台

网站首页 > 技术文章 正文

接口测试实战:Jmeter与Python结合测试异步接口场景

hfteth 2025-05-25 12:47:06 技术文章 1 ℃

一. 测试场景

压测“用户登陆”的场景,在用户登陆后,会调用大量的接口,为了测试该场景的负载

二. 采用过的方案

方案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输出


如果你有很多问题想要解决,你的测试职业规划也需要一点光亮,你也想跟着大家一起分享探讨,我给你推荐一个「Python自动化测试学习交流群」你缺的知识这里有,你少的技能这里有,你要的大牛也在这里……

请关注+私信回复:“测试”就可以免费拿到软件测试学习资料,同时进入群学习交流~~

Tags:

最近发表
标签列表