关闭

基于 k6 和 python 进行自动化性能测试

发表于:2023-7-10 09:29

字体: | 上一篇 | 下一篇 | 我要投稿

 作者:风做了云的梦    来源:华为云开发者联盟

  当我们开发完成一个应用程序时,往往需要对其进行性能测试,以帮助我们更好的优化程序以及发现程序中的一些 bug。在性能测试中,达到相应的性能指标对于一个软件来说十分重要,在本文中,将介绍一种现代化性能测试工具 k6。
  k6 是一个开源工具,基于 JavaScript 可以编写 k6 的测试脚本,测试 Web 应用程序以及 API 的性能,支持 HTTP 等多种协议,可以很好地模拟各种高负载场景,充分验证程序稳定性和性能。k6 支持 Linux、MacOS 等多个平台,通过 k6 官网根据提示即可在各个平台快速安装 k6,终端输入 k6 version 出现如下显示说明安装成功。
  以下是一个简单的 k6 测试脚本,通过 k6 的 HTTP API 模拟 Get 请求,并且休眠一秒钟:K。
  import http from 'k6/http';
  import { sleep } from 'k6';
  export default function () {
  http.get('https://test-api.com');
  sleep(1);
  }
  通过执行下面这行代码,运行脚本,即可对服务完成测试。
  k6 run test-script.js
  k6 提供了丰富的功能,以下是 k6 常用的一些 API,具体可以参考官网文档介绍:
  - http.get(url, [options]):发送GET请求。
  - http.post(url, body, [options]):发送POST请求。
  - check(res, checks):检查响应是否符合预期。
  - group(name, func):将一组请求分组并统计性能指标。
  - sleep(duration):休眠指定的时间。
  k6 的测试结果包括以下一些指标,可以根据这些指标,更好的优化程序。
  - VUs:虚拟用户的数量。
  - Iterations:迭代次数。
  - RPS:每秒钟的请求数。
  - Duration:测试持续时间。
  - Data Sent/Received:发送和接收的数据量。
  - Checks:检查的数量。
  - Status codes:响应状态码的数量。
  - Errors:错误的数量。
  - Latency distribution:延迟分布。
  通过 Python 和 k6 你可以更加高效的完成符合自己要求的自动化测试,Python 可以提供非常多的工具库,用来收集处理 k6 返回的结果。 我们可以编写以下 k6 测试脚本,并且通过 Python 去执行它,相关注释我已经标注出来,在 handleSummary 函数中,我们可以通过 metrics 来获取各种测试信息,具体如代码所示,可以参考官网关于 metrics 的介绍,同时自定义环境变量的使用也十分方便,可以参考代码中的使用方式。
  import http from 'k6/http';
  import { check, sleep} from 'k6';
  import {Rate} from 'k6/metrics';
  export default function() {
  #post请求所需要的body体
  let requestBody = {
  "xxx":[
  "xxxxx"
  ],
  "xxxx": __ENV.MyVar # MyVar为自定义的环境变量,可以通过__ENV调用,在执行脚本时可直接通过MyVar=xxx传值
  };
  #url
  const url = 'http://example.com';
  const payload = JSON.stringify(requestBody);
  const params = {
  headers: {
  'Content-Type': 'application/json',
  },
  timeout: '100s' #每个请求的超时时间
  };
  let res = http.post(url, payload, params);
  #检测结果是否是200OK
  check(res, { 'status is 200': (r) => r.status === 200 });
  }
  export function handleSummary(data) {
  #通过data.metrics中的字段可以获取你想要的一些信息,例如每个请求的持续时间和吞吐量
  const time = `${data.metrics.http_req_duration.values.avg.toFixed(3)}`;
  const rps = `${data.metrics.http_reqs.values.rate.toFixed(3)}`;
  const res = `${time} ${rps}`;
  console.log(res); # 利用console.log可以将内容打印到控制台
  return {stdout : res}; #输出到标准输出
  }
  如下是一个 Python 代码示例,相关代码已经注释,通过 Python 中的 subprocess 模块执行 k6 脚本,并且捕获 k6 脚本的输出,通过 pandas 库进行整理输出到 excel 中。还可以通过 argparse 库解析命令行参数传入 k6 脚本中,更加灵活,高效。
  # -*- coding: utf-8 -*-
  import subprocess
  from alive_progress import alive_bar # 非常丰富的进度条工具库
  from tqdm import tqdm # 进度条工具库
  import pandas as pd # 可以用来处理文本excel,csv等
  from collections import OrderedDict
  import argparse # 用来解析命令行参数
  import time
  print('测试时间 : ', time.strftime('%b %d %Y %H:%M:%S', time.gmtime(time.time())))
  print("************开始测试啦! 祈祷不出错!**************")
  # 需要测试的测试语句集合
  test_examples = [
  "aaaaaaa",
  "bbbbbbb",
  "ccccccc"
  ]
  dataMap = {'test': test_examples}
  parser = argparse.ArgumentParser()
  parser.add_argument("-d", default="60s", help="duration time", dest="duration_time") #解析命令行参数,控制测试时间
  args = parser.parse_args()
  print("每条语句测试时间 : ", args.duration_time)
  vus = ['10', '20', '30', '40'] # 并发数集合 ,分别测试并发数为10,20,30,40的场景
  cols_name = ['1-avg/ms', '1-rps/s', '10-avg/ms', '10-rps/s','20-avg/ms', '20-rps/s','50-avg/ms', '50-rps/s'] # excel的列名
  # 循环测试,可以将多个需要测试的语句集合放入到dataMap中
  for (name, data) in dataMap.items():
  print("当前测试的项目为 :", name)
  res = OrderedDict()
  res['test_examples'] = []
  for n in cols_name:
  res[n] = []
  df = pd.DataFrame(res)
  excel_name = name + ".xlsx"
  df.to_excel(excel_name, index=False)
  for query in data:
  print("当前测试语句为 :", query)
  origin = pd.read_excel(excel_name)
  with alive_bar(len(vus)) as bar:
  temp_dict = {}
  temp_dict['test_examples'] = query
  for vu in vus:
  keyRps = vu + '-rps/s'
  keyTime = vu + '-avg/ms'
  MyVar='MyVar=' + query
  #通过Popen执行k6脚本,并且捕获它的标准输出
  process = subprocess.Popen(['k6', 'run', '--quiet', 'script.js', '--env', MyVar, '--vus', vu, '--duration', args.duration_time], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  result = process.stdout.read()
  temp = result.split()
  temp_dict[keyTime] = temp[0].decode();
  temp_dict[keyRps] = temp[1].decode();
  print("并发:", vu, temp[0].decode(), temp[1].decode())
  bar()
  #将脚本输出写到excel
  save_data = origin.append(temp_dict, ignore_index=True)
  save_data.to_excel(excel_name, index=False)
  执行此 Python 脚本,可以得到类似以下输出:
  本文内容不用于商业目的,如涉及知识产权问题,请权利人联系51Testing小编(021-64471599-8017),我们将立即处理
《2023软件测试行业现状调查报告》独家发布~

关注51Testing

联系我们

快捷面板 站点地图 联系我们 广告服务 关于我们 站长统计 发展历程

法律顾问:上海兰迪律师事务所 项棋律师
版权所有 上海博为峰软件技术股份有限公司 Copyright©51testing.com 2003-2024
投诉及意见反馈:webmaster@51testing.com; 业务联系:service@51testing.com 021-64471599-8017

沪ICP备05003035号

沪公网安备 31010102002173号