给一张表加一个自动编号字段_python+requests实现接口自动化

1. 前言

今年2月调去支持项目接口测试&＃xff0c;测试过程中使用过postman、jmeter工具&＃xff0c;基本能满足使用&＃xff0c;但是部分情况下使用较为麻烦。
比如&＃xff1a;部分字段存在唯一性校验或字段间有业务性校验&＃xff0c;每次请求均需手工修改部分报文内容&＃xff0c;使用工具难以满足实际使用。
因此&＃xff0c;萌生了使用python去实现接口自动化的想法。之前未接触过接口测试&＃xff0c;但有一点编程基础&＃xff0c;经过2个多月的磕磕碰碰&＃xff0c;不断完善&＃xff0c;经历2次重构之后&＃xff0c;最后基本达成了目标。

2. 技术栈

• python语言

• requests库

• unittest单元测试框架

• HTMLTestReportCN、BeautifulReport测试报告

3. 实现的功能概述

• 支持post、get等请求类型&＃xff0c;xml、json格式的报文

• 支持使用excel编写测试用例&＃xff0c;测试用例支持涉及多接口的场景用例&＃xff1b;支持按脚本的形式编写测试用例

• 支持测试结果保存至数据库&＃xff0c;支持生成html报告&＃xff0c;支持将生成的测试结果导出到excel文件

• 支持邮件发送测试结果

• 支持多线程并发执行测试用例

4. 框架及项目结构

APIS_AutoTest

api: 主程序目录

comm&＃xff1a;公共函数&＃xff0c;包括&＃xff1a;接口请求基类、请求及相应数据操作基类等

intf_handle&＃xff1a;接口操作层&＃xff0c;包含&＃xff1a;接口初始化、断言等

business&＃xff1a;业务实现部分

utils&＃xff1a;工具类&＃xff0c;包括&＃xff1a;读取文件、发送邮件、excel操作、数据库操作、日期时间格式化等

config&＃xff1a;配置文件目录&＃xff0c;包含yaml配置文件、以及路径配置

data&＃xff1a;测试数据目录&＃xff0c;用于存放测试数据

temp&＃xff1a;临时文件目录&＃xff0c;用于存放临时文件

result&＃xff1a;结果目录

report&＃xff1a;测试报告目录&＃xff0c;用于存放生成的html报告

details&＃xff1a;测试结果详情目录&＃xff0c;用于存放生成的测试用例执行结果excel文件

log&＃xff1a;日志文件目录

test&＃xff1a;测试用例、测试集相关目录&＃xff0c;启动test_suite执行用例文件存放在此

test_case&＃xff1a;测试用例存放路径

test_suite&＃xff1a;测试模块集&＃xff0c;按模块组装用例

5. 测试用例执行流程

• 以脚本形式编写的测试用例执行流程图&＃xff1a;

  

• 以excel形式编写的测试用例执行流程图&＃xff1a;

  

6. 核心方法设计

• 接口请求基类

  RequestBase(request_type, url, header, body, data_type&＃61;None)

功能描述&＃xff1a;

根据传入的请求类型&＃xff0c;请求地址&＃xff0c;请求头&＃xff0c;请求体&＃xff0c;发送接口请求&＃xff0c;获得响应头&＃xff0c;响应体

• request_type: 请求类型&＃xff0c;只能是&＃39;GET&＃39;, &＃39;POST&＃39;, &＃39;HEAD&＃39;, &＃39;OPTIONS&＃39;, &＃39;PUT&＃39;, &＃39;DELETE&＃39;, &＃39;TRACE&＃39;, &＃39;CONNECT&＃39;中的一个&＃xff0c;无大小写要求

• url: 请求地址&＃xff0c;完整的接口地址

• head: 请求头

• body&＃xff1a;请求体&＃xff0c;xml格式请求体为字符串&＃xff1b;json格式请求体需传入json格式

• data_type: 数据类型&＃xff0c;未使用字段&＃xff0c;备用。可为&＃xff1a;xml、json

支持方法

__send_request()
发送请求&＃xff0c;类私有方法&＃xff0c;初始化时调用&＃xff0c;调用方法返回&＃xff1a;响应对象

get_respond()
获取响应对象&＃xff0c;调用方法返回&＃xff1a;响应对象

get_respond_head()
获取响应头&＃xff0c;调用方法返回&＃xff1a;响应头

get_respond_body()
获取响应体&＃xff0c;调用方法返回&＃xff1a;响应体

• 请求或响应数据操作基类

  RequestRespondHandle(data_type, body, fields, value_dict&＃61;None)

