介面測試

#!/usr/bin/env python
# -*- coding: utf_8 -*-

import threading
import requests
import time
import re
from time import sleep

# -------介面效能測試配置-------
method = "get"
# 介面型別
url = "http://www.baidu.com"
# 介面地址
#headers = {'Id': 'test', 'time': '1626833059'}
data = {"username": "admin", "password": "123456
 
"}
# 介面引數
thread_num = 20
# 執行緒數
one_work_num = 2
# 每個執行緒迴圈次數
loop_sleep = 1
# 每次請求時間間隔
response_time = []
# 平均響應時間列表
error = []
# 錯誤資訊列表


class CreateThread:
    def __init__(self):
        pass

    @classmethod
    def thread_api(cls):
        global results
        try:
            if method == "
 
post":
                results = requests.post(url, data)
            if method == "get":
                results = requests.get(url)
                #results = requests.get(url, headers)
            return results
        except requests.ConnectionError:
            return results
    # 介面函式

    @classmethod
    
 
def thread_response(cls):
        responsetime = float(CreateThread.thread_api().elapsed.microseconds) / 1000
        return responsetime
    # 獲取響應時間 單位ms

    @classmethod
    def thread_response_avg(cls):
        avg = 0.000
        l = len(response_time)
        for num in response_time:
            avg += 1.000 * num / l
        return avg
    # 獲取平均相應時間 單位ms

    @classmethod
    def thread_time(cls):
        return time.asctime(time.localtime(time.time()))
    # 獲取當前時間格式

    @classmethod
    def thread_error(cls):      #@@@@@@@@
        try:
            #pa = u"個人資訊"
            #pattern = re.compile(pa)
            #match = pattern.search(CreateThread.thread_api().text)
            if CreateThread.thread_api().status_code == 200:
                pass
                #if match.group() == pa:
                #    pass
            else:
                error.append(CreateThread.thread_api().status_code)
        except AttributeError:
            error.append("登入失敗")
    # 獲取錯誤的返回狀態碼

    @classmethod
    def thread_work(cls):
        threadname = threading.currentThread().getName()
        print ("[", threadname, "] Sub Thread Begin")
        for i in range(one_work_num):
            CreateThread.thread_api()
            print ("介面請求時間： ", CreateThread.thread_time())
            response_time.append(CreateThread.thread_response())
            CreateThread.thread_error()
            sleep(loop_sleep)
        print ("[", threadname, "] Sub Thread End")
    # 工作執行緒迴圈

    @classmethod
    def thread_main(cls):
        start = time.time()
        threads = []
        for i in range(thread_num):
            t = threading.Thread(target=CreateThread.thread_work())
            t.setDaemon(True)
            threads.append(t)
        for t in threads:
            t.start()
        # 啟動所有執行緒
        for t in threads:
            t.join()
        # 主執行緒中等待所有子執行緒退出
        end = time.time()

        print ("========================================================================")
        print ("介面效能測試開始時間：", time.asctime(time.localtime(start)))
        print ("介面效能測試結束時間：", time.asctime(time.localtime(end)))
        print ("介面地址：", url)
        print ("介面型別：", method)
        print ("執行緒數：", thread_num)
        print ("每個執行緒迴圈次數：", one_work_num)
        print ("每次請求時間間隔：", loop_sleep)
        print ("總請求數：", thread_num * one_work_num)
        print ("錯誤請求數：", len(error))
        print ("總耗時（秒）：", end - start)
        print ("每次請求耗時（秒）：", (end - start) / (thread_num * one_work_num))
        print ("每秒承載請求數（TPS)：", (thread_num * one_work_num) / (end - start))
        print ("平均響應時間（毫秒）：", CreateThread.thread_response_avg())


if __name__ == '__main__':
    CreateThread.thread_main()

介面訪問時間測試