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

在这一节中，需要封装一下连接数据库的方法，方便后续的测试用例调用，用来验证数据是否落库或者从数据库中获取数据。
因为演示用的后端demo使用的是sqlite数据库，在实际的项目中基本使用的是mysql，mysql的连接方式与sqlite类似，这里就不演示了。
附上mysql连接的链接 http://t.csdnimg.cn/aWDP6

封装数据库操作的函数

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
def execute_sql_operations(operations_dict) -> list:
    # 连接到SQLite数据库（请根据实际情况修改数据库文件路径）
    with sqlite3.connect(SQLITE_DB_PATH) as conn:
        cursor = conn.cursor()
        results = []
        for operation, sql in operations_dict.items():
            try:
                if "insert" in operation:
                    cursor.execute(sql)
                    conn.commit()
                elif "delete" in operation:
                    cursor.execute(sql)
                    conn.commit()
                elif "update" in operation:
                    cursor.execute(sql)
                    conn.commit()
                elif "select" in operation:
                    cursor.execute(sql)
                    results.append(cursor.fetchall())
                else:
                    print(f"Unsupported operation type: {operation}")
            except Exception as e:
                print(f"Error executing {operation} operation: {e}")
    if results:
        print(f'数据库验证结果：{results}')
    # 如果有select操作，返回所有查询结果
    return results if results else None

这个函数接收一个字典作为参数，字典格式为：

1
2
3
4
5
6
{
"insert": "INSERT INTO table_name (column1, column2, ...) VALUES (value1, value2, ...)",
"delete": "DELETE FROM table_name WHERE condition",
"update": "UPDATE table_name SET column1=value1, column2=value2, ... WHERE condition",
"select": "SELECT * FROM table_name WHERE condition"
}

如果查询到的结构结果不为空则返回结果，否则返回None。后续根据这个做是否落库的判断。也可以作为清理测试用例数据的函数。
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
24
25
26
27
28
29
30
31
32
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)
        # 提取变量
        if step.step_jsonpath:
            set_variable(step.step_jsonpath, resp.json(), VARIABLE)
        # 操作数据库
        # 增加判断，如果有查询的sql则断言查询结果存在，否则不执行断言仅执行sql
        if step.step_sql:
            if step.step_sql['select']:
                assert execute_sql_operations(step.step_sql)
            else:
                execute_sql_operations(step.step_sql)

        print(f"响应结果：{resp.json()}")

修改一下测试用例，完成一个登录+添加新用户的测试场景

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
40
41
42
43
# 用例文件的配置模板
- case_id: 1 # 测试用例id
  name: 登录成功并添加新用户 # 测试用例名称
  clear_data: # 测试用例清理数据
    delete: delete from user where username = 'test1'
    delete1: delete from userinfo where username = 'test1'  # 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: /register
      step_method: POST
      step_header:
        Authorization: <token>
      step_body:
        json:
          username: test1
          password: test123
          email: test1@qq.com
      step_jsonpath: #变量名及变量提取表达式
      step_assert: # 步骤断言
        $.message:
          is_not_null: # 断言表达式
      step_sql:
        select: select * from user where username = 'test1' # 步骤sql
1
2
3
4
5
6
class TestCase:
    @pytest.mark.parametrize("case_obj", create_case_object())
    def test_case(self, case_obj: CaseInfo):
        send_request(case_obj)
        if case_obj.clear_dict:
            execute_sql_operations(case_obj.clear_dict)

简单调试一下：

步骤名称:登录,请求地址:/login,请求方法:POST,请求头:None,请求体:{‘json’: {‘username’: ‘admin’, ‘password’: ‘admin123’}},JSON路径:{‘token’: ‘$.token’},断言:{‘$.token’: {‘is_not_null’: None}, ‘$.code’: {‘eq’: 1}, ‘$.message’: {‘eq’: ‘登录成功’}},SQL:None
响应结果：{‘code’: 1, ‘message’: ‘登录成功’, ‘token’: ‘eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTcxODY5MzU5NH0._xoVf-eXD7A6YA_TFIxvOxYkZJlWCgM-mAKtm2iCuLg’}

步骤名称:注册,请求地址:/register,请求方法:POST,请求头:{‘Authorization’: ‘‘},请求体:{‘json’: {‘username’: ‘test1’, ‘password’: ‘test123’, ‘email’: ‘test1@qq.com‘}},JSON路径:None,断言:{‘$.message’: {‘is_not_null’: None}},SQL:{‘select’: “select * from user where username = ‘test1’”}
数据库验证结果：[[(4, ‘test1’, ‘test123’)]]
响应结果：{‘message’: ‘创建成功’}

重复执行也依旧会成功

注意：生产环境不要使用这种数据库验证方式，也不要在生产环境中执行sql来清理测试数据。

到了现在我们的框架已经具备基础的用例执行能力了，接下来我们还需要添加多用例文件的读取和执行能力。

本文链接：https://xiamu9527.cn/2024/06/18/%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%B8%89%EF%BC%89/index.html