全部产品
弹性计算 会员服务 网络 安全 移动云 数加·大数据分析及展现 数加·大数据应用 管理与监控 云通信 阿里云办公 培训与认证 智能硬件
存储与CDN 数据库 域名与网站(万网) 应用服务 数加·人工智能 数加·大数据基础服务 互联网中间件 视频服务 开发者工具 解决方案 物联网 更多
性能测试 PTS

脚本编写

更新时间:2017-12-01 17:12:23

对于非 HTTP 协议的业务(如 TCP 协议,HTTPS 等)或者某些内容不能录制的场景(如加密算法,非浏览器访问的应用等),需要进行手工编写代码开发脚本。同时 PTS 提供了 HTTP 相关的 API 获取 Header/Cookie/Body 等信息,以下附有示例。

手工脚本语言为 Jython,目前支持 Jython 2.5.3 版本。

Jython 语法

Jython 程序由一系列语句组成,语句组成了代码块,代码块组成了方法、函数,然后再通过类把数据、方法和函数封装起来。和其它高级语言一样,Jython 的语句也是由一些最基本的词(token)组成。Token 可以是标识符(identifiers)、关键字(keywords)、字面值(literals)、操作符(operators)和分割符(delimiters),这些 token 通过 Jython 的语言执行器进行词法分析产生,而词法分析器通过字符方式读入 Jython 脚本文件。

脚本框架

性能测试的性能测试脚本是一个 TestRunner 类,这个类会被每一个并发线程初始化。

  1. 测试进程首先加载脚本并执行脚本中顶格的语句,同时定义 TestRunner 这个测试类。
  2. 然后每个线程会实例化一个 TestRunner 类,调用类中的 init 方法一次,
  3. 继而循环调用 TestRunner 类的 call 方法。
  4. 最后线程结束时会调用类中的 del 方法。

init 和 del 方法都是可选的,只有 call 方法是必需的。

框架示例解析

  1. # 第一部分:执行器声明和脚本编码声明
  2. #! /usr/bin/env python
  3. # -*- coding: utf-8 -*-
  4. # 第二部分:Jython类库、Java类库和自定义类的导入
  5. # PTS Script Version 1.0
  6. # PTS脚本SDK:框架API、常用HTTP请求/响应处理API
  7. from util import PTS
  8. from HTTPClient import NVPair
  9. from HTTPClient import Cookie
  10. from HTTPClient import HTTPRequest
  11. from HTTPClient import CookieModule
  12. # 第三部分:测试进程级别的脚本语句和初始化
  13. # 脚本初始化段,可以设置压测引擎的常用HTTP属性
  14. #PTS.HttpUtilities.setKeepAlive(False)
  15. #PTS.HttpUtilities.setUrlEncoding('GBK')
  16. #PTS.HttpUtilities.setFollowRedirects(False)
  17. #PTS.HttpUtilities.setUseCookieModule(False)
  18. PTS.HttpUtilities.setUseContentEncoding(True)
  19. PTS.HttpUtilities.setUseTransferEncoding(True)
  20. # 第四部分:TestRunner测试类
  21. # 脚本执行单元类,每个VU/压测线程会创建一个TestRunner实例对象
  22. class TestRunner:
  23. # TestRunner对象的初始化方法,每个线程在创建TestRunner后执行一次该方法
  24. def __init__(self):
  25. self.threadContext = PTS.Context.getThreadContext()
  26. self.init1()
  27. self.init_cookies = CookieModule.listAllCookies(self.threadContext)
  28. # 主体压测方法,每个线程在测试生命周期内会循环调用该方法
  29. def __call__(self):
  30. PTS.Data.delayReports = 1
  31. for c in self.init_cookies:
  32. CookieModule.addCookie(c, self.threadContext)
  33. statusCode = self.action1()
  34. PTS.Framework.setExtraData(statusCode)
  35. statusCode = self.action2()
  36. PTS.Framework.setExtraData(statusCode)
  37. PTS.Data.report()
  38. PTS.Data.delayReports = 0
  39. # TestRunner销毁方法,每个线程循环执行完成后执行一次该方法
  40. def __del__(self):
  41. for c in self.init_cookies:
  42. CookieModule.addCookie(c, self.threadContext)
  43. self.end1()
  44. # 定义请求函数
  45. def init1(self):
  46. ……
  47. def action2(self):
  48. ……
  49. def action2(self):
  50. ……
  51. # 第五部分:instrumentMethod语句
  52. # 编织压测事务
  53. PTS.Framework.instrumentMethod(u'action1', 'action1', TestRunner)
  54. PTS.Framework.instrumentMethod(u'action2', 'action2', TestRunner)

