第二次提交测试程序

APIAutoTest_V2_1/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      __init__.py
+# Author:       liulei
+# Datetime:     2024/7/15 10:30
+# Description:
+# 
+# ---------------------------------------------------------------------------

APIAutoTest_V2_1/commen/db.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      db.py
+# Author:       liulei
+# Datetime:     2024/7/15 9:04
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import pymysql
+
+from APIAutoTest_V2_1.commen.read_ini import ReadIni
+
+
+class DB:
+    def __init__(self):
+        ini = ReadIni()
+        try:
+            self.conn = pymysql.connect(
+                host=ini.get_database_connect_msg("host"),
+                port=int(ini.get_database_connect_msg("port")),
+                user=ini.get_database_connect_msg("user"),
+                password=ini.get_database_connect_msg("pwd"),
+                database=ini.get_database_connect_msg("database"),
+                charset="utf8"
+            )
+        except Exception as e:
+            print("连接数据库出现错误")
+            raise e
+        else:
+            # 建立游标对象
+            self.cur = self.conn.cursor()
+
+    def close(self):
+        # 关闭游标对象
+        self.cur.close()
+        # 关闭数据库连接
+        self.conn.close()
+
+    def delete(self, sql_sentence):
+        """执行删除的sql语句"""
+        try:
+            self.cur.execute(sql_sentence)
+        except Exception as e:
+            print("执行删除语句时出现错误")
+            raise e
+        else:
+            self.conn.commit()
+
+    def select(self, sql_sentence):
+        """执行查询的sql语句"""
+        try:
+            self.cur.execute(sql_sentence)
+        except Exception as e:
+            print("执行查询的sql语句时出现错误")
+            raise e
+        else:
+            result = self.cur.fetchall()
+            if result:
+                return result[0][0]
\ No newline at end of file

