《接口自动化测试框架》代码片段

抛砖引玉

requests模块是Python中发送HTTP请求的强大工具，它以其直观易用的API和人性化的设计赢得了广泛赞誉。 这个模块不仅提供了丰富的功能来定制HTTP请求，如设置请求头、传递URL参数等，还能够自动处理许多底层细节，如Cookie管理和会话保持，极大地简化了网络编程的复杂性。

在Python的网络编程中，requests模块凭借其高效且简洁的特性，成为了开发者处理网络请求的优选库； 无论是用于爬虫项目抓取网页数据，还是进行数据分析时从远程服务器获取数据，亦或是构建接口自动化测试框架，requests都能够轻松胜任。

在之前的文章中已经对requests模块的功能和用法进行了详细的介绍，不在重复。

本次将为大家展示一个实用的工具类，这个工具类是我们《接口自动化框架（付费）》中的一部分，它专门设计用于接口自动化测试 - 接口请求的相关功能。

通过这个工具类，可以更加高效地驱动测试数据进行接口测试，并与其他模块配合实现Allure报告记录。希望这个工具类能够为你的工作带来便利，并提升接口自动化测试的效率。

历史文章：Python 爬虫与接口自动化必备requests模块

历史文章：Allure速查表（03）报告结构详解和常用装饰器

历史文章：Allure速查表（04）静态和动态生成报告标记

框架文章：完美落地的自动化测试框架：智能生成？业务依赖？动态替换？报告构建？你来，这儿有！

注意：框架暂不开源，66R包执行，可免费讲解。

请添加图片描述

代码封装

