Python http接口自动化测试框架实现方法示例

#****************************************************************# TestFrame.py# Author   : Vince# Version  : 1.1.2# Date    : 2011-3-14# Description: 自动化测试平台#****************************************************************import os,sys, urllib, httplib, profile, datetime, timefrom xml2dict import XML2Dictimport win32com.clientfrom win32com.client import Dispatchimport xml.etree.ElementTree as et#import MySQLdb#Excel表格中测试结果底色OK_COLOR=0xffffffNG_COLOR=0xff#NT_COLOR=0xffffNT_COLOR=0xC0C0C0#Excel表格中测试结果汇总显示位置TESTTIME=[1, 14]TESTRESULT=[2, 14]#Excel模版设置#self.titleindex=3    #Excel中测试用例标题行索引#self.casebegin =4    #Excel中测试用例开始行索引#self.argbegin  =3    #Excel中参数开始列索引#self.argcount =8    #Excel中支持的参数个数class create_excel:  def __init__(self, sFile, dtitleindex=3, dcasebegin=4, dargbegin=3, dargcount=8):    self.xlApp = win32com.client.Dispatch('et.Application')  #MS:Excel WPS:et    try:      self.book = self.xlApp.Workbooks.Open(sFile)    except:      print_error_info()      print "打开文件失败"      exit()    self.file=sFile    self.titleindex=dtitleindex    self.casebegin=dcasebegin    self.argbegin=dargbegin    self.argcount=dargcount    self.allresult=[]    self.retCol=self.argbegin+self.argcount    self.xmlCol=self.retCol+1    self.resultCol=self.xmlCol+1  def close(self):    #self.book.Close(SaveChanges=0)    self.book.Save()    self.book.Close()    #self.xlApp.Quit()    del self.xlApp  def read_data(self, iSheet, iRow, iCol):    try:      sht = self.book.Worksheets(iSheet)      sValue=str(sht.Cells(iRow, iCol).Value)    except:      self.close()      print('读取数据失败')      exit()    #去除'.0'    if sValue[-2:]=='.0':      sValue = sValue[0:-2]    return sValue  def write_data(self, iSheet, iRow, iCol, sData, color=OK_COLOR):    try:      sht = self.book.Worksheets(iSheet)      sht.Cells(iRow, iCol).Value = sData.decode("utf-8")      sht.Cells(iRow, iCol).Interior.Color=color      self.book.Save()    except:      self.close()      print('写入数据失败')      exit()  #获取用例个数  def get_ncase(self, iSheet):    try:      return self.get_nrows(iSheet)-self.casebegin+1    except:      self.close()      print('获取Case个数失败')      exit()  def get_nrows(self, iSheet):    try:      sht = self.book.Worksheets(iSheet)      return sht.UsedRange.Rows.Count    except:      self.close()      print('获取nrows失败')      exit()  def get_ncols(self, iSheet):    try:      sht = self.book.Worksheets(iSheet)      return sht.UsedRange.Columns.Count    except:      self.close()      print('获取ncols失败')      exit()  def del_testrecord(self, suiteid):    try:      #为提升性能特别从For循环提取出来      nrows=self.get_nrows(suiteid)+1      ncols=self.get_ncols(suiteid)+1      begincol=self.argbegin+self.argcount      #提升性能      sht = self.book.Worksheets(suiteid)      for row in range(self.casebegin, nrows):        for col in range(begincol, ncols):          str=self.read_data(suiteid, row, col)          #清除实际结果[]          startpos = str.find('[')          if startpos>0:            str = str[0:startpos].strip()            self.write_data(suiteid, row, col, str, OK_COLOR)          else:            #提升性能            sht.Cells(row, col).Interior.Color = OK_COLOR        #清除TestResul列中的测试结果，设置为NT        self.write_data(suiteid, row, self.argbegin+self.argcount+1, ' ', OK_COLOR)        self.write_data(suiteid, row, self.resultCol, 'NT', NT_COLOR)    except:      self.close()      print('清除数据失败')      exit()#执行调用def HTTPInvoke(IPPort, url):  conn = httplib.HTTPConnection(IPPort)  conn.request("GET", url)  rsps = conn.getresponse()  data = rsps.read()  conn.close()  return data#获取用例基本信息[Interface,argcount,[ArgNameList]]def get_caseinfo(Data, SuiteID):  caseinfolist=[]  sInterface=Data.read_data(SuiteID, 1, 2)  argcount=int(Data.read_data(SuiteID, 2, 2))  #获取参数名存入ArgNameList  ArgNameList=[]  for i in range(0, argcount):    ArgNameList.append(Data.read_data(SuiteID, Data.titleindex, Data.argbegin+i))  caseinfolist.append(sInterface)  caseinfolist.append(argcount)  caseinfolist.append(ArgNameList)  return caseinfolist#获取输入def get_input(Data, SuiteID, CaseID, caseinfolist):  sArge=''  #参数组合  for j in range(0, caseinfolist[1]):    if Data.read_data(SuiteID, Data.casebegin+CaseID, Data.argbegin+j) != "None":      sArge=sArge+caseinfolist[2][j]+'='+Data.read_data(SuiteID, Data.casebegin+CaseID, Data.argbegin+j)+'&'  #去掉结尾的&字符  if sArge[-1:]=='&':    sArge = sArge[0:-1]  sInput=caseinfolist[0]+sArge  #组合全部参数  return sInput#结果判断def assert_result(sReal, sExpect):  sReal=str(sReal)  sExpect=str(sExpect)  if sReal==sExpect:    return 'OK'  else:    return 'NG'#将测试结果写入文件def write_result(Data, SuiteId, CaseId, resultcol, *result):  if len(result)>1:    ret='OK'    for i in range(0, len(result)):      if result[i]=='NG':        ret='NG'        break    if ret=='NG':      Data.write_data(SuiteId, Data.casebegin+CaseId, resultcol,ret, NG_COLOR)    else:      Data.write_data(SuiteId, Data.casebegin+CaseId, resultcol,ret, OK_COLOR)    Data.allresult.append(ret)  else:    if result[0]=='NG':      Data.write_data(SuiteId, Data.casebegin+CaseId, resultcol,result[0], NG_COLOR)    elif result[0]=='OK':      Data.write_data(SuiteId, Data.casebegin+CaseId, resultcol,result[0], OK_COLOR)    else: #NT      Data.write_data(SuiteId, Data.casebegin+CaseId, resultcol,result[0], NT_COLOR)    Data.allresult.append(result[0])  #将当前结果立即打印  print 'case'+str(CaseId+1)+':', Data.allresult[-1]#打印测试结果def statisticresult(excelobj):  allresultlist=excelobj.allresult  count=[0, 0, 0]  for i in range(0, len(allresultlist)):    #print 'case'+str(i+1)+':', allresultlist[i]    count=countflag(allresultlist[i],count[0], count[1], count[2])  print 'Statistic result as follow:'  print 'OK:', count[0]  print 'NG:', count[1]  print 'NT:', count[2]#解析XmlString返回Dictdef get_xmlstring_dict(xml_string):  xml = XML2Dict()  return xml.fromstring(xml_string)#解析XmlFile返回Dictdef get_xmlfile_dict(xml_file):  xml = XML2Dict()  return xml.parse(xml_file)#去除历史数据expect[real]def delcomment(excelobj, suiteid, iRow, iCol, str):  startpos = str.find('[')  if startpos>0:    str = str[0:startpos].strip()    excelobj.write_data(suiteid, iRow, iCol, str, OK_COLOR)  return str#检查每个item （非结构体）def check_item(excelobj, suiteid, caseid,real_dict, checklist, begincol):  ret='OK'  for checkid in range(0, len(checklist)):    real=real_dict[checklist[checkid]]['value']    expect=excelobj.read_data(suiteid, excelobj.casebegin+caseid, begincol+checkid)    #如果检查不一致测将实际结果写入expect字段，格式：expect[real]    #将return NG    result=assert_result(real, expect)    if result=='NG':      writestr=expect+'['+real+']'      excelobj.write_data(suiteid, excelobj.casebegin+caseid, begincol+checkid, writestr, NG_COLOR)      ret='NG'  return ret#检查结构体类型def check_struct_item(excelobj, suiteid, caseid,real_struct_dict, structlist, structbegin, structcount):  ret='OK'  if structcount>1: #传入的是List    for structid in range(0, structcount):      structdict=real_struct_dict[structid]      temp=check_item(excelobj, suiteid, caseid,structdict, structlist, structbegin+structid*len(structlist))      if temp=='NG':        ret='NG'  else: #传入的是Dict    temp=check_item(excelobj, suiteid, caseid,real_struct_dict, structlist, structbegin)    if temp=='NG':      ret='NG'  return ret#获取异常函数及行号def print_error_info():  """Return the frame object for the caller's stack frame."""  try:    raise Exception  except:    f = sys.exc_info()[2].tb_frame.f_back  print (f.f_code.co_name, f.f_lineno)#测试结果计数器，类似Switch语句实现def countflag(flag,ok, ng, nt):  calculation = {'OK':lambda:[ok+1, ng, nt],             'NG':lambda:[ok, ng+1, nt],             'NT':lambda:[ok, ng, nt+1]}  return calculation[flag]()