APIAutoTest_V2_1/commen/read_excel.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/13 20:19
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import openpyxl
+
+from APIAutoTest_V2_1.commen.read_ini import ReadIni
+from APIAutoTest_V2_1.commen.read_json import ReadJson
+
+
+class ReadExcel:
+    def __init__(self):
+        # 获取excel文件的路径和工作表
+        self.RI = ReadIni()
+        excel_path = self.RI.get_file("excel")
+        table_name = self.RI.get_table("table")
+        # 创建openpyxl对象
+        wb = openpyxl.load_workbook(excel_path)
+        self.ws = wb[table_name]
+
+    def __get_value_cell(self, column, row):
+        """配置一个对象内部使用的获取单元格值的方法"""
+        Value = self.ws[column + str(row)].value
+        if Value:
+            if Value.strip():
+                return Value.strip()
+            else:
+                return None
+        else:
+            return None
+
+    def get_module_name(self, row):
+        """获取模块名称"""
+        try:
+            module_name = self.__get_value_cell("B", row)
+        except Exception as e:
+            print("获取模块名称时出现错误，错误为：")
+            raise e
+        else:
+            if module_name is None:
+                return None
+            else:
+                return module_name
+
+    def get_API_name(self, row):
+        """获取接口名称"""
+        try:
+            API_name = self.__get_value_cell("C", row)
+        except Exception as e:
+            print("获取接口名称时出现错误，错误为：")
+            raise e
+        else:
+            if API_name is None:
+                return None
+            else:
+                return API_name
+
+    def get_case_title(self, row):
+        """获取用例标题"""
+        try:
+            case_title = self.__get_value_cell("D", row)
+        except Exception as e:
+            print("获取用例标题时出现错误，错误为：")
+            raise e
+        else:
+            return case_title
+
+    def get_request_method(self, row):
+        """获取请求方法"""
+        try:
+            request_method = self.__get_value_cell("F", row)
+        except Exception as e:
+            print("获取请求方法时出现错误，错误为：")
+            raise e
+        else:
+            return request_method
+
+    def get_request_path(self, row):
+        """获取请求路径"""
+        try:
+            path = self.__get_value_cell("G", row)
+        except Exception as e:
+            print("获取请求路径时出现错误，错误为：")
+            raise e
+        else:
+            try:
+                host = self.RI.get_url("test_host")
+            except Exception as E:
+                print("在获取请求路径时，读取被测系统域名时出现错误，错误为：")
+                raise E
+            else:
+                if path is None:
+                    return host
+                else:
+                    return host + path
+
+    def get_mime(self, row):
+        """获取媒体类型"""
+        try:
+            mime = self.__get_value_cell("H", row)
+        except Exception as e:
+            print("在获取媒体类型时出现错误，错误为：")
+            raise e
+        else:
+            return mime
+
+    def get_case_data(self, row):
+        """获取用例数据"""
+        try:
+            case_data = self.__get_value_cell("I", row)
+        except Exception as e:
+            print("在excel中读取用例数据时出现错误，错误为：")
+            raise e
+        else:
+                if self.get_module_name(row) is None or self.get_API_name(row) is None or case_data is None:
+                    return None
+                else:
+                    try:
+                        json_data = ReadJson("case")[self.get_module_name(row)][self.get_API_name(row)][case_data]
+                    except KeyError as k:
+                        print("读取该条json数据时出现错误，case_data为：", case_data)
+                        print("获取json中的用例数据时，找不到对应的数据，错误为：")
+                        raise k
+                    else:
+                        return json_data
+
+    def get_expect_data(self, row):
+        """获取期望数据"""
+        try:
+            expect_data = self.__get_value_cell("J", row)
+        except Exception as e:
+            print("在excel中读取期望数据时出现错误，错误为：")
+            raise e
+        else:
+            try:
+                if self.get_module_name(row) is None or self.get_API_name(row) is None:
+                    return None
+                else:
+                    json_data = ReadJson("expect")[self.get_module_name(row)][self.get_API_name(row)][expect_data]
+            except Exception as e:
+                print("根据excel中的用例数据去读取json文件中的数据时出现错误，错误为：")
+                raise e
+            else:
+                return json_data
+
+    def get_sql_type(self, row):
+        """根据key获取sql语句类型"""
+        sql_type = self.__get_value_cell("K", row)
+        if sql_type:
+            print("读取excel中获取到的语句类型为：", sql_type)
+            return sql_type
+
+    def get_sql_data(self, row):
+        """根据key读取sql语句"""
+        sql_data_key = self.__get_value_cell("L", row)
+        if sql_data_key:
+            module_name = self.get_module_name(row)
+            API_name = self.get_API_name(row)
+            sql_data = ReadJson("sql")[module_name][API_name][sql_data_key]
+            return sql_data
+
+    def get_updatekey(self, row):
+        """获取更新的key"""
+        update_key = self.__get_value_cell("M", row)
+        if update_key:
+            return update_key
+
+    def get_data_all(self):
+        """将用例数据装进一个二维列表里面"""
+        data_list = []
+        try:
+            for row in range(2, self.ws.max_row+1):
+                list1 = [self.get_request_method(row), self.get_request_path(row), self.get_mime(row), self.get_case_data(row), self.get_expect_data(row), self.get_sql_type(row), self.get_sql_data(row), self.get_updatekey(row)]
+                if self.get_module_name(row) is not None and self.get_API_name(row) is not None:
+                    data_list.append(list1)
+                else:
+                    continue
+        except Exception as e:
+            print("获取所有用例数据时出现错误，错误为：")
+            raise e
+        else:
+            return data_list
+
+if __name__ == "__main__":
+    RE = ReadExcel()
+
+    list2 = RE.get_data_all()
+    for i in range(len(list2)):
+        print(i+1, end=' ')
+        print(list2[i])
+        print("="*140)
+    #
+    # s1 = "    "
+    # print(s1.strip())
+    # if s1.strip():
+    #     print("hello")
+
+

APIAutoTest_V2_1/commen/read_ini.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/13 20:19
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import configparser
+import os
+
+
+class ReadIni:
+    def __init__(self):
+        # 获取配置文件的路径
+        self.config_path = os.path.dirname(os.path.dirname(__file__)) + r"\data_config\\"
+        # 创建configparser对象
+        self.conf = configparser.ConfigParser()
+        self.conf.read(self.config_path + r"\config.ini", encoding='utf-8')
+
+    def get_file(self, key):
+        """根据key获取文件的路径"""
+        try:
+            path = self.config_path + self.conf.get("file", key)
+        except Exception as e:
+            print("获取文件的路径时出现错误，错误为：")
+            raise e
+        else:
+            return path
+
+    def get_url(self, key):
+        """根据key获取被测系统的域名"""
+        try:
+            host = self.conf.get("host", key)
+        except Exception as e:
+            print("获取被测系统域名时出现错误，错误为：")
+            raise e
+        else:
+            return host
+
+    def get_table(self, key):
+        """根据key获取excel的工作表"""
+        try:
+            table = self.conf.get("table_name", key)
+        except Exception as e:
+            print("获取工作表名称时出现错误，错误为：")
+            raise e
+        else:
+            return table
+
+    def get_database_connect_msg(self, key):
+        """根据key获取数据库的连接信息"""
+        try:
+            msg = self.conf.get("sql", key)
+        except Exception as e:
+            print("获取数据库连接信息时出现错误，错误为：")
+            raise e
+        else:
+            return msg
+
+if __name__ == "__main__":
+    RI = ReadIni()
+
+