# -*- coding: utf-8 -*-
"""
@Author  : yangkai
@Email   : 807440781@qq.com
@Project : KKNOBUG-API
@Module  : RequestUtils.py
@DateTime: 2024/5/09 20:28
"""
import json as js
import typingimport requests
from requests import exceptions
import allurefrom services.ResponseModel import ResponseCodeError, ResponseOtherError
from services.ExtendedDecorator import request_decorator, execution_durationclass RequestUtils(object):"""代码设计思想：1.利用requests模块构建HTTP请求客户端2.利用魔术方法____new__完成单例模式，保证HTTP请求客户端不会重复实例化3.实例化时可以配置接口请求超时时间，默认120秒4.处理请求头部信息，如未上送则附带常用属性，若以上送则与常用属性合并5.处理请求参数类型和请求参数双向绑定（request分params、data和json方式传递参数）6.处理HTTP常用请求发送（GET, POST, PUT, DELETE, HEAD, OPTIONS, CONNECT, TRACE...）7.处理响应对象中常用的信息封装成字典（url, headers, method, time, cookie, params_type, params_data...）8.处理allure报告step展示接口的请求和响应过程""""用于存储该类的当一实例"__instance = None"重写父类的__new__方法，实现单例模式"def __new__(cls, *args, **kwargs) -> object:if not cls.__instance:cls.__instance = super().__new__(cls)return cls.__instancedef __init__(self, timeout: int = 120) -> None:"""客户端发起请求前的配置"""self.TIMEOUT = timeoutself.session = requests.session()@execution_duration(number=5)def request(self, *, url: str, method: str,params: typing.Optional[str] = None,data: typing.Optional[typing.Dict[str, str]] = None,json: typing.Optional[typing.Union[typing.Dict[str, str], str]] = None,headers: typing.Optional[typing.Union[typing.Dict[str, str], str]] = None,**kwargs) -> typing.Dict[str, str]:"""按照method请求方式向url地址携带params/data/json/..数据发送HTTP请求:param url:     必填项，字符类型，接口请求地址；如：http://127.0.0.1/test:param method:  必填项，字符类型，接口请求方式；如：GET、POST、PUT、DELETE等:param params:  非必填，字符类型，接口请求参数类型（参数增加到url中）:param data:    非必填，字典类型，接口请求参数类型（作为Request的内容）:param json:    非必填，JSON类型，接口请求参数类型（作为Request的内容）:param headers: 非必填，字典类型，接口请求的头部信息；:param kwargs:  非必填，字典类型，其他参数；:return:        返回requests请求对象"""# 1.GET：获取实体数据# 2.HEAD：获取响应头# 3.POST：提交数据# 4.PUT：上传数据# 5.PATCH：同PUT请求，对已知资源进行局部更新# 6.DELETE：删除数据# 7.OPTIONS：测试通信# 8.CONNECT：更改连接模式为管道方式的代理服务器# 9.TRACE：回显服务方收到的请求，用于测试和诊断# 1.检查请求方式是否允许methods = ('GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS', 'CONNECT', 'TRACE')if method.upper() not in methods:raise exceptions.HTTPError(f'不支持[{method}]请求方式，支持的请求方式有：{methods}')# 2.检查是否指定请求头信息headers = RequestUtils.headers_wrapper(headers=headers, params=params, data=data, json=json)# 3.发起请求try:response = requests.request(url=url,method=method,params=params,data=data,json=json,headers=headers,timeout=int(self.TIMEOUT),**kwargs)if response.status_code != 200:raise ResponseCodeError(message="请求失败，服务器响应状态码非200",status_code=response.status_code,reason=response.reason)except exceptions.Timeout as e:raise ResponseOtherError(message="请求失败，服务器响应超时",reason=e)except exceptions.InvalidURL as e:raise ResponseOtherError(message="请求失败，URL解析失败或无效",reason=e)except exceptions.HTTPError as e:raise ResponseOtherError(message="请求失败，服务器响应异常",reason=e)except exceptions.ConnectionError as e:raise ResponseOtherError(message="请求失败，网络连接失败或服务器拒绝连接",reason=e)params_type = "json" if json is not None else "data" if data is not None else "params"params_data = json if json is not None else data if data is not None else paramsresponse_dict: dict = RequestUtils.response_wrapper(response)self.api_allure_step(url=url,headers=headers,method=method,params_type=params_type,params_data=params_data,response=response_dict,)return response_dict@staticmethoddef response_wrapper(response) -> typing.Dict[str, str]:"""将接口响应对象中常用的信息封装成字典, url, headers, method, params_type, params_data:param response: 响应对象:return: 字典"""response_dict = {}response_dict.setdefault("response_url", response.url)response_dict.setdefault("response_code", response.status_code)response_dict.setdefault("response_info", response.reason)response_dict.setdefault("response_encoding", response.encoding)response_dict.setdefault("response_text", response.text)response_dict.setdefault("response_time", response.elapsed.total_seconds())response_dict.setdefault("response_microseconds", response.elapsed.microseconds)response_dict.setdefault("response_millisecond", response.elapsed.microseconds / 1000)response_dict.setdefault("response_headers", dict(response.headers))response_dict.setdefault("response_cookie", dict(response.cookies))response_dict.setdefault("response_other", response)try:# js.dumps(response.json(), ensure_ascii=False, separators=(",", ":"))response_dict.setdefault("response_json", response.json())except Exception as e:response_dict.setdefault("response_json", e)return response_dict@staticmethoddef headers_wrapper(headers: typing.Optional[typing.Union[typing.Dict[str, str], str]] = None,params: typing.Optional[str] = None,data: typing.Optional[typing.Dict[str, str]] = None,json: typing.Optional[typing.Union[typing.Dict[str, str], str]] = None) -> typing.Dict[str, str]:"""处理请求头信息:param headers::param params::param data::param json::return:"""# 如果headers类型不符则抛出异常if not isinstance(headers, (type(None), dict, str)):raise TypeError(f"参数headers应为None、字符串、字典类型，但得到了：「{type(headers)}」类型")final_headers: dict = {"Accept-Language": "zh-CN,zh;q=0.9","Connection": "keep-alive","User-Agent": "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36","Accept": "application/json, text/plain, */*","Accept-Encoding": "gzip, deflate, br",}# 如果传递的headers不空且是字符串类型，则尝试加载成字典类型if headers and isinstance(headers, str):try:headers_to_dict = js.loads(headers)if isinstance(headers_to_dict, dict):final_headers.update(headers_to_dict)else:raise TypeError("字符串异常，无法转化字典类型")except js.decoder.JSONDecodeError:raise TypeError("字符串异常，无法转化字典类型")# 根据参数类型设置请求头的客户端数据类型if params or data:# 将数据编码为键值对final_headers.update({"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"})elif json:# 将数据编码为JSON对象# 可以忽略，因为使用json参数时，不需要手动序列化数据或设置Content-Type头，requests会自动处理。final_headers.update({"Content-Type": "application/json;charset=utf-8"})return final_headers@staticmethoddef datagram_wrapper(params_type: typing.Literal["params", "PARAMS", "data", "DATA", "json", "JSON"] = None,params_data: typing.Any = None,):if params_type:return {params_type.lower(): params_data}return {"json": None, "data": None, "params": None}@classmethod@request_decorator(switch=True)def api_allure_step(cls, *, url, headers, method, params_type, params_data, response):with allure.step("核心步骤一：接口请求"):allure.attach(name="请求地址", body=str(url))allure.attach(name="请求头部",body=js.dumps(headers, ensure_ascii=False, indent=4),attachment_type=allure.attachment_type.JSON)allure.attach(name="请求方式", body=str(method))allure.attach(name="请求参数类型", body=str(params_type))try:allure.attach(name="请求参数内容",body=js.dumps(params_data, ensure_ascii=False, indent=4),attachment_type=allure.attachment_type.JSON)except Exception:allure.attach(name="请求参数内容", body=str(params_data))with allure.step("核心步骤二：接口响应"):del response["response_other"]allure.attach(name="响应代码", body=str(response.get("response_code")))allure.attach(name="响应文本", body=str(response.get("response_text")))try:allure.attach(name="响应内容",body=js.dumps(response.get("response_json"), ensure_ascii=False, indent=4),attachment_type=allure.attachment_type.JSON)except Exception:allure.attach(name="响应内容", body=str(response.get("response_json")))allure.attach(name="响应耗时", body=str(response.get("response_time")) + "秒")allure.attach(name="响应对象",body=js.dumps(response, ensure_ascii=False, indent=4),attachment_type=allure.attachment_type.JSON)if __name__ == '__main__':# 测试单例模式cls1 = RequestUtils()cls2 = RequestUtils()print(cls1 is cls2)# 测试接口test_get = cls2.request(url="https://httpbin.org/get", method="GET")print(test_get)print(test_get.get("response_url"))  # http://httpbin.org/getprint(test_get.get("response_encoding"))  # utf-8print(test_get.get("response_info"))  # OKprint(test_get.get("response_code"))  # 200print(test_get.get("response_time"))  # 0.781666headers = {"Content-Type": "1111"}data = {"name": "张三", "age": 20, "phone": "10086", "address": "上海市浦东新区"}test_post = cls2.request(url="http://httpbin.org/post", method="POST", json=data, headers=headers)print(test_post)