装饰器大显身手：优雅解决请求前后调试信息输出-每日运维

前言

进行接口自动化测试时，为了方便调试，通常我们会增加一些日志来打印请求 URL、方法、参数、响应状态码和内容。常见的笨办法，当然是直接在请求之后增加日志输出。但这有一个问题，会造成大量的冗余代码。那我们就想办法解决？带着这个实际场景，我们一起看看该如何优化呢？

笨办法

我们看看如果使用笨办法，写出来的代码是这样的：

import requests

def test_demo():
    response = requests.get(
        url="https://example.com/",
    )
    print(f"Request URL: {response.request.url}")
    print(f"Request Method: {response.request.method}")
    print(f"Request Body: {response.request.body}")
    print(f"Response Status Code: {response.status_code}")
    print(f"Response Content: {response.content}")

    assert response.status_code == 200

这是一个简单的案例，我们使用requests库进行HTTP请求后，使用 response.request 对象访问请求属性，使用 response.status_code 和 response.content 属性访问响应相关信息。最终打印出请求 URL、方法、参数、响应状态码和内容。

可以看到，如果每个测试case都这样写打印，很麻烦也不利于代码的维护阅读。假设我想新增打印headers还需要每个case上新增打印语句，可见重复工作量是很大的。

该如何解决优化呢？笔者这里使用装饰器来解决。

使用装饰器解决

介绍实现代码之前，有同学可能不明白装饰器该如何使用？想进一步了解的同学，可以参考笔者之前的文章：基于FastApi框架测试平台(14)-装饰器复习

首先，我们简单封装一下requests请求，比如这样：

class HttpRequest:

    @request
    def get(self, url, params=None, **kwargs):
        return requests.get(url, params=params, **kwargs)

我们看到，这里使用的装饰器@request，该装饰器将实现笨办法中的那些打印。

request装饰器实现，先附上实现代码，在做一个具体讲解

def request(func):
    def wrapper(*args, **kwargs):
        func_name = func.__name__
        try:
            url = list(args)[1]
        except IndexError:
            url = kwargs.get("url", "")
        if (Panda.base_url is not None) and (url.startswith("http") is False):
            url = Panda.base_url + url

        img_file = False
        file_type = url.split(".")[-1]
        if file_type in IMG:
            img_file = True

        log.info(f"[method]: {func_name.upper()}      [url]: {url} ")
        auth = kwargs.get("auth", None)
        headers = kwargs.get("headers", None)
        cookies = kwargs.get("cookies", None)
        params = kwargs.get("params", None)
        data = kwargs.get("data", None)
        json_ = kwargs.get("json", None)
        files = kwargs.get("files", None)
        if auth is not None:
            log.debug(f"[auth]:\n{auth}")
        if headers is not None:
            log.debug(f"[headers]:\n{formatting(headers)}")
        if cookies is not None:
            log.debug(f"[cookies]:\n{formatting(cookies)}")
        if params is not None:
            log.debug(f"[params]:\n{formatting(params)}")
        if data is not None:
            log.debug(f"[data]:\n{formatting(data)}")
        if json_ is not None:
            log.debug(f"[json]:\n{formatting(json_)}")
        if files is not None:
            log.debug(f"[files]:\n{files}")

        r = func(*args, **kwargs)

        ResponseResult.request = r.request
        ResponseResult.status_code = r.status_code
        if ResponseResult.status_code == 200 or ResponseResult.status_code == 304:
            log.info(f"successful with status {ResponseResult.status_code}")
        else:
            log.warning(f"unsuccessful with status {ResponseResult.status_code}")
        resp_time = r.elapsed.total_seconds()
        try:
            resp = r.json()
            log.debug(f"[type]: json      [time]: {resp_time}")
            log.debug(f"[response]:\n {formatting(resp)}")
            ResponseResult.response = resp
        except BaseException as msg:
            log.debug("[warning]: failed to convert res to json, try to convert to text")
            log.trace(f"[warning]: {msg}")
            if img_file is True:
                log.debug(f"[type]: {file_type}      [time]: {resp_time}")
                ResponseResult.response = r.content
            else:
                r.encoding = 'utf-8'
                log.debug(f"[type]: text      [time]: {resp_time}")
                log.debug(f"[response]:\n {r.text}")
                ResponseResult.response = r.text
        return r
    return wrapper

建议不懂的小伙伴先去看一看装饰器的使用。下面我们做一个简单解释：

func_name = func.__name__：获取被装饰函数的名称。
获取请求的 URL：

try:
    url = list(args)[1]
except IndexError:
    url = kwargs.get("url", "")

这里通过检查参数列表来获取 URL，如果没有传递 URL 参数，则从关键字参数中获取。

检查是否配置了基本URL，并根据需要拼接完整的URL：

if (Panda.base_url is not None) and (url.startswith("http") is False):
    url = Panda.base_url + url

