新增所有代码

Author: 居强

.gitignore

@@ -0,0 +1,5 @@
+.idea/**
+.cache/**
+core/.cache/**
+*.pyc
+core/*.pyc

constants.py

@@ -0,0 +1,21 @@
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+# 通用包：常量
+
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+# 脚本功能：全部变量
+
+
+CASE_NUMBER = 0  # 用例编号
+CASE_NAME = 1    # 用例名称
+CASE_DATA = 2    # 用例参数
+CASE_URL = 3     # 用例接口地址
+CASE_METHOD = 4  # 用例请求类型
+CASE_CODE = 5    # 用例code
+CASE_HEADERS = 6 # 用例headers
+
+SQL_ROW = 0      # 预执行SQL的行号
+SQL_COL = 1      # 预执行SQL的列号
+
+FILE_NAME = 'test.xlsx'

core/__init__.py

@@ -0,0 +1,50 @@
+ # !/usr/bin/python
+ # -*- coding: UTF-8 -*-
+ # author: 赫本z
+ # 基础包: excel的封装
+
+import xlrd
+workbook = None
+
+def open_excel(path):
+     """
+     打开excel
+     :param path: 打开excel文件的位置
+     """
+     global workbook
+     if (workbook == None):
+        workbook = xlrd.open_workbook(path, on_demand=True)
+
+def get_sheet(sheetName):
+     """
+     获取页名
+     :param sheetName: 页名
+     :return: workbook
+     """
+     global workbook
+     return workbook.sheet_by_name(sheetName)
+
+def get_rows(sheet):
+    """
+    获取行号
+    :param sheet: sheet
+    :return: 行数
+    """
+    return sheet.nrows
+
+def get_content(sheet, row, col):
+    """
+    获取表格中内容
+    :param sheet: sheet
+    :param row: 行
+    :param col: 列
+    :return:
+    """
+    return sheet.cell(row, col).value
+
+def release(path):
+    """释放excel减少内存"""
+    global workbook
+    workbook.release_resources()
+    del workbook
+    # todo:没有验证是否可用

core/excel.py

@@ -0,0 +1,15 @@
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+# author: 赫本z
+# 基础包: 日志服务
+import logging
+
+def get_logger():
+    global logPath
+    try:
+        logPath
+    except NameError:
+        logPath = ""
+    FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    logging.basicConfig(level=logging.INFO, format=FORMAT)
+    return logging

core/log.py

@@ -0,0 +1,57 @@
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+# author: 赫本z
+# 基础包: 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()

core/mysql.py

@@ -0,0 +1,82 @@
+#!/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: code码
+    """
+    global results
+    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)
+      # if method == "put":
+      #     results = requests.put(url, data, headers=headers)
+      # if method == "delete":
+      #     results = requests.delete(url, headers=headers)
+      # if method == "patch":
+      #     results == requests.patch(url, data, headers=headers)
+      # if method == "options":
+      #     results == requests.options(url, headers=headers)
+        response = results.json()
+        code = response.get("code")
+        return code
+    except Exception, e:
+        logging.error("service is error", e)
+
+
+def content(method, url, data, headers):
+    """
+    请求response自己可以自定义检查结果
+    :param method: 请求类型
+    :param url: 请求地址
+    :param data: 请求参数
+    :param headers: 请求headers
+    :return: message信息
+    """
+    global results
+    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)
+        if method == ("put" or "PUT"):
+            results = requests.put(url, data, headers=headers)
+        if method == ("patch" or "PATCH"):
+            results = requests.patch(url, data, headers=headers)
+        response = results.json()
+        message = response.get("message")
+        result = response.get("result")
+        content = {"message": message, "result": result}
+        return content
+    except Exception, e:
+        logging.error("请求失败 %s" % e)

core/request.py

@@ -0,0 +1,70 @@
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+# 业务包：通用函数
+
+
+import core.mysql as mysql
+import core.log as log
+import core.request as request
+import core.excel as excel
+import constants as cs
+from prettytable import PrettyTable
+
+logging = log.get_logger()
+
+
+class ApiTest:
+    """接口测试业务类"""
+    filename = cs.FILE_NAME
+
+    def __init__(self):
+        pass
+
+    def prepare_data(self, host, user, password, db, sql):
+        """数据准备，添加测试数据"""
+        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_excel_sheet(self, path, module):
+        """依据模块名获取sheet"""
+        excel.open_excel(path)
+        return excel.get_sheet(module)
+
+    def get_prepare_sql(self, sheet):
+        """获取预执行SQL"""
+        return excel.get_content(sheet, cs.SQL_ROW, cs.SQL_COL)
+
+    def run_test(self, sheet, url):
+        """再执行测试用例"""
+        rows = excel.get_rows(sheet)
+        fail = 0
+        for i in range(2, rows):
+            testNumber = str(int(excel.get_content(sheet, i, cs.CASE_NUMBER)))
+            testData = excel.get_content(sheet, i, cs.CASE_DATA)
+            testName = excel.get_content(sheet, i, cs.CASE_NAME)
+            testUrl = excel.get_content(sheet, i, cs.CASE_URL)
+            testUrl = url + testUrl
+            testMethod = excel.get_content(sheet, i, cs.CASE_METHOD)
+            testHeaders = eval(excel.get_content(sheet, i, cs.CASE_HEADERS))
+            testCode = excel.get_content(sheet, i, cs.CASE_CODE)
+            actualCode = request.api(testMethod, testUrl, testData, testHeaders)
+            expectCode = str(int(testCode))
+            failResults = PrettyTable(["Number", "Method", "Url", "Data", "ActualCode", "ExpectCode"])
+            failResults.align["Number"] = "l"
+            failResults.padding_width = 1
+            failResults.add_row([testNumber, testMethod, testUrl, testData, actualCode, expectCode])
+
+            if actualCode != expectCode:
+                logging.info("FailCase %s", testName)
+                print "FailureInfo"
+                print failResults
+                fail += 1
+            else:
+                logging.info("Number %s", testNumber)
+                logging.info("TrueCase %s", testName)
+        if fail > 0:
+            return False
+        return True

function/__init__.py

@@ -0,0 +1,27 @@
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+# 验证包：接口测试脚本
+
+import core.log as log
+from function.func import ApiTest
+
+func = ApiTest()
+logging = log.get_logger()
+
+"""1.外部输入参数"""
+
+module = 'user'
+url = 'http://127.0.0.1:8080'
+
+"""2.根据module获取Sheet"""
+logging.info("-------------- Execute TestCases ---------------")
+sheet = func.get_excel_sheet(func.filename,  module)
+
+# """3.数据准备"""
+# logging.info("-------------- Prepare data through MysqlDB --------------")
+# sql = func.get_prepare_sql(sheet)
+# func.prepare_data(host=host, user=user, password=password, db=db, sql=sql)
+
+"""4.执行测试用例"""
+res = func.run_test(sheet, url)
+logging.info("-------------- Get the result ------------ %s", res)