PTS 提供的脚本 API

PTS 提供的脚本 API 包含框架的接口,比如 Framework API,Data API。该接口一般您不需要使用,在生成脚本的模板的时候已经包含。

其余的接口需要根据业务需求进行使用。具体功能总结如下:

HttpUtilities:设置 HTTP 请求相关设置(长连或者断连、URL Encoding、超时时间)以及获取响应的 Header、Cookie、响应内容相关接口。

比如,在业务上 B 请求依赖 A 请求的响应值作为参数;针对一个请求,在响应包含某段字符串的时候才统计为成功,这个时候就需要用到此类相关接口获取响应值。具体示例在下面的例子中有。

Framework API: Framework 类主要用于测试引擎对脚本框架进行内部处理。您可以通过 API 对脚本中的函数进行性能计数(instrument),也提供了对 HTTP 返回码和响应处理时间分布的处理函数。一般您无需自己使用这个类的方法进行操作。通过录制或者模版生成的脚本会自动加入 Framework 代码,只有在您完全手工编写脚本时才会用到 Framework 内的方法。

Context API:返回线程级上下文对象。使用该线程上下文对象可以对测试脚本的线程对象进行操作。比如,获取线程上下文中的 Cookie 信息;往线程上下文中加入 Cookie 信息。

Data API:数据统计相关接口。一般您无需自己使用这个类的方法进行操作

DsvReader API:参数化文件操作相关接口。

ParamManager API:参数化功能管理类,管理您上传的参数文件。

Thread API:线程相关的信息,包括线程睡眠接口、当前调用线程的线程号、当前调用线程的迭代次数接口。

常见用法示例

HTTPS 支持

在脚本的最前面添加两行:

  1. from java.lang import System
  2. System.setProperty("https.protocols", "TLSv1.2")

简单 POST 请求

  1. def action_20140285(self):
  2. statusCode = [0L, 0L, 0L, 0L]
  3. headers = [NVPair(u'Content-Type', u'application/x-www-form-urlencoded'),]
  4. # body体设置
  5. bodyContext = u''
  6. bodyData = String(bodyContext).getBytes('utf-8')
  7. result = HTTPRequest().POST(u'http://www.test.com', bodyData, headers)
  8. # 出报表使用,统计成功失败率
  9. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  10. if(300 <= result.getStatusCode()):
  11. PTS.Data.forCurrentTest.success = False
  12. return statusCode

带有 Header、Cookie、Body 体的 POST 请求

此示例带 Header、Cookie、Body 体,如下:

  1. def action_20140285(self):
  2. statusCode = [0L, 0L, 0L, 0L]
  3. # 设置cookie
  4. CookieModule.addCookie(Cookie(u'cna', u'QycTEqahR1YCASp4Sl3upq8j', \
  5. 'www.test.com', "/", expiryDate, 0), self.threadContext)
  6. CookieModule.addCookie(Cookie(u'JSESSIONID', u'OnoanonggawaCASp4Sl911aj', \
  7. 'www.test.com', "/", expiryDate, 0), self.threadContext)
  8. # 设置header
  9. headers = [NVPair(u'Content-Type', u'application/x-www-form-urlencoded'),NVPair(u'User-Agent', u'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'),]
  10. # 请求内容主体(为字符串)
  11. bodyContext = u'BODYTEST中文'
  12. bodyData = String(bodyContext).getBytes('utf-8')
  13. result = HTTPRequest().POST(u'http://www.test.com', bodyData, headers)
  14. # 结果统计
  15. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  16. if(300 <= result.getStatusCode()):
  17. PTS.Data.forCurrentTest.success = False
  18. return statusCode

