全部产品
性能测试 PTS

脚本示例

更新时间:2017-09-13 16:30:59   分享:   

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&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&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)
  52. 附录三:UDP Socket示例脚本
  53. #! /usr/bin/env python
  54. # -*- coding: utf-8 -*-
  55. # PTS UDP Socket Script Version 1.0
  56. from util import PTS
  57. import socket
  58. class TestRunner:
  59. # TestRunner对象的初始化方法,每个线程在创建TestRunner后执行一次该方法
  60. def __init__(self):
  61. return
  62. # 主体压测方法,每个线程在测试生命周期内会循环调用该方法
  63. def __call__(self):
  64. PTS.Data.delayReports = 1
  65. statusCode = self.action1()
  66. PTS.Framework.setExtraData(statusCode)
  67. PTS.Data.report()
  68. PTS.Data.delayReports = 0
  69. # TestRunner销毁方法,每个线程循环执行完成后执行一次该方法
  70. def __del__(self):
  71. return
  72. # 业务函数
  73. def action1(self):
  74. statusCode = [0L, 0L, 0L, 0L]
  75. #创建UDP Socket
  76. s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
  77. #设置服务器IP地址及端口号
  78. port = 20000
  79. host = 'localhost'
  80. msg='''0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijkl
  81. mnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRS
  82. TUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnop
  83. qrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWX
  84. YZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrs
  85. tuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVW
  86. XYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghij
  87. klmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'''
  88. #用Connect提高效率
  89. s.connect((host, port))
  90. #将内容发送给服务器
  91. s.send(msg)
  92. #设置接收超时,单位秒
  93. s.settimeout(30)
  94. #接收服务器返回内容
  95. RecvContent,ADDR = s.recvfrom(1024)
  96. PTS.Logger.info('Content come from server:'+RecvContent)
  97. #关闭连接
  98. s.close()
  99. #socket协议脚本结束
  100. #抓返回值
  101. value = PTS.HttpUtilities.valueFromTextBetween(RecvContent,'{Status:','}')
  102. #判断事务是否成功
  103. if not value:
  104. PTS.Data.forCurrentTest.success = False
  105. PTS.Logger.info('Have no value')
  106. elif value.lower() == 'success':
  107. PTS.Data.forCurrentTest.success = True
  108. PTS.Logger.info('Success flag1:'+value)
  109. else:
  110. PTS.Data.forCurrentTest.success = False
  111. PTS.Logger.info('Success flag2:'+value)
  112. return statusCode
  113. # 编织压测事务
  114. PTS.Framework.instrumentMethod(u'事务名', 'action1', TestRunner)
本文导读目录
本文导读目录
以上内容是否对您有帮助?