API 自動化框架

時間 2019-11-17

標籤 api 自動化框架欄目測試自動化简体版

原文原文鏈接

API 自動化框架

　　我的認爲接口自動化測試使用python語言編寫更加簡單，但全部接口自動化項目代碼的思惟都是同樣的python

1.項目包結構

1.case:存放用例數據的包，將全部用例數據以配置文件形式傳入mysql

2.core:核心包web

　　1）config.py：封裝ConfigParser解析獲取配置文件數據的方法sql

　　　　python的內置模塊ConfigParser：不太瞭解的能夠百度數據庫

　　2）log.py：封裝log的模塊json

　　3）request.py：封裝接口測試的方法api

　　4）mysql.py：封裝鏈接數據庫的方法服務器

3.function：功能包，封裝執行用例的方法，以及生成測試報告的方法cookie

　　生成測試報告的原理：使用file生成.md文件，使用pip install mkdocs安裝mkdocs,本地服務器的話端口默認爲8000app

4.report：存放全部生成的測試報告

5.constant.py：存放全部全局變量的模塊

6.run.py:運行測試用例，以及生成測試報告的模塊

2.核心代碼

　config.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 基礎包：配置服務

import ConfigParser
import core.log as log

config = ConfigParser.ConfigParser()
logging = log.get_logger()

def get_config(filename):
    """
    獲取文件配置
    :param filename: 配置文件名
    :return: None
    """
    global config
    try:
        config.read(filename)
        return True
    except Exception, e:
        logging.error("讀取配置失敗 %s" % e)


def get_data(title, key):
    """
    參數配置
    :param title: 配置文件的頭信息
    :param key: 配置文件的key值
    :return: 配置文件的value
    """
    try:
        value = config.get(title, key)
        type(value)
        return value
    except Exception, e:
        logging.error("獲取配置文件參數失敗 %s" % e)


def get_title_list():
    """
    獲取全部title
    :return: title list
    """
    try:
        title = config.sections()
        return str(title).decode("string_escape")
        # return '\n'.join(title)
    except Exception, e:
        logging.error("獲取title信息失敗 %s", e)

mysql.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# author: zhizhi
# 基礎包: MySQL

import pymysql.cursors
import core.log as log


logging = log.get_logger()
conn = None

def connect(host, user, password, db, charset='utf8'):
    """
    連接Mysql
    :param host: 地址
    :param user: 用戶
    :param password: 密碼
    :param db: 數據庫名
    :param charset: 數據類型
    :return: 連接
    """
    global conn
    if conn == None:
        conn = pymysql.connect(host=host,
                               user=user,
                               password=password,
                               db=db,
                               charset=charset,
                               cursorclass=pymysql.cursors.DictCursor)
    return conn


def execute(sql):
    """
    執行SQL
    :param sql: 執行的SQL
    :return: 影響行數
    """
    global conn
    try:
        with conn.cursor() as cursor:
            res = cursor.execute(sql)
        conn.commit()
        # 這裏必定要寫commit 否則提交的sql 都會被事務回滾
        return res
    except Exception, e:
        logging.error("sql is empty or error %s" % e)


def close():
    """
    關閉MySQL鏈接
    :return: None
    """
    global conn
    conn.close()

request.py

#!/usr/bin/python
#-*- coding: UTF-8 -*-
# 基礎包：接口測試的封裝

import requests
import core.log as log
import json



logging = log.get_logger()

def change_type(value):
    """
    對dict類型進行中文識別
    :param value: 傳的數據值
    :return: 轉碼後的值
    """
    try:
        if isinstance(eval(value), str):
            return value
        if isinstance(eval(value), dict):
            result = eval(json.dumps(value))
            return result
    except Exception, e:
        logging.error("類型問題 %s", e)


def api(method, url, data ,headers):
    """
    自定義一個接口測試的方法
    :param method: 請求類型
    :param url: 地址
    :param data: 數據
    :param headers: 請求頭
    :return: success（str)
    """
    global results
    try:
        if method == ("post" or "POST"):
            results = requests.post(url, data, headers=headers,cookies=cookie)
        if method == ("get" or "GET"):
            results = requests.get(url, data, headers=headers,cookies=cookie)

        response = results.json()
        success = response.get("success")
        return success
    except Exception, e:
        logging.error("service is error", e)