功能描述:

根据传入的xml、json格式请求体或响应体&＃xff0c;读取字段的值或更新字段的值

参数描述:

• data_type: 数据类型&＃xff0c;当前仅支持xml、json&＃xff0c;不区分大小写

• body: 请求体或响应体&＃xff0c;json格式请求体支持传入字典、json格式数据(自动转换为字典)

• fields: 字段名称&＃xff0c;支持多字段传入&＃xff0c;支持数据类型&＃xff1a;字符串、元组、列表。多字段传入形式&＃xff0c;字符串&＃xff1a;多个字段名称之间逗号隔开&＃xff0c;如&＃xff1a;&＃39;a1,b2&＃39;&＃xff1b;元组、列表正常传入即可。json格式&＃xff1a;需写入完整节点路径&＃xff0c;如&＃xff1a;body.base.name&＃xff0c;对应list类型的需传入索引位置&＃xff0c;如&＃xff1a;body.baselist[0].name

• value_dict: 字段值字典&＃xff0c;以字段名称及字段值键值对的方式存储数据&＃xff0c;读取字段值时&＃xff0c;一般不需传入&＃xff0c;也支持传入(用于读取excel形式流程前后传值)&＃xff1b;更新字段值时&＃xff0c;需传入&＃xff0c;且字典中的key值需要与fields中的字段名对应。

支持方法:

get_fields_value()
获取字段值&＃xff0c;调用方法即可获取到xml或json格式的请求体或响应体对应的字段值&＃xff0c;并返回&＃xff1a;字段值字典

update_fields_value()
更新字段值&＃xff0c;调用方法即可更新请求体中对应的字段值&＃xff0c;返回更新后的请求体。
Json格式请求体返回格式为字典。

不足与改进&＃xff1a;

xml格式&＃xff0c;读取或更新字段时&＃xff0c;若存在多个相同名称字段&＃xff0c;默认只选第一个&＃xff1b;
json格式&＃xff0c;读取嵌套列表的时候&＃xff0c;未支持按列表读取&＃xff0c;当前需精确位置单个读取或更新&＃xff1b;

• 请求体初始化-接口映射类

  RequestMsgInitMapper(data_type, intf_code, request_body, **kwargs)

功能描述&＃xff1a;

根据传入的接口编号&＃xff0c;映射到对应的接口请求报文的初始化方法&＃xff0c;进行接口初始化

参数描述&＃xff1a;

• data_type: 数据类型&＃xff0c;执行xml、json&＃xff0c;不区分大小写

• intf_code: 接口编号&＃xff0c;每个接口都要一个接口编号&＃xff0c;根据接口编号可以唯一确定一个接口

• request_body: 请求体&＃xff0c;json格式请求体支持传入字典、json格式数据

• **kwargs: 可变关键字参数&＃xff0c;每个接口初始化时&＃xff0c;参数数量均不一致&＃xff0c;所有需要使用可变参数&＃xff0c;需要以a&＃61;value这种方式传入。部分字段(如id)可在方法内设置生成方法&＃xff0c;一般可变参数设置都是用于前后接口字段传值。

支持方法:

start_Intf_init_mapper()
启动 接口初始化映射&＃xff0c;该方法通过判断传入数据类型、接口编号&＃xff0c;执行对应的接口初始化方法。
调用方法后&＃xff0c;进行对应接口的初始化&＃xff0c;并返回初始化后的请求体。
Json格式的数据&＃xff0c;返回json格式的请求体(不管传入的是字典格式、还是json格式)。

适用场景说明:

接口映射类主要是针对不同接口初始化字段涉及复杂业务判断&＃xff0c;需按接口单独编写的场景。
如果一个接口初始化涉及的字段均不涉及业务相关的复杂判断处理&＃xff0c;可以直接统一使用通用接口初始化方法进行初始化。

通用接口初始化方法

intf_base_init(data_type, request_body, **kwargs)

功能描述:

根据可变关键字参数传入的键值对&＃xff0c;进行接口报文初始化&＃xff0c;返回初始化后的请求体

参数描述:

• data_type: 数据类型&＃xff0c;执行xml、json&＃xff0c;不区分大小写

• request_body: 请求体&＃xff0c;json格式请求体支持传入字典、json格式数据

• **kwargs&＃xff1a;可变关键字参数&＃xff0c;每个接口初始化时&＃xff0c;参数数量均不一致&＃xff0c;所有需要进行初始化的字段&＃xff0c;均需以键值对的方式传入。

