閱讀889 返回首頁    go 魔獸


Lite腳本編寫SDK__手工腳本編寫_Lite用戶使用手冊_性能測試-阿裏雲

訪問Lite控製台

包: PTS

PTS包封裝了大部分用於編寫HTTP手工測試腳本的類和API,類主要有:Framework、Context、Data、Thread、Logger和HttpUtilities這六個。

類: Framework

Framework類主要用於測試引擎對腳本框架進行內部處理,使用戶可以通過API對腳本中的函數進行性能計數(instrument),也提供了對HTTP返回碼和響應處理時間分布的處理函數。一般用戶無需自己使用這個類的方法進行操作。通過錄製或者模版生成的腳本會自動加入Framework代碼,隻有用戶完全手工編寫腳本時才會用到Framework內的方法。

方法:instrumentMethod

定義:def instrumentMethod(tran_name, method_name, c) 對"c"這個類(在性能測試腳本中固化為"TestRunner"這個類)的"method_name"方法進行編織,產生一個性能事務,對應性能測試平台上的事務名稱為"tran_name", 這樣在腳本執行時會自動在"method_name"方法執行前後加上時間戳,可以計算該方法的執行時間。 示例: u'事務名'是性能測試頁麵顯示的事務名,中文字符串前需要加u,viewPage1是TestRunner中定義的一 個成員方法。

PTS.Framework.instrumentMethod (u'事務名', 'viewPage1', TestRunner)

方法:addHttpCode

定義:def addHttpCode(code, codeArray)

對腳本中HTTP請求返回的狀態碼進行累加計數,code是當前需要累加的狀態碼。codeArray是長度為4的int數組,codeArray[0]表示狀態碼小於300的個數,codeArray[1]表示狀態碼在[300, 400)區間的個數,codeArray[2]表示狀態碼在[400, 500)區間的個數,codeArray[3]表示狀態碼在500及以上的個數。

示例:

statusCode = [0L, 0L, 0L, 0L]
headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
result = HTTPRequest().GET('https://item.taobao.com/item.htm?spm=a217i_h.1303267.1998177633-0.10.hx5m95&id=12817144777', None, headers)
PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)

方法:setExtraData

定義:def setExtraData(codeArray=None)

對腳本中HTTP的額外計數信息進行累加,包括狀態碼和響應時間分布。

示例:

statusCode = self.action3()
PTS.Framework.setExtraData(statusCode)
statusCode = self.action4()
PTS.Framework.setExtraData(statusCode)

類:Context

方法:getThreadContext

定義:def getThreadContext()

返回線程級上下文對象,使用該線程上下文對象可以對測試腳本的線程對象進行操作。

示例:

# 獲取線程上下文中的Cookie信息
self.threadContext = PTS.Context.getThreadContext()
self.init_cookies = CookieModule.listAllCookies(self.threadContext)
# 往線程上下文中加入Cookie信息
self.threadContext = PTS.Context.getThreadContext()
CookieModule.addCookie(c, self.threadContext)

方法:setParamDirectory

定義:def setParamDirectory(dir)

測試腳本設置參數文件目錄,通常情況不需要使用,默認為腳本目錄下的data目錄。如果用戶有自定義的參數文件路徑,可以通過該方法進行設置。

類:Data

開關值:delayReports

定義:

delayReports,"0"表示事務數據收集完之後立即匯報,"1"表示將數據收集延遲匯報,通常我們需要在腳本中根據請求返回內容對事務是否成功進行判斷,所以需要將該開關設置為"1"。

示例:

# 延遲匯報數據
PTS.Data.delayReports = 1
# 事務調用    
…………
………
# 匯報數據       
PTS.Data.report()
# 關閉延遲回報數據
PTS.Data.delayReports = 0

開關值:forCurrentTest.success 定義:

forCurrentTest.success,用於設置當前事務是否成功。True表示事務成功,False表示事務失敗。

示例:

如果HTTP請求返回碼不是200或者300,或者返回內容中不存在"test"字符串,就將當前事務標記為失敗,事務默認為成功。

if(not PTS.HttpUtilities.checkResponse([200,300], "test")):
    PTS.Data.forCurrentTest.success = False