简单 GET 请求

  1. def action_20140636(self):
  2. statusCode = [0L, 0L, 0L, 0L]
  3. headers = []
  4. result = HTTPRequest().GET(u'http://www.test.com?param1=value1&param2=value2',[],headers)
  5. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  6. if(300 <= result.getStatusCode()):
  7. PTS.Data.forCurrentTest.success = False
  8. return statusCode
  1. def action_20140636(self):
  2. statusCode = [0L, 0L, 0L, 0L]
  3. # 设置cookie
  4. CookieModule.addCookie(Cookie(u'cna', u'QycTEqahR1YCASp4Sl3upq8j', \
  5. 'www.test.com', "/", expiryDate, 0), self.threadContext)
  6. CookieModule.addCookie(Cookie(u'JSESSIONID', u'OnoanonggawaCASp4Sl911aj', \
  7. 'www.test.com', "/", expiryDate, 0), self.threadContext)
  8. # 设置header
  9. headers = [NVPair(u'Content-Type', u'application/x-www-form-urlencoded'),NVPair(u'User-Agent', u'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'),]
  10. result = HTTPRequest().GET(u'http://www.test.com?param1=value1&param2=value2',[],headers)
  11. # 结果统计
  12. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  13. if(300 <= result.getStatusCode()):
  14. PTS.Data.forCurrentTest.success = False
  15. return statusCode

获取 Body 体示例

  1. str(result.getText())
  2. result为请求的结果,如下:
  1. result = HTTPRequest().GET(u'http://www.test.com',[],headers)
  2. PTS.Logger.info(str(result.getText()))

检查点示例

  1. POST请求的响应内容包含 <div id="code">success</div>时,才认为请求成功。如下:
  1. def action_20140285(self):
  2. statusCode = [0L, 0L, 0L, 0L]
  3. headers = [NVPair(u'Content-Type', u'application/x-www-form-urlencoded'),]
  4. bodyContext = u''
  5. bodyData = String(bodyContext).getBytes('utf-8')
  6. result = HTTPRequest().POST(u'http://www.test.com', bodyData, headers)
  7. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  8. # 检查点设置(需要匹配的响应内容)
  9. checkPoint = u'<div id="code">success</div>'
  10. # 如果检查点校验通过则统计为成功
  11. if(not PTS.HttpUtilities.checkResponse(200, checkPoint)):
  12. PTS.Logger.error(u'检查点:"' + checkPoint + u'"校验失败')
  13. PTS.Data.forCurrentTest.success = False
  14. return statusCode

上一请求获取特定字段

获取响应体重 以 name=\”rsv_bp\” value 开头(引号要用转义),> 结尾的中间内容,并判断内容是否符合要求,获取的字符串为 ArrayList 需要转换为 string, resultCheck=””.join(values) 。同时更多的函数见 HttpUtilities 。具体如下:

  1. def action_20140285(self):
  2. statusCode = [0L, 0L, 0L, 0L]
  3. headers = [NVPair(u'Content-Type', u'application/x-www-form-urlencoded'),]
  4. bodyContext = u''
  5. bodyData = String(bodyContext).getBytes('utf-8')
  6. result = HTTPRequest().POST(u'http://www.test.com', bodyData, headers)
  7. # 获取body主体内容
  8. bodyRestult=str(result.getText())
  9. # 获取body主体内容的某些字段
  10. values=PTS.HttpUtilities.valuesFromTextBetween(bodyRestult, "name=\"rsv_bp\" value",">")
  11. # 将获取的内容转换为string
  12. resultCheck="".join(values)
  13. PTS.Logger.error(resultCheck)
  14. # 判断string是否满足需求
  15. if(resultCheck=="=\"1\""):
  16. PTS.Logger.error("ok")
  17. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  18. if(300 <= result.getStatusCode()):
  19. PTS.Data.forCurrentTest.success = False
  20. return statusCode

