接口框架

zx/common/db.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      db.py
+# Author:        gff
+# Datetime:      2023/8/7 18:45
+# Description:
+# ---------------------------------------------------------------------------
+import pymysql
+from APIAotu.day3.zx.common.read_ini import ReadIni
+
+
+class DB:
+    def __init__(self):
+        read_ini = ReadIni()
+        self.conn = pymysql.connect(host=read_ini.get_sql_connect_message("host"),
+                                    port=int(read_ini.get_sql_connect_message("port")),
+                                    user=read_ini.get_sql_connect_message("user"),
+                                    password=read_ini.get_sql_connect_message("pwd"),
+                                    database=read_ini.get_sql_connect_message("database"),
+                                    charset="utf8")
+        self.cursor = self.conn.cursor()
+
+    def close(self):
+        self.cursor.close()
+        self.conn.close()
+
+    def delete(self, sql_sentence):
+        self.cursor.execute(sql_sentence)
+        self.conn.commit()
+
+    def select(self, sql_sentence):
+        self.cursor.execute(sql_sentence)
+        result = self.cursor.fetchall()
+        if result:
+            return result[0][0]
\ No newline at end of file

zx/common/read_excel.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      read_excel.py
+# Author:        gff
+# Datetime:      2023/8/7 19:02
+# Description:
+# ---------------------------------------------------------------------------
+import openpyxl
+from APIAotu.day3.zx.common.read_ini import ReadIni
+from APIAotu.day3.zx.common.read_json import read_json
+from APIAotu.day3.zx.data_config.settings import *
+
+
+class ReadExcel:
+    def __init__(self):
+        self.read_ini = ReadIni()
+        case_data_path = self.read_ini.get_file_path(CASE)
+        expect_data_path = self.read_ini.get_file_path(EXPECT)
+        sql_data_path = self.read_ini.get_file_path(SQL)
+        self.case_data_dict = read_json(case_data_path)
+        self.expect_data_dict = read_json(expect_data_path)
+        self.sql_data_dict = read_json(sql_data_path)
+
+        excel_path = self.read_ini.get_file_path(EXCEL)
+        wb = openpyxl.load_workbook(excel_path)
+        table_name = self.read_ini.get_table_name(TABLE)
+        self.ws = wb[table_name]
+
+    def __get_cell_value(self, column: str, row: int) -> str:
+        value = self.ws[str(column)+str(row)].value
+        if value is None:
+            return None
+        elif value.strip():
+            return value.strip()
+
+    def module_name(self, row):
+        return self.__get_cell_value(MODULE, row)
+
+    def interface_name(self, row):
+        return self.__get_cell_value(API, row)
+
+    def case_title(self, row):
+        return self.__get_cell_value(TITLE, row)
+
+    def case_level(self, row):
+        return self.__get_cell_value(LEVEL, row)
+
+    def case_method(self, row):
+        return self.__get_cell_value(METHOD, row)
+
+    def case_url(self, row):
+        host = self.read_ini.get_host(HOST)
+        path = self.__get_cell_value(PATH, row)
+        return host+path
+
+    def case_mime(self, row):
+        value = self.__get_cell_value(MIME, row)
+        if value:
+            return value.lower()
+
+    def case_data(self, row):
+        case_data_key = self.__get_cell_value(CASEDATA, row)
+        if case_data_key:
+            module_name = self.module_name(row)
+            interface_name = self.interface_name(row)
+            return self.case_data_dict[module_name][interface_name][case_data_key]
+
+    def expect_data(self, row):
+        expect_data_key = self.__get_cell_value(EXPECTDATA, row)
+        if expect_data_key:
+            module_name = self.module_name(row)
+            interface_name = self.interface_name(row)
+            return self.expect_data_dict[module_name][interface_name][expect_data_key]
+
+    def sql_type(self, row):
+        value = self.__get_cell_value(SQLTYPE, row)
+        if value:
+            return value.lower()
+
+    def sql_sentence(self, row):
+        sql_data_key = self.__get_cell_value(SQLSENTENCE, row)
+        if sql_data_key:
+            module_name = self.module_name(row)
+            interface_name = self.interface_name(row)
+            return self.sql_data_dict[module_name][interface_name][sql_data_key]
+
+    def update_key(self, row):
+        return self.__get_cell_value(UPDATEKEY, row)
+
+    def get_data(self):
+        list_data = []
+        for row in range(2, self.ws.max_row+1):
+            module = self.module_name(row)
+            interface = self.interface_name(row)
+            title = self.case_title(row)
+            level = self.case_level(row)
+            method = self.case_method(row)
+            url = self.case_url(row)
+            mime = self.case_mime(row)
+            data = self.case_data(row)
+            expect = self.expect_data(row)
+            sql_type = self.sql_type(row)
+            sql_sentence = self.sql_sentence(row)
+            update_key = self.update_key(row)
+            list_data.append([module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key])
+        else:
+            return list_data
+
+
+if __name__ == '__main__':
+    read_excel = ReadExcel()
+    print(read_excel.get_data())

