首页 / PYTHON / python 实现接口自动化
python 实现接口自动化
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了python 实现接口自动化，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含9478字，纯文字阅读大概需要14分钟。
内容图文
目录结构：
log.py
import logging
from logging.handlers import TimedRotatingFileHandler
class Logger(object):
def __init__(self):
# 创建logger对象
self.logger=logging.getLogger()#logging.getLogger(name)函数中的name是日志记录的用例名，不指定name会返回root对象
self.logger.setLevel(logging.DEBUG)
# logging.root.setLevel(logging.NOTSET) #NOTSET会显示所有输出
self.log_file_name="E:/PyCharmWorkSpace/AutoInterfaceTest/log/testlog.log"
self.backup_count=5
self.console_output_level="INFO"
self.file_output_level="DEBUG"
self.formatter=logging.Formatter("%(asctime)s-%(name)s-%(levelname)s-%(message)s")
def get_logger(self):
#handler对象：日志对象用于输出日志，而Handler对象用于指定日志向哪里输出（文件、终端等等）
# 常用handler对象：
# 1.StreamHandler， 用于向标准输入输出流等输出日志 2.FileHandler，用于向文件输出日志
# 3.NullHandler，什么也不输出 4.RotatingFileHandler，向文件输出日志，如果文件到达指定大小，创建新文件并继续输出日志。
# 控制台日志
console_handler = logging.StreamHandler()
console_handler.setFormatter(self.formatter)
console_handler.setLevel(self.console_output_level)
self.logger.addHandler(console_handler) #将控制台日志对象添加到logger
# 文件日志
# 每天重新创建一个日志文件，最多保留backup_count份
file_handler = TimedRotatingFileHandler(filename=self.log_file_name,
when='D',
interval=1,
backupCount=self.backup_count,
delay=True,
encoding='utf-8')
file_handler.setFormatter(self.formatter)
file_handler.setLevel(self.file_output_level)
self.logger.addHandler(file_handler)#将文件日志对象添加到logger
return self.logger
if __name__ == '__main__':
log=Logger().get_logger()
# 打印优先级：critical>error>warning>info>debug
# 当等级为debug时，全部等级都能够打印出来，等级为info时，除了debug，其他都能够打印
log.debug("debug信息，最低级别，一般开发人员用来打印一些调试信息")
log.info("info信息，正常输出信息，一般用来打印一些正常的操作")
log.warning("warning，一般用来打印警信息，默认等级为warning")
log.error("error信息，一般用来打印一些错误信息")
log.critical("critical信息，一般用来打印一些致命的错误信息,等级最高")
logger=Logger().get_logger()
db_util.py
import pymysql.cursors
from util.log import logger
class DBUtil():
def __init__(self):
self.connect=None
self.cursor=None
def get_con(self):
try:
self.connect = pymysql.Connect(
host='localhost',
port=3306,
user='root',
passwd='123456',
db='test',
charset='utf8'
)
except Exception as result:
print(result)
else:
# 获取游标
self.cursor = self.connect.cursor()
def other_data(self,operations,sql):
# 插入数据/修改数据/删除数据
if (operations==1):
try:
self.cursor.execute(sql)
except Exception as e:
self.connect.rollback() # 事务回滚
logger.info('数据库事务处理失败%s'% e)
else:
result=self.cursor.fetchone()
result_str=result.__str__()
result_count=self.cursor.rowcount
logger.info('数据库查询结果：%s', result)
logger.info('数据库成功操作%d条数据'%result_count)
return result_str
elif(operations in (2,3,4)):
try:
self.cursor.execute(sql)
except Exception as e:
self.connect.rollback() # 事务回滚
logger.info('数据库事务处理失败%s' % e)
else:
self.connect.commit() # 事务提交
result_str="数据库处理成功"
result_count = self.cursor.rowcount
logger.info('数据库成功操作%d条数据' % result_count)
return result_str
else:
logger.info("输入数据库操作不正确")
def close_database(self):
# 关闭连接
self.cursor.close()
self.connect.close()
if __name__ == '__main__':
db=DBUtil()
db.get_con()
db.close_database()
主代码：
#coding="utf-8"
import xlrd
from xlutils.copy import copy
import requests
from util.log import logger
from util.db_util import DBUtil
# xlrd：读取Excel文件数据
# xlwt：写入Excel 数据，缺点是无法复用，写入会全部覆盖，无法追加数据，为了方便用户，写入的话，比较推荐xlutils模块，它可以复制原excel
# formatting_info=True，保留Excel的原格式，这样xlutils写入后格式不变
# xlrd模块0.8版本后不支持以xlsx为后缀名文件，所以excel要用xls格式，不能会打不开
if __name__ == '__main__':
db = DBUtil()
db.get_con()
dir_path=r'E:\PyCharmWorkSpace\AutoInterfaceTest\testFile\test_interface005.xls'
run_sheet_name="用例"
summary_sheet_name="总结"
# 统计成功数，失败数
all_cases =0
statistics_success=0
statistics_fail=0
# 前置标志
pre_flag = False
try:
# 目录加r可以取消转义，不加r的话\改为\\即可
workbook=xlrd.open_workbook(dir_path,formatting_info=True)
except FileNotFoundError:
print ("File is not found.") #文件不存在
except PermissionError:
print ( "You don't have permission to access this file.") #文件存在无权限访问，例如文件被打开时无法写入
else:
table = workbook.sheet_by_name(run_sheet_name) # 根据sheet名字获取sheet
new_workbook = copy(workbook) # 复制文件，这样将结果写入excel
writeSheet = new_workbook.get_sheet(run_sheet_name) # 获取写入用例sheet
writeSheet_summary=new_workbook.get_sheet(summary_sheet_name)# 获取写入总结sheet
for i in range(1, table.nrows):
request_method = table.cell(i, 2).value
url = table.cell(i, 3).value
params = table.cell(i, 4).value
expected_results=table.cell(i,5).value
db_operations = table.cell(i, 9).value
sql = table.cell(i, 10).value
pre_request_method=table.cell(i, 12).value
pre_url = table.cell(i, 13).value
pre_params = table.cell(i, 14).value
if( request_method!="" and url!="" and params!="" and expected_results!=""):
logger.info(
"********************************************************************************************************************")
all_cases +=1
if(pre_request_method!="" and pre_url!=""and pre_params!=""):
logger.info("第" + str(i) + "个用例前置请求 url：" + pre_url + " 请求参数：" + pre_params)
if (pre_request_method == "get"):
try:
res = requests.get(url=pre_url, params=pre_params)
except Exception as result:
logger.info("第" + str(i) + "个用例前置请求异常：" + str(result))
else:
logger.info("第" + str(i) + "个用例前置请求结果：" + res.text)
res_json = res.json() # 将返回参数转为json串，取某字段值，result_json[父元素1][子元素2]
pre_response_token = res_json["data"]["token"]
headers = {'Authorization-Qkids': pre_response_token}
pre_flag=True
elif (pre_request_method == "post"):
try:
res = requests.post(url=pre_url, data=pre_params.encode())
except Exception as result:
logger.info("第" + str(i) + "个用例前置请求异常：" + str(result))
else:
logger.info("第" + str(i) + "个用例前置请求结果：" + res.text)
res_json = res.json() # 将返回参数转为json串，取某字段值，result_json[父元素1][子元素2]
pre_response_token = res_json["data"]["token"]
headers = {'Authorization-Qkids':pre_response_token}
pre_flag = True
logger.info(pre_response_token)
else:
logger.info("前置请求方式格式不正确")
headers = {'Authorization-Qkids': pre_response_token}
elif(pre_request_method=="" and pre_url==""and pre_params==""):
# 不做任何操作
pass
else:
logger.info("第" + str(i) + "个用例：前置请求方式/前置url/前置请求参数未填")
# 前置请求就算报错，第二个请求也会有相应错误提示，所以下面这段代码不用放到前置后面，同级即可
logger.info("第" + str(i) + "个用例请求 url：" + url + " 请求参数：" + params)
if (request_method == "get"):
try:
if(pre_flag==True):
res = requests.get(url=url, params=params,headers=headers)
else:
res = requests.get(url=url, params=params)
except Exception as result:
statistics_fail += 1
logger.info("第" + str(i) + "个用例异常：" + str(result))
writeSheet.write(i, 8, "N")
else:
logger.info("第" + str(i) + "个用例结果：" + res.text)
writeSheet.write(i, 6, res.text) # 写入整个返回结果
res_json = res.json() # 将返回参数转为json串，取某字段值，result_json[父元素1][子元素2]
response_message = res_json["message"]
writeSheet.write(i, 7, response_message) # 写入返回的message
if (expected_results == response_message):
logger.info("结果比对：Y")
writeSheet.write(i, 8, "Y")
statistics_success += 1
else:
logger.info("结果比对：N")
writeSheet.write(i, 8, "N")
statistics_fail += 1
elif (request_method == "post"):
try:
if(pre_flag==True):
# 假如请求body里面有汉字，需对data进行encode(),仅用于post请求
res = requests.post(url=url, data=params.encode(),headers=headers)
else:
res = requests.post(url=url, data=params.encode())
except Exception as result:
statistics_fail += 1
logger.info("第" + str(i) + "个用例异常：" + str(result))
writeSheet.write(i, 8, "N")
else:
logger.info("第" + str(i) + "个用例结果：" + res.text)
writeSheet.write(i, 6, res.text) # 写入
res_json = res.json() # 将返回参数转为json串，取某字段值方式:result_json[父元素1][子元素2]，例如res_json[data][name]
response_message = res_json["message"]
writeSheet.write(i, 7, response_message)
if (expected_results == response_message): # 比对预期结果与返回结果
logger.info("结果比对：Y")
writeSheet.write(i, 8, "Y")
statistics_success += 1
else:
logger.info("结果比对：N")
writeSheet.write(i, 8, "N")
statistics_fail += 1
else:
logger.info("请求方式格式不正确")
if (db_operations != "" and sql != ""):
if (db_operations in (1, 2, 3, 4)):
db_result = db.other_data(db_operations, sql)
writeSheet.write(i, 11, db_result)
else:
logger.info("数据库操作填写不符合规则")
elif (db_operations == "" and sql == ""):
pass
else:
logger.info("数据库操作/数据库sql未填")
elif (request_method == "" and url == "" and params == "" and expected_results == ""):
# 不做任何操作
pass
else:
logger.info("第" + str(i) + "个用例：请求方式/url/请求参数/期望结果未填")
# %转义方式：%%，其他使用\
logger.info(
"---------------------------------------------------------------------------------------------------------------------------------------------")
summary_str="总的请求用例数%d, 已通过%d,不通过%d, 通过比例%.2f%%"%(all_cases,statistics_success,statistics_fail,(statistics_success/all_cases)*100)
logger.info(summary_str)
logger.info(
"---------------------------------------------------------------------------------------------------------------------------------------------")
writeSheet_summary.write(0,1,summary_str)
new_workbook.save(dir_path) #最后将写的保存
db.close_database()
class request_case():
pass
接口excel模板
内容总结
以上是互联网集市为您收集整理的python 实现接口自动化全部内容，希望文章能够帮你解决python 实现接口自动化所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】