文档章节

自动化测试接口PYTHON

o
 osc_fmg49rzg
发布于 2019/03/20 11:14
字数 2111
阅读 35
收藏 0

精选30+云产品,助力企业轻松上云!>>>

 

 在开发测试中经常会遇到接口迭代和代码重构,一个无关紧要的改动往往会引起整个项目的运行。现有的接口测试中往往只是针对单一接口的测试,可是业务的连贯性是非常紧密的,比如:用户从登陆,获取商品信息,下单,支付回调的处理,以及退款等流程的处理。发现没有好使的接口测试工具(jmeter太难用了)。于是自己使用python 实现自动化接口测试的一套脚本,该脚本可以实现单一接口的测试,和流程的测试;支持多个项目之间的测试;主要的功能如下:

  • 读取Excel
  • 接口参数(1、地址栏参数,2、body参数)动态化,可以从请求返回值中提取数据作为全局参数,供流程下一步骤使用
  • 根据接口发送请求
  • 结果的诊断,使用jsonpath
  • 结果报告邮件
#!/usr/bin/python
# -*- coding: UTF-8 -*-

import xlrd
import requests
import json
import logging
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
import jsonpath
import sys
import traceback

#日志定义
logging.basicConfig(level=logging.DEBUG,  # log level
                    format='[%(asctime)s] %(levelname)s [%(funcName)s: %(filename)s, %(lineno)d] %(message)s',  # log格式
                    datefmt='%Y-%m-%d %H:%M:%S',  # 日期格式
                    filename='log.txt',  # 日志输出文件
                    filemode='a')  # 追加模式
class Config:
    global PHP_API_DOMAIN #php接口域名
    global JAVA_API_DOMAIN #java接口域名
    global TEST_CASE_PATH  #测试用例excel的路径
    global TEST_CASE_LIST  #测试用例转换后的集合<字典>
    global PARAMS #存储结果变量
    global CURRENT_REQUEST_HEADER #存储当前请求的请求头
    global CURRENT_REQUEST_PARAM #存储当前请求的请求数据
    global ALL_REQUEST_RESULT
    global CURRENT_REQUEST_STATUS

    def __init__(self,php_api_domain,java_api_domain,test_case_path):
        self.PHP_API_DOMAIN = php_api_domain
        self.TEST_CASE_PATH = test_case_path
        self.JAVA_API_DOMAIN = java_api_domain
        self.TEST_CASE_LIST = []
        self.PARAMS = {}
        self.CURRENT_REQUEST_HEADER = {}
        self.CURRENT_REQUEST_PARAM = {}
        self.ALL_REQUEST_RESULT = []
        self.CURRENT_REQUEST_STATUS = True


headers = {'Accept': 'application/json, text/javascript, */*',
               'Accept-Encoding':'gzip, deflate, br',
               'Accept-Language':'zh-CN,zh;q=0.9',
               'Connection':'Keep-Alive',
               'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36 Edge/15.15063'}

