import json import pickle import requests from tools import loggerKit class http_util: def __init__(self): self.api_domain = None @staticmethod def __http_get(interface_url, header_data, interface_param): """ :param interface_url: 接口地址 :param header_data: 请求头文件 :param interface_param: 接口请求参数 :return: 字典形式结果 """ result = None try: if interface_url != '': req_url = interface_url response = requests.get(url=req_url, headers=header_data, verify=False, timeout=40, params=interface_param ) if response.status_code == 200: response_time = response.elapsed.microseconds / 1000 # 发起请求和响应到达的时间,单位ms result = response.text if response.status_code == 404: result = json.dumps({'接口返回状态': response.status_code}, ensure_ascii=False) loggerKit.error(result) elif interface_url == '': result = json.dumps({'接口地址参数为空': ''}, ensure_ascii=False) loggerKit.error(result) else: result = json.dumps({'接口地址错误': ''}, ensure_ascii=False) loggerKit.error(result) except Exception as e: result = json.dumps({'未知错误': e}, ensure_ascii=False) loggerKit.error(result) return result @staticmethod def __http_post(interface_url, header_data, interface_param): """ :param interface_url: 接口地址 :param header_data: 请求头文件 :param interface_param: 接口请求参数 :return: 字典形式结果 """ print("headers", type(header_data)) print("interface_param", type(interface_param)) result = None response = None try: if interface_url != '': if 'application/x-www-form-urlencoded' in header_data.get('Content-Type'): response = requests.post(url=interface_url, headers=header_data, # 字典 data=interface_param, # 字典 verify=False, timeout=40) if 'application/json' in header_data.get('Content-Type'): response = requests.post(url=interface_url, headers=header_data, # 字典 json=interface_param, # 字典 verify=False, timeout=40) if 'multipart/form-data' in header_data.get('Content-Type'): # files = {'file': open('upload.txt', 'rb')} response = requests.post(url=interface_url, headers=header_data, # 字典 files=interface_param, # 字典 verify=False, timeout=40) if response.status_code == 200: response_time = response.elapsed.microseconds / 1000 # 发起请求和响应到达的时间,单位ms loggerKit.debug('接口返回结果：%s' % response.text) result = response.text if response.status_code == 404: result = json.dumps({'接口返回状态:%s': response.status_code}, ensure_ascii=False) loggerKit.error(result) elif interface_url == '': result = json.dumps({'接口地址参数为空': ''}, ensure_ascii=False) loggerKit.error(result) else: result = json.dumps({'接口地址错误': ''}, ensure_ascii=False) loggerKit.error(result) except Exception as e: result = json.dumps({'未知错误': e}, ensure_ascii=False) loggerKit.error(result) return result @staticmethod def __http_put(interface_url, header_data, interface_param): """ :param interface_url: 请求URL :param header_data: 请求头参数 :param interface_param: 请求体参数 :return: """ result = None response = None try: if 'application/x-www-form-urlencoded' in header_data.get('Content-Type'): interface_param = interface_param elif 'application/json' in header_data.get('Content-Type'): interface_param = json.dumps(interface_param) else: pass print(type(interface_param), interface_param) if interface_url != '': req_url = interface_url response = requests.put(url=req_url, headers=header_data, data=interface_param, verify=False, timeout=40) if response.status_code == 200: result = response.text if response.status_code == 404: result = json.dumps({'接口返回状态': response.status_code}, ensure_ascii=False) elif interface_url == '': result = json.dumps({'接口地址参数为空': ''}, ensure_ascii=False) loggerKit.error(result) else: result = json.dumps({'接口地址错误': ''}, ensure_ascii=False) loggerKit.error(result) except Exception as e: result = json.dumps({'未知错误': e}, ensure_ascii=False) loggerKit.error(result) return result @staticmethod def set_http_head(header_data): """ 公共设置请求头方法 设置http请求头，参数为json格式的字符串：{"Content-Type":“application/json”} """ http_head = None if not header_data: http_head = {} else: if isinstance(header_data, dict): http_head = header_data elif isinstance(header_data, str) and header_data.startswith('{') and header_data.endswith('}'): http_head = json.loads(header_data) elif isinstance(header_data, bytes): http_head = pickle.loads(header_data) else: pass return http_head @staticmethod def set_http_body(body): """ 公共设置请求体方法 设置http请求体，参数为json格式的字符串：{"demo":123} """ http_body = None try: if isinstance(body, str) and body.startswith('{'): http_body = json.loads(body) loggerKit.debug('设置请求体成功') elif isinstance(body, dict): http_body = body elif isinstance(body, bytes): http_body = pickle.loads(body) else: pass except Exception as e: http_body = body loggerKit.error('设置请求体失败:%s' % body) return http_body def set_http_url(self, interface_url): """ 公共设置URL方法 设置http请求url,如：http://www.baidu.com """ self.api_domain = "https://www.baidu.com/" if not interface_url.startswith('https://') and not interface_url.startswith('http://'): http_url = self.api_domain + interface_url else: http_url = interface_url return http_url def common_http_request(self, request_type, request_url=None, headers=None, request_data=None, result_type="dict"): """ 发送http请求 :param request_url: :param interface_url: 接口地址 :param header_data: 请求头文件 :param interface_param: 接口请求参数 :param request_type: 请求类型,不区分大小写,GET POST :return: 字典形式结果 """ result = None # if requestUrl is not None: http_url = self.set_http_url(request_url) # if headers is not None: http_head = self.set_http_head(headers) if request_data: http_body = self.set_http_body(request_data) else: http_body = {} # try: if request_type.lower() == 'get': result = self.__http_get(http_url, http_head, http_body) loggerKit.debug('用例发送消息URL:%s\n' '用例发送消息头：%s\n' '用例发送消息体：%s\n' % (http_url, http_head, http_body)) elif request_type.lower() == 'post': result = self.__http_post(http_url, http_head, http_body) loggerKit.debug('用例发送消息URL:%s\n' '用例发送消息头：%s\n' '用例发送消息体：%s\n' % (http_url, http_head, http_body)) elif request_type.lower() == 'put': result = self.__http_put(http_url, http_head, http_body) loggerKit.debug('用例发送消息URL:%s\n' '用例发送消息头：%s\n' '用例发送消息体：%s\n' % (http_url, http_head, http_body)) else: loggerKit.error('请求类型错误') # except Exception as e: # logger.error(e) # logger.error('用例发送消息URL:%s\n' # '用例发送消息头：%s\n' # '用例发送消息体：%s\n' # % (http_url, http_head, http_body)) # return {} if result_type == "dict": try: return pickle.dumps(json.loads(result)) except Exception: return pickle.dumps(result) # return json.loads(result) else: return pickle.dumps(result)