适用场景说明:

接口初始化字段不涉及复杂逻辑判断&＃xff0c;直接传值后更新即可&＃xff1b;该方式也适用于接口请求头的初始化。

7. 数据操作类

数据操作主要是数据读取、写入&＃xff0c;主要分为如下几类&＃xff1a;

- 读取txt、json等文件内容(整个读取)、写入内容到文件
- 读取yaml文件内容
- 读取excel表格内容、导出excel表格、读取excel并作为模板
- 操作数据库表中的数据(增删改查)

读取txt/json文件内容&＃xff0c;写入内容到文件

FileHandle(file_name, file_path)

功能描述&＃xff1a;

读取文件所有内容

参数描述&＃xff1a;

• file_name&＃xff1a;读取/写入文件名称&＃xff0c;包含后缀&＃xff0c;支持txt&＃xff0c;json等

• file_path&＃xff1a;读取/写入文件所在的目录

支持方法&＃xff1a;

read_file_content()
读取文件内容&＃xff0c;并以字符串返回

write_to_file(content)
将内容写入文件

读取yaml文件内容

ReadYaml(file_path)

功能描述&＃xff1a;

读取yaml文件内&＃xff0c;支持按名称读取

参数描述&＃xff1a;

• file_path&＃xff1a;读取文件的完整路径

支持方法&＃xff1a;

get_yaml()
读取yaml文件所有内容&＃xff0c;并以字典格式返回

get_value(level_name)
读取yaml文件字段的值&＃xff0c;并返回
level_name: 节点字段名称&＃xff0c;如涉及多节点需传入对应路径&＃xff0c;如&＃xff1a;db.host

读取excel表格内容、导出excel表格、读取excel并作为模板

ExcelHandle()

功能描述&＃xff1a;

读取excel表格内容、将数据导出到excle表格中

支持方法&＃xff1a;

read_excel_data(excel_path, sheet_name&＃61;None)

读取excel表格内容&＃xff0c;并以列表嵌套列表的方式返回

excel_path: 读取excel文件的完整路径
sheet_name: 读取excel的页签名称&＃xff0c;默认读取第一个页签

export_to_excel(data, head, file_name, time_flag&＃61;None)

将数据导出到excle表格中

data: 需要导出到excel的内容&＃xff0c;元组嵌套元组(对应数据库中查询返回的结果)
head: excel表头&＃xff0c;列表、元组嵌套元组(数据库中查询表头、描述&＃xff0c;支持多表头)
file_name: excel文件名称&＃xff0c;包含后缀.xlsx&＃xff0c;excel导出路径系统默认
time_flag: 时间戳标记&＃xff0c;可传入用例执行的报告号&＃xff0c;使其对应

copy_excel_template(template_path, sheet_name&＃61;None)

复制excel表格模板&＃xff0c;返回workbook&＃xff0c; workssheet

template_path: 模板文件完整路径
sheet_name: 模板文件页签名称&＃xff0c;默认第一个页签

操作数据库数据

功能描述&＃xff1a;

• 表数据读取&＃xff1a;根据情况读取所需数据&＃xff0c;返回数据格式元组嵌套元组。

• 插入/更新表数据&＃xff1a;根据实际内容&＃xff0c;写入数据到对应的表中

备注&＃xff1a;

数据库部分封装为数据库操作类

8. 部分关键代码

接口请求基类

# create by: wyun
# create at: 2020/4/19 9:30
import json

import requests


class RequestBase:

def __init__(self, request_type, request_url, request_body, request_headers, intf_type&＃61;None):
"""
请求接口公共类
:param request_type: 请求类型&＃xff1a;post, get
:param request_url: 请求url地址
:param request_headers: 请求头
:param request_body: 请求体(xml类型传入字符串格式&＃xff0c;json类型数据必须传入json格式&＃xff0c;不能传入字典)
:param intf_type: 接口类型&＃xff1a;webservice&＃xff0c; api&＃xff0c;分别对应接口数据格式&＃xff1a;xml&＃xff0c; json
"""
self.request_type &＃61; request_type
self.request_url &＃61; request_url
self.request_body &＃61; request_body
self.request_headers &＃61; request_headers

self.intf_type &＃61; intf_type
self.res &＃61; self.__send_request()