class request_handler:
    test_case = {}
    result_obj = {}
    assert_obj = {}
    assert_result = True
    need_assert = True
    def __init__(self, test_case):
        self.test_case = test_case
        self.main()
        self.return_entity()

    def main(self):
        self.request(self.test_case)
        if self.need_assert:
            self.setting_param()

    def return_entity(self):
        return_entity = {}
        return_entity['成功状态'] = self.assert_result
        return_entity['请求结果数据'] = self.result_obj
        return_entity['断言结果'] = self.assert_obj
        return_entity['用例数据'] = self.test_case
        return_entity['请求参数'] = config.CURRENT_REQUEST_PARAM
        return_entity['请求头'] = config.CURRENT_REQUEST_HEADER
        logging.info(return_entity)
        if not self.assert_result:
            config.ALL_REQUEST_RESULT.append(return_entity)
            config.CURRENT_REQUEST_STATUS = False
        else:
            config.CURRENT_REQUEST_STATUS = True

    def request(self,test_case):
        try:
            if test_case['method'] == 'GET':
                self.result_obj = self.request_get(test_case)
            if test_case['method'] == 'POST':
                self.result_obj = self.request_post(test_case)
            if test_case['method'] == 'DELETE':
                self.result_obj = self.request_delete(test_case)
            if test_case['method'] == 'PUT':
                self.result_obj = self.request_put(test_case)
            self.assert_handler()
        except Exception as e:
            self.need_assert = False
            self.assert_result = False
            self.result_obj = {"status_code": "error", "error_message": e}
            traceback.print_exc()
            info = traceback.format_exc()
            print(info)

    def request_get(self,object):
        url = self.get_url(object['url'], object['type'], object['param_data'])
        head = headers
        if isinstance(object['header'],list):
            head = self.get_config_param(object['header'], head)
        request_body = {}
        if isinstance(json.loads(object['body_data']),dict):
            request_body = self.get_config_param(json.loads(object['body_data']), request_body)
        print("GET URL:"+url)
        config.CURRENT_REQUEST_HEADER = head
        config.CURRENT_REQUEST_PARAM = request_body
        response = requests.get(url=url, params=request_body,headers=head)
        return {"status_code" : response.status_code, "response_content": response.text}

    def request_delete(self,object):
        url = self.get_url(object['url'], object['type'], object['param_data'])
        head = headers
        if isinstance(object['header'],list):
            head = self.get_config_param(object['header'], head)
        request_body = {}
        if isinstance(json.loads(object['body_data']),dict):
            request_body = self.get_config_param(json.loads(object['body_data']), request_body)
        print("GET URL:"+url)
        config.CURRENT_REQUEST_HEADER = head
        config.CURRENT_REQUEST_PARAM = request_body
        response = requests.get(url=url, params=request_body,headers=head)
        return {"status_code" : response.status_code, "response_content": response.text}

    def request_put(self,object):
        url = self.get_url(object['url'], object['type'], object['param_data'])
        head = headers
        if isinstance(object['header'],list):
            head = self.get_config_param(object['header'], head)
        request_body = {}
        if isinstance(json.loads(object['body_data']),dict):
            request_body = self.get_config_param(json.loads(object['body_data']), request_body)
        print("GET URL:"+url + '; 参数:' + request_body)
        config.CURRENT_REQUEST_HEADER = head
        config.CURRENT_REQUEST_PARAM = request_body
        response = requests.get(url=url, params=request_body,headers=head)
        return {"status_code" : response.status_code, "response_content": response.text}

    def request_post(self,object):
        head = headers
        #将header中的数据添加到请求中
        if isinstance(json.loads(object['header']),dict):
            head = self.get_config_param(json.loads(object['header']),head)
        url = self.get_url(object['url'], object['type'], object['param_data'])
        #将data中的动态参数从共享参数中添加到数据中
        request_body = {}
        if isinstance(json.loads(object['body_data']),dict):
            request_body = self.get_config_param(json.loads(object['body_data']), request_body)
        print("POST URL:" + url+";param:"+str(request_body))
        config.CURRENT_REQUEST_HEADER = head
        config.CURRENT_REQUEST_PARAM = request_body
        response = requests.post(url, request_body, headers=head)
        return {"status_code": response.status_code, "response_content": response.text}

    def get_config_param(self, list, object):
        for param_name in list.keys():
            param_value = list[param_name]
            if not str(param_value).startswith("$P_"):
                object[param_name] = param_value
            else:
                body_value = config.PARAMS[param_value] if param_value in config.PARAMS.keys() else param_value
                object[param_name] = body_value
        return object

    def get_url(self, path, type, param_data):
        request_url = (config.PHP_API_DOMAIN + '/' + path) if type == 'php' else (config.JAVA_API_DOMAIN + '/' + path)
        param_data = json.loads(param_data)
        url_param = self.get_config_param(param_data, {})
        request_url = request_url + '?a=a'
        for key in url_param.keys():
            request_url = request_url + "&" + key + '=' + url_param[key]
        return request_url

    def assert_handler(self):
        test_case = self.test_case
        result_obj = self.result_obj
        self.assert_obj = {"测试用例名称:": test_case['case_name']}
        # 访问接口直接报错直接
        if result_obj['status_code'] == 'error':
            self.assert_obj = dict(self.assert_obj.items(), {"请求状态:": "错误", "错误信息:": result_obj['error_message']}.items())
            self.need_assert = False
            self.assert_result = False
            return
        # 状态大于300的异常直接处理
        if result_obj['status_code'] >= 300:
            self.assert_obj['请求状态'] = result_obj['status_code']
            self.assert_obj['错误信息'] = result_obj['response_content']
            self.need_assert = False
            self.assert_result = False
            return
        # 请求状态为成功,判断业务状态
        buiess_content = json.loads(result_obj['response_content'])
        expect_res = json.loads(test_case['expect_res'])
        # 校验规则
        for ruler in expect_res:
            matcher = jsonpath.jsonpath(buiess_content, ruler['rule'])
            if isinstance(matcher,bool):
                self.assert_obj['请求状态'] = result_obj['status_code']
                self.assert_obj['错误信息'] = '规则:' + ruler['rule'] + '值不匹配,期望是:' + str(ruler['value']) + ',返回是:' + str(
                    matcher)
                self.assert_result = False
                break
            elif len(matcher) == 0:
                self.assert_obj['请求状态'] = result_obj['status_code']
                self.assert_obj['错误信息'] = '规则:' + ruler['rule'] + '提取出的数组长度为0'
                self.assert_result = False
                break
            elif ruler['type'] == 'value':  # 对值进行比较
                if matcher[0] == ruler['value']:
                    continue
                else:
                    self.assert_obj['请求状态'] = result_obj['status_code']
                    self.assert_obj['错误信息'] = '规则:' + ruler['rule'] + '值不匹配,期望是:' + str(ruler['value']) + ',返回是:' + str(
                        matcher[0])
                    self.assert_result = False
                    break
            elif ruler['type'] == 'length':  # 对数组长度进行比较
                if len(matcher) == ruler['value']:
                    continue
                else:
                    self.assert_obj['请求状态'] = result_obj['status_code']
                    self.assert_obj['错误信息'] = '规则:' + ruler['rule'] + '大小不匹配,期望是:' + str(ruler['value']) + ',返回是:' + str(
                        len(matcher))
                    self.assert_result = False
                    break
            else:
                self.assert_obj['请求状态'] = result_obj['status_code']
                self.assert_obj['错误信息'] = '规则:' + ruler['rule'] + '错误'
                self.assert_result = False
                break

    def setting_param(self):
        actives = json.loads(self.test_case['active'])
        buiess_content = json.loads(self.result_obj['response_content'])
        for active in actives:
            try:
                p_name = active['p_name']
                p_value = jsonpath.jsonpath(buiess_content, active['p_value'])
                if isinstance(p_value, bool):
                    p_value = ''
                elif len(p_value) == 0:
                    p_value = ''
                else:
                    p_value = p_value[0]
                config.PARAMS[p_name] = p_value
            except Exception as e:
                traceback.print_exc()
                info = traceback.format_exc()
                print(info)