zx/common/read_ini.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      read_ini.py
+# Author:        gff
+# Datetime:      2023/8/7 18:25
+# Description:
+# ---------------------------------------------------------------------------
+import configparser
+import os
+
+
+class ReadIni:
+    def __init__(self):
+        self.data_config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data_config")
+        ini_path = os.path.join(self.data_config_path, "config.ini")
+        self.conf = configparser.ConfigParser()
+        self.conf.read(ini_path, encoding="utf-8")
+
+    def get_file_path(self, key):
+        return os.path.join(self.data_config_path, self.conf.get("file", key))
+
+    def get_host(self, key):
+        return self.conf.get("host", key)
+
+    def get_table_name(self, key):
+        return self.conf.get("table_name", key)
+
+    def get_sql_connect_message(self, key):
+        return self.conf.get("sql", key)
+
+
+if __name__ == '__main__':
+    read_ini = ReadIni()
+    print(read_ini.get_file_path("excel"))
\ No newline at end of file

zx/common/read_json.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      read_json.py
+# Author:        gff
+# Datetime:      2023/8/7 18:43
+# Description:
+# ---------------------------------------------------------------------------
+import json
+
+
+def read_json(filename):
+    with open(filename, mode="r", encoding="utf-8") as fp:
+        return json.load(fp)
\ No newline at end of file

zx/data_config/case_data.json

+{
+    "认证接口":{
+        "登录系统":{
+            "LoginSuccess": {"password": "MTIzNDU2","username": "admin"},
+            "LoginFailUsernameIsNone":{"password": "MTIzNDU2","username": ""},
+            "LoginFailUsernameIsLong":{"password": "MTIzNDU2","username": "adminadminadminadminadminadminadminadminadmin"},
+            "LoginFailUsernameIsShort":{"password": "MTIzNDU2","username": "a"},
+            "LoginFailUsernameIsSpecialChar":{"password": "MTIzNDU2","username": "❀❁❀❁❀❁❀❁"},
+            "LoginFailUsernameIsError":{"password": "MTIzNDU2","username": "tomcat"},
+            "LoginFailPWDIsNone":{"password": "","username": "admin"},
+            "LoginFailPWDIsLong":{"password": "MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2","username": "admin"},
+            "LoginFailPWDIsShort":{"password": "1","username": "admin"},
+            "LoginFailPWDIsSpecial":{"password": "❀❁❀❁❀❁❀❁","username": "admin"}
+        }
+    },
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": {
+                "code": "hmjz",
+                "description": "黑猫警长",
+                "isDefault": 0,
+                "name": "黑猫警长"
+            }
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"ids": "不知道"}
+        }
+    },
+    "组织管理": {
+        "添加组织": {
+            "AddOrgSuccess": {
+                "code": "djn",
+                "demId": "不知道",
+                "exceedLimitNum": 0,
+                "grade": "",
+                "limitNum": 0,
+                "name": "大角牛",
+                "nowNum": 0,
+                "orderNo": 0,
+                "parentId": "0"
+            }
+        },
+        "删除组织": {
+            "DeleteOrgSuccess": "djn"
+        },
+        "保存组织参数": {
+            "SaveOrgParamSuccess": {
+                "query": {"orgCode": "djn"},
+                "body": [
+                {"alias":"sz","value":0},
+                {"alias":"zy","value":"math"},
+                {"alias":"rq","value":"2023-08-04T16:00:00.000Z"},
+                {"alias":"ah","value":"lq"},
+                {"alias":"yyyy","value":"越高越好"},
+                {"alias":"xb","value":"2"}]
+            }
+        }
+    }
+}
\ No newline at end of file

