接口自动化框架搭建（二）
2024-06-16 / 刘辉

在上一节中完成了接口的基本调试，接下来需要根据响应来做断言

2.2、断言封装

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def assert_json_path(param, step_assert):
    """
    断言jsonpath
    :param param:
    :param step_assert:
    :return:
    """
    # 验证输入参数的类型和结构
    if not isinstance(param, dict) or not isinstance(step_assert, dict):
        raise ValueError("参数类型错误，param和step_assert都应为字典类型")

    result_cache = {}  # 用于缓存jsonpath查询结果

    for key, value in step_assert.items():
        for key1, value1 in value.items():
            try:
                # 使用缓存结果或进行查询
                result = result_cache.get(key, jsonpath.jsonpath(param, key))
                result_cache[key] = result  # 更新缓存

                # 分别处理不同的断言
                if key1 == 'is_not_null':
                    assert result, f"{key} 的结果为 None，预期非空"
                elif key1 == 'in':
                    assert value1 in result, f"{value1} 不在 {key} 的结果中"
                elif key1 == 'eq':
                    # 确保结果非空
                    if result:
                        assert value1 == result[0], f"{value1} 不等于 {key} 的第一个结果"
                    else:
                        assert False, f"{key} 的结果为空，无法进行等于比较"
                elif key1 == 'not_in':
                    assert value1 not in result, f"{value1} 在 {key} 的结果中，预期不在"
            except AssertionError as e:
                # 提供更详细的错误信息
                raise AssertionError(f"在处理 {key} 时失败: {e}")
            except Exception as e:
                # 处理其他潜在异常
                raise RuntimeError(f"在查询 {key} 时发生错误: {e}")

2.3、封装设置变量的方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def set_variable(step_jsonpath: dict, step_body: dict, variable: dict=VARIABLE) -> dict:
    """
    提取变量并更新给定的变量字典
    :param step_jsonpath: 包含变量名和对应JSON路径的字典
    :param step_body: 待提取变量的JSON结构体
    :param variable: 用于存储提取变量的字典
    :return: 更新后的变量字典
    :raises ValueError: 当step_jsonpath或step_body的格式不正确时
    :raises IndexError: 当jsonpath查询结果为空列表且尝试访问第一个元素时
    """
    # 输入验证
    if not isinstance(step_jsonpath, dict):
        raise ValueError("step_jsonpath must be a dictionary")
    if not isinstance(step_body, dict):
        raise ValueError("step_body must be a dictionary")
    if not isinstance(variable, dict):
        raise ValueError("variable must be a dictionary")

    try:
        # 更新变量字典
        for key, value in step_jsonpath.items():
            # jsonpath查询并更新变量
            result = jsonpath.jsonpath(step_body, value)
            if result:
                variable[key] = result[0]
            else:
                # 如果查询结果为空，可以考虑记录日志、抛出异常或其他处理方式
                print(f"Warning: JSON path '{value}' did not match any value.")
    except Exception:
        raise ValueError(f"JSON path '{value}' error.")

    return variable

在send_request方法中来尝试断言并保存变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def send_request(case_obj: 'CaseInfo'):
    """
    发送请求
    :param case_obj:
    :return:
    """
    for step in case_obj.step_list:
        print()# 遍历步骤列表发送多个请求
        print(step)
        # 根据请求方法发送不同body体
        resp = requests.request(method=step.step_method, url=HOST+step.step_url, headers=step.step_header, **step.step_body)
        # 断言jsonpath
        assert_json_path(resp.json(), step.step_assert)
        # 提取变量
        print(set_variable(step.step_jsonpath, resp.json(), VARIABLE))
        return resp

简单跑一下

1
2
3
4
5
6
class TestCase:
    @pytest.mark.parametrize("case_obj", create_case_object())
    def test_case(self, case_obj):
        resp=send_request(case_obj)
        print(resp)

得到结果：

步骤名称:登录,请求地址:/login,请求方法:POST,请求头:None,请求体:{‘json’: {‘username’: ‘admin’, ‘password’: ‘admin123’}},JSON路径:{‘token’: ‘$.token’},断言:{‘$.token’: {‘is_not_null’: None}, ‘$.code’: {‘eq’: 1}, ‘$.message’: {‘eq’: ‘登录成功’}},SQL:{‘select’: None}
{‘token’: ‘eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTcxODU5MzE0NH0.jSa529DvdS929Ig9Z7PfW4ZQLVDxC42SC7lprTB5eW8’}
{‘code’: 1, ‘message’: ‘登录成功’, ‘token’: ‘eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTcxODU5MzE0NH0.jSa529DvdS929Ig9Z7PfW4ZQLVDxC42SC7lprTB5eW8’}
可以看到已经断言成功了，并且保存了token，后续就可以在请求头中使用了
接下来可以尝试完成两个接口的关联

  1. 登录接口
  2. 查询当前用户信息接口

在完成接口关联之前还要处理两个事情，其中一个就是目前的token在实际使用中需要再前面加一个Bearer ，所以需要在设置变量的时候单独处理一下，更新set_variable方法
其次就是我们需要在用例文件中来引用变量，我们用格式来表示变量，封装一个替换变量的方法

