接口自动化测试之调用excel实现接口数据依赖-编程知识

背景

我们把接口的信息按照规定的格式都维护在excel文件中，通过代码实现调用excel，完成接口的自动化测试。这样，只需要负责人将主要逻辑写好之后，公司其他不会写代码的员工，也可以通过维护excel中的接口数据，完成整体项目的接口自动化。

本篇文章中可能代码会偏多，原谅我文笔不是很好，不知道怎么优化文字描述。。。直接上代码，哈哈哈。。。在写的过程中部分逻辑判断不是很完整，但是主要流程是可以实现的，大家就将就着看吧。

excel数据

excel文件中的数据为下图所示：

简单说明一下：

（1）id : 测试用例的编号，格式尽量为：case_01；

（2）请求名称：自己写，能看明白就行；

（3）url：接口的路径，也可以加上域名，我没写域名，因为域名在代码中配置了，这样可以在不同环境上运行同一套代码；

（4）是否运行：当前用例是否运行；

（5）请求类型：接口的请求方式；

（6）header：当前接口是否需要header；

（7）依赖id：当前接口需要依赖哪个接口，也就是依赖的这个接口的响应结果中的某个字段是当前接口的请求参数；

（8）依赖接口的返回数据：这个后面没有用到，大家自己看情况；

（9）数据依赖字段：依赖的数据字段名；

（10）请求数据：当前接口的请求数据，页面操作之后可在F12中复制即可；

（11）预期结果：可用来做断言

（12）实际结果：写入接口的响应结果，目前也没用，大家自行写入吧

业务流程

业务名称叫散件入库（这是我自己当时公司项目中的）

该业务流程接口有4个：先新增一个散件入库单（接口1），给这个散件入库单添加需要入库的产品（接口2），此时为待提交的状态，然后提交该散件入库单（接口3），提交之后为待审核的状态，再将该散件入库单审核通过（接口4）。其中每个接口都需要token验证，token在代码中使用登录接口单独提取，全局传入。

代码实现

下面是项目下的目录，方便大家看导入的包都在哪找，目录接口其实也不是特别的合理，大家按照自己项目中的习惯排列就好。

封装的请求函数send_requests.py

import requests'''封装一个接口请求方法'''def send_requests(method, url, data=None, header=None):result = ''if method == "get":res = requests.get(url=url, json=data, headers=header)result = res.json()elif method == "post":res = requests.post(url=url, json=data, headers=header)result = res.json()elif method == "delete":res = requests.delete(url=url, json=data, headers=header)result = res.json()else:print("没有包含当前的请求方式！！！")return result

封装的日志函数logger.py

在测试用例的case上面加上装饰器（@decorate_log）实现就行

import logging, time, os
from functools import wraps
import traceback"""handlers是什么？"""
# logging模块中包含的类
# 用来自定义日志对象的规则（比如：设置日志输出格式、等级等）
# 常用子类：StreamHandler、FileHandler
# StreamHandler 控制台输出日志
# FileHandler 日志输出到文件BASE_PATH = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
print(BASE_PATH)
# 日志文件路径
LOG_PATH = os.path.join(BASE_PATH, "log")
if not os.path.exists(LOG_PATH):os.mkdir(LOG_PATH)class Logger():def __init__(self):# 创建日志路径和时间"""  知识点解析      #time strftime() 函数接收以时间元组，并返回以可读字符串表示的当地时间，格式由参数 format 决定。"""self.logname = os.path.join(LOG_PATH, "{}.log".format(time.strftime("%Y%m%d")))# 创建一个logger日志对象self.logger = logging.getLogger("log")# 设置默认的日志级别self.logger.setLevel(logging.DEBUG)# 创建日志格式对象self.formater = logging.Formatter('[%(asctime)s][%(filename)s %(lineno)d][%(levelname)s]: %(message)s')# 创建FileHandler对象"""mode = 'a'"""# a=append=追加，即 给文件保存写入内容，不是覆盖之前已有文件中内容，而是放在最后，追加到最后。# 一般append追加，适用于 log日志的处理：保留之前log，追加写入新log。self.filelogger = logging.FileHandler(self.logname, mode='a', encoding="UTF-8")# 创建StreamHandler对象self.console = logging.StreamHandler()# FileHandler对象自定义日志级别self.console.setLevel(logging.DEBUG)self.filelogger.setLevel(logging.DEBUG)# 设置两个地方的格式self.filelogger.setFormatter(self.formater)self.console.setFormatter(self.formater)# logger日志对象加载FileHandler对象self.logger.addHandler(self.filelogger)# logger日志对象加载StreamHandler对象self.logger.addHandler(self.console)Logger = Logger().loggerdef decorate_log(func):@wraps(func)def log(*args, **kwargs):Logger.info(f'------开始执行{func.__name__}用例------')try:func(*args, **kwargs)except Exception as e:Logger.error(f'------{func.__name__}用例执行失败，失败原因：{e}------')Logger.error(f"{func.__name__} 用例 is error,here are details:{traceback.format_exc()}")raise eelse:Logger.info(f'------{func.__name__}执行成功------')return log@decorate_log
def funmane_wangxiaoyu():print("************")assert 2 == 2if __name__ == '__main__':funmane_wangxiaoyu()# Logger.info("---测试开始---")# Logger.error("---测试结束---")# Logger.debug("---测试结束---")