HTTP 示例

  1. #! /usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # PTS Script Version 1.0
  4. # PTS脚本SDK:框架API、常用HTTP请求/响应处理API
  5. from util import PTS
  6. from HTTPClient import NVPair
  7. from HTTPClient import Cookie
  8. from HTTPClient import HTTPRequest
  9. from HTTPClient import CookieModule
  10. # 脚本初始化段,可以设置压测引擎的常用HTTP属性
  11. # PTS.HttpUtilities.setKeepAlive(False)
  12. # PTS.HttpUtilities.setUrlEncoding('GBK')
  13. # PTS.HttpUtilities.setFollowRedirects(False)
  14. # PTS.HttpUtilities.setUseCookieModule(False)
  15. # 脚本执行单元类,每个VU/压测线程会创建一个TestRunner实例对象
  16. class TestRunner:
  17. # TestRunner对象的初始化方法,每个线程在创建TestRunner后执行一次该方法
  18. def __init__(self):
  19. self.threadContext = PTS.Context.getThreadContext()
  20. self.action1()
  21. self.action2()
  22. self.init_cookies = CookieModule.listAllCookies(self.threadContext)
  23. # 主体压测方法,每个线程在测试生命周期内会循环调用该方法
  24. def __call__(self):
  25. PTS.Data.delayReports = 1
  26. for c in self.init_cookies:
  27. CookieModule.addCookie(c, self.threadContext)
  28. statusCode = self.action3()
  29. PTS.Framework.setExtraData(statusCode)
  30. statusCode = self.action4()
  31. PTS.Framework.setExtraData(statusCode)
  32. PTS.Data.report()
  33. PTS.Data.delayReports = 0
  34. # TestRunner销毁方法,每个线程循环执行完成后执行一次该方法
  35. def __del__(self):
  36. for c in self.init_cookies:
  37. CookieModule.addCookie(c, self.threadContext)
  38. self.action5()
  39. self.action6()
  40. # 定义请求函数
  41. def action1(self):
  42. headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
  43. result = HTTPRequest().GET('http://www.ptstest.com', None, headers)
  44. 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'), ]
  45. result = HTTPRequest().POST('http://www.ptstest.com/s', 'wd=test', headers)
  46. def action2(self):
  47. headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
  48. result = HTTPRequest().GET('http://www.ptstest.com/s?wd=test', None, headers)
  49. 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'), ]
  50. result = HTTPRequest().POST('http://www.ptstest.com/s', 'wd=test', headers)
  51. def action3(self):
  52. statusCode = [0L, 0L, 0L, 0L]
  53. headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
  54. result = HTTPRequest().GET('http://item.taobao.com/item.htm?spm=a217i_h.1303267.1998177633-0.10.hx5m95&amp;id=12817144777', None, headers)
  55. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  56. 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'), ]
  57. result = HTTPRequest().POST('http://www.ptstest.com/s', 'wd=test', headers)
  58. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  59. if(not PTS.HttpUtilities.checkResponse(200)):
  60. PTS.Data.forCurrentTest.success = False
  61. return statusCode
  62. def action4(self):
  63. statusCode = [0L, 0L, 0L, 0L]
  64. headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
  65. result = HTTPRequest().GET('http://item.taobao.com/item.htm?spm=a217i_h.1303267.1998177633-0.10.hx5m95&amp;id=12817144777', None, headers)
  66. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  67. 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'), ]
  68. result = HTTPRequest().GET('http://www.ptstest.com/s?wd=test', None, headers)
  69. PTS.Framework.addHttpCode(result.getStatusCode(), statusCode)
  70. if(not PTS.HttpUtilities.checkResponse(200)):
  71. PTS.Data.forCurrentTest.success = False
  72. return statusCode
  73. def action5(self):
  74. headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
  75. result = HTTPRequest().GET('http://www.ptstest.com/s?wd-test', None, headers)
  76. 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'), ]
  77. result = HTTPRequest().POST('http://www.ptstest.com/s', 'wd=test', headers)
  78. def action6(self):
  79. headers = [ NVPair('Accept', '*/*'), NVPair('Connection', 'Keep-Alive'), NVPair('User-Agent', 'PTS-HTTP-CLIENT'), ]
  80. result = HTTPRequest().GET('http://www.ptstest.com', None, headers)
  81. 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'), ]
  82. result = HTTPRequest().POST('http://www.ptstest.com/s', 'wd=test', headers)
  83. # 编织压测事务
  84. PTS.Framework.instrumentMethod(u'事务1', 'action3', TestRunner)
  85. PTS.Framework.instrumentMethod(u'事务2', 'action4', TestRunner)

