本文介绍一个接口自动化测试框架。
Python+unittest+requests
实现结果：读取Excel接口测试用例并执行，输出测试报告。
框架脑图
如图，各个模块及作用如上。
处理数据库
db_funcs用来处理数据库，实现数据库数据的读取操作。（如果没有用到则不需要）
import sqlite3from config.ProjectConfig import ETConfigfrom common.logger import write_logdef 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_workbookclass 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 requestsfrom common.logger import write_logclass 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 eETReq = HttpReq()日志模块
本模块用来生成执行测试用例时产生的日志
import loggingimport osfrom logging import handlersfrom config.ProjectConfig import ETConfigdef 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 logwrite_log = logger()write_log.info("你好")发送邮件
本模块实现发送邮件功能，通过附件将html报告发送给对应的邮箱。
from email.mime.text import MIMETextfrom email.mime.multipart import MIMEMultipartfrom config.ProjectConfig import ETConfigfrom common.logger import write_logimport smtplibimport timeclass 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 osclass 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 unittestimport requestsimport jsonfrom config.ProjectConfig import ETConfigfrom testcase.data.DepartmentData import ADD_DATAfrom ddt import ddt,data,unpackfrom common.wrapers import *#这里竟然可以只导入一个对象from common.HttpReq import ETReqfrom common.doExcel import DoExcelfrom common.HttpReq import HttpReqfrom common.is_json import IsJsonfrom common.logger import write_logimport warnings@ddtclass 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 unittestimport platformimport os.pathfrom common.logger import write_logfrom config.ProjectConfig import ETConfigfrom common.HTMLTestRunnerCNs import HTMLTestRunnerfrom common.send_email import sendEmailclass 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() #调用发送邮件函数测试报告
