建网站公司 快云,网站更换ip地址,蓝色网站素材,WordPress对象储存Step01: Python脚本开发
文件路径#xff1a;D://wl//testproject//Fone-grpc//project1//test_client.py
Python 脚本作用#xff1a;
通过 grpc 调用底层 c 的接口#xff0c;做数据库的数据插入与查询操作#xff0c;然后将返回的结果进行拼接与输出。
2.代码里面将…
Step01: Python脚本开发
文件路径D://wl//testproject//Fone-grpc//project1//test_client.py
Python 脚本作用
通过 grpc 调用底层 c 的接口做数据库的数据插入与查询操作然后将返回的结果进行拼接与输出。
2.代码里面将每一次调用后返回的内容进行拼接后并做了成功信息的统计输出成功的次数为3输出后会转给 jmeter 里面的脚本获取。
from __future__ import print_function
import logging
import sys
import grpc
import Storage_pb2 # Storage_pb2.py which contains our generated request and response classes
import StorageService_pb2_grpc # StorageService_pb2_grpc.py which contains our generated client and server classes.host10.10.1.117:50066def run():resultwith grpc.insecure_channel(host) as channel:resp0StorageService_pb2_grpc.FOneStorageStub(channel).InsertKVS(Storage_pb2.PUpsertKVS(DBNameMergeLog,TableNameMergeLog,KeyValues[{Key:b,Key:b}],UpsertTrue,TransactionTrue))resultstr(resp0)with grpc.insecure_channel(host) as channel:stud1StorageService_pb2_grpc.FOneStorageStub(channel)resp1stud1.InsertV(Storage_pb2.PInsertV(DBNameMergeLog, TableNameMergeLog, Valueb1))result str(resp1)with grpc.insecure_channel(host) as channel:stud2StorageService_pb2_grpc.FOneStorageStub(channel)resp2stud2.FindOne(Storage_pb2.PFindK(DBNameMergeLog, TableNameMergeLog, Key1))result str(resp2)resultresult.replace(\n, )resultresult.count(Successful return: 0)print(result)if __name__ __main__:logging.basicConfig()run()PS在本地调试好脚本确保符合预期。
Step02: Jmeter 创建 beanshell 取样器
脚本内容如下
import java.io.BufferedReader;
import java.io.InputStreamReader;log.info(----------------------------start to exe);
String command cmd /c python D://wl//testproject//Fone-grpc//project1//test_client.py; //定义要执行的python文件路径
String var;
Runtime rt Runtime.getRuntime();
Process pr rt.exec(command); //执行上面的python脚本
pr.waitFor()
BufferedReader b new BufferedReader(new InputStreamReader(pr.getInputStream())); // 转换执行脚本后的返回响应
String line ;
StringBuilder response new StringBuilder();
while ((line b.readLine()) ! null) {response.append(line);
}
String response_data response.toString();
log.info(response_data);
if(response_data.equals(3)){ // 判断响应的内容是否有3次如果不是3次就表示执行失败了log.info(---------------------------success end exe);
}else{log.error(--------------------------failed end exe); //这里的输出一次会记录一次失败次数
}
b.close();Step03: 执行效果
通过 jmeter 工具进行并发操作对后端底层接口进行压力测试效果不错效果图如下所示。 Step04: Jmeter日志输出
2023-02-22 17:26:08,539 INFO o.a.j.t.JMeterThread: Stopping because end time detected by thread: 线程组--python 1-33
2023-02-22 17:26:08,539 INFO o.a.j.t.JMeterThread: Thread finished: 线程组--python 1-33
2023-02-22 17:26:08,597 INFO o.a.j.u.BeanShellTestElement: 3
2023-02-22 17:26:08,597 INFO o.a.j.u.BeanShellTestElement: ---------------------------success end exe
2023-02-22 17:26:08,598 INFO o.a.j.t.JMeterThread: Stopping because end time detected by thread: 线程组--python 1-82
2023-02-22 17:26:08,598 INFO o.a.j.t.JMeterThread: Thread finished: 线程组--python 1-82
2023-02-22 17:26:08,608 INFO o.a.j.u.BeanShellTestElement: 2
2023-02-22 17:26:08,608 ERROR o.a.j.u.BeanShellTestElement: --------------------------failed end exe最后感谢每一个认真阅读我文章的人礼尚往来总是要有的虽然不是什么很值钱的东西如果你用得到的话可以直接拿走 这些资料对于【软件测试】的朋友来说应该是最全面最完整的备战仓库这个仓库也陪伴上万个测试工程师们走过最艰难的路程希望也能帮助到你
