mirror of https://gitee.com/a529548204/apitest.git
230 lines
8.7 KiB
Python
230 lines
8.7 KiB
Python
# coding:utf-8
|
||
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
|
||
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
|
||
|
||
Log()
|
||
|
||
|
||
class apiSend(object):
|
||
|
||
def __init__(self):
|
||
self.http_type = host_manage(hos="${http_type}$")
|
||
|
||
@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, request_parameter_type="json", timeout=8, data=None, files=None,host="host"):
|
||
"""
|
||
post请求
|
||
:param host:
|
||
:param address: 请求地址
|
||
:param header: 请求头
|
||
:param request_parameter_type: 请求参数格式（form_data,raw）
|
||
:param timeout: 超时时间
|
||
:param data: 请求参数
|
||
:param files: 文件路径
|
||
:return:
|
||
"""
|
||
|
||
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + address
|
||
logging.info("请求地址：%s" % "" + url)
|
||
logging.info("请求地址：%s" % "" + str(address))
|
||
logging.info("请求头: %s" % str(header))
|
||
|
||
if 'form-data' in request_parameter_type:
|
||
with allure.step("POST上传文件"):
|
||
allure.attach(name="请求地址", body=url)
|
||
allure.attach(name="请求头", body=str(header))
|
||
allure.attach(name="请求参数", body=str(data))
|
||
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:
|
||
# print(type(data))
|
||
multipart = MultipartEncoder(data)
|
||
|
||
response = requests.post(url=url, data=multipart, headers={'Content-Type': multipart.content_type},
|
||
timeout=timeout)
|
||
else:
|
||
data_random = self.iniDatas(data)
|
||
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()))
|
||
return response.json(), response.elapsed.total_seconds()
|
||
except json.decoder.JSONDecodeError:
|
||
return ''
|
||
except simplejson.errors.JSONDecodeError:
|
||
return ''
|
||
except Exception as e:
|
||
logging.exception('ERROR')
|
||
logging.error(e)
|
||
raise
|
||
|
||
def get(self, address, header, data, timeout=8,host="host"):
|
||
"""
|
||
get请求
|
||
:param host:
|
||
:param address:
|
||
:param header: 请求头
|
||
:param data: 请求参数
|
||
:param timeout: 超时时间
|
||
:return:
|
||
"""
|
||
data_random = self.iniDatas(data)
|
||
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + address
|
||
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()))
|
||
return response.json(), response.elapsed.total_seconds()
|
||
# except json.decoder.JSONDecodeError:
|
||
# return ''
|
||
# except simplejson.errors.JSONDecodeError:
|
||
# return ''
|
||
except Exception as e:
|
||
logging.exception('ERROR')
|
||
logging.error(e)
|
||
raise
|
||
|
||
def put(self, address, header, request_parameter_type="json", timeout=8, data=None, files=None,host="host"):
|
||
"""
|
||
put请求
|
||
:param host:
|
||
:param address:
|
||
:param header: 请求头
|
||
:param request_parameter_type: 请求参数格式（form_data,raw）
|
||
:param timeout: 超时时间
|
||
:param data: 请求参数
|
||
:param files: 文件路径
|
||
:return:
|
||
"""
|
||
|
||
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + address
|
||
logging.info("请求地址：%s" % "" + url)
|
||
logging.info("请求头: %s" % str(header))
|
||
|
||
with allure.step("PUT请求接口"):
|
||
allure.attach(name="请求地址", body=url)
|
||
allure.attach(name="请求头", body=str(header))
|
||
allure.attach(name="请求参数", body=str(data))
|
||
|
||
if request_parameter_type == 'raw':
|
||
data = self.iniDatas(data)
|
||
else:
|
||
data = data
|
||
response = requests.put(url=url, data=data, headers=header, timeout=timeout, files=files)
|
||
try:
|
||
logging.info("请求接口结果： %s" % str(response.json()))
|
||
return response.json(), response.elapsed.total_seconds()
|
||
except json.decoder.JSONDecodeError:
|
||
return ''
|
||
except simplejson.errors.JSONDecodeError:
|
||
return ''
|
||
except Exception as e:
|
||
logging.exception('ERROR')
|
||
logging.error(e)
|
||
raise
|
||
|
||
def delete(self, address, header, data, timeout=8,host="host"):
|
||
"""
|
||
get请求
|
||
:param host:
|
||
:param address:
|
||
:param header: 请求头
|
||
:param data: 请求参数
|
||
:param timeout: 超时时间
|
||
:return:
|
||
"""
|
||
data_random = self.iniDatas(data)
|
||
url = str(self.http_type) + "://" + host_manage(hos='${{{}}}$'.format(host)) + address
|
||
logging.info("请求地址：%s" % "" + url)
|
||
logging.info("请求头: %s" % str(header))
|
||
|
||
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()))
|
||
return response.json(), response.elapsed.total_seconds()
|
||
except json.decoder.JSONDecodeError:
|
||
return ''
|
||
except simplejson.errors.JSONDecodeError:
|
||
return ''
|
||
except Exception as e:
|
||
logging.exception('ERROR')
|
||
logging.error(e)
|
||
raise
|
||
|
||
def __call__(self, address, method, headers, data, **kwargs):
|
||
try:
|
||
if method == "post" or method == 'POST':
|
||
return self.post(address=address, data=data, header=headers, **kwargs)
|
||
elif method == "get" or method == 'GET':
|
||
return self.get(address=address, data=data, header=headers, **kwargs)
|
||
elif method == "delete" or method == 'DELETE':
|
||
return self.delete(address=address, data=data, header=headers ** kwargs)
|
||
elif method == "put" or method == 'PUT':
|
||
return self.put(address=address, data=data, header=headers, **kwargs)
|
||
else:
|
||
raise TypeError(f"请求异常,检查yml文件method")
|
||
except Exception:
|
||
raise
|
||
|
||
|
||
apisend = apiSend()
|
||
if __name__ == '__main__':
|
||
apisend("/123123","get",data="er",headers={
|
||
"Content-Type": "application/json"
|
||
},host="host2") |