搭建私有PTS性能测试服务(四)

#! /usr/bin/env python
# -*- coding:utf-8 -*-

# A simple example using the HTTP plugin that shows the retrieval of a
# single page via HTTP.
#
# This script is automatically generated by ngrinder.
#
# @author admin
from net.grinder.script.Grinder import grinder
from net.grinder.script import Test
from net.grinder.plugin.http import HTTPRequest
from net.grinder.plugin.http import HTTPPluginControl
from HTTPClient import NVPair

control = HTTPPluginControl.getConnectionDefaults()
# if you don't want that HTTPRequest follows the redirection, please modify the following option 0.
# control.followRedirects = 1
# if you want to increase the timeout, please modify the following option.
control.timeout = 6000

test1 = Test(1, "www.baidu.com")
request1 = HTTPRequest()

# Make any method call on request1 increase TPS
test1.record(request1)

class TestRunner:
    # initlialize a thread
    def __init__(self):
        grinder.statistics.delayReports=True
        pass

    # test method
    def __call__(self):
        result = request1.GET("http://www.baidu.com")

        # You get the message body using the getText() method.
        # if result.getText().find("HELLO WORLD") != -1 :
        #    grinder.statistics.forLastTest.success = 1
        # else :
        #    grinder.statistics.forLastTest.success = 0

        # if you want to print out log.. Don't use print keyword. Instead, use following.
        # grinder.logger.info("Hello World")

        if result.getStatusCode() == 200 :
            grinder.statistics.forLastTest.success = 1
        elif result.getStatusCode() in (301, 302) :
            grinder.logger.warn("Warning. The response may not be correct. The response code was %d." %  result.getStatusCode())
            grinder.statistics.forLastTest.success = 1
        else :
            grinder.statistics.forLastTest.success = 0

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# PTS Script Version 1.0
# Trunner auto-generated test script at Fri Oct 09 16:34:42 CST 2015

from java.lang import String
from java.util import Random
from java.util import Date

# 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
from HTTPClient import ShutdownException

# PTS对参数化的相关支持
from com.aliyun.pts import DsvReader
from com.aliyun.pts import ParamManager

# 脚本初始化段，可以设置压测引擎的常用HTTP属性
PTS.HttpUtilities.setUrlEncoding('utf-8')
PTS.HttpUtilities.setFollowRedirects(True)
PTS.HttpUtilities.setTimeout(120000)
# PTS.HttpUtilities.setKeepAlive(False)
# PTS.HttpUtilities.setUseCookieModule(False)
# PTS.HttpUtilities.setProxyServer('localhost', 8888)
# PTS.Context.setParamDirectory("/Users/fei/Work/trunner/data")

# 支持socket测试, 如TCP\UDP等协议
# import socket

# 设置系统编码
import sys
reload(sys)
sys.setdefaultencoding('utf-8')


params = ParamManager.getInstance()

expiryDate = Date()
expiryDate.year += 10

class TestRunner:

    # TestRunner对象的初始化方法，每个线程在创建TestRunner后执行一次该方法
    def __init__(self):
        self.threadContext = PTS.Context.getThreadContext()


    # 主体压测方法，每个线程在测试生命周期内会循环调用该方法
    def __call__(self):
        PTS.Data.delayReports = 1

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

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

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

        headers = []
        result = HTTPRequest().GET(u'http://www.baidu.com',[],headers)
        PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
        ## statusCode[0]代表http code < 300 个数,    statusCode[1] 代表 300<=http code<400 个数
        # statusCode[2]代表400<=http code<500个数，  statusCode[3] 代表 http code >=500个数
        # 如果http code 300 到 400 之间是正常的
        # 那么判断事务失败，请将statusCode[1:3] 改为   statusCode[2:3] 即可
        if(sum(statusCode[1:3]) > 0):
            PTS.Data.forCurrentTest.success = False
            PTS.Logger.error(u'事务请求中http 返回状态大于300，请检查请求是否正确!')

        return statusCode

# 调用施压引擎施压。第一个参数是事务名，可以为中文；第二个参数是执行事务方法的方法名；第三个统一写TestRunner
PTS.Framework.instrumentMethod(u'百度首页', 'action_20015009', TestRunner)

from net.grinder.script import Test
from net.grinder.script.Grinder import grinder
from org.json.simple import JSONValue
from java.util import Random
from java.lang import String
# 下面这个库是位于 lib 文件夹中的。
from mpcCaller import NpcCaller
 
random = Random()Z
 
caller = NpcCaller("xx.xx.xx.15", 7090);
module = "tchat"
procedure = "chatBO/RelayCommand"…

# -*- coding:utf-8 -*-

# 性能测试框架公共方法
# @author 归根落叶
# @blog http://this.ispenn.com

#统计单页http code数量
def addHttpCode(statusCode,statusCodeList=[0L,0L,0L,0L]):
    if statusCode < 300:
        statusCodeList[0] += 1
    if 300 <= statusCode < 400:
        statusCodeList[1] += 1
    if 400 <= statusCode < 500:
        statusCodeList[2] += 1
    if statusCode >= 500:
        statusCodeList[3] += 1
    return statusCodeList

#统计所有页http code数量
def sumHttpCode(statusCodeList,sumStatusCodeList=[0L,0L,0L,0L]):
    for i in range(len(statusCodeList)):
        sumStatusCodeList[i] += statusCodeList[i]
    return sumStatusCodeList

#统计通过检查点的数量
def sumCheckPointStatus(checkPointStatus,checkPointStatusList=[0L,0L]):
    if checkPointStatus == 1:
        checkPointStatusList[0] += 1
    else:
        checkPointStatusList[1] += 1
    return checkPointStatusList

# -*- coding:utf-8 -*-

# A simple example using the HTTP plugin that shows the retrieval of a
# single page via HTTP.
#
# This script is automatically generated by ngrinder.
#
# @author admin
from net.grinder.script.Grinder import grinder
from net.grinder.script import Test
from net.grinder.plugin.http import HTTPRequest
from net.grinder.plugin.http import HTTPPluginControl
from HTTPClient import CookieModule
from HTTPClient import NVPair
import PTS    #这里引入自定义模块