httprunner3.x总结24 - hrun3常见的测试前置、测试后置处理

1、关于hrun3中setup、teardown的使用

import json
import pymysql
import requests
from pymysql.constants import CLIENT
from login import get_config_value

"""
hrun3中一些常见的测试前置后置处理：
（1）直接数据库处理表格数据
（2）根据 response 调用相关增删改查接口
"""


header = {
    "accept-language": get_config_value("Header", "accept-language"),
    "Authorization": get_config_value(role="expert_001", key="token"),
    "Content-Type": "application/json; charset=utf-8",
}
host = get_config_value(key1="Host", key2="rem_host_001")


def setup_banner_test_005(name):
    """
    清理可能遗留的脏数据
    @param name: png格式的图片名字
    @return:
    """
    conn_base_admin = pymysql.connect(
        host="xxx",
        port=123,
        user="xxx",
        password="xxx",
        db="xxx",
        charset="utf8",
        client_flag=CLIENT.MULTI_STATEMENTS,
    )
    curs_base_admin = conn_base_admin.cursor()
    sql = f"delete from tb where filename = xxx;"
    curs_base_admin.execute(sql)
    conn_base_admin.commit()


def tear_down_test_004(phone_number):
    """
    批量执行sql处理数据
    :param phone_number: 手机号
    :return: 
    """
    # （1）从第一个db库中获取 id
    conn_base_admin = pymysql.connect(
        host="xxx",
        port=123,
        user="xxx",
        password="xxx",
        db="xxx",
        charset="utf8",
        client_flag=CLIENT.MULTI_STATEMENTS,
    )
    curs_base_admin = conn_base_admin.cursor()
    sql = f"select id from user where phone_number = {phone_number};"
    curs_base_admin.execute(sql)
    conn_base_admin.commit()
    id = curs_base_admin.fetchall()[0][0]
    
    # （2）从第二个db库处理相关数据
    conn = pymysql.connect(
        host="xxx",
        port=123,
        user="xxx",
        password="xxx",
        db="xxx2",
        charset="utf8",
        client_flag=CLIENT.MULTI_STATEMENTS,
    )
    # 2.通过数据库连接,创建游标(可以通过游标执行sql语句)
    curs = conn.cursor()
    sql = f"""
    SET @openId = '{str(id)}';
    delete from tb where tb_expert.xx_id = @openId;
    delete from tb where tb_expert.xx_id = @openId;
    delete from tb where tb_expert.xx_id = @openId;
    """
    curs.execute(sql)
    conn.commit()
    data = json.dumps({})
    url = host + "/xx/yy"
    with requests.put(url=url, headers=header, data=data) as response:
        res = response.json()
        print(res)


def teardown_hook_clean_file(response):
    """
    删除接口执行后的相关数据
    @param response:
    @return:
    """
    data_content = response.body["data"]
    if data_content == []:
        return None
    file_id = [x["id"] for x in data_content]
    for i in file_id:
        url = (host + f"/diagn/xxx/{i}")
        with requests.delete(url=url, headers=header) as response:
            print("成功删除用户操作手册界面的内容")
    return response.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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106