類:Thread

方法:sleep

定義:def sleep(millis)

當前調用方法的線程進入睡眠狀態,millis毫秒。

示例:

# 線程睡眠1秒
PTS.Thread.sleep(1000)

方法:getThreadNumber

定義:def getThreadNumber()

返回當前調用線程的線程號,注意該線程號是當前施壓進程啟動線程的邏輯線程號,從0開始往上遞增,不代表實際操作係統的線程號。

方法:getRunNumber

定義:def getRunNumber()

返回當前調用線程的迭代次數。

類:Logger

定義:

腳本日誌類,所有的輸出請使用Logger類進行,Logger可以輸出級別為INFO、WARN、ERROR的三種日誌。INFO包含WARN包含ERROR。根據用戶在性能測試場景裏設置的日誌級別,用戶可以看到不同的日誌,比如用戶在場景設置了INFO,則可以看到通過所有三種級別方法打印出的日誌,如果設置為WARN,則隻能看到通過warn和error方法打出的日誌,如果設置成ERROR,則隻能看到通過error方法打出的日誌。建議用戶在調試階段可以將日誌級別設為INFO,正式壓測時設成WARN或者ERROR。

示例:

# 打印INFO級別的日誌
PTS.Logger.info(u'INFO級別的日誌')


# 打印WARN級別的日誌
PTS.Logger.warn(u'WARN級別的日誌')

# 打印ERROR級別的日誌
PTS.Logger.error(u'ERROR級別的日誌')

類:HttpUtilities

方法:setKeepAlive

定義:def setKeepAlive(flag)

設置底層HTTP客戶端引擎在發送請求時使用長連接還是短連接,默認使用長連接,flag為True或者False。

示例:

# 強製HTTP客戶端引擎使用短連接,每次循環之後關閉連接
PTS.HttpUtilities.setKeepAlive(False)

方法:setUrlEncoding

定義:def setUrlEncoding(code)

設置底層HTTP客戶端引擎在發送請求時使用的URL編碼方式,默認使用GBK編碼。如果用戶的URL已經經過URL編碼,則修改設置無效。

示例:

# 使用UTF-8來進行URL編碼
PTS.HttpUtilities.setUrlEncoding('UTF-8')

法:setTimeout

定義:def setTimeout(time)

設置底層HTTP客戶端引擎控製HTTP請求的Connection和reading Response超時時間,時間單位為毫秒。

示例:

# 設置超時時間120000毫秒
PTS.HttpUtilities. setTimeout (120000)

方法:setFollowRedirects

定義:def setFollowRedirects(flag)

設置底層HTTP客戶端引擎針對類似302這樣的跳轉響應的處理方式。如果設置成True,則引擎會自動跟隨跳轉內容進行繼續請求,否則不再跟隨。按照模板編寫和手工編寫引擎默認為True,錄製下來的腳本引擎默認為False。 示例:

# 跟隨跳轉請求
PTS.HttpUtilities.setFollowRedirects(True)

方法:valueFromCookie

定義:def valueFromCookie(cookieName, threadContext=None)

獲取Cookie管理器中cookieName對應的值,當HTTP響應返回使用SET-COOKIE來設置Cookie時,這些Cookie會自動存儲到Cookie管理器。

示例:

# 獲取Cookie管理器中名字為JSESSIONID的Cookie值
PTS.HttpUtilities.valueFromCookie('JESSIONID')

# 獲取Cookie管理器中名字為JSESSIONID的Cookie值(僅在當前線程中查找)
PTS.HttpUtilities.valueFromCookie('JESSIONID', self.threadContext)

方法:valueFromHeader

定義:def valueFromHeader(header)

獲取HTTP響應中Header名為header的值。

示例:

# 獲取HTTP響應中Header名為Location的值
PTS.HttpUtilities.valueFromHeader('Location')

方法:valueFromBodyInput

定義:def valueFromBodyInput(tokenName)

獲取HTTP響應中標簽名tokenName的input的值。

示例:

# 獲取HTTP響應表單中input名稱為username的控件值
PTS.HttpUtilities.valueFromBodyInput('username')