填写yaml文件内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
- case_id: 1 # 测试用例id
  name: 登录成功 # 测试用例名称
  clear_data: # 测试用例清理数据
    clean_sql: # delete from table where id = XXX # 测试用例清理sql
  steps:
    - step_name: 登录 #步骤名称
      step_url: /login # 步骤url
      step_method: POST # 步骤请求方法
      step_header:  # 步骤请求头
      step_body: #步骤请求体
        json: # 步骤请求体格式
          username: admin # 步骤请求体参数
          password: admin123
      step_jsonpath:
        token: $.token #变量名及变量提取表达式
      step_assert: # 步骤断言
        $.token:
          is_not_null:  # 断言表达式
        $.code:
          eq: 1
        $.message:
          eq: 登录成功
      step_sql:
        select: # select * from table where name = 'XXX' # 步骤sql

    - step_name: 查看当前用户信息
      step_url: /api/user
      step_method: GET
      step_header:
        Authorization: <token>
      step_body:
      step_jsonpath:
        id: $.id #变量名及变量提取表达式
      step_assert: # 步骤断言
        $.username:
          is_not_null: # 断言表达式
      step_sql:
        select:  # 步骤sql

更新set_variable方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def set_variable(step_jsonpath: dict, step_body: dict, variable: dict = VARIABLE) -> dict:
    """
    提取变量并更新给定的变量字典
    :param step_jsonpath: 包含变量名和对应JSON路径的字典
    :param step_body: 待提取变量的JSON结构体
    :param variable: 用于存储提取变量的字典
    :return: 更新后的变量字典
    :raises ValueError: 当step_jsonpath或step_body的格式不正确时
    :raises IndexError: 当jsonpath查询结果为空列表且尝试访问第一个元素时
    """
    # 输入验证
    if not isinstance(step_jsonpath, dict):
        raise ValueError("step_jsonpath must be a dictionary")
    if not isinstance(step_body, dict):
        raise ValueError("step_body must be a dictionary")
    if not isinstance(variable, dict):
        raise ValueError("variable must be a dictionary")

    try:
        # 更新变量字典
        for key, value in step_jsonpath.items():
            # jsonpath查询并更新变量
            result = jsonpath.jsonpath(step_body, value)
            if result:
                if key == 'token':
                    variable[key] = "Bearer " + result[0]
                
                variable[key] = result[0]
            else:
                # 如果查询结果为空，可以考虑记录日志、抛出异常或其他处理方式
                print(f"Warning: JSON path '{value}' did not match any value.")
    except Exception:
        raise ValueError(f"JSON path '{value}' error.")

    return variable

封装变量替换方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def replace_variable(step_obj:'StepInfo'):
    """
    替换变量
    :param step_obj:
    :return:
    """
    for attr_name in dir(step_obj):
        # 使用 dir 获取属性的值，注意过滤掉特殊方法（以__开头和结尾的）
        if not attr_name.startswith("__") and not attr_name.endswith("__"):
            data=json.dumps(getattr(step_obj, attr_name))#将属性值转换为字符串
            variable_list = re.findall(r'<(.*?)>',data)  # 获取所有变量名
            for i in variable_list:
                try:
                    data = data.replace('<' + i + '>', VARIABLE[i])  # 替换变量值
                except KeyError:
                    raise ValueError(f"变量 {i} 在变量字典中不存在，请检查变量名是否正确。")
                setattr(step_obj,attr_name,json.loads(data))#将替换后的字符串转换为字典

这个时候需要更新一下之前封装的send_request方法，在发送请求之前替换变量，并且还要判断step_body是否为空，如果为空则不携带请求体
更新send_request方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def send_request(case_obj: 'CaseInfo'):
    """
    发送请求
    :param case_obj:
    :return:
    """
    for step in case_obj.step_list:
        print()  # 遍历步骤列表发送多个请求
        print(step)
        # 替换变量
        replace_variable(step)
        # 根据请求方法发送不同body体
        if step.step_body==None:
            resp = requests.request(method=step.step_method, url=HOST + step.step_url, headers=step.step_header,
                                    )
        else:
            resp = requests.request(method=step.step_method, url=HOST + step.step_url, headers=step.step_header,
                                    **step.step_body)
        # 断言jsonpath
        assert_json_path(resp.json(), step.step_assert)
        # 提取变量
        set_variable(step.step_jsonpath, resp.json(), VARIABLE)
        print(resp.json())

简单调试一下得到了响应结果

步骤名称:登录,请求地址:/login,请求方法:POST,请求头:None,请求体:{‘json’: {‘username’: ‘admin’, ‘password’: ‘admin123’}},JSON路径:{‘token’: ‘$.token’},断言:{‘$.token’: {‘is_not_null’: None}, ‘$.code’: {‘eq’: 1}, ‘$.message’: {‘eq’: ‘登录成功’}},SQL:{‘select’: None}
{‘code’: 1, ‘message’: ‘登录成功’, ‘token’: ‘eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTcxODYwMzM4NH0.FaPPPEVml1RtwsSZ9-yJmXXF3vgBJM7isIA3UA4XUVA’}

步骤名称:查看当前用户信息,请求地址:/api/user,请求方法:GET,请求头:{‘Authorization’: ‘‘},请求体:None,JSON路径:{‘id’: ‘$.id’},断言:{‘$.username’: {‘is_not_null’: None}},SQL:{‘select’: None}
{‘email’: ‘admin@example.com‘, ‘id’: 0, ‘username’: ‘admin’}

到这里我们就完成了接口的关联啦，接下来我们需要封装用来操作数据库的类，用来获取数据库中的数据，然后进行断言。

本文链接：https://xiamu9527.cn/2024/06/16/%E6%8E%A5%E5%8F%A3%E8%87%AA%E5%8A%A8%E5%8C%96%E6%A1%86%E6%9E%B6%E6%90%AD%E5%BB%BA%EF%BC%88%E4%BA%8C%EF%BC%89/index.html