阅读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用户使用手册_性能测试-阿里云