方法:valueFromHiddenInput

定義:def valueFromHiddenInput(tokenName)

獲取HTTP響應中標簽名為tokenName的隱藏控件的值

方法:valueFromTextBetween

定義:def valueFromTextBetween(text, left, right)

獲取text中被left和right字符串包圍的中間字符串,如果有多個匹配則返回第一個。

示例:

<html> ……<body>……<img token='tb_token'>KDJAK</img></body></html>

比如上麵這段HTML代碼,調用下麵這個方法會返回"KDJAK"

# 返回KDJAK
PTS.HttpUtilities.valueFromTextBetween(text, "'tb_token'>;", "<;")

方法:valuesFromTextBetween

定義:def valuesFromTextBetween(text, left, right)

功能和valueFromTextBetween類似,返回被left和right包圍的字符串列表。

示例:

<xml><cars><car name="VW"/> <car name="BMW"/> <car name="BENZ"/> </cars></xml>

比如上麵這段xml代碼,調用下麵這個方法會返回:["VW", "BMW", "BENZ"]

PTS.HttpUtilities.valuesFromTextBetween(text, "name="", """)

方法:valueFromBodyBetween

定義:def valueFromBodyBetween(left,right)

獲取HTTP響應中被left和right字符串包圍的中間字符串,如果有多個匹配則返回第一個。

示例:

<html> ……<body>……<img token='tb_token'>KDJAK</img></body></html>

比如上麵這段HTML代碼,調用下麵這個方法會返回"KDJAK"

# 返回KDJAK
PTS.HttpUtilities.valueFromBodyBetween("'tb_token'>;", "<;")

方法:valuesFromBodyBetween

定義:def valuesFromBodyBetween(left, right)

功能和valueFromBodyBetween類似,返回被left和right包圍的字符串列表。

示例:

<xml><cars><car name="VW"/> <car name="BMW"/> <car name="BENZ"/> </cars></xml>

比如上麵這段xml代碼,調用下麵這個方法會返回:["VW", "BMW", "BENZ"]

PTS.HttpUtilities.valuesFromBodyBetween("name="", """)

方法:setProxyServer

定義:def setProxyServer(ip, port)

設置底層HTTP引擎使用IP為ip,端口為port的代理進行請求。

示例:

PTS.HttpUtilities.setProxyServer('localhost', 8888)

方法:setUseCookieModule

定義:def setUseCookieModule(flag)

設置腳本中是否使用CookieModule來進行cookie管理,默認使用。如果用戶想在HTTP的header裏自己設置Cookie,則需要通過調用這個方法來將CookieModule禁用。

示例:

PTS.HttpUtilities.setUseCookieModule(False)

方法:checkResponse

定義:def checkResponse(code, expectedText=None)

對HTTP響應進行腳本,返回碼是否為code,響應body裏麵是否包含expectedText。隻有當返回碼符合且expectedText存在時返回True。當expectedText為None時(可以不傳入該參數),不進行文本校驗。

示例:

# 當響應返回碼為200,且返回body包含test則返回True
PTS.HttpUtilities.checkResponse(200, "test")

方法:checkResponseByRegex

定義:def checkResponseByRegex(code, expectedPattern=None)

對HTTP響應進行腳本,返回碼是否為code,響應body裏麵是否包含expectedPattern這樣的正則字符串。隻有當返回碼符合且expectedPattern能夠匹配時,返回True。當expectedPattern為None時(可以不傳入該參數),不進行文本校驗。

示例:

# 當響應返回碼為200,且返回body包含IP地址則返回True
PTS.HttpUtilities.checkResponseByRegex(200, "^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5]).){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$")

附錄一:HTTP示例腳本

#! /usr/bin/env python   
# -*- coding: utf-8 -*-
# PTS Script Version 1.0

# PTS腳本SDK:框架API、常用HTTP請求/響應處理API
from util import PTS
from HTTPClient import NVPair
from HTTPClient import Cookie
from HTTPClient import HTTPRequest
from HTTPClient import CookieModule

