接口自动化测试之调用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().logger
def 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 e
else:
Logger.info(f'------{func.__name__}执行成功------')
return log
@decorate_log
def funmane_wangxiaoyu():
print("************")
assert 2 == 2
if __name__ == '__main__':
funmane_wangxiaoyu()
# Logger.info("---测试开始---")
# Logger.error("---测试结束---")
# Logger.debug("---测试结束---")
封装的工具函数operation_json.py
# include json library
import json
class 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_object
def dict_to_str(self):
# json.dumps()函数将字典转化为字符串
json_info = json.dumps(self.data_str)
return json_info
if __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 yaml
class HandleYaml:
"""读取yaml"""
def read_yaml(self, filename):
path = os.path.dirname(os.path.dirname(__file__))
datapath = path + filename
with open(datapath, encoding='utf-8') as f:
result = yaml.load(f, Loader=yaml.SafeLoader)
return result
do_yaml = HandleYaml()
封装获取常量data_config.py
# 封装获取常量
from common.send_requests import send_requests
from util.read_yaml import do_yaml
class global_var:
# case_id
Id = '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.Id
def get_request_name():
return global_var.request_name
def get_url():
return global_var.url
def get_run():
return global_var.run
def get_request_way():
return global_var.request_way
def get_header():
return global_var.header
def get_case_dependid():
return global_var.case_dependid
def get_data_depend():
return global_var.data_depend
def get_field_depend():
return global_var.field_depend
def get_data():
return global_var.data
def get_expect():
return global_var.expect
def get_result():
return global_var.result
def 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 header
if __name__ == '__main__':
a = get_header_value()
print(type(a))
print(a)
操作excel函数:operation_excel.py
import os
import xlrd
class OperationExcel:
def __init__(self, file_name=None, sheet_id=None):
if file_name:
self.file_name = file_name
self.sheet_id = sheet_id
self.data = self.get_data()
else:
self.file_name = os.path.abspath(os.path.join(os.getcwd(), "..")) + "\data\PartsPurchase.xls"
self.sheet_id = 0
self.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.data
return 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 = 0
clols_data = self.get_cols_data()
for col_data in clols_data:
if case_id in col_data:
return num
num = num + 1
# 根据对应行号，找到该行的内容
def get_row_values(self, row):
tables = self.data
row_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 cols
if __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 OperationJson
class GetData:
opera_excel = None
def __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 = None
col = int(data_config.get_run())
run_model = self.opera_excel.get_cell_value(row, col)
if run_model == 'yes':
flag = True
else:
flag = False
return flag
# 是否携带header
def 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
# 获取url
def 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 None
return 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 None
return 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 None
return expect
def 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 None
else:
return depend_case_id
# 获取依赖数据的key
def 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 None
else:
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 None
else:
return data
有依赖case的处理逻辑：dependent_data.py
from common.send_requests import send_requests
from util.read_yaml import do_yaml
class DependentDataclass:
# 通过依赖的caseid找到所在行row, 把行传进去之后可以获取依赖接口的响应结果
def __init__(self, data):
self.data = data
self.opera_excel = data.opera_excel
def 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中是否也依赖另外的caseID
if 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 time
from 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)) # True
print(id(cls1))
print(id(cls2))
单例类装饰器：Singleton.py
import threading
# 双重检查加锁单例类装饰器
class Singleton(object):
_instance_lock = threading.Lock()
def __init__(self, cls):
self._cls = cls
self.uniqueInstance = None
def __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_home
config._metadata["项目名称"] = report_subject # 添加项目名称
config._metadata["接口地址"] = report_url # 添加接口地址
@pytest.mark.optionalhook
def pytest_html_results_table_header(cells):
cells.insert(2, html.th("Description")) # 表头添加Description
cells.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 = yield
report = 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 MIMEText
from util.read_yaml import do_yaml
cwd_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 = 465
sender = "*******@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_subject
msg = MIMEText(mail_body, 'html', 'utf8')
msg['from'] = sender
msg['to'] = ",".join(receiver)
msg['subject'] = subject
server = 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_token
self.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)
# 获取该行header
header = self.data.is_header(i)
# 获取该行是否依赖
depend_case = self.data.is_depend(i)
# 当前执行接口行的caseid
caseid = 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) # 假如依赖的是id
depend_keys = depend_key.split(',')
for key in depend_keys:
print("当前依赖的字段:", key)
depend_value = depend_response_data["data"][key] # 得到所依赖接口的id对应的值
request_data_dict = request_data
if depend_value != None and key in request_data_dict:
# 用所依赖接口响应结果中取到的值替换当前接口请求中的依赖字段的值
request_data_dict[key] = depend_value
request_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_data
else:
print("没有header，这是没有header的执行请求，没写没写没写没写！！！！！！")
# return depend_response_data
else:
print("该case不依赖其他接口！独立执行！没写没写没写")
else:
print("用例不执行！")
@decorate_log
def test_sparePartsPurchaseByexcel():
''' 测试散件入库 '''
cls1: MyClass = MyClass() # 全局参数单例对象初始化
cls1.header_token = get_header_value()
header_token = MyClass().header_token
file_name = os.path.abspath(os.path.join(os.getcwd(), "..")) + "\wxy_test_chelingzhuByExcel\data\PartsPurchase.xls"
print(file_name)
sheet_id = 0
runn = test_PartsPurchaseByexcel(header_token, file_name, sheet_id)
runn.sparePartsPurchaseByexcel()
运行入口run.py
import pytest
import send_email
if __name__ == '__main__':
pytest.main(["-sv", "--html=./report/report.html", "--self-contained-html"])
# pytest.main()
send_email.send_mail()
大功告成
运行之后会给你的QQ邮箱发送测试报告，如下图，虽然丑点但是能看明白！！！
行动吧，在路上总比一直观望的要好，未来的你肯定会感 谢现在拼搏的自己！如果想学习提升找不到资料，没人答疑解惑时，请及时加入扣群： 320231853，里面有各种软件测试+开发资料和技术可以一起交流学习哦。
最后感谢每一个认真阅读我文章的人，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走：
这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，希望也能帮助到你！
原文地址：https://blog.csdn.net/2301_78276982/article/details/137782519
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！