# 发送请求
def __send_request(self):
if not isinstance(self.request_type, str):
print(&＃39;请求类型格式错误。&＃39;)
return None
if self.request_type.upper() not in [&＃39;GET&＃39;, &＃39;POST&＃39;, &＃39;HEAD&＃39;, &＃39;OPTIONS&＃39;, &＃39;PUT&＃39;, &＃39;DELETE&＃39;, &＃39;TRACE&＃39;, &＃39;CONNECT&＃39;]:
print(&＃39;请求类型不存在。&＃39;)
return None
return requests.request(method&＃61;self.request_type, url&＃61;self.request_url, data&＃61;self.request_body.encode(&＃39;utf-8&＃39;),
headers&＃61;self.request_headers)

# 获取响应
def get_respond(self):
return self.res

# 获取响应头
def get_respond_head(self):
return self.res.headers

# 获取响应体
def get_respond_body(self):
if self.res.content:
return self.res.text



请求或响应数据操作基类

# create by: wyun
# create at: 2020/4/19 12:58
import json
import re

"""
对请求体或响应体进行处理&＃xff1a;
1. 支持读取请求体或响应体中字段的值
2. 支持更新请求体中字段的值
"""


# 递归调用&＃xff0c;更新json中的字段值
def update_json_step(node, json_str, value, i&＃61;0):
# 当前节点索引(负向)
node_index &＃61; -len(node) &＃43; i

# 如果包含[n]形式&＃xff0c;说明节点为列表&＃xff0c;需处理
if &＃39;[&＃39; in node[node_index] and &＃39;]&＃39; in node[node_index]:
list_node, list_index &＃61; node[node_index].split(&＃39;[&＃39;)
index &＃61; list_index.split(&＃39;]&＃39;)[0]
if index is None or index &＃61;&＃61; &＃39;&＃39;:
print(&＃39;参数传入错误&＃xff0c;请指定列表[%s]索引&＃39; % list_node)
else:
index &＃61; int(index)
return update_json_step(node, json_str[list_node][index], value, i &＃43; 1)

# 判断如果当前节点为最后一个节点&＃xff0c;则更新value值
if node_index &＃61;&＃61; -1:
json_str[node[-1]] &＃61; value
return json_str

return update_json_step(node, json_str[node[node_index]], value, i &＃43; 1)


class DataHandle:

def __init__(self, data_type, data_msg, fields, value_dict&＃61;None):
"""
处理请求体或响应体数据&＃xff0c;读取或更新字段值
:param data_type: 数据类型&＃xff1a;xml&＃xff0c;json
:param data_msg: 请求体或响应体
:param fields: 字段值&＃xff0c;支持多字读方式&＃xff0c;字段间逗号隔开&＃xff1b;或传入列表、元组
json格式&＃xff1a;需写入完整节点路径&＃xff0c;如&＃xff1a;body.base.name&＃xff0c;对应list类型的需传入索引位置&＃xff0c;如&＃xff1a;body.baselist[0].name
:param value_dict: 以字典键值对保存字段值
"""
self.data_type &＃61; data_type
self.data &＃61; data_msg
# fields支持str&＃xff0c;list方式&＃xff0c;str自动转换为list
if isinstance(fields, str):
self.fields &＃61; fields
self.fields_list &＃61; []
if &＃39;,&＃39; in fields:
self.fields_list &＃61; fields.strip().split(&＃39;,&＃39;)
else:
self.fields_list.append(fields.strip())
elif isinstance(fields, list):
self.fields_list &＃61; fields
else:
self.fields_list &＃61; list(fields)
# 初始化字典值
if value_dict is None:
self.value_dict &＃61; dict()
else:
self.value_dict &＃61; value_dict