zx/data_config/config.ini

+[file]
+excel=APIAutoTest.xlsx
+case=case_data.json
+expect=expect_data.json
+sql=sql.json
+
+
+[host]
+test_bpm=http://120.46.172.186:8080
+
+
+[table_name]
+table=BPM接口
+
+
+[sql]
+host=120.46.172.186
+port=3306
+user=root
+pwd=root@2023
+database=eipsaas
\ No newline at end of file

zx/data_config/expect_data.json

+{
+    "认证接口":{
+        "登录系统":{
+            "LoginSuccess": {"username": "超级管理员","account": "admin"},
+            "LoginFailUsernameIsNone":{"state": false,"message": "账号或密码错误"},
+            "LoginFailUsernameIsLong":{"state": false,"message": "账号或密码错误"},
+            "LoginFailUsernameIsShort":{"state": false,"message": "账号或密码错误"},
+            "LoginFailUsernameIsSpecialChar":{"state": false,"message": "账号或密码错误"},
+            "LoginFailUsernameIsError":{"state": false,"message": "账号或密码错误"},
+            "LoginFailPWDIsNone":{"state": false,"message": "账号或密码错误"},
+            "LoginFailPWDIsLong":{"state": false,"message": "账号或密码错误"},
+            "LoginFailPWDIsShort":{"state": false,"message": "账号或密码错误"},
+            "LoginFailPWDIsSpecial":{"state": false,"message": "账号或密码错误"}
+        },
+        "刷新token": {
+            "RefreshSuccess": {"message": "刷新token成功"}
+        }
+    },
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": {"state": true, "message": "添加维度成功！"}
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"state": true, "message": "删除维度成功"}
+        }
+    },
+    "组织管理": {
+        "添加组织": {
+            "AddOrgSuccess": {"state": true, "message": "添加组织成功！"}
+        },
+        "删除组织": {
+            "DeleteOrgSuccess": {"state": true, "message": "删除组织成功！"}
+        },
+        "保存组织参数": {
+            "SaveOrgParamSuccess": {"state": true, "message": "保存组织参数成功！"}
+        }
+    }
+}
+

zx/data_config/settings.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   test_56_env
+# FileName:      column.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 17:27
+# Description:
+# 
+# ---------------------------------------------------------------------------
+"""excel列号的配置文件"""
+
+MODULE = "b"
+API = "C"
+TITLE = "D"
+LEVEL = "E"
+PATH = "F"
+METHOD = "G"
+MIME = "H"
+CASEDATA = "I"
+EXPECTDATA = "J"
+SQLTYPE = "K"
+SQLSENTENCE = "L"
+UPDATEKEY = "M"
+
+
+"""配ini文件中file节点下的key"""
+EXCEL = "excel"
+CASE = "case"
+EXPECT = "expect"
+SQL = "sql"
+
+"""配置table_name节点下的key"""
+TABLE = "table"
+
+"""配置host节点下的key"""
+HOST = "test_bpm"
+
+
+"""配置sql节点下的key"""
\ No newline at end of file

zx/data_config/sql.json

+{
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": "DELETE FROM uc_demension WHERE `CODE_`=\"test56_dem\";"
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": "SELECT ID_ FROM uc_demension WHERE `CODE_`=\"test56_dem\";"
+        }
+    },
+    "组织管理": {
+        "添加组织": {
+            "AddOrgSuccess": {
+                "select": "SELECT ID_ FROM uc_demension WHERE `CODE_`=\"test56_dem\";",
+                "delete": "DELETE FROM uc_org WHERE `CODE_`=\"test56_org\";"
+            }
+        }
+    }
+}
\ No newline at end of file

zx/report/log/日志.log