def set_cookieApi(method, url, data ,headers):
    """
    自定義一個登陸接口測試的方法
    :param method: 請求類型
    :param url: 地址
    :param data: 數據
    :param headers: 請求頭
    :return: success（bool)
    """
    global cookie
    try:
        if method == ("post" or "POST"):
            results = requests.post(url, data, headers=headers)
        if method == ("get" or "GET"):
            results = requests.get(url, data, headers=headers)
        response = results.json()
        success = str(response.get("success"))
        cookie = requests.utils.dict_from_cookiejar(results.cookies)
        return success
    except Exception, e:
        logging.info("LoginApi請求失敗", e)

func.py: 能夠經過接口返回數據與本身寫的sql查詢的數據進行對比，判斷用例是否經過，因此提供了mysql通用模塊

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 業務包：通用函數


import core.mysql as mysql
import core.log as log
import core.request as request
import core.config as conf
import constants as cs
import os


logging = log.get_logger()


class ApiTest:
    """接口測試業務類"""

    def __init__(self):
        pass

    def prepare_data(self, host, user, password, db, sql):
        """
        數據準備，添加測試數據
        :param host: 服務地址
        :param user: 用戶
        :param password: 密碼
        :param db: 數據庫名
        :param sql: 執行的SQL
        :return:
        """
        mysql.connect(host, user, password, db)
        res = mysql.execute(sql)
        mysql.close()
        logging.info("Run sql: the row number affected is %s" % res)
        return res

    def get_prepare_sql(self, filename, key):
        """
        獲取預備執行的SQL
        :param title: 配置文件頭信息
        :param key: 配置文件值
        :return: Value
        """
        try:
            conf.get_config(filename)
            value = conf.get_data(title=cs.TITLE, key=key)
            return value
        except Exception, e:
            logging.error("獲取用例參數值失敗 %s" % e)

    def new_report_menu(self, filename):
        """
        這個方法主要是經過寫入文件的方法，先打開cs.YML_REPORT也就是
        mkdocs.yml文件，判斷文件中是否存在當前寫入的內容。
        :param filename: 測試用例文件
        :return: 測試報告內容
        """
        try:
            result = os.path.exists(cs.REPORT_PATH)
            if result == True:
                conf.get_config(filename)
                reportName = eval(conf.get_data(title=cs.REPORT_NAME, key=cs.REPORT))
                report_name = eval(conf.get_data(title=cs.REPORT_NAME, key=cs.R_NAME))
                file = open(cs.YML_REPORT, 'r')
                list_con = file.readlines()
                content = str(list_con).decode("string_escape")
                fileContent = "- %s"
                row = "\n"
                _content = fileContent % (reportName + cs.NOW)
                con = row + _content

                if _content not in content:
                    f = open(cs.YML_REPORT, 'a+')
                    f.write(con)
                else:
                    logging.info("內容已經存在 %s" % _content)
        except Exception, e:
            logging.error("文件路徑不存在 %s", e)

    def write_report(self, content):
        """
        這個方法用於書寫測試報告從而解決以前的經過
        logging方式寫入致使其餘的日誌沒法實現寫入
        :param content: 傳入文件的內容
        :return: None
        """
        reportName = eval(conf.get_data(title=cs.REPORT_NAME, key=cs.REPORT))
        _reportName = reportName + cs.NOW
        filename = cs.REPORT_PATH + _reportName
        try:
            file = open(filename, 'a+')
            file.writelines(content)
        except Exception, e:
            logging.error("文件路徑不存在 %s", e)



    def execute_case(self, filename):
        """
        執行接口測試用例的方法
        :param filename: 用例文件名稱
        :return: 測試結果
        """
        conf.get_config(filename)
        list = eval(conf.get_title_list())

        for i in range(1, len(list)):
            title = list[i]

            number = conf.get_data(title, key=cs.NUMBER)

            name = conf.get_data(title, key=cs.NAME)
            method = conf.get_data(title, key=cs.METHOD)
            url = conf.get_data(title, key=cs.URL)
            data = conf.get_data(title, key=cs.DATA)
            _data = request.json.dumps(data,ensure_ascii=False,indent=4)
            headers = eval(conf.get_data(title, key=cs.HEADERS))
            # headers['Cookie']=cookie
            _headers = request.json.dumps(headers,ensure_ascii=False,indent=4)

            testUrl = cs.DOMAIN + url
            login=cs.LOGIN
            if(title==login):
                actualCode = request.set_cookieApi(method, testUrl, data, headers)
            else:
                actualCode = str(request.api(method, testUrl, data, headers))

            expectCode = str(conf.get_data(title, key=cs.CODE))


            if actualCode != expectCode:
                logging.info("新增一條接口失敗報告")
                self.write_report(cs.API_TEST_FAIL % (name, number, method, testUrl, headers,data, expectCode, actualCode))
            else:
                logging.info("新增一條接口成功報告")
                self.write_report(cs.API_TEST_SUCCESS % (name, number, method, testUrl, headers,data, expectCode, actualCode))

    def run_test(self, filename):
        """
        普通接口測試類方法
        :param filename: 接口的用例name
        :return: 測試報告
        """
        reportName =eval( conf.get_data(title=cs.REPORT_NAME, key=cs.REPORT))

        _filename = cs.REPORT_PATH + reportName + cs.NOW
        try:
            if os.path.exists(_filename):
                os.remove(_filename)
                self.execute_case(filename)
            else:
                self.execute_case(filename)
        except Exception, e:
            logging.error("執行接口測試失敗 %s", e)

    def write_report_result(self):
        """
        這個方法用於書寫測試報告結果
        :return: None
        """
        reportName = eval(conf.get_data(title=cs.REPORT_NAME, key=cs.REPORT))
        _filename = cs.REPORT_PATH + reportName + cs.NOW
        try:
            f = file(_filename)
            content = f.read()
            if content != None:
                _count = content.count("Number")
                _fail = content.count("Case Fail")
                _pass = content.count("Case Pass")
                space = content.split('\n')
                space.insert(0,cs.RESULT_CONTENT % (_count, _pass, _fail))
                _content_ = '\n'.join(space)
                fp = file(_filename,'r+')
                fp.write(_content_)
        except Exception, e:
            logging.error("文件路徑不存在 %s", e)