def read_excel():
    wb = xlrd.open_workbook(config.TEST_CASE_PATH)
    sh = wb.sheet_by_name('Sheet1')
    #处理 Excel的数据
    for row_ndex in range(sh.nrows):
        test_case = dict(zip(sh.row_values(0), sh.row_values(row_ndex)))
        try:
            if row_ndex == 0:
                continue
            request_handler(test_case)
            if test_case['error_continue'] == 0 and not config.CURRENT_REQUEST_STATUS:
                break

        except Exception as e:
            traceback.print_exc()
            info = traceback.format_exc()
            print(info)
            #config.ALL_REQUEST_RESULT.append({"测试用例": test_case, "请求异常": e})
    #发送邮件通知成功与否
    message  = ''
    for result_record in  config.ALL_REQUEST_RESULT:
        message += json.dumps(result_record,ensure_ascii=False) +'\n\n\n'

    send_email(message)

def send_email(message):
    if len(message) == 0:
        return
    my_sender = 'xxx@xxx.com'  # 发件人邮箱账号
    my_pass = 'xxxxxx'  # 发件人邮箱密码
    my_user = 'xxx@xxx.com'  # 收件人邮箱账号,我这边发送给自己
    ret = True
    try:
        msg = MIMEText(message, 'plain', 'utf-8')
        msg['From'] = formataddr(["小课接口测试报告", my_sender])  # 括号里的对应发件人邮箱昵称、发件人邮箱账号
        msg['To'] = formataddr(["250", my_user])  # 括号里的对应收件人邮箱昵称、收件人邮箱账号
        msg['Subject'] = "小课接口测试报告"  # 邮件的主题,也可以说是标题

        server = smtplib.SMTP_SSL("imap.exmail.qq.com", 465)  # 发件人邮箱中的SMTP服务器,端口是25
        server.login(my_sender, my_pass)  # 括号中对应的是发件人邮箱账号、邮箱密码
        server.sendmail(my_sender, [my_user, ], msg.as_string())  # 括号中对应的是发件人邮箱账号、收件人邮箱账号、发送邮件
        server.quit()  # 关闭连接

    except Exception as e:  # 如果 try 中的语句没有执行,则会执行下面的 ret=False
        traceback.print_exc()
        info = traceback.format_exc()
        print(info)
        ret = False
    if ret:
        print("发送邮件成功")
    else:
        print('发送邮件失败')

