mirror of https://gitee.com/a529548204/apitest.git
优化请求头
This commit is contained in:
parent
17c05ae963
commit
92a94461b4
|
@ -27,9 +27,9 @@
|
||||||
|
|
||||||
``` yaml
|
``` yaml
|
||||||
login:
|
login:
|
||||||
- info: "用户名登录-成功" \
|
- info: "用户名登录-成功"
|
||||||
headers: {
|
headers: {
|
||||||
"Content-Type": "application/json"\
|
"Content-Type": "application/json"
|
||||||
}
|
}
|
||||||
data:
|
data:
|
||||||
param: {
|
param: {
|
||||||
|
|
|
@ -33,7 +33,7 @@ class apiSend(object):
|
||||||
dataran = replace_random(data)
|
dataran = replace_random(data)
|
||||||
return dataran
|
return dataran
|
||||||
|
|
||||||
def post(self, address, header,rel,request_parameter_type="json", timeout=8, data=None, files=None, host="host"):
|
def post(self, address, header, rel, timeout=8, data=None, files=None, host="host"):
|
||||||
"""
|
"""
|
||||||
post请求
|
post请求
|
||||||
:param host:
|
:param host:
|
||||||
|
@ -43,7 +43,6 @@ class apiSend(object):
|
||||||
name: 'code'
|
name: 'code'
|
||||||
:param address: 请求地址
|
:param address: 请求地址
|
||||||
:param header: 请求头
|
:param header: 请求头
|
||||||
:param request_parameter_type: 请求参数格式(form_data,raw)
|
|
||||||
:param timeout: 超时时间
|
:param timeout: 超时时间
|
||||||
:param data: 请求参数
|
:param data: 请求参数
|
||||||
:param files: 文件路径
|
:param files: 文件路径
|
||||||
|
@ -56,7 +55,7 @@ class apiSend(object):
|
||||||
logging.info("请求地址:%s" % "" + url)
|
logging.info("请求地址:%s" % "" + url)
|
||||||
logging.info("请求头: %s" % str(header))
|
logging.info("请求头: %s" % str(header))
|
||||||
logging.info("请求参数: %s" % str(data_random))
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
if 'form-data' in request_parameter_type:
|
if "multipart/form-data" in header.values:
|
||||||
with allure.step("POST上传文件"):
|
with allure.step("POST上传文件"):
|
||||||
allure.attach(name="请求地址", body=url)
|
allure.attach(name="请求地址", body=url)
|
||||||
allure.attach(name="请求头", body=str(header))
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
@ -77,16 +76,18 @@ class apiSend(object):
|
||||||
boundary='-----------------------------' + str(random.randint(int(1e28), int(1e29 - 1)))
|
boundary='-----------------------------' + str(random.randint(int(1e28), int(1e29 - 1)))
|
||||||
)
|
)
|
||||||
header['Content-Type'] = multipart.content_type
|
header['Content-Type'] = multipart.content_type
|
||||||
|
|
||||||
response = requests.post(url=url, data=data, headers=header, timeout=timeout)
|
response = requests.post(url=url, data=data, headers=header, timeout=timeout)
|
||||||
else:
|
else:
|
||||||
# print(type(data))
|
|
||||||
multipart = MultipartEncoder(data)
|
multipart = MultipartEncoder(data)
|
||||||
|
|
||||||
response = requests.post(url=url, data=multipart, headers={'Content-Type': multipart.content_type},
|
response = requests.post(url=url, data=multipart, headers={'Content-Type': multipart.content_type},
|
||||||
timeout=timeout)
|
timeout=timeout)
|
||||||
|
elif 'application/json' in header.values:
|
||||||
|
with allure.step("POST请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.post(url=url, json=data_random, headers=header, timeout=timeout)
|
||||||
else:
|
else:
|
||||||
|
|
||||||
with allure.step("POST请求接口"):
|
with allure.step("POST请求接口"):
|
||||||
allure.attach(name="请求地址", body=url)
|
allure.attach(name="请求地址", body=url)
|
||||||
allure.attach(name="请求头", body=str(header))
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
@ -98,14 +99,13 @@ class apiSend(object):
|
||||||
return response.text
|
return response.text
|
||||||
else:
|
else:
|
||||||
logging.info("请求接口结果: %s" % str(response.json()))
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
self.rel.relevance(rel,bodys=data_random,res=response.json())
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
return response.json(), response.elapsed.total_seconds()
|
return response.json(), response.elapsed.total_seconds()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.exception('ERROR')
|
|
||||||
logging.error(e)
|
logging.error(e)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
def get(self, address,rel, header, data, timeout=8, host="host"):
|
def get(self, address, rel, header, data, timeout=8, host="host"):
|
||||||
"""
|
"""
|
||||||
get请求
|
get请求
|
||||||
:param host:
|
:param host:
|
||||||
|
@ -128,7 +128,6 @@ class apiSend(object):
|
||||||
logging.info("请求地址:%s" % "" + url)
|
logging.info("请求地址:%s" % "" + url)
|
||||||
logging.info("请求头: %s" % str(header))
|
logging.info("请求头: %s" % str(header))
|
||||||
logging.info("请求参数: %s" % str(data_random))
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
|
||||||
with allure.step("GET请求接口"):
|
with allure.step("GET请求接口"):
|
||||||
allure.attach(name="请求地址", body=url)
|
allure.attach(name="请求地址", body=url)
|
||||||
allure.attach(name="请求头", body=str(header))
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
@ -141,11 +140,10 @@ class apiSend(object):
|
||||||
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
return response.json(), response.elapsed.total_seconds()
|
return response.json(), response.elapsed.total_seconds()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.exception('ERROR')
|
|
||||||
logging.error(e)
|
logging.error(e)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
def put(self, address,rel, header, request_parameter_type="json", timeout=8, data=None, files=None, host="host"):
|
def put(self, address, rel, header, timeout=8, data=None, files=None, host="host"):
|
||||||
"""
|
"""
|
||||||
put请求
|
put请求
|
||||||
:param host:
|
:param host:
|
||||||
|
@ -155,7 +153,7 @@ class apiSend(object):
|
||||||
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
name: 'code'
|
name: 'code'
|
||||||
:param header: 请求头
|
:param header: 请求头
|
||||||
:param request_parameter_type: 请求参数格式(form_data,raw)
|
|
||||||
:param timeout: 超时时间
|
:param timeout: 超时时间
|
||||||
:param data: 请求参数
|
:param data: 请求参数
|
||||||
:param files: 文件路径
|
:param files: 文件路径
|
||||||
|
@ -165,12 +163,15 @@ class apiSend(object):
|
||||||
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
logging.info("请求地址:%s" % "" + url)
|
logging.info("请求地址:%s" % "" + url)
|
||||||
logging.info("请求头: %s" % str(header))
|
logging.info("请求头: %s" % str(header))
|
||||||
|
|
||||||
if request_parameter_type == 'raw':
|
|
||||||
data_random = self.iniDatas(data["param"])
|
data_random = self.iniDatas(data["param"])
|
||||||
else:
|
|
||||||
data_random = data
|
|
||||||
logging.info("请求参数: %s" % str(data_random))
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
if 'application/json' in header.values:
|
||||||
|
with allure.step("PUT请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.put(url=url, json=data_random, headers=header, timeout=timeout, files=files)
|
||||||
|
else:
|
||||||
with allure.step("PUT请求接口"):
|
with allure.step("PUT请求接口"):
|
||||||
allure.attach(name="请求地址", body=url)
|
allure.attach(name="请求地址", body=url)
|
||||||
allure.attach(name="请求头", body=str(header))
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
@ -182,11 +183,10 @@ class apiSend(object):
|
||||||
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
return response.json(), response.elapsed.total_seconds()
|
return response.json(), response.elapsed.total_seconds()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.exception('ERROR')
|
|
||||||
logging.error(e)
|
logging.error(e)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
def delete(self, address, rel,header, data, timeout=8, host="host"):
|
def delete(self, address, rel, header, data, timeout=8, host="host"):
|
||||||
"""
|
"""
|
||||||
get请求
|
get请求
|
||||||
:param host:
|
:param host:
|
||||||
|
@ -217,20 +217,19 @@ class apiSend(object):
|
||||||
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
return response.json(), response.elapsed.total_seconds()
|
return response.json(), response.elapsed.total_seconds()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.exception('ERROR')
|
|
||||||
logging.error(e)
|
logging.error(e)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
def __call__(self,rel, address, method, headers, data,**kwargs):
|
def __call__(self, rel, address, method, headers, data, **kwargs):
|
||||||
try:
|
try:
|
||||||
if method == "post" or method == 'POST':
|
if method == "post" or method == 'POST':
|
||||||
return self.post(address=address, data=data, header=headers,rel=rel, **kwargs)
|
return self.post(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
elif method == "get" or method == 'GET':
|
elif method == "get" or method == 'GET':
|
||||||
return self.get(address=address, data=data, header=headers,rel=rel, **kwargs)
|
return self.get(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
elif method == "delete" or method == 'DELETE':
|
elif method == "delete" or method == 'DELETE':
|
||||||
return self.delete(address=address, data=data, header=headers,rel=rel, ** kwargs)
|
return self.delete(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
elif method == "put" or method == 'PUT':
|
elif method == "put" or method == 'PUT':
|
||||||
return self.put(address=address, data=data, header=headers,rel=rel, **kwargs)
|
return self.put(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
else:
|
else:
|
||||||
raise TypeError(f"请求异常,检查yml文件method")
|
raise TypeError(f"请求异常,检查yml文件method")
|
||||||
except Exception:
|
except Exception:
|
||||||
|
@ -239,8 +238,7 @@ class apiSend(object):
|
||||||
|
|
||||||
apisend = apiSend()
|
apisend = apiSend()
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
ress = apisend("/$url(home_id)$", "get", data={"param":None,"urlparam": {"home_id": 123}}, headers={
|
ress = apisend("/$url(home_id)$", "get", data={"param": None, "urlparam": {"home_id": 123}}, headers={
|
||||||
"Content-Type": "application/json"
|
"Content-Type": "application/json"
|
||||||
}, host="host")
|
}, host="host")
|
||||||
print(ress)
|
print(ress)
|
||||||
que
|
|
|
@ -0,0 +1,447 @@
|
||||||
|
# coding:utf-8
|
||||||
|
"""
|
||||||
|
@author: 井松
|
||||||
|
@contact: 529548204@qq.com
|
||||||
|
@file: basePageSession.py
|
||||||
|
@time: 2021/11/19 9:57
|
||||||
|
"""
|
||||||
|
# coding:utf-8
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import random
|
||||||
|
import time
|
||||||
|
|
||||||
|
import allure
|
||||||
|
import requests
|
||||||
|
import simplejson
|
||||||
|
from requests_toolbelt import MultipartEncoder
|
||||||
|
|
||||||
|
from scripts.log import Log
|
||||||
|
from scripts.randomData import replace_random
|
||||||
|
from config.confManage import host_manage
|
||||||
|
from scripts.relevance import Relevance
|
||||||
|
|
||||||
|
Log()
|
||||||
|
|
||||||
|
|
||||||
|
class apiSend(object):
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.http_type = host_manage(hos="${http_type}$")
|
||||||
|
self.rel = Relevance()
|
||||||
|
self.session = requests.Session()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def iniDatas(data):
|
||||||
|
if isinstance(data, dict):
|
||||||
|
data = json.dumps(data)
|
||||||
|
if data is None:
|
||||||
|
return data
|
||||||
|
dataran = replace_random(data)
|
||||||
|
return dataran
|
||||||
|
|
||||||
|
def post(self, address, header, rel, timeout=8, data=None, files=None, host="host"):
|
||||||
|
"""
|
||||||
|
post请求
|
||||||
|
:param host:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param address: 请求地址
|
||||||
|
:param header: 请求头
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:param data: 请求参数
|
||||||
|
:param files: 文件路径
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
if "multipart/form-data" in header.values:
|
||||||
|
with allure.step("POST上传文件"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
if files is not None:
|
||||||
|
for i in files:
|
||||||
|
value = files[i]
|
||||||
|
if isinstance(value, int):
|
||||||
|
files[i] = str(value)
|
||||||
|
pass
|
||||||
|
elif '/' in value:
|
||||||
|
file_parm = i
|
||||||
|
files[file_parm] = (os.path.basename(value), open(value, 'rb'), 'application/octet-stream')
|
||||||
|
else:
|
||||||
|
pass
|
||||||
|
multipart = MultipartEncoder(
|
||||||
|
fields=files,
|
||||||
|
boundary='-----------------------------' + str(random.randint(int(1e28), int(1e29 - 1)))
|
||||||
|
)
|
||||||
|
header['Content-Type'] = multipart.content_type
|
||||||
|
response = requests.post(url=url, data=data, headers=header, timeout=timeout)
|
||||||
|
else:
|
||||||
|
multipart = MultipartEncoder(data)
|
||||||
|
response = requests.post(url=url, data=multipart, headers={'Content-Type': multipart.content_type},
|
||||||
|
timeout=timeout)
|
||||||
|
elif 'application/json' in header.values:
|
||||||
|
with allure.step("POST请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.post(url=url, json=data_random, headers=header, timeout=timeout)
|
||||||
|
else:
|
||||||
|
with allure.step("POST请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.post(url=url, data=data_random, headers=header, timeout=timeout)
|
||||||
|
try:
|
||||||
|
if response.status_code != 200:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.text))
|
||||||
|
return response.text
|
||||||
|
else:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def get(self, address, rel, header, data, timeout=8, host="host"):
|
||||||
|
"""
|
||||||
|
get请求
|
||||||
|
:param host:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param address:
|
||||||
|
:param header: 请求头
|
||||||
|
:param data: 请求参数
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
# if isinstance(data, dict):
|
||||||
|
# if "urlparam" in data.keys():
|
||||||
|
# address = replace_random(address, param=data["urlparam"])
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
with allure.step("GET请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.get(url=url, params=data_random, headers=header, timeout=timeout)
|
||||||
|
if response.status_code == 301:
|
||||||
|
response = requests.get(url=response.headers["location"])
|
||||||
|
try:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def put(self, address, rel, header, timeout=8, data=None, files=None, host="host"):
|
||||||
|
"""
|
||||||
|
put请求
|
||||||
|
:param host:
|
||||||
|
:param address:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param header: 请求头
|
||||||
|
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:param data: 请求参数
|
||||||
|
:param files: 文件路径
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
if 'application/json' in header.values:
|
||||||
|
with allure.step("PUT请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.put(url=url, json=data_random, headers=header, timeout=timeout, files=files)
|
||||||
|
else:
|
||||||
|
with allure.step("PUT请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
|
||||||
|
response = requests.put(url=url, data=data_random, headers=header, timeout=timeout, files=files)
|
||||||
|
try:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def delete(self, address, rel, header, data, timeout=8, host="host"):
|
||||||
|
"""
|
||||||
|
get请求
|
||||||
|
:param host:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param address:
|
||||||
|
:param header: 请求头
|
||||||
|
:param data: 请求参数
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
with allure.step("DELETE请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.delete(url=url, params=data_random, headers=header, timeout=timeout)
|
||||||
|
|
||||||
|
try:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def post_session(self, address, header, rel, timeout=8, data=None, files=None, host="host"):
|
||||||
|
"""
|
||||||
|
post请求
|
||||||
|
:param host:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param address: 请求地址
|
||||||
|
:param header: 请求头
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:param data: 请求参数
|
||||||
|
:param files: 文件路径
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
if "multipart/form-data" in header.values:
|
||||||
|
with allure.step("POST上传文件"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
if files is not None:
|
||||||
|
for i in files:
|
||||||
|
value = files[i]
|
||||||
|
if isinstance(value, int):
|
||||||
|
files[i] = str(value)
|
||||||
|
pass
|
||||||
|
elif '/' in value:
|
||||||
|
file_parm = i
|
||||||
|
files[file_parm] = (os.path.basename(value), open(value, 'rb'), 'application/octet-stream')
|
||||||
|
else:
|
||||||
|
pass
|
||||||
|
multipart = MultipartEncoder(
|
||||||
|
fields=files,
|
||||||
|
boundary='-----------------------------' + str(random.randint(int(1e28), int(1e29 - 1)))
|
||||||
|
)
|
||||||
|
header['Content-Type'] = multipart.content_type
|
||||||
|
response = self.session.post(url=url, data=data, headers=header, timeout=timeout)
|
||||||
|
else:
|
||||||
|
multipart = MultipartEncoder(data)
|
||||||
|
response = self.session.post(url=url, data=multipart, headers={'Content-Type': multipart.content_type},
|
||||||
|
timeout=timeout)
|
||||||
|
elif 'application/json' in header.values:
|
||||||
|
with allure.step("POST请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = self.session.post(url=url, json=data_random, headers=header, timeout=timeout)
|
||||||
|
else:
|
||||||
|
with allure.step("POST请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = self.session.post(url=url, data=data_random, headers=header, timeout=timeout)
|
||||||
|
try:
|
||||||
|
if response.status_code != 200:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.text))
|
||||||
|
return response.text
|
||||||
|
else:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def get_session(self, address, rel, header, data, timeout=8, host="host"):
|
||||||
|
"""
|
||||||
|
get请求
|
||||||
|
:param host:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param address:
|
||||||
|
:param header: 请求头
|
||||||
|
:param data: 请求参数
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
# if isinstance(data, dict):
|
||||||
|
# if "urlparam" in data.keys():
|
||||||
|
# address = replace_random(address, param=data["urlparam"])
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
with allure.step("GET请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = requests.get(url=url, params=data_random, headers=header, timeout=timeout)
|
||||||
|
if response.status_code == 301:
|
||||||
|
response = self.session.get(url=response.headers["location"])
|
||||||
|
try:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def put_session(self, address, rel, header, timeout=8, data=None, files=None, host="host"):
|
||||||
|
"""
|
||||||
|
put请求
|
||||||
|
:param host:
|
||||||
|
:param address:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param header: 请求头
|
||||||
|
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:param data: 请求参数
|
||||||
|
:param files: 文件路径
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
if 'application/json' in header.values:
|
||||||
|
with allure.step("PUT请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = self.session.put(url=url, json=data_random, headers=header, timeout=timeout, files=files)
|
||||||
|
else:
|
||||||
|
with allure.step("PUT请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
|
||||||
|
response = self.session.put(url=url, data=data_random, headers=header, timeout=timeout, files=files)
|
||||||
|
try:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def delete_session(self, address, rel, header, data, timeout=8, host="host"):
|
||||||
|
"""
|
||||||
|
get请求
|
||||||
|
:param host:
|
||||||
|
:param rel: 关联情况
|
||||||
|
- cachefrom: 'body' # response : 从结果中获取 body : 从参数中获取
|
||||||
|
path: '$.code' # body如果是 "id=2&path=haha" 会转换成字典 然后根据path使用jsonpath取值
|
||||||
|
name: 'code'
|
||||||
|
:param address:
|
||||||
|
:param header: 请求头
|
||||||
|
:param data: 请求参数
|
||||||
|
:param timeout: 超时时间
|
||||||
|
:return:
|
||||||
|
"""
|
||||||
|
iniaddress = replace_random(address, param=data["urlparam"])
|
||||||
|
data_random = self.iniDatas(data["param"])
|
||||||
|
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + iniaddress
|
||||||
|
logging.info("请求地址:%s" % "" + url)
|
||||||
|
logging.info("请求头: %s" % str(header))
|
||||||
|
logging.info("请求参数: %s" % str(data_random))
|
||||||
|
with allure.step("DELETE请求接口"):
|
||||||
|
allure.attach(name="请求地址", body=url)
|
||||||
|
allure.attach(name="请求头", body=str(header))
|
||||||
|
allure.attach(name="请求参数", body=str(data_random))
|
||||||
|
response = self.session.delete(url=url, params=data_random, headers=header, timeout=timeout)
|
||||||
|
|
||||||
|
try:
|
||||||
|
logging.info("请求接口结果: %s" % str(response.json()))
|
||||||
|
self.rel.relevance(rel, bodys=data_random, res=response.json())
|
||||||
|
return response.json(), response.elapsed.total_seconds()
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(e)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def __call__(self, rel, address, method:str, headers, data, **kwargs):
|
||||||
|
try:
|
||||||
|
if method.lower() == "post":
|
||||||
|
return self.post(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
elif method.lower() == "get":
|
||||||
|
return self.get(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
elif method.lower() == "delete":
|
||||||
|
return self.delete(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
elif method.lower() == "put" or method == 'PUT':
|
||||||
|
return self.put(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
if method.lower() == "post_session":
|
||||||
|
return self.post(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
elif method.lower() == "get_session":
|
||||||
|
return self.get(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
elif method.lower() == "delete_session":
|
||||||
|
return self.delete(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
elif method.lower() == "put_session":
|
||||||
|
return self.put(address=address, data=data, header=headers, rel=rel, **kwargs)
|
||||||
|
else:
|
||||||
|
raise TypeError(f"请求异常,检查yml文件method")
|
||||||
|
except Exception:
|
||||||
|
raise
|
||||||
|
|
||||||
|
|
||||||
|
apisend = apiSend()
|
||||||
|
if __name__ == '__main__':
|
||||||
|
ress = apisend("/$url(home_id)$", "get", data={"param": None, "urlparam": {"home_id": 123}}, headers={
|
||||||
|
"Content-Type": "application/json"
|
||||||
|
}, host="host")
|
||||||
|
print(ress)
|
|
@ -49,3 +49,6 @@ class Test_alarm(object):
|
||||||
casedata["headers"]["Authorization"] = "JWT " + setup_login["data"]["token"]
|
casedata["headers"]["Authorization"] = "JWT " + setup_login["data"]["token"]
|
||||||
res, restime = alarmDistribute(casedata)
|
res, restime = alarmDistribute(casedata)
|
||||||
asserting(hope_res=casedata["assert"], real_res=res, re_time=restime)
|
asserting(hope_res=casedata["assert"], real_res=res, re_time=restime)
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
print(111)
|
Loading…
Reference in New Issue