python實現(xiàn)接口并發(fā)測試腳本

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import requests,time,json,threading,random

class Presstest(object):
  headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36',
    'Content-Type': 'application/json; charset=UTF-8',
  }
  def __init__(self,login_url,press_url,phone="1376193000",password="123456"):
    self.login_url = login_url
    self.press_url = press_url
    self.phone = phone
    self.password = password
    self.session = requests.Session()
    self.session.headers = self.headers

  def login(self):
    '''登陸獲取session'''
    data = data = {'t': int(time.time() * 1000), 'userName': self.phone, 'passWord': self.password}
    res = self.session.post(self.login_url,data=json.dumps(data))
    XToken = res.json().get('data').get('companyToken')
    self.session.headers['X-Token'] = XToken

  def testinterface(self):
    '''壓測接口'''
    self.session.headers['X-UnionId'] = 'of6uw1CUVhP533sQok'
    data = {"id": int(''.join(str(random.choice(range(10))) for _ in range(10))),
        "openId": "oMr0c5LGJjlTc", "addressId": 10, "shipType": "SELF", "totalAmount": 5,
        "receivable": 5, "carts": [
        {"amount": 1, "barcode": "1234567890", "skuId": 1, "spec": "34", "itemAmount": 5, "price": 0,
         "cover": "xxxx-dd.oss-cn-shanghai.aliyuncs.com/dfc91fd067ac464c096c90af33a196a5.png",
         "name": "沙宣洗發(fā)水", "packingType": "瓶", "placeOfOrigin": "上海", "productId": "310153323435134976",
         "retailPrice": 5, "suitableAge": "1-100"}], "formId": "the formId is a mock one", "comments": "aa"}
    global ERROR_NUM
    try:
      html = self.session.post(self.press_url, data=json.dumps(data))
      if html.json().get('code') != 0:
        print(html.json())
        ERROR_NUM += 1
    except Exception as e:
      print(e)
      ERROR_NUM += 1

  def testonework(self):
    '''一次并發(fā)處理單個任務(wù)'''
    i = 0
    while i < ONE_WORKER_NUM:
      i += 1
      self.work()
    time.sleep(LOOP_SLEEP)

  def run(self):
    '''使用多線程進程并發(fā)測試'''
    t1 = time.time()
    Threads = []

    for i in range(THREAD_NUM):
      t = threading.Thread(target=self.testonework, 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("===============壓測結(jié)果===================")
    print("URL:", self.press_url)
    print("任務(wù)數(shù)量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
    print("總耗時(秒):", t2 - t1)
    print("每次請求耗時(秒):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
    print("每秒承載請求數(shù):", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
    print("錯誤數(shù)量:", ERROR_NUM)


if __name__ == '__main__':
  login_url = 'https://ds.xxxxx.com/sys/sysUser/login'
  press_url = 'https://ds.xxxxx.com/weshop/order/checkout'
  phone = "1376193000"
  password = "123456"
  
  THREAD_NUM = 1     # 并發(fā)線程總數(shù)
  ONE_WORKER_NUM = 5   # 每個線程的循環(huán)次數(shù)
  LOOP_SLEEP = 0.1    # 每次請求時間間隔(秒)
  ERROR_NUM = 0      # 出錯數(shù)
  
  obj = Presstest(login_url=login_url,press_url=press_url,phone=phone,password=password)
  obj.login()
  obj.run()

===============壓測結(jié)果===================
URL: https://ds.xxxxx.com/weshop/order/checkout
任務(wù)數(shù)量: 1 * 5 = 5
總耗時(秒): 1.9810078144073486
每次請求耗時(秒): 0.39620156288146974
每秒承載請求數(shù): 2.5239678327547805
錯誤數(shù)量: 0