这里主要是为了结合输出日志,及测试报告的呈现

普通接口请求方法封装

  1. #!/usr/bin/python
  2. # -*- coding: UTF-8 -*-
  3. # @Author :zaygee
  4. # @time :2020/12/3 16:30
  5. # @file :api_method.py
  6. import json
  7. import os
  8. import allure
  9. import jsonpath
  10. import requests
  11. from common import logger
  12. from common.path import RootPath
  13. from config import WEB_LOGIN_URL, LOGIN_DATA, OTHER_LOGIN_DATA
  14. requests.packages.urllib3.disable_warnings()
  15. headers = {
  16. "Connection": "keep-alive",
  17. "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) "
  18. "Chrome/58.0.3029.110 Safari/537.36",
  19. "Content-type": "application/json;charset=UTF-8",
  20. "Accept": "*/*",
  21. "DNT": "1",
  22. "Accept-Encoding": "gzip, deflate, br",
  23. "Accpet-Language": "zh-CN,zh;q=0.8,en;q=0.6"
  24. }
  25. class BaseRequests:
  26. """
  27. 封装通用调用方法,支持get、post、put、delete请求
  28. 封装常用请求方法
  29. """
  30. def __init__(self):
  31. self.req = requests.session()
  32. self.req.headers = headers
  33. @classmethod
  34. def _data_to_json(cls, data):
  35. """
  36. Args:
  37. data:接口请求参数/返回参数
  38. Return:
  39. 美化的json格式数据
  40. """
  41. if isinstance(data, str):
  42. return json.dumps(json.loads(data), indent=4, ensure_ascii=False)
  43. elif isinstance(data, dict):
  44. return json.dumps(data, indent=4, ensure_ascii=False)
  45. @classmethod
  46. def _api_data_log(cls, data=None, response=None):
  47. try:
  48. logger.info('*******用例开始执行~*********')
  49. if response is not None:
  50. logger.info('***Request-Url**:' + str(response.url))
  51. logger.info('******Request-Header**:' + '\n' + str(response.request.headers))
  52. allure.attach(name="请求接口", body=str(response.url))
  53. # allure.attach(name="请求头", body=str(response.request.headers))
  54. if data is not None:
  55. logger.info('******Request-Content**:' + '\n' + cls._data_to_json(data))
  56. allure.attach(name="请求参数", body=cls._data_to_json(data))
  57. else:
  58. logger.info('******Request-Content***:' + "无请求参数")
  59. allure.attach(name="请求参数", body=str("无请求参数"))
  60. if response is not None:
  61. logger.info('******Response-Code**:' + str(response.status_code))
  62. logger.info('******Response-Cookie**:' + '\n' + str(response.cookies.get_dict()))
  63. # allure.attach(name="响应码", body=str(response.status_code))
  64. # allure.attach(name="响应头cookie", body=str(response.cookies.get_dict()))
  65. if response.text == '':
  66. logger.info('******Response-Content***:' + "无响应参数")
  67. allure.attach(name="响应参数", body=str('无响应参数'))
  68. else:
  69. logger.info('******Response-Content**:' + '\n' + cls._data_to_json(response.text))
  70. allure.attach(name="响应参数", body=cls._data_to_json(response.text))
  71. logger.info('*******用例执行结束~*********')
  72. except ValueError as e:
  73. raise ValueError(e, '参数不存在,请检查~')
  74. def post_req(self, url, data=None, **args):
  75. """
  76. Args:
  77. url:post方法的请求URL
  78. data:请求参数
  79. args:字典类型的其他传参
  80. Return
  81. res:响应参数
  82. """
  83. res = self.req.post(url=url, json=data, verify=False, **args)
  84. self._api_data_log(data=data, response=res, )
  85. return res
  86. def get_req(self, url, params=None, **args):
  87. """
  88. Args:
  89. url:get方法的请求URL
  90. params:请求参数
  91. args:字典类型的其他传参
  92. Return
  93. res:响应参数
  94. """
  95. res = self.req.get(url=url, params=params, verify=False, **args)
  96. self._api_data_log(data=params, response=res)
  97. return res
  98. def put_req(self, url, data=None, **kwargs):
  99. """
  100. Args:
  101. url:put方法的请求URL
  102. params:请求参数
  103. args:字典类型的其他传参
  104. Return
  105. res:响应参数
  106. """
  107. res = self.req.put(url=url, json=data, verify=False, **kwargs)
  108. self._api_data_log(data=data, response=res, )
  109. return res
  110. def delete_req(self, url, **kwargs):
  111. """
  112. Args:
  113. url:delete方法的请求URL
  114. args:字典类型的其他传参
  115. Return
  116. res:响应参数
  117. """
  118. res = self.req.delete(url=url, verify=False, **kwargs)
  119. self._api_data_log(response=res)
  120. return res
  121. @classmethod
  122. def get_cookie(cls, url=WEB_LOGIN_URL, data=LOGIN_DATA):
  123. """
  124. Args:
  125. url: r'请求url,默认login url'
  126. data: r'请求参数,默认 login data'
  127. Return:
  128. r'返回响应头cookie'
  129. """
  130. res = base.post_req(url=url, data=data)
  131. try:
  132. if res.status_code == 200:
  133. return res.headers['Set-Cookie']
  134. except BaseException as e:
  135. raise (e, '获取cookie失败~!')
  136. @classmethod
  137. def get_cookie_val(cls, url=WEB_LOGIN_URL, data=LOGIN_DATA):
  138. """
  139. Args:
  140. url: r'请求url,默认login url'
  141. data: r'请求参数,默认 login data'
  142. Return:
  143. r'返回响应头cookie的值'
  144. """
  145. res = base.post_req(url=url, data=data)
  146. try:
  147. if res.status_code == 200:
  148. for key, value in res.cookies.get_dict().items():
  149. if key.find("authorization") != -1:
  150. return value
  151. break
  152. except BaseException as e:
  153. raise (e, '获取cookie的值失败~!')
  154. @classmethod
  155. def get_cookie_key_val(cls, url=WEB_LOGIN_URL, data=LOGIN_DATA):
  156. """
  157. Args:
  158. url: r'请求url,默认login url'
  159. data: r'请求参数,默认 login data'
  160. Return:
  161. r'返回响应头cookie'
  162. """
  163. res = base.post_req(url=url, data=data)
  164. try:
  165. if res.status_code == 200:
  166. for key, value in res.cookies.get_dict().items():
  167. if key.find("authorization") != -1:
  168. return key + '=' + value
  169. break
  170. except BaseException as e:
  171. raise (e, '获取cookie失败~!')
  172. @classmethod
  173. def write_cookie(cls):
  174. """
  175. 写入两个账号的cookie到json文件
  176. """
  177. cookie_file = os.path.join(RootPath.root_path, 'cookie.json')
  178. with open(cookie_file, 'w', encoding="utf-8") as f:
  179. cookie = cls.get_cookie_key_val()
  180. other_cookie = cls.get_cookie_key_val(data=OTHER_LOGIN_DATA)
  181. cookie = {'cookie': cookie, 'other_cookie': other_cookie}
  182. json.dump(cookie, f, indent=4)
  183. @classmethod
  184. def read_cookie(cls):
  185. """从json读取读取两个账号的cookie"""
  186. cookie_file = os.path.join(RootPath.root_path, 'cookie.json')
  187. with open(cookie_file, 'r', encoding='utf-8') as ff:
  188. data = ff.read()
  189. data = json.loads(data)
  190. return data
  191. @classmethod
  192. def write_cookie_value(cls):
  193. """写入两个账号的cookie的值到json文件"""
  194. cookie_file = os.path.join(RootPath.root_path, 'cookie_value.json')
  195. with open(cookie_file, 'w', encoding="utf-8") as f:
  196. cookie = cls.get_cookie_val()
  197. other_cookie = cls.get_cookie_val(data=OTHER_LOGIN_DATA)
  198. cookie = {'value': cookie, 'other_value': other_cookie}
  199. json.dump(cookie, f, indent=4)
  200. @classmethod
  201. def read_cookie_value(cls):
  202. """从json读取读取两个账号的cookie的值"""
  203. cookie_file = os.path.join(RootPath.root_path, 'cookie_value.json')
  204. with open(cookie_file, 'r', encoding='utf-8') as ff:
  205. data = ff.read()
  206. data = json.loads(data)
  207. return data
  208. @classmethod
  209. def extract(cls, data, extract):
  210. """
  211. Args:
  212. data:响应参数
  213. extract: jsonpath提取
  214. eg:({"code": 0,}, $.code, 0)
  215. Return:
  216. 提取期望响应参数
  217. """
  218. try:
  219. if data != '':
  220. result = jsonpath.jsonpath(json.loads(data), extract)
  221. logger.info("提取响应参数内容:" + str(result))
  222. return result[0]
  223. except ValueError as e:
  224. raise (e, "提取响应参数失败~")
  225. base = BaseRequests()