封装的工具函数operation_json.py

# include json library
import jsonclass OperationJson:def __init__(self, data_str):self.data_str = data_str# 将传入的str类型转化为json格式def get_data_json(self):json_object = json.loads(self.data_str)return json_objectdef dict_to_str(self):# json.dumps()函数将字典转化为字符串json_info = json.dumps(self.data_str)return json_infoif __name__ == '__main__':b = '{"supplierSn": "1", "sparePartsType": "451776602776477696", "remarks": ""}'a = OperationJson(b).get_data_json()print(type(a))print(a)dict1 = {"age": "12"}c = OperationJson(dict1).dict_to_str()print(c)

封装的工具函数read_yaml.py

import os
import yamlclass HandleYaml:"""读取yaml"""def read_yaml(self, filename):path = os.path.dirname(os.path.dirname(__file__))datapath = path + filenamewith open(datapath, encoding='utf-8') as f:result = yaml.load(f, Loader=yaml.SafeLoader)return resultdo_yaml = HandleYaml()

封装获取常量data_config.py

# 封装获取常量
from common.send_requests import send_requests
from util.read_yaml import do_yamlclass global_var:# case_idId = '0'request_name = '1'url = '2'run = '3'request_way = '4'header = '5'case_dependid = '6'data_depend = '7'field_depend = '8'data = '9'expect = '10'result = '11'# 获取caseid
def get_id():return global_var.Iddef get_request_name():return global_var.request_namedef get_url():return global_var.urldef get_run():return global_var.rundef get_request_way():return global_var.request_waydef get_header():return global_var.headerdef get_case_dependid():return global_var.case_dependiddef get_data_depend():return global_var.data_dependdef get_field_depend():return global_var.field_dependdef get_data():return global_var.datadef get_expect():return global_var.expectdef get_result():return global_var.resultdef get_header_value():# 获取header，主要是需要包含变化的token"""登录接口"""url_data = '\configs\url_data.yaml'url = do_yaml.read_yaml(url_data)["baseurl"] + '/auth/login'case_data_file = '\configs\login_data.yaml'username = do_yaml.read_yaml(case_data_file)['test_login']['login_data']['username']password = do_yaml.read_yaml(case_data_file)['test_login']['login_data']['password']data = {"username": username,"password": password}header1 = {"Content-Type": "application/json;charset=UTF-8"}result = send_requests("post", url, data, header=header1)access_token = result["data"]["access_token"]header = {"Content-Type": "application/json;charset=UTF-8","access-token": access_token}return headerif __name__ == '__main__':a = get_header_value()print(type(a))print(a)

操作excel函数:operation_excel.py

import osimport xlrdclass OperationExcel:def __init__(self, file_name=None, sheet_id=None):if file_name:self.file_name = file_nameself.sheet_id = sheet_idself.data = self.get_data()else:self.file_name = os.path.abspath(os.path.join(os.getcwd(), "..")) + "\data\PartsPurchase.xls"self.sheet_id = 0self.data = self.get_data()# 获取sheets的内容def get_data(self):data = xlrd.open_workbook(self.file_name)tables = data.sheets()[self.sheet_id]return tables# 获取单元格的行数def get_lines(self):tables = self.datareturn tables.nrows# 获取某一个单元格的内容def get_cell_value(self, row, col):return self.data.cell_value(row, col)# 写入excel数据def write_value(self, row, col, value):read_data = xlrd.open_workbook(self.file_name)write_data = copy(read_data)sheet_data = write_data.get_sheet(0)sheet_data.write(row, col, value)write_data.save(self.file_name)# 根据对应的caseid 找到对应行的内容def get_rows_data(self, case_id):row_num = self.get_row_num(case_id)rows_data = self.get_row_values(row_num)return rows_data# 根据对应的caseid 找到对应的行号``````````````````def get_row_num(self, case_id):num = 0clols_data = self.get_cols_data()for col_data in clols_data:if case_id in col_data:return numnum = num + 1# 根据对应行号，找到该行的内容def get_row_values(self, row):tables = self.datarow_data = tables.row_values(row)return row_data# 根据某一个获取某一列的内容def get_cols_data(self, col_id=None):if col_id != None:cols = self.data.col_values(col_id)else:cols = self.data.col_values(0)return colsif __name__ == '__main__':filename = os.path.abspath(os.path.join(os.getcwd(), "..")) + "\data\PartsPurchase.xls"opers = OperationExcel(file_name=filename, sheet_id=0)print(opers.get_cell_value(1, 1))