APIAutoTest_V2_1/commen/read_json.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/13 20:19
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import json
+
+from APIAutoTest_V2_1.commen.read_ini import ReadIni
+
+
+def ReadJson(key):
+    """获取json文件的路径"""
+    read_ini = ReadIni()
+    json_path = read_ini.get_file(key)
+    # 根据传入的key决定取哪个json文件的数据
+    try:
+        with open(json_path, 'r', encoding='utf-8') as f:
+            json_data = json.loads(f.read())
+    except Exception as e:
+        print("获取json文件数据时出现错误，错误为：")
+        raise e
+    else:
+        return json_data
+
+

APIAutoTest_V2_1/data_config/case_data.json

+{
+    "认证接口":{
+        "登录系统":{
+            "LoginSuccess":{"username":"admin","password":"F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="},
+            "LoginFailUsernameIsNone":{"username":"","password":"F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="},
+            "LoginFailUsernameIsShort":{"username":"a","password":"F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="},
+            "LoginFailUsernameIsLong":{"username":"adminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadmin","password":"F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="},
+            "LoginFailUsernameIsSpecial":{"username":"♡♣♤♥♦♧♨♩ε","password":"F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="},
+            "LoginFailPwdIsNone":{"username":"admin","password":""},
+            "LoginFailPwdIsShort":{"username":"admin","password":"1"},
+            "LoginFailPwdIsLong":{"username":"admin","password":"F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8=F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="}
+        }
+    },
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": {"code": "test_dem_xyz_123","description": "测试维度","isDefault": 0, "name": "xyz测试维度"}
+        },
+        "设置默认维度": {
+            "SetDefaultDemSuccess": {"code": "test_dem_xyz_123"}
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"ids": "需要更新"}
+        }
+    }
+}
\ No newline at end of file

APIAutoTest_V2_1/data_config/config.ini

+
+[节点名称]
+键=值
+
+
+# 配置数据层中文件的节点
+[file]
+# 用例管理文件
+excel=APIAutoTest.xlsx
+# 用例数据文件
+case=case_data.json
+# 期望数据文件
+expect=expect.json
+# sql语句文件
+sql=sql_data.json
+
+
+# 配置被测系统的域名
+[host]
+test_host=http://36.139.193.99:8088
+
+
+# 配置excel中工作表的名称
+[table_name]
+table=BPM
+
+
+[sql]
+host=36.139.193.99
+port=3306
+user=root
+pwd=Rhrc@2024
+database=eip8
\ No newline at end of file

APIAutoTest_V2_1/data_config/expect.json

+{
+    "认证接口":{
+        "登录系统":{
+            "LoginSuccess": {"username":"超级管理员","account":"admin","userId":"1","expiration":86400},
+            "LoginFailUsernameIsNone":{"state":false,"message":"账户错误或该租户未启用"},
+            "LoginFailUsernameIsShort":{"state":false,"message":"账户错误或该租户未启用"},
+            "LoginFailUsernameIsLong":{"state":false,"message":"账户错误或该租户未启用"},
+            "LoginFailUsernameIsSpecial":{"state":false,"message":"账户错误或该租户未启用"},
+            "LoginFailPwdIsNone":{"state":false,"message":"账户错误或该租户未启用"},
+            "LoginFailPwdIsShort":{"state":false,"message":"账户错误或该租户未启用"},
+            "LoginFailPwdIsLong":{"state":false,"message":"账户错误或该租户未启用"}
+        },
+        "刷新token": {
+            "RefreshSuccess": {"message": "刷新成功"}
+        }
+    },
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": {"message": "添加维度成功！"}
+        },
+        "设置默认维度": {
+            "SetDefaultDemSuccess": {"message": "设置默认维度成功！"}
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"message": "删除维度成功"}
+        }
+    }
+}
\ No newline at end of file