+ERROR--2023-08-08 10:20:26--断言失败, 模块名称为：认证接口，接口名称为：刷新token，用例标题：正向用例，用例为：None，期望结果为：{'message': '刷新token成功'}，服务器返回数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE2OTE1NDc2MjYsImlhdCI6MTY5MTQ2MTIyNn0.SZPJIQdGB7nmAkv-nu5Bf7GOtZq0ctpe_AiT-7VHBvJBQubR-8M1Hr3L_bYiA9TPx5GLKxolCjpL3I1hbwvw5A","username":"","account":"","userId":"","loginStatus":true,"userAttrs":{}}
+ERROR--2023-08-08 10:20:26--断言失败, 模块名称为：认证接口，接口名称为：刷新token，用例标题：正向用例，用例为：None，期望结果为：{'message': '刷新token成功'}，服务器返回数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE2OTE1NDc2MjYsImlhdCI6MTY5MTQ2MTIyNn0.SZPJIQdGB7nmAkv-nu5Bf7GOtZq0ctpe_AiT-7VHBvJBQubR-8M1Hr3L_bYiA9TPx5GLKxolCjpL3I1hbwvw5A","username":"","account":"","userId":"","loginStatus":true,"userAttrs":{}}
+ERROR--2023-08-08 10:20:27--断言失败, 模块名称为：组织管理，接口名称为：添加组织，用例标题：正向用例，用例为：{'code': 'djn', 'demId': None, 'exceedLimitNum': 0, 'grade': '', 'limitNum': 0, 'name': '大角牛', 'nowNum': 0, 'orderNo': 0, 'parentId': '0'}，期望结果为：{'state': True, 'message': '添加组织成功！'}，服务器返回数据为：{"state":false,"message":"添加组织失败，维度id【demId】不能为空！","value":"保存失败"}
+ERROR--2023-08-08 10:20:27--断言失败, 模块名称为：组织管理，接口名称为：保存组织参数，用例标题：正向用例，用例为：{'query': {'orgCode': 'djn'}, 'body': [{'alias': 'sz', 'value': 0}, {'alias': 'zy', 'value': 'math'}, {'alias': 'rq', 'value': '2023-08-04T16:00:00.000Z'}, {'alias': 'ah', 'value': 'lq'}, {'alias': 'yyyy', 'value': '越高越好'}, {'alias': 'xb', 'value': '2'}]}，期望结果为：{'state': True, 'message': '保存组织参数成功！'}，服务器返回数据为：{"state":false,"message":"保存失败,请将参数填写完整","value":"根据输入的组织编码没有找到对应的组织信息！"}
+ERROR--2023-08-08 10:20:27--断言失败, 模块名称为：组织管理，接口名称为：删除组织，用例标题：正向用例，用例为：djn，期望结果为：{'state': True, 'message': '删除组织成功！'}，服务器返回数据为：{"state":true,"message":"部分删除成功，其中编码为【djn】的组织不存在；","value":""}
+ERROR--2023-08-08 10:20:27--断言失败, 模块名称为：维度管理，接口名称为：根据维度编码删除维度，用例标题：正向用例，用例为：{'ids': None}，期望结果为：{'state': True, 'message': '删除维度成功'}，服务器返回数据为：{"state":false,"message":"删除维度失败，维度ids必填","logId":"1688736902460489728"}

zx/requests_method/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   test_56
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 15:20
+# Description:
+# 
+# ---------------------------------------------------------------------------