Websocket接口请求方法封装

  1. # -*- coding: utf-8 -*-#
  2. # file: socket_method.py
  3. # Author: zaygee
  4. # Date: 2021/5/17
  5. import allure
  6. import requests
  7. from websocket import create_connection
  8. from common import logger
  9. requests.packages.urllib3.disable_warnings()
  10. header = {
  11. 'Accept-Encoding': 'gzip, deflate, br',
  12. 'Connection': 'keep-alive',
  13. 'Content-Type': 'application/json;charset=UTF-8',
  14. 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) '
  15. 'Chrome/87.0.4280.88 Safari/537.36 '
  16. }
  17. def websocket_log(url=None, ws=None, count=1):
  18. try:
  19. logger.info('*******Socket用例开始执行~*********')
  20. if url is not None:
  21. logger.info("***Socket请求URL*** :" + '\n' + str(url))
  22. allure.attach(name="请求URL", body=str(url))
  23. if ws:
  24. logger.info("***连接状态*** :" + str(ws.status))
  25. allure.attach(name="连接状态", body=str(ws.status))
  26. if count is not None:
  27. for i in range(count):
  28. res = ws.recv()
  29. logger.info("***下发数据*** :" + str(res))
  30. allure.attach(name=f"下发数据{i}", body=str(res))
  31. if i == count - 1:
  32. return res
  33. logger.info('*******Socket用例执行结束~*********')
  34. except ValueError as e:
  35. raise (e, 'error!')
  36. class SocketCon:
  37. @classmethod
  38. def con(cls, url, cookie=None):
  39. """
  40. 开始链接
  41. Args:
  42. url-->请求url
  43. cookie-->cookie,默认不传
  44. Return:
  45. websocket连接成功对象
  46. """
  47. ws = create_connection(url=url, cookie=cookie)
  48. # websocket.enableTrace(True)
  49. logger.info('****Socket链接开始****')
  50. logger.info('***Socket请求Url***:' + '\n' + str(url))
  51. allure.attach(name=f"请求Url", body=str(url))
  52. logger.info('***连接状态***:' + str(ws.status))
  53. return ws
  54. @staticmethod
  55. def recv_log(ws, count):
  56. """
  57. 接收服务端消息并输出日志
  58. Args:
  59. ws-->websocket连接成功对象
  60. count-->接收下发数据次数
  61. Return:
  62. 最后一次下发数据,主要用于断言
  63. """
  64. for i in range(count):
  65. res = ws.recv()
  66. logger.info('***下发数据***:' + str(res))
  67. allure.attach(name=f"下发数据{i}", body=str(res))
  68. if i == count - 1:
  69. logger.info("*****Socket链接结束*****")
  70. return res