APIAutoTest_V2_1/data_config/sql_data.json

+{
+  "维度管理": {
+    "添加维度": {
+      "AddDemSuccess": "DELETE FROM uc_demension WHERE `CODE_`=\"test_dem_xyz_123\";"
+    },
+    "根据维度编码删除维度": {
+            "DeleteDemSuccess": "SELECT ID_ FROM uc_demension WHERE `CODE_`=\"test_dem_xyz_123\";"
+    }
+  }
+}
+
+

APIAutoTest_V2_1/request_method/request_method.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      request_method.py
+# Author:       liulei
+# Datetime:     2024/7/15 9:23
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import requests
+
+from APIAutoTest_V2_1.commen.read_ini import ReadIni
+
+
+class RequestMethod:
+    def __init__(self):
+        """关联登录成功状态的token"""
+        ini = ReadIni()
+        login_data = {"username": "admin", "password": "F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="}
+        login_url = ini.get_url("test_host") + "/auth"
+        # 创建session对象
+        self.session = requests.sessions.Session()
+        token = self.session.request(method="post", url=login_url, json=login_data).json().get("token")
+        self.session.headers["Authorization"] = "Bearer " + token
+
+    def request_all(self, req_method, req_url, req_mime, case_data):
+        """根据媒体类型的不同，来决定请求方式"""
+        req_mime = req_mime.lower()
+        if req_mime == "application/json" or req_mime == "json":
+           return self.session.request(method=req_method, url=req_url, json=case_data)
+
+        elif req_mime == "application/x-www-form-urlencoded" or req_mime == "x-www-form-urlencoded":
+           return self.session.request(method=req_method, url=req_url, data=case_data)
+
+        elif req_mime == "multipart/form-data" or req_mime == "form-data":
+           return self.session.request(method=req_method, url=req_url, files=case_data)
+
+        elif req_mime == "query" or req_mime == "params":
+           return self.session.request(method=req_method, url=req_url, params=case_data)
+
+        elif req_mime is None:
+           return self.session.request(method=req_method, url=req_url)
+
+        else:
+            raise ValueError("传入的参数错误")
\ No newline at end of file

APIAutoTest_V2_1/test_case/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      __init__.py
+# Author:       liulei
+# Datetime:     2024/7/15 11:02
+# Description:
+# 
+# ---------------------------------------------------------------------------

APIAutoTest_V2_1/test_case/conftest.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      conftest.py
+# Author:       liulei
+# Datetime:     2024/7/15 9:43
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import pytest
+
+from APIAutoTest_V2_1.commen.db import DB
+from APIAutoTest_V2_1.request_method.request_method import RequestMethod
+
+
+@pytest.fixture(scope="session")
+def req_fix():
+    req = RequestMethod()
+    yield req
+
+
+@pytest.fixture(scope="session")
+def db_fix():
+    db = DB()
+    yield db
+    db.close()

APIAutoTest_V2_1/test_case/test_bpm.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      test_bpm.py
+# Author:       liulei
+# Datetime:     2024/7/15 9:23
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pytest
+
+from APIAutoTest_V2_1.commen.read_excel import ReadExcel
+
+
+class TestBPM:
+    @pytest.mark.parametrize("req_method, req_url, req_mime, case_data, expect_data, sql_type, sql_data, update_key", ReadExcel().get_data_all())
+    def test_request_bpm(self, req_fix, db_fix, req_method, req_url, req_mime, case_data, expect_data, sql_type, sql_data, update_key):
+        # print(req_method, req_url, req_mime, case_data, expect_data, sql_type, sql_data, update_key)
+        if sql_type == "delete":
+            db_fix.delete(sql_data)
+            result = req_fix.request_all(req_method, req_url, req_mime, case_data)
+            print(result.json())
+
+        elif sql_type == "select":
+           select_result = db_fix.select(sql_data)
+           print("查询结果为：", select_result)
+           case_data[update_key] = select_result
+
+           result = req_fix.request_all(req_method, req_url, req_mime, case_data)
+           print(result.json())
+
+
+        else:
+            result = req_fix.request_all(req_method, req_url, req_mime, case_data)
+            print(result.json())
+
+        try:
+            for key in expect_data.keys():
+                assert expect_data[key] == result.json().get(key)
+        except:
+            raise AssertionError("断言失败")
+        else:
+            print("断言成功")
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+