本文介绍一个接口自动化测试框架。
Python+unittest+requests
实现结果:读取Excel接口测试用例并执行,输出测试报告。
框架脑图如图,各个模块及作用如上。
处理数据库db_funcs用来处理数据库,实现数据库数据的读取操作。(如果没有用到则不需要)
import sqlite3
from config.ProjectConfig import ETConfig
from common.logger import write_log
def execute_db(sql):
"""
连接接口项目sqlite数据库,并执行sql语句
:param sql: sql语句
:return:
"""
try:
# 打开数据库连接
conn = sqlite3.connect("{0}\studentManagementSystem\db.sqlite3".format(ETConfig.PROJECT_DIR))
# 新建游标
cursor = conn.cursor()
# 执行sql
cursor.execute(sql)
# 获取执行结果
result = cursor.fetchall()
# 关闭游标、提交连接、关闭连接
cursor.close()
conn.commit()
conn.close()
return result
except sqlite3.OperationalError as e:
write_log.error("数据库连接,执行失败:{}".format(e))
def init_db():
"""
初始化数据库,删除掉departments的所有数据
:return:
"""
execute_db("delete from departments;")
if __name__ == '__main__':
init_db()
处理Excel
用来对写在Excel中的测试用例进行读取,和测试结果的写回。
from openpyxl import load_workbook
class DoExcel:
"""读写Excel文件"""
def get_data(self,filename,sheetname):
wb=load_workbook(filename)
sheet=wb[sheetname]
test_data=[]
#读取Excel数据存入列表
for i in range(2,sheet.max_row+1):
row_data={}
row_data['case_id'] = sheet.cell(i, 1).value #第一列为case_id
row_data['module']=sheet.cell(i,2).value
row_data['title']=sheet.cell(i,3).value
row_data['headers'] = sheet.cell(i, 4).value
row_data['cookie']=sheet.cell(i,5).value
row_data['method'] = sheet.cell(i, 6).value
row_data['url'] = sheet.cell(i, 7).value
row_data['data'] = sheet.cell(i,8).value
row_data['expected_code'] = sheet.cell(i, 9).value #预期状态码
row_data['actual_code'] = sheet.cell(i, 10).value #实际状态码
row_data['response']=sheet.cell(i,11).value
row_data['auth']=sheet.cell(i,13).value
row_data['file']=sheet.cell(i,14).value
test_data.append((row_data))
return test_data
#写入Excel方法
def write_back(self,filename,sheet_name,i,k,value):
#i为写入的行k为写入的列
wb=load_workbook(filename)
sheet=wb[sheet_name]
sheet.cell(i,k).value=value#写入表格第14列
wb.save(filename)
if __name__ == '__main__':
test_data=DoExcel().get_data("D:/接口实战/接口自动化用例.xlsx",'storm')
# print(test_data)
DoExcel().write_back("D:/接口实战/接口自动化用例.xlsx",'storm',2,5)
封装请求方法
对请求方法进行封装,通过传入的方法,来调用对于的request方法
#封装请求,让代码更简洁,更具有可读性,并且更好维护
import requests
from common.logger import write_log
class HttpReq(object):
"""利用requests封装get请求和post请求需要传递的参数"""
def __init__(self):
self.headers = {"Content-Type": "application/json",
# "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36" 伪装user-agent
}
@staticmethod
def http_request(url,data, http_method, headers=None, cookies=None, auth=None, file=None):
try:
if http_method.upper() == "GET":
res = requests.get(url, data=data, headers=headers, cookies=cookies, auth=auth, files=file)
return res
elif http_method.upper() == "POST":
res = requests.post(url, data=data, headers=headers, cookies=cookies, auth=auth, files=file)
return res
else:
# print("输入的请求方法不对")
write_log.info("输入的请求方法不对")
except Exception as e:
# print("请求报错了:{0}".format(e))
write_log.error("请求报错了:{0}".format(e))
raise e
ETReq = HttpReq()
日志模块
本模块用来生成执行测试用例时产生的日志
import logging
import os
from logging import handlers
from config.ProjectConfig import ETConfig
def logger():
# os.makedirs("{}logs".format(ETConfig.Log_DIR), exist_ok=True)#os.makedirs() 方法用于递归创建目录。
log = logging.getLogger("{}\et.log".format(ETConfig.Log_DIR))
format_str = logging.Formatter('%(asctime)s [%(module)s] %(levelname)s [%(lineno)d] %(message)s', '%Y-%m-%d %H:%M:%S')
# 按天录入日志,最多保存7天的日志
handler = handlers.TimedRotatingFileHandler(filename=("{}/et.log".format(ETConfig.Log_DIR)), when='D', backupCount=7, encoding='utf-8')
log.addHandler(handler)
log.setLevel(logging.INFO)
handler.setFormatter(format_str)
return log
write_log = logger()
write_log.info("你好")
发送邮件
本模块实现发送邮件功能,通过附件将html报告发送给对应的邮箱。
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from config.ProjectConfig import ETConfig
from common.logger import write_log
import smtplib
import time
class sendEmail:
# 发送邮件函数
def send_mail(self):
"""
发送邮件
:return:
"""
# 打开报告文件
# 邮箱信息
self.smtpserver = ETConfig.EMAIL_CONFIG['EMAIL_SERVER']
self.user = ETConfig.EMAIL_CONFIG['EMAIL_USER']
self.password = ETConfig.EMAIL_CONFIG['EMAIL_PWD']
self.sender = ETConfig.EMAIL_CONFIG['EMAIL_SENDER']
self.receiver = ETConfig.EMAIL_CONFIG['EMAIL_RECEIVER']
self.report_dir=ETConfig.Report_DIR
# # html报告路径
# self.report_dir = "{}report".format((ETConfig.Report_DIR))
with open('{}\et_result.html'.format(self.report_dir), 'rb') as f:
mail_body = str(f.read(), encoding="utf-8")
msg = MIMEMultipart('mixed')
msg_html = MIMEText(mail_body, 'html', 'utf-8')
msg_html["Content-Disposition"] = 'attachment; filename="TestReport.html"'
msg.attach(msg_html)
msg_html1 = MIMEText(mail_body, 'html', 'utf-8')
msg.attach(msg_html1)#附件
msg['Subject'] = u'自动化测试报告 {}'.format(time.strftime("%Y-%m-%d", time.localtime()))
msg['From'] = u'AutoTest <%s>' % self.sender
msg['To'] = self.receiver
# msg['Cc'] = self.cc
try:
smtp = smtplib.SMTP()
smtp.connect(self.smtpserver)
smtp.login(self.user, self.password)
smtp.sendmail(self.sender, self.receiver, msg.as_string())
smtp.quit()
write_log.info("发送邮件成功!")
except Exception as e:
write_log.error("发送邮件失败:{}".format(e))
封装配置
ProjectConfig.py用来对项目用到的配置进行进行封装。
import os
class ProjectConfig(object):
#封装配置
version="v1.0"
url="XXX"
# 替换为你本地的接口项目路径(注意不是自动化项目路径)
PROJECT_DIR = "C:\Users\010702\PycharmProjects\easytest\接口环境\"
# 自动化测试项目目录
TEST_DIR = "D:APItest"
Log_DIR="D:\APItest\log"
Report_DIR="D:\APItest\report"
#邮件配置信息
EMAIL_ConFIG={"EMAIL_SERVER":"smtp.qq.com",#服务器
"EMAIL_USER":"XXX",
"EMAIL_PWD":"XXXXXXmdhje",#授权码
"EMAIL_SENDER":"XXX",
"EMAIL_RECEIVER":"XXX"
}
ETConfig=ProjectConfig()
测试用例模块
getUserStorm.py测试用例文件。
#storm项目获取用户信息 DDT+Excel Excel有几条数据,就执行几次用例
import unittest
import requests
import json
from config.ProjectConfig import ETConfig
from testcase.data.DepartmentData import ADD_DATA
from ddt import ddt,data,unpack
from common.wrapers import *
#这里竟然可以只导入一个对象
from common.HttpReq import ETReq
from common.doExcel import DoExcel
from common.HttpReq import HttpReq
from common.is_json import IsJson
from common.logger import write_log
import warnings
@ddt
class GetUserTest(unittest.TestCase):
test_data = DoExcel().get_data("D:/接口实战/接口自动化用例.xlsx", 'getuser')
#只要哪一行有值,就会被读为一条用例
@classmethod
def setUpClass(cls):
write_log.info("------------------")
"""获取用户信息"""
def setUp(self):
warnings.simplefilter('ignore',ResourceWarning)
def tearDown(self):
pass
@write_case_log()
@data(*test_data)
def test_get_user_info(self,item):
print("正在执行测试用例{0}".format(item['title']))
r=HttpReq().http_request(url=item['url'],data=(item['data']),http_method=item['method'])
# print("响应文本为:"+r.text)
try:
self.assertEqual(item['expected_code'], r.json()['code']) # 预期结果和直接返回的状态码比较
TestResult='PASS'
# print("测试用例执行成功")
write_log.info("测试用例执行成功")
except AssertionError as e:
TestResult='FAIL'
# print("执行用例出错(0)".format(e))
write_log.error("执行用例出错{0}".format(e))
raise e
finally:
DoExcel().write_back("D:/接口实战/接口自动化用例.xlsx", 'getuser',item['case_id']+1,10,r.json()['code'])
DoExcel().write_back("D:/接口实战/接口自动化用例.xlsx", 'getuser',item['case_id']+1,11,r.text)
DoExcel().write_back("D:/接口实战/接口自动化用例.xlsx", 'getuser',item['case_id']+1,12,TestResult)#写入结果
# r=IsJson(r)
# self.assertEqual(item['expected'],r['code'])#结果和响应结果中的code边角
if __name__ == '__main__':
# unittest.main()
suite=unittest.TestSuite()
suite.addTest(GetUserTest("test_get_user_info"))
# suite.addTest(AddDepartmentTest("test_add_department_2"))
runner=unittest.TextTestResult()
test_result=runner.run(suite)
执行用例
run.py用来执行用例,并生成测试报告。
import unittest
import platform
import os.path
from common.logger import write_log
from config.ProjectConfig import ETConfig
from common.HTMLTestRunnerCNs import HTMLTestRunner
from common.send_email import sendEmail
class RunCase(object):
report_dir=ETConfig.Report_DIR
#执行用例函数
def run_case(self):
# 运行测试用例并生成html测试报告
with open('{}//et_result.html'.format(self.report_dir), 'wb') as fp:
try:
write_log.info("RunCase执行用例--开始")
suite = unittest.TestSuite()
tests = unittest.defaultTestLoader.discover('..\testcase', pattern='*Storm.py')
suite.addTest(tests)
runner = HTMLTestRunner(stream=fp, title=u'自动化测试报告', description=u'运行环境:{}'.format(platform.platform()),
tester="istester")
runner.run(suite)
write_log.info("RunCase执行用例--结束")
except Exception as e:
write_log.error("RunCase执行用例,生成报告失败:{}".format(e))
if __name__ == '__main__':
test = RunCase()#创建对象
test.run_case()#调用测试用例执行函数
sendEmail().send_mail()
#调用发送邮件函数
测试报告