封装获取数据函数get_data.py

from handle_data import data_config
from handle_data.operation_excel import OperationExcel
from util.operation_json import OperationJsonclass GetData:opera_excel = Nonedef __init__(self, filename, sheetid):self.opera_excel = OperationExcel(file_name=filename, sheet_id=sheetid)# def __init__(self):#     self.opera_excel = OperationExcel(#         file_name=os.path.abspath(os.path.join(os.getcwd(), "..")) + "\data\PartsPurchase.xls", sheet_id=0)# 获取excel行数，就是case的个数def get_case_lines(self):return self.opera_excel.get_lines()# 是否执行def get_is_run(self, row):flag = Nonecol = int(data_config.get_run())run_model = self.opera_excel.get_cell_value(row, col)if run_model == 'yes':flag = Trueelse:flag = Falsereturn flag# 是否携带headerdef is_header(self, row):col = int(data_config.get_header())header = self.opera_excel.get_cell_value(row, col)return header# def header1(self, row):#     col = int(data_config.get_header())#     header = self.opera_excel.get_cell_value(row, col)#     header = data_config.get_header_value()#     return header# def headervalue(self, row):#     col = int(data_config.get_header())#     header = self.opera_excel.get_cell_value(row, col)#     header = data_config.get_aut_value()#     return header# return header, data_config.get_header_value()# def is_header(self,row):#     header = data_config.get_header_value()#     return header#     if header != '':#         return header#     else:#         return None# 获取请求方式def get_request_method(self, row):col = int(data_config.get_request_way())request_method = self.opera_excel.get_cell_value(row, col)return request_method# 获取urldef get_request_url(self, row):col = int(data_config.get_url())url = self.opera_excel.get_cell_value(row, col)return url# 获取请求数据def get_request_data(self, row):col = int(data_config.get_data())data = self.opera_excel.get_cell_value(row, col)if data == '':return Nonereturn data# 通过获取关键字拿到  searchjson data数据def get_data_for_json(self, row):# opera_json = OperationJson() # 源代码# request_data = opera_json.get_data(self.get_request_data(row))opera_json = OperationJson(self.get_request_data(row))request_data = opera_json.get_data_json()if request_data == '':return Nonereturn request_data# 获取预期结果def get_expect_data(self, row):col = int(data_config.get_expect())expect = self.opera_excel.get_cell_value(row, col)if expect == '':return Nonereturn expectdef write_result(self, row, value):col = int(data_config.get_result())self.opera_excel.write_value(row, col, value)# 判断是否有case依赖def is_depend(self, row):col = int(data_config.get_case_dependid())depend_case_id = self.opera_excel.get_cell_value(row, col)if depend_case_id == "":return Noneelse:return depend_case_id# 获取依赖数据的keydef get_depend_key(self, row):col = int(data_config.get_data_depend())depend_key = self.opera_excel.get_cell_value(row, col)if depend_key == "":return Noneelse:return depend_key# 获取数据依赖字段def get_depend_field(self, row):col = int(data_config.get_field_depend())data = self.opera_excel.get_cell_value(row, col)if data == "":return Noneelse:return data

有依赖case的处理逻辑：dependent_data.py

