51Testing软件测试论坛

 找回密码
 (注-册)加入51Testing

QQ登录

只需一步,快速开始

测试开发精英班,通向高级软件测试工程师【周活动】 找茬--心里圈的故事 !【长期招募】博为峰网校招聘兼职讲师!横扫BAT,Python全栈测试开发技能大全
【专家101期】:自动化测试可以这样学! 【专题】有远见的测试员已经开始学MySQL了 【干货】各大公司测试大牛职场晋升宝典 自学软件测试那点事
查看: 2783|回复: 3

[原创] 简单api接口测试框架

[复制链接]
  • TA的每日心情
    奋斗
    2016-2-16 14:03
  • 签到天数: 1 天

    连续签到: 1 天

    [LV.1]测试小兵

    发表于 2016-2-16 11:26:33 | 显示全部楼层 |阅读模式
      最近公司开发了一个app项目,需要进行api接口测试,为了让不会写代码的同事也能参与进来,我写了一个简单的接口测试框架,本身这是第一次写,自己的代码经验也不多,想和大家分享下,评论下不足。
      首先这个简易框架有四部分组成:framework公共函数、case测试用例、data数据驱动、report测试报告。整个框架的流程是这样的,用一个循环执行一个api的所有测试用例,测试数据源从excel中读取,执行其中一个case的时候,会调用一个公共的流程,最后将执行结果写入文档。这边重点介绍下公共流程,首先是发送请求,然后对服务器的反馈进行判断,如果是一个有效操作,例如code=0,预期msg和实际一致,那么进行数据库插入数据比对,如果都正确,那么结果正确(当然这边有缺陷,如果是不需要查询数据库的,这个流程就不能用了,需要改进),如果是一个非法操作,那么只判断code和msg,具体代码如下:
    1. #!/usr/bin/env python
    2. #-*- coding:utf-8 -*-

    3. import hashlib
    4. import requests
    5. import MySQLdb
    6. from openpyxl import load_workbook
    7. import sys
    8. import time

    9. class FrameWork(object):
    10.         def __init__(self):
    11.                 self.CaseSuccessful = 'TRUE'
    12.                 self.CaseFalse = 'FALSE'
    13.                 self.reportPath = '***'

    14.         #字符串md5
    15.         def toMd5(self,s):
    16.                 return hashlib.md5(s).hexdigest()

    17.         #高亮
    18.         def highlight(self,s):  
    19.                 return "%s[30;2m%s%s[1m"%(chr(27), s, chr(27))

    20.         #红色字体
    21.         def inRed(self,s):  
    22.                 return self.highlight('') + "%s[31;2m%s%s[0m"%(chr(27), s, chr(27))

    23.         #绿色字体
    24.         def inGreen(self,s):  
    25.                 return self.highlight('') + "%s[32;2m%s%s[0m"%(chr(27), s, chr(27))

    26.         #写入测试结果
    27.         def report(self,fileName,caseTitle,describe,resultFlag):
    28.                 with open(self.reportPath+fileName,'a+') as fp:
    29.                         if resultFlag == 0:
    30.                                 fp.write(caseTitle+':\t'+self.inGreen(self.CaseSuccessful)+'\t'+describe+'\n')
    31.                         else:
    32.                                 fp.write(caseTitle+':\t'+self.inRed(self.CaseFalse)+'\t'+describe+'\n')

    33.         #get请求
    34.         def get(self,url,others):
    35.                 s = requests.Session()
    36.                 try:
    37.                         if others.has_key('paramms') and others.has_key('headers'):
    38.                                 r = s.get(url=url,params=others['params'],heads=others['headers'])
    39.                         elif others.has_key('params'):
    40.                                 r = s.get(url=url,params=others['params'])
    41.                         elif others.has_key('headers'):
    42.                                 r = s.get(url=url,heads=others['headers'])
    43.                         else:
    44.                                 r = s.get(url=url)
    45.                         if r.status_code == requests.codes.ok:
    46.                                 return s,r
    47.                 except (requests.excepctions.ConnectionError,requests.exceptions.Timeout) as e:
    48.                         print 'connect error:%s' %(e)
    49.                         return 'flag',-1

    50.         #post上传文件
    51.         def post(self,url,others):
    52.                 s = requests.Session()
    53.                 try:
    54.                         if others.has_key('headers') and others.has_key('files'):
    55.                                 r = s.post(url=url,data=others['data'],files=others['files'],headers=others['headers'])
    56.                         elif others.has_key('files'):
    57.                                 r = s.post(url=url,data=others['data'],files=others['files'])
    58.                         elif others.has_key('headers'):
    59.                                 r = s.post(url=url,data=others['data'],headers=others['headers'])
    60.                         else:
    61.                                 r = s.post(url=url,data=others['data'])
    62.                         if r.status_code == requests.codes.ok:
    63.                                 return s,r
    64.                         if r.status_code == 500:
    65.                                 return s,500
    66.                 except (requests.exceptions.ConnectionError,requests.exceptions.Timeout) as e:
    67.                         print 'connect error:%s' %(e)
    68.                         return 'flag',-1

    69.         #数据库操作
    70.         def mysqlConnect(self):
    71.                 try:
    72.                         con = MySQLdb.connect(host='***',port=3306,user='***',passwd='***',db='***')
    73.                         return con
    74.                 except _mysql_exceptions.OperationalError,e:
    75.                         print 'connect error:%s' %(e)
    76.                         return -1

    77.         #获取excel内容
    78.         def getExcel(self,filename,sheetname):
    79.                 inwb = load_workbook(filename)
    80.                 sheet = inwb[sheetname]
    81.                 return sheet

    82.         def executeSql():
    83.                 pass
    84.        
    85.         def getSql(keys,dic,table):
    86.                 sql = 'select '
    87.                 l = dic.keys()
    88.                 for key in l:
    89.                         if key in keys:
    90.                                 l.remove(key)
    91.                 for key in l:
    92.                         sql = sql + key + ','
    93.                 return sql + 'id from ' + table + ' order by id desc limit 0,1'

    94.         def inParamsChange(self,dic):
    95.                 keys = dic.keys()
    96.                 for key in keys:
    97.                         if dic[key] == None:
    98.                                 dic[key] = ''
    99.                         elif isinstance(dic[key],unicode):
    100.                                 dic[key] = dic[key].encode('utf-8')
    101.                 return dic

    102.         #流程
    103.         def flow(self,method,url,**others):
    104.                 others = self.inParamsChange(others)
    105.                 if method == 'get':
    106.                         s,r = self.get(url,others)
    107.                 else:
    108.                         s,r = self.post(url,others)
    109.                 if r==-1 or others['cursor']==-1:
    110.                         print 'requests or mysql connect error'
    111.                         sys.exit()
    112.                 elif r == 500:
    113.                         self.report(reportFile,caseTitle,'500 service error',-1)
    114.                 else:
    115.                         r = r.json()
    116.                         print r
    117.                         if r['code']==others['code'] and r['msg'].encode('utf-8')==others['msg']:
    118.                                 if r['code'] == 0:
    119.                                         p = ['code','msg','caseTitle','reportFile','params','data','headers','files','cursor','table']
    120.                                         sql == getSql(p,others,others['table'])
    121.                                         cursor.execute(sql)
    122.                                         ret = cursor.fetchall()[0][:-1]
    123.                                         flag = True
    124.                                         for index,i in enumerate(ret):
    125.                                                 if str(i) != str(others[l[index]]):
    126.                                                         print i,others[l[index]],l[index]
    127.                                                         flag = False
    128.                                         if flag:
    129.                                                 self.report(others['reportFile'],others['caseTitle'],r['msg'].encode('utf-8')+',valid data case test ture',0)
    130.                                         else:
    131.                                                 self.report(others['reportFile'],others['caseTitle'],r['msg'].encode('utf-8')+',valid data case test false',-1)
    132.                                 else:
    133.                                         self.report(others['reportFile'],others['Title'],r['msg'].encode('utf-8')+',invalid data case test true',0)
    134.                         else:
    135.                                 self.report(others['reportFile'],others['caseTitle'],r['msg'].encode('utf-8')+',valid or invalid data case test false',-1)
    复制代码
    以上代码是framework中的公共函数,其中flow就是执行case时的公共函数。具体的case代码实现就不贴了,设计到公司代码,可以贴一下最后的report

    以上就是我测试api接口的一个简单框架,这个框架给不会写代码的测试人员也能进行。

    本帖子中包含更多资源

    您需要 登录 才可以下载或查看,没有帐号?(注-册)加入51Testing

    x
    回复

    使用道具 举报

  • TA的每日心情
    无聊
    2018-5-10 09:16
  • 签到天数: 172 天

    连续签到: 2 天

    [LV.7]测试师长

    发表于 2016-3-22 13:30:31 | 显示全部楼层
    还是不会api接口 平时测试服务器接口
    回复 支持 反对

    使用道具 举报

  • TA的每日心情
    开心
    2017-4-6 15:56
  • 签到天数: 1 天

    连续签到: 1 天

    [LV.1]测试小兵

    发表于 2017-11-16 19:51:14 | 显示全部楼层
    请问这是数据驱动还是关键字驱动框架啊 ? 如果是给不会代码得人用, 那应该是关键字吧 /
    回复 支持 反对

    使用道具 举报

    本版积分规则

    关闭

    站长推荐上一条 /3 下一条

    小黑屋|手机版|Archiver|51Testing软件测试网 ( 沪ICP备05003035号 关于我们

    GMT+8, 2019-4-27 00:16 , Processed in 0.063064 second(s), 24 queries .

    Powered by Discuz! X3.2

    © 2001-2019 Comsenz Inc.

    快速回复 返回顶部 返回列表