#config = Config(r'php域名', r'java域名',r'excel路径.xlsx')
if __name__ == '__main__':
    argv = sys.argv
    if len(sys.argv) != 4:
        print('参数个数不正确')
    php_api = argv[1]
    java_api = argv[2]
    excel_path = argv[3]
    config = Config(php_api, java_api, excel_path)
    read_excel()

  

EXCEL模板案列

case_name header type  url method param_data body_data execp_res memo active error_continue
 获取用户token {}  java  student-service/get-student-token  GET  {"student_id":25009}  {}  [{"rule":"$.code","type":"value","value":0}]  

 [{"p_name":"$P_student_id","p_type":"int",

"p_value":"$.data.student_id"},{"p_name":"$P_token","p_type":"string",

"p_value":"$.data.token"}]

 1
 获取商品信息  {"token":"$P_token"}  php  product/info  POST  {}  {"product_id":3100052}  [{"rule":"$.code","type":"value","value":0}]  

[{"p_name":"$P_product_id",

"p_value":"$.data.id"},{"p_name":"$P_course_id",

"p_value":"$.data.course_id"},{"p_name":"$P_total_money",

"p_value":"$.data.total_money"},{"p_name":"$P_actual_money",

"p_value":"$.data.actual_money"}]

 0
下单 {"token":"$P_token"} php order/create POST {}

{"product_id":"$P_product_id","student_id":"$P_student_id",

"course_id":"$P_course_id",

"total_money":"$P_total_money",

"actual_money":"$P_actual_money","group_buy_order_id":null}

[{"rule":"$.code","type":"value","value":0}]  

[{"p_name":"$P_order_id",

"p_value":"$.data.order_id"}]  

1

 

 

字段说明:

case_name:测试用例名称

type:接口提供者(java/php)

url:请求地址path

method:请求接口类型(POST/GET/PUT/DELETE)

param_data:请求参数(需要放在地址栏上的参数),格式json

body_data:请求接口类型不是GET时,请求的body参数

expect_res:对接口返回数据的断言 ;rule:用jsonpath提取返回字段,type:value/length 标识判断字段值的长度还是值匹配,value:如果type == value,则判断返回字段值和value值是否相同;type==length 判断返回的数组或字符串长度是否相同

  [{"rule":"$.code","type":"value","value":0},{"rule":"$.code","type":"length","value":1}]

active:将返回的数据值保存,作为下一步骤的请求参数,如下含义案例的含义是:存储返回值的json中的data下的student_id的值,使用时用 $P_student_id参数使用

  [{"p_name":"$P_student_id","p_type":"int","p_value":"$.data.student_id"}]

error_continue:当前步骤诊断错误时,是否继续执行后面步骤

案列说明:

上述excel流程是,获取用户token  ---》请求商品参数 ----》从商品中获取参数 ----》创建订单 流程,当然还有取消,支付,退款就不一一演示了

o
粉丝 0
博文 500
码字总数 0
作品 0
私信 提问
加载中
请先登录后再评论。

暂无文章

unity的UGUI之中锚点(Anchors)和中心点(Pivot)、Shift和Alt键功能

在UGUI的控件属性之中,最上方的Rect Transform一栏可以看到锚点和中心点: 锚点Anchors 控件用于定位自身的基准点 可以点击左上角的方框,在其中选择锚点的不同方式: 注意图中,黄色的小点...

路过暴风
51分钟前
7
0
如何将div放置在其容器的底部? - How can I position my div at the bottom of its container?

问题: Given the following HTML: 鉴于以下HTML: <div id="container"> <!-- Other elements here --> <div id="copyright"> Copyright Foo web designs </div> </div> I would like #cop......

富含淀粉
今天
10
0
Distinct()与lambda? - Distinct() with lambda?

问题: Right, so I have an enumerable and wish to get distinct values from it. 是的,所以我有一个可枚举的,并希望从中获得不同的值。 Using System.Linq , there's of course an ext......

法国红酒甜
今天
16
0
学习编写编译器[关闭] - Learning to write a compiler [closed]

问题: Preferred languages : C/C++, Java, and Ruby. 首选语言 :C / C ++,Java和Ruby。 I am looking for some helpful books/tutorials on how to write your own compiler simply for......

技术盛宴
今天
17
0
OSChina 周一乱弹 —— 毛巾又怎么样?!我在乎的是大姐姐温柔的怀抱!

Osc乱弹歌单(2020)请戳(这里) 【今日歌曲】 @薛定谔的兄弟 :分享洛神有语创建的歌单「我喜欢的音乐」: 《雨 因你而下,于你而止》- Seto 手机党少年们想听歌,请使劲儿戳(这里) @Dan...

小小编辑
今天
48
2

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部