358
財經資訊
腳本SDK參考__SDK使用手冊_性能測試-阿裏雲
包: 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])$")
類:DsvReader
定義:
參數化文件讀取操作類。
方法:DsvReader
定義:public DsvReader(fileName)
DsvReader類構造函數,初始化類成員變量
類:ParamManager
定義:
參數化功能管理類,提供對用戶上傳的參數文件進行管理。
方法:getInstance
定義:public static ParamManager getInstance()
返回參數化功能管理類靜態實例對象。
示例:
一般情況下,下列代碼放在全局位置,既class TestRunner上麵
params = ParamManager.getInstance()
方法:addProvider
定義:public void addProvider(ParamProvider_Instance)
添加參數讀操作類實例對象。
示例:
一般情況下,下列代碼放在全局位置,既class TestRunner上麵
添加mailaddr.csv文件讀取操作實例對象
params.addProvider(DsvReader(u'mailaddr.csv'))
方法:nextRecord
定義:public boolean nextRecord (paramFileName)
public boolean nextRecord(paramFileName,bCircle)
根據參數文件名加載當前參數位置的下一個位置的參數值
示例:
一般情況下,下列代碼放在全局位置,既class TestRunner上麵
添加mailaddr.csv文件讀取操作實例對象
params = ParamManager.getInstance()
params.addProvider(DsvReader(u'mailaddr.csv'))
一般情況下,下列代碼放在 class TestRunner 成員函數裏麵,例如action裏麵
加載mailaddr.csv下一個位置的參數值,參數值循環,如果已到最後一個值,會繼續取第一個值
params.nextRecord(u'mailaddr.csv')
加載mailaddr.csv下一個位置的參數值,參數值不循環,如果已到最後一個值,此函數值返回False
參數化數據用完後,要想腳本停止運行請添加下列代碼(需要導入:from HTTPClient import ShutdownException):
if (not params.nextRecord(u'mailaddr.csv', False)):
raise ShutdownException("param file EOF")
方法:getParamValue
定義:public String getParamValue(paramName)
根據參數化文件和參數名取當前位置的參數值。
示例:
讀取mailaddr.csv參數文件的sendmail_addr字段當前位置的值
params. getParamValue (u'mailaddr.csv:sendmail_addr')
附錄一: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&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&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)
最後更新:2016-05-06 10:44:37
上一篇:
在線幫助__使用手冊_性能測試-阿裏雲
下一篇:
腳本模塊使用說明__SDK使用手冊_性能測試-阿裏雲
JavaSDK手冊__SDK使用手冊_郵件推送-阿裏雲
OSSWriter__Writer插件_使用手冊_數據集成-阿裏雲
BatchGetRow__API 概覽_API 參考_表格存儲-阿裏雲
網絡分析__使用手冊(new)_機器學習-阿裏雲
ALIYUN::ESS::ScalingConfiguration__資源列表_資源編排-阿裏雲
KMS地域分布__API 參考_密鑰管理服務-阿裏雲
設置對象/文件元信息__管理文件_開發人員指南_對象存儲 OSS-阿裏雲
6.3 使用ACL進行權限管理__第六章 用戶與權限_使用手冊_分析型數據庫-阿裏雲
發送定時消息__Java SDK_TCP 接入(專業)_消息隊列 MQ-阿裏雲
查詢 Topic 當前位點__Topic 相關接口_Open API_消息隊列 MQ-阿裏雲
相關內容
常見錯誤說明__附錄_大數據計算服務-阿裏雲
發送短信接口__API使用手冊_短信服務-阿裏雲
接口文檔__Android_安全組件教程_移動安全-阿裏雲
運營商錯誤碼(聯通)__常見問題_短信服務-阿裏雲
設置短信模板__使用手冊_短信服務-阿裏雲
OSS 權限問題及排查__常見錯誤及排除_最佳實踐_對象存儲 OSS-阿裏雲
消息通知__操作指南_批量計算-阿裏雲
設備端快速接入(MQTT)__快速開始_阿裏雲物聯網套件-阿裏雲
查詢API調用流量數據__API管理相關接口_API_API 網關-阿裏雲
使用STS訪問__JavaScript-SDK_SDK 參考_對象存儲 OSS-阿裏雲