from common.send_requests import send_requests
from util.read_yaml import do_yamlclass DependentDataclass:# 通过依赖的caseid找到所在行row, 把行传进去之后可以获取依赖接口的响应结果def __init__(self, data):self.data = dataself.opera_excel = data.opera_exceldef DependentData(self, depend_case, header_token):print("此接口依赖:" + depend_case)# 通过caseid获取所在的行号，找到该行所对应的值depend_row = self.opera_excel.get_row_num(depend_case)url_data = '\configs\url_data.yaml'url1 = self.data.get_request_url(depend_row)url = do_yaml.read_yaml(url_data)["baseurl"] + url1# print(url)method = self.data.get_request_method(depend_row)request_data = self.data.get_data_for_json(depend_row)header = self.data.is_header(depend_row)# 获取该行是否依赖depend_case = self.data.is_depend(depend_row)# 判断依赖行的case中是否也依赖另外的caseIDif depend_case != None:print("依赖的caseid也依赖其他的接口！暂未写")else:print("该case不依赖其他接口！独立执行！")# 判断依赖行的case中header的取值if header == "yes":header = header_token# print("1header:" + str(header))depend_response_data = send_requests(method=method, url=url, data=request_data, header=header)else:depend_response_data = send_requests(method=method, url=url, data=request_data)return depend_response_data

域名配置url_data.yaml

下面的地址是项目中的地址，外网可以访问，所以我不能透露，抱歉哦，嘻嘻嘻

#预发布环境
#预发布环境测试地址
baseurl: http://yufabu****************
subject: "这是系统预发布环境的接口测试报告"##练习环境
##练习环境地址
#baseurl: http://lianxi****************
#subject: "这是系统练习环境的接口测试报告"

登录参数：login_data.yaml

# 登录参数
test_login:header:"Content-Type": "application/json"login_data:"username": "admin""password": "123456""expected": "登录成功"

单例对象：MyClass.py

# 单例对象
import timefrom configs.Singleton import Singleton@Singleton
class MyClass(object):# 请求头header_token = None# 创建字典，用于存储每个已执行接口的返回对象dict = {}def __init__(self):time.sleep(1)if __name__ == "__main__":cls1 = MyClass()cls2 = MyClass()print(id(cls1) == id(cls2))  # Trueprint(id(cls1))print(id(cls2))

单例类装饰器：Singleton.py

import threading# 双重检查加锁单例类装饰器
class Singleton(object):_instance_lock = threading.Lock()def __init__(self, cls):self._cls = clsself.uniqueInstance = Nonedef __call__(self):if self.uniqueInstance is None:with self._instance_lock:if self.uniqueInstance is None:self.uniqueInstance = self._cls()return self.uniqueInstance

报告的配置conftest.py

from datetime import datetime
from py.xml import html
import pytest
from util.read_yaml import do_yaml@pytest.mark.optionalhook
def pytest_html_results_summary(prefix, summary, postfix):prefix.extend([html.p("测试人: wangxiaoyu")])@pytest.mark.parametrize
def pytest_configure(config):url_data = '\configs\url_data.yaml'report_url = do_yaml.read_yaml(url_data)["baseurl"]report_subject = do_yaml.read_yaml(url_data)["subject"]# config._metadata.pop("JAVA_HOME")  # 删除java_homeconfig._metadata["项目名称"] = report_subject  # 添加项目名称config._metadata["接口地址"] = report_url  # 添加接口地址@pytest.mark.optionalhook
def pytest_html_results_table_header(cells):cells.insert(2, html.th("Description"))  # 表头添加Descriptioncells.insert(3, html.th("Time", class_="sortable time", col="time"))cells.pop(-1)  # 删除link@pytest.mark.optionalhook
def pytest_html_results_table_row(report, cells):cells.insert(2, html.td(report.description))  # 表头对应的内容cells.insert(3, html.td(datetime.now(), class_="col-time"))cells.pop(-1)  # 删除link@pytest.mark.hookwrapper
def pytest_runtest_makereport(item, call):  # Description取值为用例说明__doc__outcome = yieldreport = outcome.get_result()report.description = str(item.function.__doc__)# report.nodeid = report.nodeid.encode("utf-8").decode("unicode_escape")report.nodeid = report.nodeid.encode("unicode_escape").decode("utf-8")

发送电子邮件的配置send_email.py

import smtplib,os
from email.mime.text import MIMETextfrom util.read_yaml import do_yamlcwd_path = os.getcwd()
result_path = os.path.join(cwd_path, "report")
report_path = os.path.join(result_path, "report.html")
# print(report_path)def send_mail():'''发送测试报告邮件'''smtps_server = "smtp.qq.com"port = 465sender = "*******@qq.com"  # 涉及隐私psw = "**w********beaa"  # 涉及隐私receiver = ["*******@qq.com"]  # 涉及隐私f = open(report_path, 'rb')mail_body = f.read().decode('gbk')f.close()url_data = '\configs\url_data.yaml'report_subject = do_yaml.read_yaml(url_data)["subject"]subject = report_subjectmsg = MIMEText(mail_body, 'html', 'utf8')msg['from'] = sendermsg['to'] = ",".join(receiver)msg['subject'] = subjectserver = smtplib.SMTP_SSL(smtps_server, port)server.connect(smtps_server, port)server.login(sender, psw)server.sendmail(sender, receiver, msg.as_string())server.quit()print("测试邮件发送成功")if __name__ == '__main__':send_mail()

