Python装饰器的进阶应用：实现测试接口过程可观测性

# test_example.py
import requests
import pytest
import functools
import os
import inspect
import json
import allure
from typing importAny, Dict, Optional


# 假设的 ResponseObject 类型
class ResponseObject:
    def  __init__(self, details):
        self.details = details


# 记录消息到 Allure
def allure_record_msg(description: str, message: Dict[str, Any]):
    """记录结构化消息到 Allure 报告"""
    with allure.step(description):
        allure.attach(
            json.dumps(message, indent=4, ensure_ascii=False),
            name="Function Details",
            attachment_type=allure.attachment_type.JSON
        )


def allure_operation(desc: str, record=True):
    """增强版装饰器，支持捕获函数内部最终变量"""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 为每次调用创建独立的存储容器
            internal_vars = {}
            setattr(wrapper, "_internal_vars", internal_vars)

            # 设置描述信息
            setattr(wrapper, "desc", desc)

            # 执行环境控制
            env_var = os.environ.get("setup_teardown", "")
            os.environ["setup_teardown"] = "true"
            try:
                ret = func(*args, **kwargs)
            finally:
                os.environ["setup_teardown"] = env_var

            # 获取捕获的内部变量
            captured_vars = getattr(wrapper, "_internal_vars", {})

            # 清理临时属性
            if  hasattr(wrapper, "_internal_vars"):
                delattr(wrapper, "_internal_vars")

            # 处理函数返回结果
            result = ret.details if  isinstance(ret, ResponseObject) else ret

            # 处理参数序列化问题
            args_new = []
            for item in args:
                try:
                    json.dumps({"item": item}, indent=4, ensure_ascii=False)
                    args_new.append(item)
                except Exception:
                    args_new.append(str(item))

            # 处理默认参数和关键字参数
            try:
                func_detail = inspect.getfullargspec(func)
            except AttributeError:
                func_detail = inspect.getargspec(func)

            param_names = func_detail.args
            if func_detail.defaults:
                defaults_len = len(func_detail.defaults)
                required_args = param_names[:-defaults_len]

                # 处理位置参数覆盖默认参数的情况
                if len(args) > len(required_args):
                    extra_args = args[len(required_args):]
                    extra_params = param_names[len(required_args):len(required_args) + len(extra_args)]
                    kwargs.update(dict(zip(extra_params, extra_args)))
                    args = args[:len(required_args)]

                # 填充未提供的默认参数
                default_params = param_names[-defaults_len:]
                default_values = func_detail.defaults
                for param, value inzip(default_params, default_values):
                    if param notin kwargs:
                        kwargs[param] = value

            # 记录执行详情
            if not os.environ.get("setup_teardown") and record:
                message = {
                    "function": func.__name__,
                    "description": desc,
                    "arguments": {
                        "positional": args_new,
                        "keyword": kwargs
                    },
                    "return_value": result,
                    "internal_state": captured_vars
                }

                # 添加函数文档
                if func.__doc__:
                    message["documentation"] = func.__doc__.strip()

                # 获取步骤描述
                step_desc = desc
                if  hasattr(wrapper, "desc"):
                    step_desc = getattr(wrapper, "desc")

                # 记录到 Allure
                allure_record_msg(step_desc, message)

            return ret

        return wrapper

    return decorator


@allure_operation("调用考勤管理 API[AttendanceManagementList] - 获取指定列表")
def  attendance_manage_list(body):
    # 比如这里模拟处理点逻辑
    if body['user.age']>20:
        body['user.age'] = 18

    # ========= 关键部分 =========
    # 捕获最终 body 信息到装饰器
    try:
        # 获取当前函数的包装器
        func_wrapper = attendance_manage_list

        # 确保装饰器属性存在
        if  hasattr(func_wrapper, "_internal_vars"):
            # 存储最终 body 和任何其他需要的信息
            getattr(func_wrapper, "_internal_vars")["final_body"] = body
            getattr(func_wrapper, "_internal_vars")["processed_nas_infos"] = body['user.age']
    except Exception as e:
        print(f"捕获内部变量失败: {e}")
    # ==========================
    response = requests.post(url = "http://httpbin.org/post", json=body)
    print(response.text)
    return response.json()

def  test_api():
    attendance_manage_list(body={
        "user.name": "拾光",
        "user.age": 28,
        "details.hobbies": ["reading", "swimming"]
    })
if __name__ == '__main__':
    pytest.main(['-vs','--alluredir',"report",__file__])