快捷導(dǎo)航

如何用python做簡(jiǎn)單的接口壓力測(cè)試

更新時(shí)間：2023年09月05日 14:07:23 作者：星旅zl

這篇文章主要介紹了如何用python做簡(jiǎn)單的接口壓力測(cè)試問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

一、引入所需庫(kù)和設(shè)置需要到的參數(shù)

import threading
import time
import http.client
import urllib
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
pararms = urllib.parse.urlencode({'token':'xxx'})
rqheaders={ #'DeviceToken':'xxxxxxxxx','OSVersion':'1.0.3','AppVersion':'14',
'Accept':'application/json, text/plain, */*',
'Accept-Encoding':'gzip, deflate, br',
'Accept-Language':'zh-CN,zh;q=0.9',
'Host':'deve.xxx.com',
'Origin':'https://www.xxx.com',
'Referer':'https://www.xxx.com/xxx_test/',
'token':'xxx',
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.157 Safari/537.36'}
HOST1 = "www.baidu.com"; #主機(jī)地址 例如192.168.1.101   nanning.anjuke.com,
HOST_L = "deve.xxx.com"
HOST_IP = "xxx.xx.xxx.xx"  #ip地址
PORT = 9999 #端口
METHOD = {1:"GET",2:"POST",3:"OPTIONS"}
URI = "/api/web/XXX" #相對(duì)地址
TOTAL = 0 #總數(shù)
SUCC = 0 #響應(yīng)成功數(shù)
FAIL = 0 #響應(yīng)失敗數(shù)
EXCEPT = 0 #響應(yīng)異常數(shù)
MAXTIME=0 #最大響應(yīng)時(shí)間
MINTIME=100 #最小響應(yīng)時(shí)間，初始值為100秒
GT3=0 #統(tǒng)計(jì)3秒內(nèi)響應(yīng)的
LT3=0 #統(tǒng)計(jì)大于3秒響應(yīng)的
# 并發(fā)的線程數(shù)
thread_count = 5

這里請(qǐng)根據(jù)真實(shí)的接口地址進(jìn)行配置，有需要到token的請(qǐng)自行拿到token值，不需要頭部的的可以在下面中寫(xiě)headers = {}

二、創(chuàng)建一個(gè) threading.Thread 的派生類(lèi)

class RequestThread(threading.Thread):
    # 構(gòu)造函數(shù)
    def __init__(self, thread_name):
        threading.Thread.__init__(self)
        self.test_count = 0
        print ("===========task init===========METHOD=",METHOD[3])
    # 線程運(yùn)行的入口函數(shù)
    def run(self):
        self.test_performace()
    def test_performace(self):
            global TOTAL
            global SUCC
            global FAIL
            global EXCEPT
            global GT3
            global LT3
            try:
                st = time.time()
                conn = http.client.HTTPConnection(HOST1,PORT)  #若沒(méi)有PORT請(qǐng)刪掉這個(gè)參數(shù)
                req = conn.request(METHOD[1], URI, body = {}, headers = rqheaders)  # or POST  headers=rqheaders
                res = conn.getresponse()
                #print ('msg:', res.msg)
                #print ('headers:', res.getheaders())
                print("res.status =",res.status)
                print('res.read =',res.read().decode('utf-8'))
                start_time
                if res.status == 200:
                    TOTAL = TOTAL+1
                    SUCC = SUCC+1
                    print("TOTAL = ",TOTAL)
                else:
                    TOTAL = TOTAL+1
                    FAIL = FAIL+1
                    print("TOTAL = ",TOTAL)
                time_span = time.time()-st
                print ("%s:%f\n"%(self.name,time_span))
                self.maxtime(time_span)
                self.mintime(time_span)
                if time_span>3:
                    GT3 = GT3+1
                else:
                    LT3 = LT3+1
            except Exception as e:
                print("e =",e)
                TOTAL = TOTAL+1
                EXCEPT = EXCEPT+1
            conn.close()
    def maxtime(self,ts):
            global MAXTIME
            print(ts)
            if ts>MAXTIME:
                MAXTIME=ts
    def mintime(self,ts):
            global MINTIME
            if ts<MINTIME:
                MINTIME=ts

res.read().decode('utf-8')這個(gè)方法會(huì)返回接口響應(yīng)的所有內(nèi)容

三、使用派生類(lèi)方法寫(xiě)執(zhí)行

# main 代碼開(kāi)始
print ("===========task start===========")
# 開(kāi)始的時(shí)間
start_time = time.time()
i = 1
while i <= thread_count:
    t = RequestThread("thread" + str(i))
    t.start()
    i = i + 1
t=0
#并發(fā)數(shù)所有都完成或大于60秒就結(jié)束
while TOTAL<thread_count|t>60:
        print ("total:%d,succ:%d,fail:%d,except:%d\n"%(TOTAL,SUCC,FAIL,EXCEPT))
        print (HOST,URI)
        t = t+1
        time.sleep(2)
time.sleep(2)
print ("===========task end===========")
print ("total:%d,succ:%d,fail:%d,except:%d"%(TOTAL,SUCC,FAIL,EXCEPT))
print ("response maxtime:",MAXTIME)
print ("response mintime:",MINTIME)
print ("great than 3 seconds:%d,percent:%0.2f"%(GT3,float(GT3)/TOTAL))
print ("less than 3 seconds:%d,percent:%0.2f"%(LT3,float(LT3)/TOTAL))

四、上面就是所有的代碼

然后我們打開(kāi)cmd來(lái)運(yùn)行這個(gè)腳本，這里設(shè)置thread_count = 1，即執(zhí)行一次請(qǐng)求

cmd進(jìn)入到相應(yīng)的py文件夾下，運(yùn)行腳本結(jié)果如下所示：

F:\pythonTest>python testInterfacePressure.py
===========task start===========
===========task init===========METHOD= OPTIONS
res.status = 200
res.read = {..........，"code":1,"message":"數(shù)據(jù)讀取成功","status":200}

TOTAL = 1
Thread-1:0.149927

0.14992666244506836
===========task end===========
total:1,succ:1,fail:0,except:0
response maxtime: 0.14992666244506836
response mintime: 0.14992666244506836
great than 3 seconds:0,percent:0.00
less than 3 seconds:1,percent:1.00