测试用例test_PartsPurchaseByexcel.py

import ast
import json
import os
import pytest
import pytest_html
from common.logger import decorate_log
from common.send_requests import send_requests
from configs.MyClass import MyClass
from handle_data.data_config import get_header_value
from handle_data.dependent_data import DependentDataclass
from handle_data.get_data import GetData
from util.read_yaml import do_yaml'''通过调用excel中的数据测试散件入库'''class test_PartsPurchaseByexcel:def __init__(self, header_token, file_name, sheet_id):self.header_token = header_tokenself.data = GetData(file_name, sheet_id)self.DependentDataclass = DependentDataclass(self.data)def sparePartsPurchaseByexcel(self):# data = GetData()rows_count = self.data.get_case_lines()for i in range(1, rows_count):is_run = self.data.get_is_run(i)if is_run:# print("用例执行！")# 获取该行的URL# url = self.data.get_request_url(i)url_data = '\configs\url_data.yaml'url1 = self.data.get_request_url(i)url = do_yaml.read_yaml(url_data)["baseurl"] + url1# 获取该行的请求方式method = self.data.get_request_method(i)# 获取该行的请求数据，json格式request_data = self.data.get_data_for_json(i)# 获取该行预期结果expect = self.data.get_expect_data(i)# 获取该行headerheader = self.data.is_header(i)# 获取该行是否依赖depend_case = self.data.is_depend(i)# 当前执行接口行的caseidcaseid = self.data.opera_excel.get_cell_value(i, 0)# 判断依赖的id是否为空if depend_case != None:print("依赖caseid不为空，继续！")request_data_dict = None# self.depend_data = DependentData(depend_case)# 判断id对应的返回值字典是否存在，存在则表示之前执行过该接口，不再重复执行if depend_case in MyClass().dict:depend_response_data = MyClass().dict[depend_case]else:# 通过依赖的caseid找到所在行row,把行传进去之后可以获取依赖接口的响应结果depend_response_data = self.DependentDataclass.DependentData(depend_case, self.header_token)MyClass().dict[depend_case] = depend_response_data# 获取依赖字段depend_key = self.data.get_depend_field(i)  # 假如依赖的是iddepend_keys = depend_key.split(',')for key in depend_keys:print("当前依赖的字段:", key)depend_value = depend_response_data["data"][key]  # 得到所依赖接口的id对应的值request_data_dict = request_dataif depend_value != None and key in request_data_dict:# 用所依赖接口响应结果中取到的值替换当前接口请求中的依赖字段的值request_data_dict[key] = depend_valuerequest_data_str = json.dumps(request_data_dict)request_data = ast.literal_eval(request_data_str)# request_data2 = json.dumps({depend_key: depend_value})  # 请求数据=依赖的返回数据# request_data = json.dumps(request_data2)if header == "yes":header = self.header_token# print("2header:" + str(header))depend_response_data = send_requests(method=method, url=url, data=request_data, header=header)MyClass().dict[caseid] = depend_response_data# return depend_response_dataelse:print("没有header，这是没有header的执行请求，没写没写没写没写！！！！！！")# return depend_response_dataelse:print("该case不依赖其他接口！独立执行！没写没写没写")else:print("用例不执行！")@decorate_log
def test_sparePartsPurchaseByexcel():''' 测试散件入库 '''cls1: MyClass = MyClass()  # 全局参数单例对象初始化cls1.header_token = get_header_value()header_token = MyClass().header_tokenfile_name = os.path.abspath(os.path.join(os.getcwd(), "..")) + "\wxy_test_chelingzhuByExcel\data\PartsPurchase.xls"print(file_name)sheet_id = 0runn = test_PartsPurchaseByexcel(header_token, file_name, sheet_id)runn.sparePartsPurchaseByexcel()

运行入口run.py

import pytestimport send_emailif __name__ == '__main__':pytest.main(["-sv", "--html=./report/report.html", "--self-contained-html"])# pytest.main()send_email.send_mail()