run.py

# !/usr/bin/python
# -*- coding: UTF-8 -*-
# 執行包：runscript

import function.func as func

ApiTest = func.ApiTest()

FILENAME = 'login.ini'

"""1.新建測試報告目錄"""
ApiTest.new_report_menu(filename=func.cs.CASE_PATH+FILENAME)

"""2.執行測試用例"""
ApiTest.run_test(filename=func.cs.CASE_PATH+FILENAME)

"""3.統計測試報告結果"""
ApiTest.write_report_result()

配置文件：login.ini

[Test Report]
report = 'Enterprise Version'
reportName = 'Regression Testing Report'

[login]
number = 1
name = login
method = post
url = /s1/web/login
data = {data:{'userCode':'admin','password':'admin'}}
headers = {'Content-Type': 'application/json;charset=UTF-8;'}
code = True

[brand]
number = 2
name = get_brand
method = post
url = /s1/brand/getList
data = {"pageSize":10,"pageNum":1}
headers = {'Content-Type': 'application/json; charset=UTF-8;'}
code = True

constant.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 腳本功能：所有變量

import sys
import time
import os
reload(sys)
sys.setdefaultencoding('utf8')

DOMAIN = 'http://****'

REPORT_NAME = 'Test Report'
TITLE = 'All Data prepare the SQL'

METHOD = 'method'
URL = 'url'
DATA = 'data'
NAME = 'name'
NUMBER = 'number'
CODE = 'code'
HEADERS = 'headers'
REPORT = 'report'
R_NAME = 'reportName'
LOGIN='login'

REPORT_PATH = "../api4code/report/docs/"
YML_REPORT = "../api4code/report/mkdocs.yml"

CASE_PATH = "../api4code/case/"


#測試報告內容
API_TEST_FAIL = """
```
%s: Case Fail
 Number: %s
 Method: %s
 Url: %s
 Headers:
 %s
 Data : 
 %s
 Expect : %s
 Actual : %s
```
"""

API_TEST_SUCCESS = """
```
%s: Case Pass
 Number: %s
 Method: %s
 Url: %s
 Headers:
 %s
 Data : 
 %s
 Expect : %s
 Actual : %s
```
"""

#報告結果統計
RESULT_CONTENT = """
<p>Result：</p>
<table border="3" width="500px">
  <tr>
    <th style="color: #787878">All</th>
    <th style="color: #3cc8b4">Pass</th>
    <th style="color: #FFB5C5">Fail</th>
  </tr>
  <tr>
    <th style="color: #787878">%s</th>
    <th style="color: #3cc8b4">%s</th>
    <th style="color: #FFB5C5">%s</th>
  </tr>
</table>
"""

NOW = '_' + time.strftime('%Y%m%d', time.localtime(time.time())) + '.md'
PROJECT_TIME = time.strftime('%Y%m%d', time.localtime(time.time()))

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。