1. 框架结构说明
2. 框架代码实现
1. 框架结构说明
整个测试框架分为四层，通过分层的方式，测试代码更容易理解，维护起来较为方便。
第一层是“测试工具层”：
- util 包：用于实现测试过程中调用的工具类方法，例如读取配置文件、接口请求方法、生成测试报告、发送邮件等。
- conf 目录：存放配置文件。
- log 目录：日志输出文件。
第二层是“服务层”：相当于对测试对象的一个业务封装。对于接口测试，是对远程方法的一个实现；对于 UI 测试，是对页面元素或操作的一个封装。
- action 包：实现了对一个接口的请求，包含请求数据的参数化预处理及响应数据的关联提取。
第三层是“测试用例逻辑层”：该层主要是将服务层封装好的各个业务对象，组织成测试逻辑，进行校验。
- bussiness_process 包：基于关键字的形式，实现具体模块或测试用例集的测试脚本逻辑。
- test_data 目录：Excel 测试数据文件，包含请求地址、请求方法、请求数据、关联关键字等。
第四层是“测试场景层”：将测试用例组织成测试场景，实现各种级别 cases 的管理，如冒烟，回归等测试场景。
- main.py：本框架工程的运行主入口。
2. 框架代码实现
本框架的 github 地址：https://github.com/juno3550/InterfaceAutoTest
action 包
action 包为框架第二层“服务层”，相当于对测试对象的一个业务封装。对于接口测试，是对远程方法的一个实现；对于 UI 测试，是对页面元素或操作的一个封装。
case_action.py
本模块实现了对一个接口的请求，包含请求数据的参数化预处理及响应数据的关联提取。
关联
在 excel 测试数据文件的具体模块用例 sheet 中，添加了“关联数据获取”列用于存放关联的正则表达式，如 userid=userid”: (\d+) 表示从该行用例的接口返回结果中，获取 userid 的值，获取到 userid 的值之后把它放到全局变量 global_vars 中。
参数化及函数化
- 对于 ${unique_num1} 标识则调用函数获取一个每次递增的唯一数供注册用户名使用；对于 ${md5(‘密码’)} 则调用 md5() 函数实现请求数据的加密处理。
- 在后续用例的请求数据中，用标识符 ${userid} 和 ${token} 表示从关联结果中获取 userid 和 token 的值。
本模块的函数说明
- data_preprocessor()：对请求数据进行预处理：参数化及函数化。
- data_postprocessor()：将响应数据需要关联的参数保存进全局变量，供后续接口使用。
- execute_case()：执行接口用例：请求数据预处理——>请求接口——>响应数据关联提取。
1 import re
2 import traceback
3 import time
4 import requests
5 from util.excel_util import Excel
6 from util.request_util import http_client, get_request_url, get_unique_num, assert_keyword
7 from util.datetime_util import *
8 from util.global_var import *
9 from util.log_util import *
10
11
12 # 对请求数据进行预处理：参数化、函数化
13 def data_preprocessor(data):
14 # 匹配需要调用唯一数函数的参数
15 if re.search(r"\$\{unique_\w*\}", data):
16 unique_num = get_unique_num()
17 # 从用例中获取唯一数的变量名，供后续接口关联使用
18 global_num_name = re.search(r"\$\{unique_(\w*)\}", data).group(1)
19 # 将调用获取的唯一数的变量名和值存入全局变量中，供后续接口关联使用
20 PARAM_GLOBAL_DICT[global_num_name] = unique_num
21 data = re.sub(r"\$\{unique_\w*\}", unique_num, data)
22 # 匹配需要进行关联的参数
23 if re.search(r"\$\{\w+\}", data):
24 var = re.search(r"\$\{(\w+)\}", data).group(1)
25 data = re.sub(r"\$\{\w+\}", PARAM_GLOBAL_DICT[var], data)
26 # 匹配需要进行函数话的参数
27 if re.search(r"\$\{\w+?\(.+?\)\}", data):
28 func_var = re.search(r"\$\{(\w+?\(.+?\))\}", data).group(1)
29 func_result = eval(func_var)
30 data = re.sub(r"\$\{(\w+?\(.+?\))\}", func_result, data)
31 return data
32
33
34 # 将响应数据需要关联的参数保存进全局变量，供后续接口使用
35 def data_postprocessor(response_data, revelant_param):
36 if not isinstance(revelant_param, str):
37 error("数据格式有误！【%s】" % revelant_param)
38 error(traceback.format_exc())
39 param, regx = revelant_param.split("=")
40 # none标识为该条测试数据没有关联数据，因此无需处理
41 if regx.lower() == "none":
42 return
43 if re.search(regx, response_data):
44 var_result = re.search(regx, response_data).group(1)
45 final_regx = re.sub(r"\(.+\)", var_result, regx)
46 info("关联数据【%s】获取成功！" % final_regx)
47 PARAM_GLOBAL_DICT[param] = var_result
48 return "%s" % final_regx
49 else:
50 error("关联数据【%s】在响应数据中找不到！" % regx)
51
52
53 # 执行接口用例：参数化——>请求接口——>响应数据关联提取
54 def execute_case(data):
55 if not isinstance(data, (list, tuple)):
56 error("测试用例数据格式有误！测试数据应为列表或元组类型！【%s】" % data)
57 data[CASE_EXCEPTION_INFO_COL_NO] = "测试用例数据格式有误！应为列表或元组类型！【%s】" % data
58 data[CASE_TEST_RESULT_COL_NO] = "fail"
59 return data
60 # 该用例无需执行
61 if data[CASE_IS_EXECUTE_COL_NO].lower() == "n":
62 return
63 # 获取请求地址
64 url = get_request_url(data[CASE_SERVER_COL_NO])
65 # 获取请求接口名称
66 api_name = data[CASE_API_NAME_COL_NO]
67 info("*" * 40 + " 开始执行接口用例【%s】 " % api_name + "*" * 40)
68 # 获取请求方法
69 method = data[CASE_METHOD_COL_NO]
70 # 替换测试数据
71 request_data = data[CASE_REQUEST_DATA_COL_NO]
72 info("data before process: %s" % request_data)
73 try:
74 request_data = data_preprocessor(request_data)
75 except:
76 error("请求数据【%s】预处理失败！" % request_data)
77 error(traceback.format_exc())
78 data[CASE_EXCEPTION_INFO_COL_NO] = "请求数据【%s】预处理失败！\n%s" % (request_data, traceback.format_exc())
79 data[CASE_TEST_RESULT_COL_NO] = "fail"
80 return data
81 else:
82 info("data after process: %s" % request_data)
83 # 数据回写到测试结果
84 data[CASE_REQUEST_DATA_COL_NO] = request_data
85 # 请求接口并获取响应数据
86 start_time = time.time()
87 response = http_client(url, api_name, method, request_data)
88 api_request_time = time.time() - start_time
89 data[CASE_TIME_COST_COL_NO] = int(api_request_time*1000)
90 if not isinstance(response, requests.Response):
91 error("接口用例【%s】返回的响应对象类型有误！" % api_name)
92 data[CASE_TEST_RESULT_COL_NO] = "fail"
93 data[CASE_EXCEPTION_INFO_COL_NO] = "接口用例【%s】返回的响应对象类型【%s】有误！" % (api_name, response)
94 return data
95 else:
96 info("接口用例【%s】请求成功！" % api_name)
97 data[CASE_API_NAME_COL_NO] = response.url
98 try:
99 data[CASE_RESPONSE_DATA_COL_NO] = response.text
100 info("接口响应数据：{}".format(response.text))
101 # 进行断言
102 assert_keyword(response, data[CASE_ASSERT_KEYWORD_COL_NO])
103 info("接口用例【%s】断言【%s】成功！" % (api_name, data[CASE_ASSERT_KEYWORD_COL_NO]))
104 data[CASE_TEST_RESULT_COL_NO] = "pass"
105 except:
106 error("接口用例【%s】断言【%s】失败！" % (api_name, data[CASE_ASSERT_KEYWORD_COL_NO]))
107 error(traceback.format_exc())
108 data[CASE_TEST_RESULT_COL_NO] = "fail"
109 data[CASE_EXCEPTION_INFO_COL_NO] = traceback.format_exc()
110 return data
111 try:
112 if data[CASE_RELEVANT_PARAM_COL_NO]:
113 data[CASE_RELEVANT_PARAM_COL_NO] = data_postprocessor(response.text, data[CASE_RELEVANT_PARAM_COL_NO])
114 except:
115 error("关联数据【%s】处理失败！" % data[CASE_RELEVANT_PARAM_COL_NO])
116 error(traceback.format_exc())
117 data[CASE_EXCEPTION_INFO_COL_NO] = "关联数据【%s】处理失败！\n%s" % \
118 (data[CASE_RELEVANT_PARAM_COL_NO], traceback.format_exc())
119 data[CASE_TEST_RESULT_COL_NO] = "fail"
120 return data
121 data[CASE_TEST_RESULT_COL_NO] = "pass"
122 return data
123
124
125 if __name__ == "__main__":
126 excel = Excel(EXCEL_FILE_PATH)
127 excel.get_sheet("注册")
128 datas = excel.get_all_row_data(False)
129 for row_data in datas:
130 execute_case(row_data)
131 # excel.write_row_data(row_data)
132 # excel.save()
business_process 包
business_process 包是框架第三层“测试用例逻辑层”，该层主要是将服务层封装好的各个业务对象，组织成测试逻辑，进行校验。
main_process.py
- suite_process()：基于 excel 测试数据文件，执行具体模块的用例 sheet（注册 sheet、登录 sheet 等）。
- main_suite_process()：基于 excel 测试数据文件，执行主 sheet “测试用例集”。
- main_process()：根据入参区分执行模块 sheet 或 主 sheet 的用例，并完成测试报告的生成与邮件发送。
1 from action.case_action import execute_case
2 from util.excel_util import Excel
3 from util.global_var import *
4 from util.datetime_util import *
5 from util.report_util import *
6 from util.log_util import *
7 from util.email_util import send_mail
8
9
10 # 执行具体模块的用例sheet（注册sheet、登录sheet等）
11 def suite_process(excel_file_path, sheet_name):
12 # 记录测试结果统计
13 global TOTAL_CASE
14 global PASS_CASE
15 global FAIL_CASE
16 # 只要有一条用例失败，则本测试用例集结果算失败
17 suite_test_flag = True
18 # 第一条接口用例的执行时间则为本测试用例集的执行时间
19 first_test_time = False
20 # 初始化excel对象
21 if isinstance(excel_file_path, Excel):
22 excel = excel_file_path
23 else:
24 excel = Excel(excel_file_path)
25 if not excel.get_sheet(sheet_name):
26 error("sheet名称【%s】不存在！" % sheet_name)
27 return
28 # 获取所有行数据
29 all_row_datas = excel.get_all_row_data()
30 if len(all_row_datas) <= 1:
31 error("sheet【】数据不大于1行，停止执行！" % sheet_name)
32 return
33 # 标题行数据
34 head_line = all_row_datas[0]
35 # 切换到“测试结果”sheet，以写入测试执行结果
36 excel.get_sheet("测试结果明细")
37 # 写入标题行
38 excel.write_row_data(head_line, None, True, "green")
39 # 遍历执行用例
40 for data in all_row_datas[1:]:
41 # 跳过不需要执行的用例
42 if data[MAIN_SHEET_IS_EXECUTE_COL_NO].lower() == "n":
43 info("接口用例【%s】无需执行！" % data[CASE_API_NAME_COL_NO])
44 continue
45 TOTAL_CASE += 1
46 # 记录测试时间
47 execute_time = get_english_datetime()
48 # 用例集测试执行同步模块sheet的第一条用例的执行时间
49 if not first_test_time:
50 first_test_time = execute_time
51 case_result_data = execute_case(data)
52 # 标识为n的用例返回None
53 if not case_result_data:
54 continue
55 if case_result_data[CASE_TEST_RESULT_COL_NO] == "fail":
56 suite_test_flag = False
57 # 获取html测试报告的所需数据
58 if case_result_data[CASE_TEST_RESULT_COL_NO] == "pass":
59 PASS_CASE += 1
60 TEST_RESULT_FOR_REPORT.append([case_result_data[CASE_API_NAME_COL_NO], case_result_data[CASE_REQUEST_DATA_COL_NO],
61 case_result_data[CASE_RESPONSE_DATA_COL_NO], case_result_data[CASE_TIME_COST_COL_NO],
62 case_result_data[CASE_ASSERT_KEYWORD_COL_NO], "成功", case_result_data[CASE_EXCEPTION_INFO_COL_NO]])
63 else:
64 FAIL_CASE += 1
65 TEST_RESULT_FOR_REPORT.append([case_result_data[CASE_API_NAME_COL_NO], case_result_data[CASE_REQUEST_DATA_COL_NO],
66 case_result_data[CASE_RESPONSE_DATA_COL_NO], case_result_data[CASE_TIME_COST_COL_NO],
67 case_result_data[CASE_ASSERT_KEYWORD_COL_NO], "失败", case_result_data[CASE_EXCEPTION_INFO_COL_NO]])
68 # 写入excel测试结果明细
69 case_result_data[CASE_TEST_TIME_COL_NO] = execute_time
70 excel.write_row_data(case_result_data)
71 # 写入excel测试结果统计
72 excel.get_sheet("测试结果统计")
73 excel.insert_row_data(1, [TOTAL_CASE, PASS_CASE, FAIL_CASE])
74 # 返回excel对象，是为了在生成测试报告时再调用save()生成excel版测试结果文件，以此同步和html测试报告相同的时间戳命名
75 # suite_false_count 返回本次测试用例集的执行结果
76 # first_test_time 作为本测试用例集的执行时间
77 return excel, suite_test_flag, first_test_time
78
79
80 # 执行主sheet“测试用例集”
81 def main_suite_process(excel_file_path, sheet_name):
82 # 初始化excel对象
83 excel = Excel(excel_file_path)
84 if not excel:
85 error("excel数据文件【%s】不存在！" % excel_file_path)
86 return
87 if not excel.get_sheet(sheet_name):
88 error("sheet名称【%s】不存在！" % sheet_name)
89 return
90 # 获取所有行数据
91 all_row_datas = excel.get_all_row_data()
92 if len(all_row_datas) <= 1:
93 error("sheet【%s】数据不大于1行，停止执行！" % sheet_name)
94 return
95 # 标题行数据
96 head_line = all_row_datas[0]
97 # 用例步骤行数据
98 for data in all_row_datas[1:]:
99 # 跳过不需要执行的测试用例集
100 if data[MAIN_SHEET_IS_EXECUTE_COL_NO].lower() == "n":
101 info("#" * 50 + " 测试用例集【%s】无需执行！ " % data[MAIN_SHEET_SUITE_COL_NO] + "#" * 50 + "\n")
102 continue
103 if data[MAIN_SHEET_SUITE_COL_NO] not in excel.get_all_sheet():
104 error("#" * 50 + " 测试用例集【%s】不存在！ " % data[MAIN_SHEET_SUITE_COL_NO] + "#" * 50 + "\n")
105 continue
106 info("#" * 50 + " 测试用例集【%s】开始执行 " % data[MAIN_SHEET_SUITE_COL_NO] + "#" * 50)
107 excel, suite_test_flag, first_test_time = suite_process(excel, data[MAIN_SHEET_SUITE_COL_NO])
108 if suite_test_flag:
109 info("#" * 50 + " 测试用例集【%s】执行成功！ " % data[MAIN_SHEET_SUITE_COL_NO] + "#" * 50 + "\n")
110 data[MAIN_SHEET_TEST_RESULT_COL_NO] = "pass"
111 else:
112 info("#" * 50 + " 测试用例集【%s】执行失败！ " % data[MAIN_SHEET_SUITE_COL_NO] + "#" * 50 + "\n")
113 data[MAIN_SHEET_TEST_RESULT_COL_NO] = "fail"
114 data[MAIN_SHEET_TEST_TIME_COL_NO] = first_test_time
115 # 切换到“测试结果明细”sheet，以写入测试执行结果
116 excel.get_sheet("测试结果明细")
117 # 写入标题行
118 excel.write_row_data(head_line, None, True, "red")
119 # 写入测试结果
120 excel.write_row_data(data)
121 # 返回excel对象，是为了在生成测试报告时再调用save()生成excel版测试结果文件，以此同步和html测试报告相同的时间戳命名
122 return excel
123
124
125 # 区分模块sheet与主sheet的用例执行，生成测试报告，发送测试报告邮件
126 def main_process(excel_file_path, sheet_name, excel_report_name, html_report_name, receiver, subject, content):
127 if sheet_name == "测试用例集":
128 excel_obj = main_suite_process(excel_file_path, sheet_name)
129 else:
130 excel_obj = suite_process(excel_file_path, sheet_name)[0]
131 if not isinstance(excel_obj, Excel):
132 error("测试执行失败，已停止生成测试报告！")
133 timestamp = get_timestamp()
134 xlsx, html = create_xlsx_and_html_report(TEST_RESULT_FOR_REPORT, excel_report_name, html_report_name, excel_obj, timestamp)
135 send_mail([xlsx, html], receiver, subject+"_"+timestamp, content)
136
137
138 if __name__ == "__main__":
139 # excel_obj = main_suite_process(EXCEL_FILE_PATH, "测试用例集")
140 # create_xlsx_and_html_report(TEST_RESULT_FOR_REPORT, "接口测试报告", excel_obj, get_timestamp())
141 # suite_process(EXCEL_FILE_PATH, "注册&登录")
142 # main_suite_process(EXCEL_FILE_PATH, "测试用例集")
143 html_report_name = "接口自动化测试报告"
144 receiver = "182230124@qq.com"
145 subject = "接口自动化测试报告"
146 content = "接口自动化测试报告excel版和html版 请查收附件~"
147 main_process(EXCEL_FILE_PATH, "测试用例集", html_report_name, receiver, subject, content)
util 包
util 包属于框架第一层“测试工具库”：用于实现测试过程中调用的工具类方法，例如读取配置文件、接口请求、生成测试报告、发送邮件等。
global_var.py
该模块存放了整个框架中所用到的全局变量。
1 import os
2
3
4 # 工程根目录
5 PROJECT_ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
6
7 # 维护唯一数参数的文件路径
8 UNIQUE_NUM_FILE_PATH = os.path.join(PROJECT_ROOT_DIR, "config", "unique_num.txt")
9
10 # 维护接口服务端信息的ini文件路径
11 INI_FILE_PATH = os.path.join(PROJECT_ROOT_DIR, "config", "server_info.ini")
12
13 # excel数据文件
14 EXCEL_FILE_PATH = os.path.join(PROJECT_ROOT_DIR, "test_data", "interface_test_case.xlsx")
15
16 # 维护一个参数化全局变量：供接口关联使用
17 PARAM_GLOBAL_DICT = {}
18
19 # 测试报告保存目录
20 TEST_REPORT_SAVE_PATH = os.path.join(PROJECT_ROOT_DIR, "test_report")
21
22 # excel数据文件用例数据列号
23 CASE_API_NAME_COL_NO = 1
24 CASE_SERVER_COL_NO = 2
25 CASE_METHOD_COL_NO = 3
26 CASE_REQUEST_DATA_COL_NO = 4
27 CASE_RESPONSE_DATA_COL_NO = 5
28 CASE_TIME_COST_COL_NO = 6
29 CASE_ASSERT_KEYWORD_COL_NO = 7
30 CASE_RELEVANT_PARAM_COL_NO = 8
31 CASE_IS_EXECUTE_COL_NO = 9
32 CASE_TEST_TIME_COL_NO = 10
33 CASE_TEST_RESULT_COL_NO = 11
34 CASE_EXCEPTION_INFO_COL_NO = 12
35
36 # excel主sheet的用例数据列号
37 MAIN_SHEET_SUITE_COL_NO = 1
38 MAIN_SHEET_IS_EXECUTE_COL_NO = 2
39 MAIN_SHEET_TEST_TIME_COL_NO = 3
40 MAIN_SHEET_TEST_RESULT_COL_NO = 4
41
42 # 存储测试报告需要用的测试结果数据
43 TEST_RESULT_FOR_REPORT = []
44
45 # 日志配置文件路径
46 LOG_CONF_FILE_PATH = os.path.join(PROJECT_ROOT_DIR, "config", "logger.conf")
47
48 # 测试结果统计
49 TOTAL_CASE = 0
50 PASS_CASE = 0
51 FAIL_CASE = 0
52
53
54 if __name__ == "__main__":
55 print(PROJECT_ROOT_DIR)
56 print(LOG_CONF_FILE_PATH)
request_util.py
该模块实现了对接口请求的所需工具函数，如获取递增唯一数（供注册用户名使用）、md5 加密（用于登录密码加密）、响应数据断言等功能。
get_unique_num()：用于获取每次递增的唯一数
该函数的目标是解决注册用户名重复的问题。
虽然可以在赋值注册用户名变量时，采用前缀字符串拼接随机数的方式，但是用随机数的方式仍然是有可能出现用户名重复的情况。因此，可以在单独的一个文件中维护一个数字，每次请求注册接口之前，先读取该文件中的数字，拼接用户名前缀字符串。读取完之后，再把这个数字进行加一的操作并保存，即每读取一次这个数字之后，就做一次修改，进而保证每次拼接的用户名都是唯一的，避免出现因为用户名重复导致用例执行失败的情况。
1 import requests
2 import json
3 import hashlib
4 import traceback
5 from util.global_var import *
6 from util.ini_reader import IniParser
7 from util.log_util import *
8
9
10 # 获取递增的唯一数参数
11 def get_unique_num():
12 with open(UNIQUE_NUM_FILE_PATH, "r+") as f:
13 # 先从文件中获取当前的唯一数
14 num = f.read()
15 # 将唯一数+1再写回文件
16 f.seek(0, 0)
17 f.write(str(int(num)+1))
18 return num
19
20
21 # MD5加密
22 def md5(string):
23 # 创建一个md5 hash对象
24 m = hashlib.md5()
25 # 对字符串进行md5加密的更新处理，需要指定编码
26 m.update(string.encode("utf-8"))
27 # 返回十六进制加密结果
28 return m.hexdigest()
29
30
31 # 根据接口主机名称映射获取主机的IP和端口
32 def get_request_url(server_name):
33 p = IniParser(INI_FILE_PATH)
34 ip = p.get_value(server_name, "ip")
35 port = p.get_value(server_name, "port")
36 del p
37 return "http://%s:%s" % (ip, port)
38
39
40 # 接口请求函数
41 def http_client(url, api_name, method, data, headers=None, cookies=None):
42 # 校验数据是否符合json格式
43 try:
44 # 字典对象转json字符串
45 if isinstance(data, dict):
46 data = json.dumps(data)
47 elif isinstance(data, str):
48 data = json.loads(data)
49 data = json.dumps(data)
50 except:
51 error("接口【%s】json格式有误！" % (url+"/%s/"%api_name))
52 traceback.print_exc()
53 return traceback.format_exc()
54 if method.lower() == "post":
55 response = requests.post(url+"/%s/" % api_name, data=data, headers=headers, cookies=cookies)
56 elif method.lower() == "get":
57 response = requests.get(url+"/%s" % api_name, params=data, headers=headers, cookies=cookies)
58 else:
59 error("接口【%s】请求方法【%s】有误！" % (url+"/%s/", method))
60 return False
61 return response
62
63
64 # 断言
65 def assert_keyword(response, keyword):
66 assert keyword in response.text
excel_util.py
该模块封装了对 excel 的读写操作（openpyxl 版本：3.0.4）。
1 import os
2 from openpyxl import load_workbook
3 from openpyxl.styles import PatternFill, Font, Side, Border
4 from util.datetime_util import *
5 from util.global_var import *
6 from util.log_util import *
7
8
9 # 支持excel读写操作的工具类
10 class Excel:
11
12 # 初始化读取excel文件
13 def __init__(self, file_path):
14 if not os.path.exists(file_path):
15 return
16 self.wb = load_workbook(file_path)
17 # 初始化默认sheet
18 self.ws = self.wb.active
19 self.data_file_path = file_path
20 # 初始化颜色字典，供设置样式用
21 self.color_dict = {"red": "FFFF3030", "green": "FF008B00"}
22
23 def get_all_sheet(self):
24 return self.wb.get_sheet_names()
25
26 # 打开指定sheet
27 def get_sheet(self, sheet_name):
28 if sheet_name not in self.get_all_sheet():
29 error("sheet名称【%s】不存在，切换sheet失败！" % sheet_name)
30 return
31 self.ws = self.wb.get_sheet_by_name(sheet_name)
32 return True
33
34 # 获取最大行号
35 def get_max_row_no(self):
36 # openpyxl的API的行、列索引默认都从1开始
37 return self.ws.max_row
38
39 # 获取最大列号
40 def get_max_col_no(self):
41 return self.ws.max_column
42
43 # 获取所有行数据
44 def get_all_row_data(self, head_line=True):
45 # 是否需要标题行数据的标识，默认需要
46 if head_line:
47 min_row = 1 # 行号从1开始，即1为标题行
48 else:
49 min_row = 2
50 result = []
51 # min_row=None：默认获取标题行数据
52 for row in self.ws.iter_rows(min_row=min_row, max_row=self.get_max_row_no(), max_col=self.get_max_col_no()):
53 # 如果单元格为空（None），则转成空字符串
54 result.append([cell.value for cell in row])
55 return result
56
57 # 获取指定行数据
58 def get_row_data(self, row_num):
59 # 0 为标题行
60 return [cell.value for cell in self.ws[row_num+1]]
61
62 # 获取指定列数据
63 def get_col_data(self, col_num):
64 # 索引从0开始
65 return [cell.value for cell in tuple(self.ws.columns)[col_num]]
66
67 # 追加行数据且可以设置样式
68 def write_row_data(self, data, font_color=None, border=True, fill_color=None):
69 if not isinstance(data, (list, tuple)):
70 print("写入数据失败：数据不为列表或元组类型！【%s】" % data)
71 self.ws.append(data)
72 # 设置字体颜色
73 if font_color:
74 if font_color.lower() in self.color_dict.keys():
75 font_color = self.color_dict[font_color]
76 # 设置单元格填充颜色
77 if fill_color:
78 if fill_color.lower() in self.color_dict.keys():
79 fill_color = self.color_dict[fill_color]
80 # 设置单元格边框
81 if border:
82 bd = Side(style="thin", color="000000")
83 # 记录数据长度（否则会默认与之前行最长数据行的长度相同，导致样式超过了该行实际长度）
84 count = 0
85 for cell in self.ws[self.get_max_row_no()]:
86 # 设置完该行的实际数据长度样式后，则退出
87 if count > len(data) - 1:
88 break
89 if font_color:
90 cell.font = Font(color=font_color)
91 # 如果没有设置字体颜色，则默认给执行结果添加字体颜色
92 else:
93 if cell.value is not None and isinstance(cell.value, str):
94 if cell.value.lower() == "pass" or cell.value == "成功":
95 cell.font = Font(color=self.color_dict["green"])
96 elif cell.value.lower() == "fail" or cell.value == "失败":
97 cell.font = Font(color=self.color_dict["red"])
98 if border:
99 cell.border = Border(left=bd, right=bd, top=bd, bottom=bd)
100 if fill_color:
101 cell.fill = PatternFill(fill_type="solid", fgColor=fill_color)
102 count += 1
103
104 # 指定行插入数据（行索引从0开始）
105 def insert_row_data(self, row_no, data, font_color=None, border=True, fill_color=None):
106 if not isinstance(data, (list, tuple)):
107 print("写入数据失败：数据不为列表或元组类型！【%s】" % data)
108 for idx, cell in enumerate(self.ws[row_no+1]): # 此处行索引从1开始
109 cell.value = data[idx]
110
111 # 保存写入了测试结果的excel数据文件
112 def save(self, file_save_name, timestamp):
113 save_dir = os.path.join(TEST_REPORT_SAVE_PATH, get_chinese_date())
114 if not os.path.exists(save_dir):
115 os.mkdir(save_dir)
116 excel_result_file_path = os.path.join(save_dir, file_save_name + "_" + timestamp + ".xlsx")
117 self.wb.save(excel_result_file_path)
118 return excel_result_file_path
119
120
121 if __name__ == "__main__":
122 from util.global_var import *
123 excel = Excel(EXCEL_FILE_PATH)
124 excel.get_sheet("测试用例集")
125 # print(excel.get_all_row_data())
126 excel.write_row_data(["4", None, "失败"], None, True, "green")
127 excel.save()
ini_reader.py
该模块封装对 ini 配置文件的读取操作。
1 import configparser
2 import os
3
4
5 # 读取ini文件的工具类
6 class IniParser:
7
8 # 初始化打开ini文件
9 def __init__(self, file_path):
10 if not os.path.exists(file_path):
11 print("ini文件【%s】不存在！" % file_path)
12 return
13 self.cf = configparser.ConfigParser()
14 self.cf.read(file_path, encoding="utf-8")
15
16 # 获取所有分组
17 def get_sections(self):
18 return self.cf.sections()
19
20 # 获取指定分组的所有键
21 def get_options(self, section):
22 return self.cf.options(section)
23
24 # 获取指定分组的所有键值对
25 def get_items(self, section):
26 return self.cf.items(section)
27
28 # 获取指定分组指定键的值
29 def get_value(self, section, option):
30 return self.cf.get(section, option)
31
32
33 if __name__ == "__main__":
34 from util.global_var import *
35 p = IniParser(INI_FILE_PATH)
36 print(p.get_sections())
37 print(p.get_options("server1"))
38 print(p.get_items("server1"))
39 print(p.get_value("server1", "ip"))
email_util.py
该模块封装了邮件发送功能。
1 import yagmail
2 import traceback
3 from util.log_util import *
4
5
6 def send_mail(attachments_report_name, receiver, subject, content):
7 try:
8 # 连接邮箱服务器
9 # 注意：若使用QQ邮箱，则password为授权码而非邮箱密码；使用其它邮箱则为邮箱密码
10 # encoding设置为GBK，否则中文附件名会乱码
11 yag = yagmail.SMTP(user="******@163.com", password="******", host="smtp.163.com", encoding='GBK')
12
13 # 收件人、标题、正文、附件（若多个收件人或多个附件，则可使用列表）
14 yag.send(to=receiver, subject=subject, contents=content, attachments=attachments_report_name)
15
16 # 可简写：yag.send("****@163.com", subject, contents, report)
17
18 info("测试报告邮件发送成功!【邮件标题：%s】【收件人：%s】" % (subject, receiver))
19 except:
20 error("测试报告邮件发送失败!【邮件标题：%s】【收件人：%s】" % (subject, receiver))
21 error(traceback.format_exc())
22
23
24 if __name__ == "__main__":
25 send_mail("e:\\code.txt", "182230124@qq.com", "测试邮件", "正文")
datetime_util.py
该模块封装了获取各种格式的当前日期时间。
1 import time
2
3
4 # 返回中文格式的日期：xxxx年xx月xx日
5 def get_chinese_date():
6 year = time.localtime().tm_year
7 if len(str(year)) == 1:
8 year = "0" + str(year)
9 month = time.localtime().tm_mon
10 if len(str(month)) == 1:
11 month = "0" + str(month)
12 day = time.localtime().tm_mday
13 if len(str(day)) == 1:
14 day = "0" + str(day)
15 return "{}年{}月{}日".format(year, month, day)
16
17
18 # 返回英文格式的日期：xxxx/xx/xx
19 def get_english_date():
20 year = time.localtime().tm_year
21 if len(str(year)) == 1:
22 year = "0" + str(year)
23 month = time.localtime().tm_mon
24 if len(str(month)) == 1:
25 month = "0" + str(month)
26 day = time.localtime().tm_mday
27 if len(str(day)) == 1:
28 day = "0" + str(day)
29 return "{}/{}/{}".format(year, month, day)
30
31
32 # 返回中文格式的时间：xx时xx分xx秒
33 def get_chinese_time():
34 hour = time.localtime().tm_hour
35 if len(str(hour)) == 1:
36 hour = "0" + str(hour)
37 minute = time.localtime().tm_min
38 if len(str(minute)) == 1:
39 minute = "0" + str(minute)
40 second = time.localtime().tm_sec
41 if len(str(second)) == 1:
42 second = "0" + str(second)
43 return "{}时{}分{}秒".format(hour, minute, second)
44
45
46 # 返回英文格式的时间：xx:xx:xx
47 def get_english_time():
48 hour = time.localtime().tm_hour
49 if len(str(hour)) == 1:
50 hour = "0" + str(hour)
51 minute = time.localtime().tm_min
52 if len(str(minute)) == 1:
53 minute = "0" + str(minute)
54 second = time.localtime().tm_sec
55 if len(str(second)) == 1:
56 second = "0" + str(second)
57 return "{}:{}:{}".format(hour, minute, second)
58
59
60 # 返回中文格式的日期时间
61 def get_chinese_datetime():
62 return get_chinese_date() + " " + get_chinese_time()
63
64
65 # 返回英文格式的日期时间
66 def get_english_datetime():
67 return get_english_date() + " " + get_english_time()
68
69
70 # 返回时间戳
71 def get_timestamp():
72 year = time.localtime().tm_year
73 if len(str(year)) == 1:
74 year = "0" + str(year)
75 month = time.localtime().tm_mon
76 if len(str(month)) == 1:
77 month = "0" + str(month)
78 day = time.localtime().tm_mday
79 if len(str(day)) == 1:
80 day = "0" + str(day)
81 hour = time.localtime().tm_hour
82 if len(str(hour)) == 1:
83 hour = "0" + str(hour)
84 minute = time.localtime().tm_min
85 if len(str(minute)) == 1:
86 minute = "0" + str(minute)
87 second = time.localtime().tm_sec
88 if len(str(second)) == 1:
89 second = "0" + str(second)
90 return "{}{}{}_{}{}{}".format(year, month, day, hour, minute, second)
91
92
93 if __name__ == "__main__":
94 print(get_chinese_datetime())
95 print(get_english_datetime())
log_util.py
该模块封装了日志功能。
1 import logging
2 import logging.config
3 from util.global_var import *
4
5
6 # 日志配置文件：多个logger，每个logger指定不同的handler
7 # handler：设定了日志输出行的格式
8 # 以及设定写日志到文件（是否回滚）？还是到屏幕
9 # 还定了打印日志的级别
10 logging.config.fileConfig(LOG_CONF_FILE_PATH)
11 logger = logging.getLogger("example01")
12
13
14 def debug(message):
15 logging.debug(message)
16
17
18 def info(message):
19 logging.info(message)
20
21
22 def warning(message):
23 logging.warning(message)
24
25
26 def error(message):
27 logging.error(message)
28
29
30 if __name__ == "__main__":
31 debug("hi")
32 info("gloryroad")
33 warning("hello")
34 error("这是一个error日志")
report_util.py
该模块实现了 html 及 excel 测试报告的生成。
1 from bottle import template
2 import os
3 from util.global_var import *
4 from util.datetime_util import *
5 from util.log_util import *
6
7
8 # 生成测试报告html模板文件
9 def report_html(data, html_name, timestamp):
10 """
11 :param data: 保存测试结果的列表对象
12 :param html_name: 报告名称
13 """
14 template_demo = """
15 <!-- CSS goes in the document HEAD or added to your external stylesheet -->
16 <style type="text/css">
17 table.hovertable {
18 font-family: verdana,arial,sans-serif;
19 font-size:10px;
20 color:#333333;
21 border-width: 1px;
22 border-color: #999999;
23 border-collapse: collapse;
24 }
25 table.hovertable th {
26 background-color:#ff6347;
27 border-width: 1px;
28 padding: 15px;
29 border-style: solid;
30 border-color: #a9c6c9;
31 }
32 table.hovertable tr {
33 background-color:#d4e3e5;
34 }
35 table.hovertable td {
36 border-width: 1px;
37 padding: 15px;
38 border-style: solid;
39 border-color: #a9c6c9;
40 }
41 </style>
42
43 <!-- Table goes in the document BODY -->
44
45 <head>
46
47 <meta http-equiv="content-type" content="txt/html; charset=utf-8" />
48
49 </head>
50
51 <table class="hovertable">
52 <tr>
53 <th>接口 URL</th><th>请求数据</th><th>接口响应数据</th><th>接口调用耗时(ms)</th><th>断言词</th><th>测试结果</th><th>异常信息</th>
54 </tr>
55 % for url,request_data,response_data,test_time,assert_word,result,exception_info in items:
56 <tr onmouseover="this.style.backgroundColor='#ffff66';" onmouseout="this.style.backgroundColor='#d4e3e5';">
57
58 <td>{{url}}</td><td>{{request_data}}</td><td>{{response_data}}</td><td>{{test_time}}</td><td>{{assert_word}}</td><td>
59 % if result == '失败':
60 <font color=red>
61 % elif result == '成功':
62 <font color=green>
63 % end
64 {{result}}</td>
65 <td>{{exception_info}}</td>
66 </tr>
67 % end
68 </table>
69 """
70 html = template(template_demo, items=data)
71 """
72 :param template_demo: 渲染的模板名称（可以是字符串对象，也可以是模板文件名）
73 :param items: 保存测试结果的列表对象
74 :return: 渲染之后的模板（字符串对象）
75 """
76 # 生成测试报告
77 save_dir = os.path.join(TEST_REPORT_SAVE_PATH, get_chinese_date())
78 if not os.path.exists(save_dir):
79 os.mkdir(save_dir)
80 report_name = os.path.join(save_dir, html_name+"_"+timestamp+".html")
81 with open(report_name, 'wb') as f:
82 f.write(html.encode('utf-8'))
83 return os.path.join(save_dir, html_name+"_"+timestamp+".html")
84
85
86 # 将测试结果写入html测试报告
87 def create_html_report(test_result_data, html_name, timestamp):
88 html_name = html_name
89 return report_html(test_result_data, html_name, timestamp)
90
91
92 # 将测试结果写入excel数据文件
93 def create_excel_report(excel_obj, save_name, timestamp):
94 return excel_obj.save(save_name, timestamp)
95
96
97 # 同时生成两种测试报告（html和html）
98 def create_xlsx_and_html_report(test_result_data, excel_name, html_name, excel_obj, timestamp):
99 html_report_file_path = create_html_report(test_result_data, html_name, timestamp)
100 excel_file_path = create_excel_report(excel_obj, excel_name, timestamp)
101 info("生成excel测试报告：{}".format(excel_file_path))
102 info("生成html测试报告：{}".format(html_report_file_path))
103 return excel_file_path, html_report_file_path
conf 目录
unique_num.txt
该文件维护了一个数字，供 request_util.py 的 get_unique_num() 函数使用，该函数用于获取当前文件中的数值，并写入递增 +1 的数值。
30179
server_info.ini
该配置文件维护了请求接口主机的地址信息。
[server1]
ip = 39.100.104.214
port = 8080
logger.conf
该配置文件维护日志功能的相关配置信息。
###############################################
[loggers]
keys=root,example01,example02
[logger_root]
level=DEBUG
handlers=hand01,hand02[logger_example01]
handlers=hand01,hand02
qualname=example01
propagate=0[logger_example02]
handlers=hand01,hand03
qualname=example02
propagate=0###############################################
[handlers]
keys=hand01,hand02,hand03[handler_hand01]
class=StreamHandler
level=INFO
formatter=form01
args=(sys.stderr,)[handler_hand02]
class=FileHandler
level=DEBUG
formatter=form01
args=('E:\\pycharm_project_dir\\InterfaceAutoTest\\log\\interface_test.log', 'a')[handler_hand03]
class=handlers.RotatingFileHandler
level=INFO
formatter=form01
args=('E:\\pycharm_project_dir\\InterfaceAutoTest\\log\\interface_test.log', 'a', 10*1024*1024, 5)###############################################
[formatters]
keys=form01,form02[formatter_form01]
format=%(asctime)s [%(levelname)s] %(message)s
datefmt=%Y-%m-%d %H:%M:%S[formatter_form02]
format=%(name)-12s: [%(levelname)-8s] %(message)s
datefmt=%Y-%m-%d %H:%M:%S
test_data 目录
该目录用于存放 excel 测试数据文件。
main.py
main.py 模块是本框架的运行主入口，属于框架第四层“测试场景层”，将测试用例组织成测试场景，实现各种级别 cases 的管理，如冒烟，回归等测试场景。
- 基于 business_process/main_process.py 中的模块用例 sheet 执行函数或主 sheet 执行函数，组装测试场景。
- 可直接用代码组装测试场景，也可根据 excel 数据文件对用例集合和用例步骤的维护来组装测试场景。
- 完成测试执行后生成测试结果文件并发送邮件。
1 from bussiness_process.main_process import *
2 from util.global_var import *
3 from util.report_util import *
4 from util.datetime_util import *
5 from util.email_util import send_mail
6
7
8 # 冒烟场景
9 def smoke_test(report_file_name):
10 """
11 :param report_file_name: 报告名称与邮件标题
12 :return:
13 """
14 excel_obj, _, _ = suite_process(EXCEL_FILE_PATH, "注册")
15 excel_obj, _, _ = suite_process(excel_obj, "登录")
16 excel_obj, _, _ = suite_process(excel_obj, "查询博文")
17 timestamp = get_timestamp()
18 # 生成excel和html的两份测试报告
19 excel_file, html_file = create_xlsx_and_html_report(TEST_RESULT_FOR_REPORT, report_file_name,
20 report_file_name, excel_obj, timestamp)
21 receiver = "182230124@qq.com" # 收件人
22 subject = report_file_name + "_" + timestamp # 邮件标题
23 content = "接口自动化测试报告excel版和html版 请查收附件~" # 邮件正文
24 send_mail([excel_file, html_file], receiver, subject, content)
25
26
27 # 测试用例集测试
28 def suite_test(report_file_name):
29 """
30 :param report_file_name: 报告名称与邮件标题
31 :return:
32 """
33 receiver = "182230124@qq.com" # 收件人
34 subject = "接口自动化测试报告_全量测试" # 邮件标题
35 content = "接口自动化测试报告excel版和html版 请查收附件~" # 邮件正文
36 main_process(EXCEL_FILE_PATH, "测试用例集", report_file_name, report_file_name, receiver, subject, content)
37
38
39 if __name__ == "__main__":
40 # smoke_test("接口自动化测试报告_冒烟测试")
41 suite_test("接口自动化测试报告_全量测试")
test_report 目录
该目录用于存放 excel 和 html 版的测试报告。
log 目录
该目录用于存放测试执行日志文件（日志内容同时也会输出到控制台）。
log/interface_test.log：