# -*- coding: utf-8 -*-#****************************************************************# xxx_server_case.py# Author   : Vince# Version  : 1.0# Date    : 2011-3-10# Description: 内容服务系统测试代码#****************************************************************from testframe import *from common_lib import *httpString='http://xxx.com/xxx_product/test/'expectXmldir=os.getcwd()+'/TestDir/expect/'realXmldir=os.getcwd()+'/TestDir/real/'def run(interface_name, suiteid):  print '【'+interface_name+'】' + ' Test Begin,please waiting...'  global expectXmldir, realXmldir  #根据接口名分别创建预期结果目录和实际结果目录  expectDir=expectXmldir+interface_name  realDir=realXmldir+interface_name  if os.path.exists(expectDir) == 0:    os.makedirs(expectDir)  if os.path.exists(realDir) == 0:    os.makedirs(realDir)  excelobj.del_testrecord(suiteid) #清除历史测试数据  casecount=excelobj.get_ncase(suiteid) #获取case个数  caseinfolist=get_caseinfo(excelobj, suiteid) #获取Case基本信息  #遍历执行case  for caseid in range(0, casecount):    #检查是否执行该Case    if excelobj.read_data(suiteid,excelobj.casebegin+caseid, 2)=='N':      write_result(excelobj, suiteid, caseid, excelobj.resultCol, 'NT')      continue #当前Case结束，继续执行下一个Case    #获取测试数据    sInput=httpString+get_input(excelobj, suiteid, caseid, caseinfolist)    XmlString=HTTPInvoke(com_ipport, sInput)   #执行调用    #获取返回码并比较    result_code=et.fromstring(XmlString).find("result_code").text    ret1=check_result(excelobj, suiteid, caseid,result_code, excelobj.retCol)    #保存预期结果文件    expectPath=expectDir+'/'+str(caseid+1)+'.xml'    #saveXmlfile(expectPath, XmlString)    #保存实际结果文件    realPath=realDir+'/'+str(caseid+1)+'.xml'    saveXmlfile(realPath, XmlString)    #比较预期结果和实际结果    ret2= check_xmlfile(excelobj, suiteid, caseid,expectPath, realPath)    #写测试结果    write_result(excelobj, suiteid, caseid, excelobj.resultCol, ret1, ret2)  print '【'+interface_name+'】' + ' Test End!'

# -*- coding: utf-8 -*-#****************************************************************# main.py# Author   : Vince# Version  : 1.0# Date    : 2011-3-16# Description: 测试组装，用例执行入口#****************************************************************from testframe import *from xxx_server_case import *import xxx_server_case#产品系统接口测试#设置测试环境xxx_server_case.excelobj=create_excel(os.getcwd()+'/TestDir/xxx_Testcase.xls')xxx_server_case.com_ipport=xxx.com'#Add testsuite beginrun("xxx_book_list", 4)#Add other suite from here#Add testsuite endstatisticresult(xxx_server_case.excelobj)xxx_server_case.excelobj.close()