zx/requests_method/requests_method.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      requests_method.py
+# Author:        gff
+# Datetime:      2023/8/7 19:49
+# Description:
+# ---------------------------------------------------------------------------
+import base64
+import requests
+from APIAotu.day3.zx.common.read_ini import ReadIni
+
+
+class RequestsMethod:
+    def __init__(self):
+        url = ReadIni().get_host("test_bpm") + "/auth"
+        data = {"username": "admin", "password": base64.b64encode("123456".encode("utf-8")).decode("utf-8")}
+        self.bpm_session = requests.sessions.Session()
+        self.bpm_session.headers["Authorization"] = "Bearer "+self.bpm_session.post(url=url, json=data).json().get("token")
+
+    def requests_all(self, req_method, req_url, req_mime, case_data):
+        if req_mime == "json" or req_mime == "application/json":
+            return self.bpm_session.request(method=req_method, url=req_url, json=case_data)
+        elif req_mime == "x-www-form-urlencoded" or req_mime == "application/x-www-form-urlencoded":
+            return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
+        elif req_mime == "form-data" or req_mime == "multipart/form-data":
+            return self.bpm_session.request(method=req_method, url=req_url, files=case_data)
+        elif req_mime == "query" or req_mime == "params":
+            return self.bpm_session.request(method=req_method, url=req_url, params=case_data)
+        elif req_mime == "application/json|query" or req_mime == "json|query" or req_mime == "query|json":
+            return self.bpm_session.request(method=req_method, url=req_url, params=case_data["query"], json=case_data["body"])
+
+        elif req_mime == "text/plain" or req_mime == "text":
+            return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
+        elif req_mime is None:
+            return self.bpm_session.request(method=req_method, url=req_url)
+        else:
+            raise NameError("传入的媒体类型的名称错误")

zx/test_case/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   test_56
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 15:20
+# Description:
+# 
+# ---------------------------------------------------------------------------

zx/test_case/conftest.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      conftest.py
+# Author:        gff
+# Datetime:      2023/8/7 20:03
+# Description:
+# ---------------------------------------------------------------------------
+import pytest
+from APIAotu.day3.zx.common.db import DB
+from APIAotu.day3.zx.requests_method.requests_method import RequestsMethod
+
+
+@pytest.fixture(scope="session")
+def req_fix():
+    req = RequestsMethod()
+    yield req
+
+
+@pytest.fixture(scope="session")
+def db_fix():
+    db = DB()
+    yield db
+    db.close()
+
+
+def pytest_collection_modifyitems(items):
+    for item in items:
+        item.name = item.name.encode("utf-8").decode("unicode-escape")
+        item._nodeid = item._nodeid.encode("utf-8").decode("unicode-escape")
\ No newline at end of file

zx/test_case/pytest.ini

+[pytest]
+log_cli=true
+log_level=NOTSET
+log_format=%(levelname)s--%(asctime)s--%(message)s
+log_date_format=%Y-%m-%d %H:%M:%S
+log_file=../report/log/日志.log
+log_file_level=error
+log_file_format=%(levelname)s--%(asctime)s--%(message)s
+log_file_date_format=%Y-%m-%d %H:%M:%S
\ No newline at end of file

zx/test_case/test_bpm.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   study_py
+# FileName:      test_bpm.py
+# Author:        gff
+# Datetime:      2023/8/7 20:00
+# Description:
+# ---------------------------------------------------------------------------
+import logging
+import allure
+import pytest
+from APIAotu.day3.zx.common.read_excel import ReadExcel
+
+
+class TestBPM:
+
+    @allure.epic("BPM系统")
+    @pytest.mark.parametrize("module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key", ReadExcel().get_data())
+    def test_bpm(self, req_fix, db_fix, module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key):
+        allure.dynamic.feature(module)
+        allure.dynamic.story(interface)
+        allure.dynamic.title(title)
+        allure.dynamic.severity(level)
+
+        if sql_type == "delete":
+            db_fix.delete(sql_sentence)
+        elif sql_type == "select":
+            select_result = db_fix.select(sql_sentence)
+            data[update_key] = select_result
+        elif sql_type == "delete|select" or sql_type == "select|delete":
+            db_fix.delete(sql_sentence["delete"])
+            select_result = db_fix.select(sql_sentence["select"])
+            data[update_key] = select_result
+
+        result = req_fix.requests_all(req_method=method, req_url=url, req_mime=mime, case_data=data)
+
+        try:
+            for key in expect:
+                assert expect[key] == result.json().get(key)
+        except:
+            print("断言失败")
+            logging.error("断言失败, 模块名称为：{}，接口名称为：{}，用例标题：{}，用例为：{}，期望结果为：{}，服务器返回数据为：{}".format(module, interface, title, str(data), str(expect), result.text))
+            raise AssertionError("断言失败")
+        else:
+            print("断言成功")
+
+
+if __name__ == '__main__':
+    pytest.main()
\ No newline at end of file