判断请求的文件类型：

file_type = url.split(".")[-1]
if file_type in IMG:
    img_file = True

这里将 URL 中的文件类型与预定义的图片文件类型列表 IMG 进行对比，如果匹配则将 img_file 设置为 True。

输出请求的方法和 URL：

log.info(f"[method]: {func_name.upper()}      [url]: {url} ")

获取请求的认证、头部信息、Cookie、参数、数据、JSON 和文件等内容：

auth = kwargs.get("auth", None)
headers = kwargs.get("headers", None)
cookies = kwargs.get("cookies", None)
params = kwargs.get("params", None)
data = kwargs.get("data", None)
json_ = kwargs.get("json", None)
files = kwargs.get("files", None)

这里使用 kwargs.get() 方法获取关键字参数的值。

如果存在认证信息、头部信息、Cookie、参数、数据、JSON 或文件，输出对应的调试信息：

if auth is not None:
    log.debug(f"[auth]:\n{auth}")
if headers is not None:
    log.debug(f"[headers]:\n{formatting(headers)}")
if cookies is not None:
    log.debug(f"[cookies]:\n{formatting(cookies)}")
if params is not None:
    log.debug(f"[params]:\n{formatting(params)}")
if data is not None:
    log.debug(f"[data]:\n{formatting(data)}")
if json_ is not None:
    log.debug(f"[json]:\n{formatting(json_)}")
if files is not None:
    log.debug(f"[files]:\n{files}")

这里使用了一个辅助函数 formatting() 来格式化输出。

调用被装饰函数，并将返回结果保存在变量 r 中：

r = func(*args, **kwargs)

设置响应结果的请求和状态码：

ResponseResult.request = r.request
ResponseResult.status_code = r.status_code

根据状态码输出成功或警告信息：

if ResponseResult.status_code == 200 or ResponseResult.status_code == 304:
    log.info(f"successful with status {ResponseResult.status_code}")
else:
    log.warning(f"unsuccessful with status {ResponseResult.status_code}")

获取响应时间，并尝试将响应内容解析为 JSON 格式。如果解析失败，尝试将响应内容转换为文本格式或保持二进制格式（如果是图片文件）：

resp_time = r.elapsed.total_seconds()
try:
    resp = r.json()
    log.debug(f"[type]: json      [time]: {resp_time}")
    log.debug(f"[response]:\n {formatting(resp)}")
    ResponseResult.response = resp
except BaseException as msg:
    log.debug("[warning]: failed to convert res to json, try to convert to text")
    log.trace(f"[warning]: {msg}")
    if img_file is True:
        log.debug(f"[type]: {file_type}      [time]: {resp_time}")
        ResponseResult.response = r.content
    else:
        r.encoding = 'utf-8'
        log.debug(f"[type]: text      [time]: {resp_time}")
        log.debug(f"[response]:\n {r.text}")
        ResponseResult.response = r.text

这里使用了一个全局变量 ResponseResult 来保存响应结果。

返回最初被装饰的函数的执行结果 r。

总结一下，整个装饰器函数的作用是在请求发出之前和响应返回之后，打印一些调试信息，并将响应结果记录到全局变量中。这样可以方便地对请求和响应进行日志记录和调试。

那我们看看成果：

http_request = HttpRequest()

def test_demo():
    http_request.get(
        url="https://example.com/",
    )

执行之后，会发现已经打印出了想要的信息

2023-09-28 16:25:33 | INFO     | request.py | [method]: GET      [url]: https://example.com/
2023-09-28 16:25:36 | INFO     | request.py | successful with status 200
2023-09-28 16:25:36 | DEBUG    | request.py | [type]: json      [time]: 3.249797
2023-09-28 16:25:36 | DEBUG    | request.py | [response]:

太棒了，信息一目了然。

最后

这样我们就使用装饰器优雅的解决了问题，这样以后想要变更打印内容，只需要更改装饰器即可，测试用例都不受影响，提高了代码的复用性、扩展性和可维护性，同时保持代码的透明性和灵活性，使得代码更加清晰、简洁和易于理解。

装饰器大显身手：优雅解决请求前后调试信息输出

前言

笨办法

使用装饰器解决

最后

相关文章

发布评论取消回复

法医

百度搜索：蓝易云 – JDK1.8在LINUX下安装教程

万字长文 | MatrixOne 的 MySQL 协议实现

一个简单小工具obtool，助你成为Oceanbase高级专家

装饰器大显身手：优雅解决请求前后调试信息输出

前言

笨办法

使用装饰器解决

最后

相关文章

发布评论 取消回复

法医

百度搜索：蓝易云 – JDK1.8在LINUX下安装教程

万字长文 | MatrixOne 的 MySQL 协议实现

一个简单小工具obtool，助你成为Oceanbase高级专家

发布评论取消回复