# 腳本初始化段,可以設置壓測引擎的常用HTTP屬性
#PTS.HttpUtilities.setKeepAlive(False)
#PTS.HttpUtilities.setUrlEncoding('GBK')
#PTS.HttpUtilities.setFollowRedirects(False)
#PTS.HttpUtilities.setUseCookieModule(False)


# 腳本執行單元類,每個VU/壓測線程會創建一個TestRunner實例對象
class TestRunner:

    # TestRunner對象的初始化方法,每個線程在創建TestRunner後執行一次該方法
    def __init__(self):
        self.threadContext = PTS.Context.getThreadContext()

        self.action1()

        self.action2()

        self.init_cookies = CookieModule.listAllCookies(self.threadContext)

    # 主體壓測方法,每個線程在測試生命周期內會循環調用該方法
    def __call__(self):
        PTS.Data.delayReports = 1

        for c in self.init_cookies:
            CookieModule.addCookie(c, self.threadContext)

        statusCode = self.action3()
        PTS.Framework.setExtraData(statusCode)

        statusCode = self.action4()
        PTS.Framework.setExtraData(statusCode)

        PTS.Data.report()
        PTS.Data.delayReports = 0

    # TestRunner銷毀方法,每個線程循環執行完成後執行一次該方法
    def __del__(self):

        for c in self.init_cookies:
            CookieModule.addCookie(c, self.threadContext)

        self.action5()

        self.action6()

    # 定義請求函數
    def action1(self):

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://www.ptstest.com', None, headers)

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('Content-Type', 'application/x-www-form-urlencoded'), NVPair('Host', 'www.ptstest.com'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().POST('https://www.ptstest.com/s', 'wd=test', headers)


    def action2(self):

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://www.ptstest.com/s?wd=test', None, headers)

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('Content-Type', 'application/x-www-form-urlencoded'), NVPair('Host', 'www.ptstest.com'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().POST('https://www.ptstest.com/s', 'wd=test', headers)


    def action3(self):
        statusCode = [0L, 0L, 0L, 0L]

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://item.taobao.com/item.htm?spm=a217i_h.1303267.1998177633-0.10.hx5m95&amp;id=12817144777', None, headers)
        PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('Content-Type', 'application/x-www-form-urlencoded'), NVPair('Host', 'www.ptstest.com'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().POST('https://www.ptstest.com/s', 'wd=test', headers)
        PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)

        if(not PTS.HttpUtilities.checkResponse(200)):
            PTS.Data.forCurrentTest.success = False

        return statusCode

    def action4(self):
        statusCode = [0L, 0L, 0L, 0L]

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://item.taobao.com/item.htm?spm=a217i_h.1303267.1998177633-0.10.hx5m95&amp;id=12817144777', None, headers)
        PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('Content-Type', 'application/x-www-form-urlencoded'), NVPair('Host', 'www.ptstest.com'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://www.ptstest.com/s?wd=test', None, headers)
        PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)

        if(not PTS.HttpUtilities.checkResponse(200)):
            PTS.Data.forCurrentTest.success = False

        return statusCode


    def action5(self):

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://www.ptstest.com/s?wd-test', None, headers)

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('Content-Type', 'application/x-www-form-urlencoded'), NVPair('Host', 'www.ptstest.com'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().POST('https://www.ptstest.com/s', 'wd=test', headers)


    def action6(self):

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().GET('https://www.ptstest.com', None, headers)

        headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('Content-Type', 'application/x-www-form-urlencoded'), NVPair('Host', 'www.ptstest.com'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
        result = HTTPRequest().POST('https://www.ptstest.com/s', 'wd=test', headers)


# 編織壓測事務
PTS.Framework.instrumentMethod(u'事務1', 'action3', TestRunner)
PTS.Framework.instrumentMethod(u'事務2', 'action4', TestRunner)

附錄二:TCP Socket示例腳本

#! /usr/bin/env python   
# -*- coding: utf-8 -*-
# PTS TCP Socket Script Template Version 1.0

from util import PTS
import socket

class TestRunner:
    # TestRunner對象的初始化方法,每個線程在創建TestRunner後執行一次該方法
    def __init__(self):
       return 

    # 主體壓測方法,每個線程在測試生命周期內會循環調用該方法        
    def __call__(self):
        PTS.Data.delayReports = 1

        statusCode = self.action1()
        PTS.Framework.setExtraData(statusCode)                
        PTS.Data.report()
        PTS.Data.delayReports = 0
    # TestRunner銷毀方法,每個線程循環執行完成後執行一次該方法
    def __del__(self):

        return 
    # 業務函數    
    def action1(self):

        statusCode = [0L, 0L, 0L, 0L]

        #以下為socket協議腳本
        #創建TCP Socket
        sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)       
        #連接到服務器,服務器IP及端口號
        sk.connect(('localhost',2000))        
        #將發送內容送到服務器
        sendContent = 'This is tcp socket test,{Status:success}'
        sk.send(sendContent);
        #接收服務器返回內容
        RecvContent = sk.recv(1024)
        PTS.Logger.info('Content come from server is:'+RecvContent)

        #關閉連接
        sk.close()
        #socket協議腳本結束 

        #抓返回值           
        value = PTS.HttpUtilities.valueFromTextBetween(RecvContent,'{Status:','}')        

        #判斷事務是否成功
        if not value:
            PTS.Data.forCurrentTest.success = False
            PTS.Logger.info('Have no value')
        elif value.lower() == 'success':
            PTS.Data.forCurrentTest.success = True
            PTS.Logger.info('Success flag1:'+value)
        else:
            PTS.Data.forCurrentTest.success = False
            PTS.Logger.info('Success flag2:'+value)            

        return statusCode

# 編織壓測事務
PTS.Framework.instrumentMethod(u'事務名', 'action1', TestRunner)

附錄三:UDP Socket示例腳本

#! /usr/bin/env python   
# -*- coding: utf-8 -*-
# PTS UDP Socket Script Version 1.0

from util import PTS
import socket

class TestRunner:
    # TestRunner對象的初始化方法,每個線程在創建TestRunner後執行一次該方法
    def __init__(self):

        return

    # 主體壓測方法,每個線程在測試生命周期內會循環調用該方法
    def __call__(self):
        PTS.Data.delayReports = 1

        statusCode = self.action1()
        PTS.Framework.setExtraData(statusCode)                
        PTS.Data.report()
        PTS.Data.delayReports = 0
    # TestRunner銷毀方法,每個線程循環執行完成後執行一次該方法
    def __del__(self):

        return 
    # 業務函數    
    def action1(self):

        statusCode = [0L, 0L, 0L, 0L]


        #創建UDP Socket
        s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
        #設置服務器IP地址及端口號
        port = 20000
        host = 'localhost'
        msg='''0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijkl
        mnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRS
        TUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnop
        qrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWX
        YZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrs
        tuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVW
        XYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghij
        klmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'''

        #用Connect提高效率
        s.connect((host, port))
        #將內容發送給服務器

        s.send(msg)
        #設置接收超時,單位秒
        s.settimeout(30)
        #接收服務器返回內容
        RecvContent,ADDR = s.recvfrom(1024)
        PTS.Logger.info('Content come from server:'+RecvContent)

        #關閉連接
        s.close()

        #socket協議腳本結束 

        #抓返回值           
        value = PTS.HttpUtilities.valueFromTextBetween(RecvContent,'{Status:','}')        

        #判斷事務是否成功
        if not value:
            PTS.Data.forCurrentTest.success = False
            PTS.Logger.info('Have no value')
        elif value.lower() == 'success':
            PTS.Data.forCurrentTest.success = True
            PTS.Logger.info('Success flag1:'+value)
        else:
            PTS.Data.forCurrentTest.success = False
            PTS.Logger.info('Success flag2:'+value)            

        return statusCode

# 編織壓測事務
PTS.Framework.instrumentMethod(u'事務名', 'action1', TestRunner)

訪問Lite控製台

最後更新:2016-05-06 10:44:43

  上一篇:go 手工腳本編寫指南__手工腳本編寫_Lite用戶使用手冊_性能測試-阿裏雲
  下一篇:go 線程數設置__測試配置_Lite用戶使用手冊_性能測試-阿裏雲