def get_fields_value(self):
"""
获取字段值
1. 支持获取多个字段值&＃xff0c;输入字符串或列表&＃xff0c; 元组
"""
step_dict &＃61; self.value_dict
# 字段列表循环获取
for p in self.fields_list:
if p &＃61;&＃61; &＃39;&＃39;:
continue
# 处理xml格式报文
if self.data_type.upper() &＃61;&＃61; &＃39;XML&＃39;:
pattern &＃61; &＃39; &＃43; p &＃43; &＃39;>.*&＃39; &＃43; p &＃43; &＃39;>&＃39;search_result &＃61; re.search(pattern, self.data)if search_result is not None:# 包括标签和值都匹配上field_and_value &＃61; search_result.group()# 去除标签获取字段值&＃xff0c;并存入字典step_dict[p] &＃61; (field_and_value.split(&＃39;&＃39;)[0]).split(&＃39;>&＃39;)[-1]else:# print(&＃39;参数[%s]提取失败&＃xff0c;无匹配值。&＃39; % p)pass# 处理json格式报文elif self.data_type.upper() &＃61;&＃61; &＃39;JSON&＃39;:# 获取层级(使用.隔开)node &＃61; p.split(&＃39;.&＃39;)# json转换字典if isinstance(self.data, dict):temp &＃61; self.dataelse:temp &＃61; json.loads(self.data)# 逐层读取数据for per_node in node:if &＃39;[&＃39; in per_node and &＃39;]&＃39; in per_node:per_node, index_str &＃61; per_node.split(&＃39;[&＃39;)index &＃61; int(index_str.split(&＃39;]&＃39;)[0])temp &＃61; temp.get(per_node)[index]if temp is None:breakelse:temp &＃61; temp.get(per_node)if temp is None:break# 讲读取的结果写入字典step_dict[p] &＃61; temp# 非 xml&＃xff0c;json的数据格式&＃xff0c;报错退出else:print(&＃39;ERROR: 不支持此数据类型[%s]&＃39; % self.data_type)breakreturn step_dictdef update_fields_value(self):"""
更新字段值
支持更新单个字段值&＃xff0c;多个字段值(字段名 字符串或列表、元组&＃xff0c;字段值 字典)
"""req_body &＃61; self.data# 字段列表循环更新for p in self.fields_list:# 入参类型判断if isinstance(self.value_dict, dict):# 参数在字典中不存在&＃xff0c;则跳过if p not in self.value_dict.keys():print(&＃39;ERROR&＃xff1a; 字典中不存在参数[%s]。&＃39; % p)continue# 获取字典中参数的值value &＃61; self.value_dict[p]elif isinstance(self.value_dict, str):value &＃61; self.value_dictelse:print(&＃39;函数入参[%s]类型不支持&＃xff0c;请传入字符串或字典。&＃39; % self.value_dict)break# 更新数据类型为 xml 的参数值if self.data_type.upper() &＃61;&＃61; &＃39;XML&＃39;:# 优先寻找是否存在指定更新参数&＃xff0c;若有&＃xff0c;则按指定参数更新if &＃39;${&＃39; &＃43; p &＃43; &＃39;}&＃39; in req_body:req_body &＃61; req_body.replace(&＃39;${&＃39; &＃43; p &＃43; &＃39;}&＃39;, value)# 其次寻找标签参数值&＃xff0c;若有&＃xff0c;则按标签更新值else:# 检查标签是否存在&＃xff0c;若存在if &＃39; &＃43; p &＃43; &＃39;>&＃39; in req_body and &＃39;&＃39; &＃43; p &＃43; &＃39;>&＃39; in req_body:# 正则匹配pattern &＃61; &＃39; &＃43; p &＃43; &＃39;>.*&＃39; &＃43; p &＃43; &＃39;>&＃39;new_field_and_value &＃61; &＃39; &＃43; p &＃43; &＃39;>&＃39; &＃43; str(value) &＃43; &＃39;&＃39; &＃43; p &＃43; &＃39;>&＃39;search_result &＃61; re.search(pattern, req_body)# 若正则匹配结果存在&＃xff0c;则进行字段值更新if search_result is not None:old_field_and_value &＃61; search_result.group()req_body &＃61; req_body.replace(old_field_and_value, new_field_and_value)else:print(&＃39;参数[%s]匹配标签失败&＃xff0c;请检查报文中标签格式。&＃39; % p)else:# print(&＃39;参数[%s]匹配标签失败&＃xff0c;无此标签。&＃39; % p)continue# 更新数据类型为 xml 的参数值elif self.data_type.upper() &＃61;&＃61; &＃39;JSON&＃39;:# json转换字典if isinstance(self.data, dict):req_body &＃61; self.dataelse:req_body &＃61; json.loads(self.data)# 获取层级node &＃61; p.split(&＃39;.&＃39;)# 调用递归函数更新字段值update_json_step(node, req_body, value)# 非 xml&＃xff0c;json的数据格式&＃xff0c;报错退出else:print(&＃39;ERROR: 不支持此数据类型[%s]&＃39; % self.data_type)breakreturn req_body

接口初始化通用类

# create by: wyun
# create at: 2020/4/23 22:42
from api.comm.data_handle import DataHandle


def intf_base_init(data_type, request_body, **kwargs):
"""
接口初始化通用类
:param data_type: 数据类型&＃xff0c;如&＃xff1a;xml&＃xff0c;json
:param request_body: 请求体
:param kwargs: 可变关键字参数
:return: 请求体
"""
return DataHandle(data_type, request_body, kwargs.keys(), kwargs).update_fields_value()