Python API接口压力测试简单实现

发表于:2021-7-01 09:27

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

 作者:小小晓晓阳    来源:CSDN

分享:
   近期使用fastapi实现了一个webapi接口,通用的请求工具无法满足的我测试需求,简单实现了一个压力测试方法,可以测试接口的并发能力,方便调整客户端请求参数。
  import base64
  import os
  import urllib
  import numpy as np
  import requests, time, json, threading, random
   
   
  class Presstest(object):
      """
      并发压力测试
      """
      def __init__(self,  press_url):
          self.press_url = press_url
   
      def test_interface(self):
          '''压测接口'''
          global INDEX
          INDEX += 1
   
          global ERROR_NUM
          global TIME_LENS
          try:
              start = time.time()
              payload = {'img_datas': img_datas}
              response_content = self.do_request(self.press_url, payload)
              result = json.loads(response_content)
              end = time.time()
              TIME_LENS.append(end - start)
              print('end')
          except Exception as e:
              ERROR_NUM += 1
              print(e)
   
      def test_onework(self):
          '''一次并发处理单个任务'''
          i = 0
          while i < ONE_WORKER_NUM:
              i += 1
              self.test_interface()
          time.sleep(LOOP_SLEEP)
   
      def do_request(self, url, payload):
           '''通用http获取webapi请求结果方法'''
          headers = {
              'Content-Type': 'application/json; charset=UTF-8',
          }
          request = urllib.request.Request(url, json.dumps(payload).encode("utf-8"), headers=headers)
          retry_num = 0
          while retry_num < 3:
              response = urllib.request.urlopen(request, timeout=300)
              if not response or response.status == 421:
                  time.sleep(1)
                  retry_num = retry_num + 1
                  continue
              else:
                  break
          response_content = response.read()
          if hasattr(response_content, 'decode'):
              response_content = response_content.decode('utf-8')
          return response_content
   
      def run(self):
          '''使用多线程进程并发测试'''
          t1 = time.time()
          Threads = []
   
          for i in range(THREAD_NUM):
              t = threading.Thread(target=self.test_onework, name="T" + str(i))
              t.setDaemon(True)
              Threads.append(t)
   
          for t in Threads:
              t.start()
          for t in Threads:
              t.join()
          t2 = time.time()
   
   
          print("===============压测结果===================")
          print("URL:", self.press_url)
          print("任务数量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
          print("总耗时(秒):", t2 - t1)
          print("每次请求耗时(秒):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
          print("每秒承载请求数:", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
          print("错误数量:", ERROR_NUM)
          print(INDEX)
   
   
  if __name__ == '__main__':
      press_url = 'http://107.0.0.1:8001/test/'
      TIME_LENS = []
      INDEX = 0
      THREAD_NUM = 1  # 并发线程总数
      ONE_WORKER_NUM = 100  # 每个线程的循环次数
      LOOP_SLEEP = 0  # 每次请求时间间隔(秒)
      ERROR_NUM = 0  # 出错数
   
      obj = Presstest(press_url)
      obj.run()

      本文内容不用于商业目的,如涉及知识产权问题,请权利人联系51Testing小编(021-64471599-8017),我们将立即处理
《2023软件测试行业现状调查报告》独家发布~

关注51Testing

联系我们

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

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

沪ICP备05003035号

沪公网安备 31010102002173号