简单api接口测试框架 - 51Testing软件测试论坛

#!/usr/bin/env python
#-*- coding:utf-8 -*-
import hashlib
import requests
import MySQLdb
from openpyxl import load_workbook
import sys
import time
class FrameWork(object):
def __init__(self):
self.CaseSuccessful = 'TRUE'
self.CaseFalse = 'FALSE'
self.reportPath = '***'
#字符串md5
def toMd5(self,s):
return hashlib.md5(s).hexdigest()
#高亮
def highlight(self,s):
return "%s[30;2m%s%s[1m"%(chr(27), s, chr(27))
#红色字体
def inRed(self,s):
return self.highlight('') + "%s[31;2m%s%s[0m"%(chr(27), s, chr(27))
#绿色字体
def inGreen(self,s):
return self.highlight('') + "%s[32;2m%s%s[0m"%(chr(27), s, chr(27))
#写入测试结果
def report(self,fileName,caseTitle,describe,resultFlag):
with open(self.reportPath+fileName,'a+') as fp:
if resultFlag == 0:
fp.write(caseTitle+':\t'+self.inGreen(self.CaseSuccessful)+'\t'+describe+'\n')
else:
fp.write(caseTitle+':\t'+self.inRed(self.CaseFalse)+'\t'+describe+'\n')
#get请求
def get(self,url,others):
s = requests.Session()
try:
if others.has_key('paramms') and others.has_key('headers'):
r = s.get(url=url,params=others['params'],heads=others['headers'])
elif others.has_key('params'):
r = s.get(url=url,params=others['params'])
elif others.has_key('headers'):
r = s.get(url=url,heads=others['headers'])
else:
r = s.get(url=url)
if r.status_code == requests.codes.ok:
return s,r
except (requests.excepctions.ConnectionError,requests.exceptions.Timeout) as e:
print 'connect error:%s' %(e)
return 'flag',-1
#post上传文件
def post(self,url,others):
s = requests.Session()
try:
if others.has_key('headers') and others.has_key('files'):
r = s.post(url=url,data=others['data'],files=others['files'],headers=others['headers'])
elif others.has_key('files'):
r = s.post(url=url,data=others['data'],files=others['files'])
elif others.has_key('headers'):
r = s.post(url=url,data=others['data'],headers=others['headers'])
else:
r = s.post(url=url,data=others['data'])
if r.status_code == requests.codes.ok:
return s,r
if r.status_code == 500:
return s,500
except (requests.exceptions.ConnectionError,requests.exceptions.Timeout) as e:
print 'connect error:%s' %(e)
return 'flag',-1
#数据库操作
def mysqlConnect(self):
try:
con = MySQLdb.connect(host='***',port=3306,user='***',passwd='***',db='***')
return con
except _mysql_exceptions.OperationalError,e:
print 'connect error:%s' %(e)
return -1
#获取excel内容
def getExcel(self,filename,sheetname):
inwb = load_workbook(filename)
sheet = inwb[sheetname]
return sheet
def executeSql():
pass
def getSql(keys,dic,table):
sql = 'select '
l = dic.keys()
for key in l:
if key in keys:
l.remove(key)
for key in l:
sql = sql + key + ','
return sql + 'id from ' + table + ' order by id desc limit 0,1'
def inParamsChange(self,dic):
keys = dic.keys()
for key in keys:
if dic[key] == None:
dic[key] = ''
elif isinstance(dic[key],unicode):
dic[key] = dic[key].encode('utf-8')
return dic
#流程
def flow(self,method,url,**others):
others = self.inParamsChange(others)
if method == 'get':
s,r = self.get(url,others)
else:
s,r = self.post(url,others)
if r==-1 or others['cursor']==-1:
print 'requests or mysql connect error'
sys.exit()
elif r == 500:
self.report(reportFile,caseTitle,'500 service error',-1)
else:
r = r.json()
print r
if r['code']==others['code'] and r['msg'].encode('utf-8')==others['msg']:
if r['code'] == 0:
p = ['code','msg','caseTitle','reportFile','params','data','headers','files','cursor','table']
sql == getSql(p,others,others['table'])
cursor.execute(sql)
ret = cursor.fetchall()[0][:-1]
flag = True
for index,i in enumerate(ret):
if str(i) != str(others[l[index]]):
print i,others[l[index]],l[index]
flag = False
if flag:
self.report(others['reportFile'],others['caseTitle'],r['msg'].encode('utf-8')+',valid data case test ture',0)
else:
self.report(others['reportFile'],others['caseTitle'],r['msg'].encode('utf-8')+',valid data case test false',-1)
else:
self.report(others['reportFile'],others['Title'],r['msg'].encode('utf-8')+',invalid data case test true',0)
else:
self.report(others['reportFile'],others['caseTitle'],r['msg'].encode('utf-8')+',valid or invalid data case test false',-1)

复制代码