TCP Socket 示例脚本

  1. #! /usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # PTS TCP Socket Script Template Version 1.0
  4. from util import PTS
  5. import socket
  6. class TestRunner:
  7. # TestRunner对象的初始化方法,每个线程在创建TestRunner后执行一次该方法
  8. def __init__(self):
  9. return
  10. # 主体压测方法,每个线程在测试生命周期内会循环调用该方法
  11. def __call__(self):
  12. PTS.Data.delayReports = 1
  13. statusCode = self.action1()
  14. PTS.Framework.setExtraData(statusCode)
  15. PTS.Data.report()
  16. PTS.Data.delayReports = 0
  17. # TestRunner销毁方法,每个线程循环执行完成后执行一次该方法
  18. def __del__(self):
  19. return
  20. # 业务函数
  21. def action1(self):
  22. statusCode = [0L, 0L, 0L, 0L]
  23. # 以下为socket协议脚本
  24. # 创建TCP Socket
  25. sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
  26. # 连接到服务器,服务器IP及端口号
  27. sk.connect(('localhost',2000))
  28. # 将发送内容送到服务器
  29. sendContent = 'This is tcp socket test,{Status:success}'
  30. sk.send(sendContent);
  31. # 接收服务器返回内容
  32. RecvContent = sk.recv(1024)
  33. PTS.Logger.info('Content come from server is:'+RecvContent)
  34. # 关闭连接
  35. sk.close()
  36. # socket协议脚本结束
  37. # 抓返回值
  38. value = PTS.HttpUtilities.valueFromTextBetween(RecvContent,'{Status:','}')
  39. # 判断事务是否成功
  40. if not value:
  41. PTS.Data.forCurrentTest.success = False
  42. PTS.Logger.info('Have no value')
  43. elif value.lower() == 'success':
  44. PTS.Data.forCurrentTest.success = True
  45. PTS.Logger.info('Success flag1:'+value)
  46. else:
  47. PTS.Data.forCurrentTest.success = False
  48. PTS.Logger.info('Success flag2:'+value)
  49. return statusCode
  50. # 编织压测事务
  51. PTS.Framework.instrumentMethod(u'事务名', 'action1', TestRunner)

UDP Socket 示例脚本

  1. #! /usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # PTS UDP Socket Script Version 1.0
  4. from util import PTS
  5. import socket
  6. class TestRunner:
  7. # TestRunner对象的初始化方法,每个线程在创建TestRunner后执行一次该方法
  8. def __init__(self):
  9. return
  10. # 主体压测方法,每个线程在测试生命周期内会循环调用该方法
  11. def __call__(self):
  12. PTS.Data.delayReports = 1
  13. statusCode = self.action1()
  14. PTS.Framework.setExtraData(statusCode)
  15. PTS.Data.report()
  16. PTS.Data.delayReports = 0
  17. # TestRunner销毁方法,每个线程循环执行完成后执行一次该方法
  18. def __del__(self):
  19. return
  20. # 业务函数
  21. def action1(self):
  22. statusCode = [0L, 0L, 0L, 0L]
  23. # 创建UDP Socket
  24. s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
  25. # 设置服务器IP地址及端口号
  26. port = 20000
  27. host = 'localhost'
  28. msg='''0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijkl
  29. mnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRS
  30. TUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnop
  31. qrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWX
  32. YZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrs
  33. tuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVW
  34. XYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghij
  35. klmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'''
  36. # 用Connect提高效率
  37. s.connect((host, port))
  38. # 将内容发送给服务器
  39. s.send(msg)
  40. # 设置接收超时,单位秒
  41. s.settimeout(30)
  42. # 接收服务器返回内容
  43. RecvContent,ADDR = s.recvfrom(1024)
  44. PTS.Logger.info('Content come from server:'+RecvContent)
  45. # 关闭连接
  46. s.close()
  47. # socket协议脚本结束
  48. # 抓返回值
  49. value = PTS.HttpUtilities.valueFromTextBetween(RecvContent,'{Status:','}')
  50. # 判断事务是否成功
  51. if not value:
  52. PTS.Data.forCurrentTest.success = False
  53. PTS.Logger.info('Have no value')
  54. elif value.lower() == 'success':
  55. PTS.Data.forCurrentTest.success = True
  56. PTS.Logger.info('Success flag1:'+value)
  57. else:
  58. PTS.Data.forCurrentTest.success = False
  59. PTS.Logger.info('Success flag2:'+value)
  60. return statusCode
  61. # 编织压测事务
  62. PTS.Framework.instrumentMethod(u'事务名', 'action1', TestRunner)
本文导读目录