接口

APIAutoTest_liu/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/2 19:46
+# Description:
+# 
+# ---------------------------------------------------------------------------

APIAutoTest_liu/common/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   test_58
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/4 9:33
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import functools
+import os
+
+from APIAutoTest_liu.common.log import log
+
+# 日志文件的路径，需要先创建好日志文件
+log_path = os.path.join(os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "report"), "log"), "log_template.log")
+# 获取loger
+loger = log(log_path)
+
+
+# 装饰器
+def log_decorator(func):
+    @functools.wraps(func)  # 解决被装饰函数和方法名重名的问题，它会将被装饰的函数或方法使用原来的名称
+    def inner(*args, **kwargs):
+        # 在执行func之前需要获取func的名称和功能，并写入到日志
+        loger.info("功能的名称为："+str(func.__name__)+" 功能的描述为："+str(func.__doc__))
+        try:
+            # 执行func
+            result = func(*args, **kwargs)
+        except Exception as e:
+            # 如果有错误，需要写入到日志
+            loger.error(str(e))
+            # 将错误，原路返回
+            raise e
+        else:
+            return result
+    return inner
\ No newline at end of file

APIAutoTest_liu/common/db.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      db.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/4 19:31
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pymysql
+
+from APIAutoTest_liu.common.read_ini import ReadIni
+
+
+class DB:
+    def __init__(self):
+        # 创建ReadIni的对象，调用get_sql_connect_message的方法来获取数据库的连接信息
+        read_ini = ReadIni()
+        try:
+            # 创建连接对象
+            self.connect = 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"
+            )
+        except:
+            raise ConnectionError("连接数据库失败，配置文件数据错误")
+        else:
+            # 获取游标对象
+            self.cursor = self.connect.cursor()
+
+    def close(self):
+        # 关闭数据库，先关闭游标对象，在关闭连接对象
+        self.cursor.close()
+        self.connect.close()
+
+    def delete(self, sql_sentence):
+        # 执行sql删除语句
+        try:
+            self.cursor.execute(sql_sentence)
+        except:
+            raise ValueError("sql语句是否正确")
+        else:
+            # 连接对象提交
+            self.connect.commit()
+
+    def select(self, sql_sentence):
+        try:
+            self.cursor.execute(sql_sentence)
+        except:
+            raise ValueError("sql语句是否正确")
+        else:
+            select_result = self.cursor.fetchall()
+            if select_result:
+                return select_result[0][0]
+
+if __name__ == '__main__':
+    db = DB()
+    str2 = "seleCT ID_,DEM_NAME_ FROM uc_demension limit 1"
+    print(db.select(str2))
+
+
+
+
+
+
+

APIAutoTest_liu/common/log.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   test_58
+# FileName:      log.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/4 11:48
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import logging
+
+
+def log(filename):
+    # 创建logger对象
+    loger = logging.getLogger()
+    # 创建文件的控制器
+    handler = logging.FileHandler(filename, mode="a", encoding="utf-8")
+    # 设置日志文件中内容的格式
+    formatter = logging.Formatter('%(levelname)s--%(asctime)s--%(message)s')
+    # 格式添加到文件的控制器中
+    handler.setFormatter(formatter)
+    # logger对象添加文件控制器
+    loger.addHandler(handler)
+    # 设置日志文件的级别
+    loger.level = logging.INFO
+
+    return loger

APIAutoTest_liu/common/read_excel.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      read_excel.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/3 19:03
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import openpyxl
+
+from APIAutoTest_liu.common.read_ini import ReadIni
+from APIAutoTest_liu.common.read_json import read_json
+
+
+class ReadExcel:
+    def __init__(self):
+        # 先创建ReadIni对象
+        self.read_ini = ReadIni()
+        # 使用ReadIni对象的get_file_path方法获取excel文件路径
+        excel_path = self.read_ini.get_file_path("excel")
+        # 调用ReadInid对象的get_file_path方法获取case文件路径
+        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")
+
+        # 在调用read_json方法读取excel文件内容
+        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工作薄
+        wb = openpyxl.load_workbook(excel_path)
+        # 调用read_ini对象的get_table_name方法获取sheet表名
+        table_name = self.read_ini.get_table_name("name")
+        self.ws = wb[table_name]
+
+    def __get_cell_value(self, colum: str, row: int):
+        # 获取单元格值
+        try:
+            cell_value = self.ws[colum + str(row)].value
+        except:
+            raise ValueError("excel中没有该列和行号")
+        else:
+            # 判断单元格值是否为空，为空返回None，否则返回去除空格的字符串
+            if cell_value is None:
+                return None
+            elif cell_value.strip():
+                return cell_value.strip()
+
+    def module_name(self, row):
+        # 根据行号获取模块名称
+        return self.__get_cell_value("B", row)
+
+    def api_name(self, row):
+        # 根据行号获取接口名称
+        return self.__get_cell_value("C", row)
+
+    def case_url(self, row):
+        # 先调用read_ini文件中ReadIni的get_host方法调用bpm域名
+        host = self.read_ini.get_host("bpm_host")
+        # 在根据行号获取接口路径
+        path = self.__get_cell_value("f", row)
+        # 非空为true，为空为false返回None
+        if path:
+            return host + path
+
+    def case_req(self, row):
+        # 根据行号获取用例的请求方法
+        return self.__get_cell_value("G", row)
+
+    def case_mime(self, row):
+        # 根据行号获取用例的媒体类型
+        mime = self.__get_cell_value("H", row)
+        # 非空为true，为空为false返回None
+        if mime:
+            return mime.lower()
+
+    def case_data(self, row):
+        # 根据行号获取用例数据
+        case_data_key = self.__get_cell_value("I", row)
+        if case_data_key:
+            # 获取模块名称和接口名称
+            module_name = self.module_name(row)
+            api_name = self.api_name(row)
+            return self.case_data_dict[module_name][api_name][case_data_key]
+
+    def expect_data(self, row):
+        # 根据行号获取期望结果
+        expect_data_key = self.__get_cell_value("J", row)
+        if expect_data_key:
+            # 获取模块名称和接口名称
+            module_name = self.module_name(row)
+            api_name = self.api_name(row)
+            return self.expect_data_dict[module_name][api_name][expect_data_key]
+
+    def sql_type(self, row):
+        # 根据行号获取sql语句类型
+        sql_type_value = self.__get_cell_value("K", row)
+        if sql_type_value:
+            # 判断sql语句是否为None，返回None。如果不为None，返回sql语句类型的值小写
+            return sql_type_value.lower()
+
+    def sql_data(self, row):
+        # 根据行号获取sql语句
+        sql_data_key = self.__get_cell_value("L", row)
+        if sql_data_key:
+            # 获取模块名称和接口名称
+            module_name = self.module_name(row)
+            api_name = self.api_name(row)
+            return self.sql_data_dict[module_name][api_name][sql_data_key]
+
+    def update_key(self, row):
+        # 根据行号获取更新的key
+        return self.__get_cell_value("M", row)
+
+    def case_title(self, row):
+        # 根据行号获取用例的标题
+        return self.__get_cell_value("D", row)
+
+    def case_level(self, row):
+        # 根据行号获取用例的等级
+        return self.__get_cell_value("E", row)
+
+    def get_data(self):
+        # 新建一个空列表
+        list_data = []
+        for row in range(2, self.ws.max_row + 1):
+            module_name = self.module_name(row)
+            api_name = self.api_name(row)
+            case_level = self.case_level(row)
+            case_title = self.case_title(row)
+            case_req = self.case_req(row)
+            case_url = self.case_url(row)
+            case_mime = self.case_mime(row)
+            case_data = self.case_data(row)
+            expect_data = self.expect_data(row)
+            sql_type = self.sql_type(row)
+            sql_data = self.sql_data(row)
+            update_key = self.update_key(row)
+            # 将所有数据存放在列表中
+            row_list = [case_url, case_req, case_mime, case_data, expect_data, sql_type, sql_data, update_key]
+            # 规避空值===每行空数据不得超过5个
+            if row_list.count(None) <= 5:
+                # if判断满足条件将数据追加到前面新疆的空列表中
+                list_data.append(row_list)
+        else:
+            # 空值数据大于5个，返回空列表
+            return list_data
+
+if __name__ == '__main__':
+    excel = ReadExcel()
+    print(excel.get_data())
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+

APIAutoTest_liu/common/read_ini.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      read_ini.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/2 19:55
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import os
+import configparser
+
+
+class ReadIni:
+    def __init__(self):
+       # 先获取ini文件路径，1、先获取目录data_config路径
+        self.data_config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data_config")
+        #2、再获取config.ini文件路径路径之间的拼接
+        ini_path = os.path.join(self.data_config_path, "config.ini")
+        #创建ConfigParser对象
+        self.config = configparser.ConfigParser()
+        #读取ini配置文件
+        self.config.read(ini_path, encoding="utf-8")
+
+    def get_file_path(self, key):
+        # 获取file对应的key值，返回对应的路径
+        try:
+            file_name = self.config.get("file", key)
+        except:
+            raise KeyError("传入的key错误")
+        else:
+            return os.path.join(self.data_config_path, file_name)
+
+    def get_table_name(self, key):
+        # 获取table对应的key值
+        try:
+            table_name = self.config.get("table", key)
+        except:
+            raise KeyError("传入的key错误")
+        else:
+            return table_name
+
+    def get_host(self, key):
+        # 获取host对应的key值的域名
+        try:
+            host = self.config.get("host", key)
+        except:
+            raise KeyError("传入的key错误")
+        else:
+            return host
+
+    def get_sql_connect_message(self, key):
+        #  获取sql连接数据库的信息
+        try:
+            value = self.config.get("sql", key)
+        except:
+            raise KeyError("传入的key错误")
+        else:
+            return value
+
+if __name__ == '__main__':
+    ini = ReadIni()
+    print(ini.get_host("bpm_host"))
+
+
+
+
+
+
+
+
+
+

APIAutoTest_liu/common/read_json.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      read_json.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/3 18:51
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import json
+import os.path
+
+
+def read_json(filename):
+    # 判断json文件的路径是否是一个文件并且以.json结尾
+    if os.path.isfile(filename) and filename.endswith(".json"):
+        with open(filename, mode="r", encoding="utf-8") as f:
+            try:
+                res = json.loads(f.read())
+            except:
+                raise FileNotFoundError("json文件格式错误")
+            else:
+                return res
+    else:
+        raise FileNotFoundError("json文件的路径错误")

APIAutoTest_liu/data_config/case_data.json

+{
+    "认证接口":{
+        "登录系统":{
+            "LoginSuccess":{"username":"admin","password":"MTIzNDU2"},
+            "LoginErrorUsernameIsNone":{"username":"","password":"MTIzNDU2"},
+            "LoginErrorUsernameIsShort":{"username":"a","password":"MTIzNDU2"},
+            "LoginErrorUsernameIsLong":{"username":"adminadminadminadminadminadminadminadminadminadmin","password":"MTIzNDU2"},
+            "LoginErrorUsernameIsSpecial":{"username":"♠♣▣▤▥▦▩◘◙◈","password":"MTIzNDU2"},
+            "LoginErrorUsernameIsError":{"username":"Admin","password":"MTIzNDU2"},
+            "LoginErrorPasswordIsNone":{"username":"admin","password":""},
+            "LoginErrorPasswordIsLong":{"username":"admin","password":"MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2"},
+            "LoginErrorPasswordIsShort":{"username":"admin","password":"m"},
+            "LoginErrorPasswordIsSpecial":{"username":"admin","password":"♠♣▣▤▥▦▩◘◙◈"}
+        }
+    },
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": {
+                "code": "test58_dem_abc",
+                "description": "测试维度-58",
+                "isDefault": 0,
+                "name": "测试维度-58"
+            }
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"ids": "需要更新"}
+        }
+    }
+}
+

APIAutoTest_liu/data_config/config.ini

+[file]
+# 数据配置层中文件的名称
+excel=APIAutoTest.xlsx
+case=case_data.json
+expect=expect_data .json
+sql=sql_data.json
+
+[table]
+# excel工作表的名称
+name=BPM
+
+[host]
+# 被测系统的域名
+bpm_host=http://120.46.172.186:8080
+
+[sql]
+# 数据库链接信息
+host=120.46.172.186
+port=3306
+user=root
+pwd=root@2023
+database=eipsaas
\ No newline at end of file

APIAutoTest_liu/data_config/expect_data .json

+{
+    "认证接口":{
+        "登录系统":{
+            "LoginSuccess":{"username":"超级管理员","account":"admin","loginStatus":true},
+            "LoginErrorUsernameIsNone":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorUsernameIsShort":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorUsernameIsLong":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorUsernameIsSpecial":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorUsernameIsError":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorPasswordIsNone":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorPasswordIsLong":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorPasswordIsShort":{"state":false,"message":"账号或密码错误"},
+            "LoginErrorPasswordIsSpecial":{"state":false,"message":"账号或密码错误"}
+        },
+        "刷新token": {
+            "RefreshTokenSuccess": {"message": "刷新成功"}
+        }
+    },
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": {"message": "添加维度成功！"}
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"message": "删除维度成功！"}
+        }
+    }
+}
+

APIAutoTest_liu/data_config/sql_data.json

+{
+    "维度管理": {
+        "添加维度": {
+            "AddDemSuccess": " delete from uc_demension WHERE `CODE_`=\"test58_dem_abc\";"
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": "select ID_ from uc_demension WHERE `CODE_`=\"test58_dem_abc\";"
+        }
+    }
+}
+

APIAutoTest_liu/report/log/log_template.log

+INFO--2024-01-06 16:59:38,929--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:38,988--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,040--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,136--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,229--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,317--功能的名称为：test_bpm_api 功能的描述为：None
+ERROR--2024-01-06 16:59:39,414--断言失败，用例数据为：{'username': 'Admin', 'password': 'MTIzNDU2'} 期望数据为：{'state': False, 'message': '账号或密码错误'} 服务器返回的数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE3MDQ2MTc5ODEsImlhdCI6MTcwNDUzMTU4MX0.7cKnfVzP6DXhyTHHj5UIHnUYrCjc0KpQOiHfytM0TDp0hEO7CPO1Bci1uhQc5JszOpCakMPe9b0NAzAgsGGNOQ","username":"超级管理员","account":"admin","userId":"1","expiration":86400,"loginStatus":true,"userAttrs":{"tenantId":"-1"}}
+ERROR--2024-01-06 16:59:39,414--断言失败
+INFO--2024-01-06 16:59:39,618--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,673--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,771--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,870--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:39,967--功能的名称为：test_bpm_api 功能的描述为：None
+ERROR--2024-01-06 16:59:40,061--断言失败，用例数据为：None 期望数据为：{'message': '刷新成功'} 服务器返回的数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE3MDQ2MTc5ODIsImlhdCI6MTcwNDUzMTU4Mn0.orqGZ4O1p1B86EgwdkL8Cpg7z2V0FfESTPvAuApF-tB43_l4l_qyzi_a_vkwffgyOx1Msgq0b5ZJB0QcigATdg","username":"","account":"","userId":"","loginStatus":true,"userAttrs":{}}
+ERROR--2024-01-06 16:59:40,061--断言失败
+INFO--2024-01-06 16:59:40,073--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 16:59:40,222--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,267--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,324--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,377--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,473--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,568--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,659--功能的名称为：test_bpm_api 功能的描述为：None
+ERROR--2024-01-06 17:00:34,758--断言失败，用例数据为：{'username': 'Admin', 'password': 'MTIzNDU2'} 期望数据为：{'state': False, 'message': '账号或密码错误'} 服务器返回的数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE3MDQ2MTgwMzYsImlhdCI6MTcwNDUzMTYzNn0.PF1diiPrtkK4GVjDYr7p1XzTVqramGFQuh34QGGcrHfVg0A2stn4EXwX1S1GaIXXDZeHhDcZIrG2BCduLxZGOQ","username":"超级管理员","account":"admin","userId":"1","expiration":86400,"loginStatus":true,"userAttrs":{"tenantId":"-1"}}
+ERROR--2024-01-06 17:00:34,758--断言失败
+INFO--2024-01-06 17:00:34,929--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:34,984--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:35,080--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:35,166--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:35,261--功能的名称为：test_bpm_api 功能的描述为：None
+ERROR--2024-01-06 17:00:35,352--断言失败，用例数据为：None 期望数据为：{'message': '刷新成功'} 服务器返回的数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE3MDQ2MTgwMzcsImlhdCI6MTcwNDUzMTYzN30.1D_GYMC7T_XezWS3nMEE-zeH9XX8y1MDiK7CwBwvHeXeXb9aGOB6OG9f7qLcS37Wm-Pp-k0-XU5FkMv_u85hVQ","username":"","account":"","userId":"","loginStatus":true,"userAttrs":{}}
+ERROR--2024-01-06 17:00:35,352--断言失败
+INFO--2024-01-06 17:00:35,362--功能的名称为：test_bpm_api 功能的描述为：None
+INFO--2024-01-06 17:00:35,494--功能的名称为：test_bpm_api 功能的描述为：None

APIAutoTest_liu/report/log/result_log.log

+ERROR--2024-01-06 16:49:07,069--断言失败，用例数据为：{'username': 'Admin', 'password': 'MTIzNDU2'} 期望数据为：{'state': False, 'message': '账号或密码错误'} 服务器返回的数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE3MDQ2MTczNDksImlhdCI6MTcwNDUzMDk0OX0.F_sFNM7F7G5JDpm9XjYJY23mZTWoMhcqEvM394FS_55LS2slzgVt3hQ0dmXmBq91TsclMSr-i24RStFXx5brCA","username":"超级管理员","account":"admin","userId":"1","expiration":86400,"loginStatus":true,"userAttrs":{"tenantId":"-1"}}
+ERROR--2024-01-06 16:49:07,688--断言失败，用例数据为：None 期望数据为：{'message': '刷新成功'} 服务器返回的数据为：{"token":"eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhZG1pbiIsInRlbmFudElkIjoiLTEiLCJleHAiOjE3MDQ2MTczNDksImlhdCI6MTcwNDUzMDk0OX0.F_sFNM7F7G5JDpm9XjYJY23mZTWoMhcqEvM394FS_55LS2slzgVt3hQ0dmXmBq91TsclMSr-i24RStFXx5brCA","username":"","account":"","userId":"","loginStatus":true,"userAttrs":{}}
+INFO--2024-01-06 16:56:09,780--方法或函数的名称为：test_bpm_api 方法或函数的功能为：None
+ERROR--2024-01-06 16:56:09,780--方法或函数的名称为：test_bpm_api 报错为：TestBPM.test_bpm_api() missing 10 required positional arguments: 'db_fix', 'req_fix', 'case_url', 'case_req', 'case_mime', 'case_data', 'expect_data', 'sql_type', 'sql_data', and 'update_key'

APIAutoTest_liu/requests_method/requests_method.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      requests_method.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/4 20:26
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import base64
+
+import requests.sessions
+
+from APIAutoTest_liu.common.read_ini import ReadIni
+
+
+class RequestsMethod:
+    def __init__(self):
+        # url路径的拼接
+        login_url = ReadIni().get_host("bpm_host") + "/auth"
+        # 登录数据
+        login_data = {"username": "admin", "password": base64.b64encode("123456".encode()).decode()}
+        # 创建session对象提取token
+        self.bpm_session = requests.sessions.Session()
+        self.bpm_session.headers["Authorization"] = "Bearer "+self.bpm_session.post(url=login_url, json=login_data).json()["token"]
+
+    def request_all(self, req_method, req_url, req_mime, case_data):
+        """
+        封装requests请求方法
+        :param req_method: 请求方法
+        :param req_url: 请求路径
+        :param req_mime: 请求类型
+        :param case_data: 请求数据
+        :return: Response type
+        """
+        # 判断媒体类型是否为表单类型
+        if req_mime == "application/x-www-form-urlencoded" or req_mime == "x-www-form-urlencoded":
+            return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
+
+        # 判断媒体类型是否为json类型
+        elif req_mime == "application/json" or req_mime == "json":
+            return self.bpm_session.request(method=req_method, url=req_url, json=case_data)
+
+        # 判断媒体类型是否为上传文件类型
+        elif req_mime == "multipart/from-data" or req_mime == "from-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 is None:
+            return self.bpm_session.request(method=req_method, url=req_url)
+
+        # 判断媒体类型是否为query|json同时传参
+        elif req_mime == "query|json" or req_mime =="json|query" or req_mime == "params|json":
+            return self.bpm_session.request(method=req_method, url=req_url, params=case_data["query"], json=case_data["json"])
+
+        # 判断媒体类型是否为text或者text/plain
+        elif req_mime == "text" or req_mime == "text/plain":
+            return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
+
+        else:
+            raise NameError("不支持的请求类型")
+if __name__ == '__main__':
+    req = RequestsMethod()
+
+

APIAutoTest_liu/test_case/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/2 19:49
+# Description:
+# 
+# ---------------------------------------------------------------------------

APIAutoTest_liu/test_case/conftest.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      conftest.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/6 11:02
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pytest
+
+from APIAutoTest_liu.common.db import DB
+from APIAutoTest_liu.requests_method.requests_method import RequestsMethod
+
+
+@pytest.fixture(scope="session")
+def req_fix():
+    """创建RequestsMethod对象，并返回"""
+    req = RequestsMethod()
+    yield req
+
+@pytest.fixture(scope="session")
+def db_fix():
+    """创建DB对象，并返回,"""
+    db = DB()
+    yield db
+    db.close()

APIAutoTest_liu/test_case/test_bpm.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   db.py
+# FileName:      test_bpm.py
+# Author:       xxxxxxx
+# Datetime:     2024/1/6 11:09
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import logging
+
+import pytest
+
+from APIAutoTest_liu.common import log_decorator
+from APIAutoTest_liu.common.read_excel import ReadExcel
+
+
+class TestBPM:
+    @log_decorator
+    @pytest.mark.parametrize("case_url, case_req, case_mime, case_data, expect_data, sql_type, sql_data, update_key", ReadExcel().get_data())
+
+    def test_bpm_api(self, db_fix, req_fix, case_url, case_req, case_mime, case_data, expect_data, sql_type, sql_data, update_key):
+        print(case_url, case_req, case_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=case_req, req_url=case_url, req_mime=case_mime, case_data=case_data)
+            print("="*100)
+            print(result.text)
+
+            # 2：判断sql语句的类型是否为select，如果是，使用DB类对象，调用select方法执行查询的sql语句，并将查询的结果更新到用例数据中
+        elif sql_type == "select":
+            # 2.1: 使用db_fix自定义固件调用select方法执行查询的sql语句，并接收查询的结果
+            select_result = db_fix.select(sql_data)
+            print("查询结果为：", select_result)
+            print("用例数据为：", case_data, type(case_data))
+            print("更新的key：", update_key)
+            # 2.2 更新用例数据
+            case_data[update_key] = select_result
+            print("更新之后的用例数据", case_data)
+            # 2.3 将更新之后的用例数据，发送给服务器，需要使用RequestsMethod类对象的request_all方法
+            result = req_fix.request_all(req_method=case_req, req_url=case_url, req_mime=case_mime, case_data=case_data)
+            print("=" * 100)
+            print(result.text)
+        else:
+            # 没有sql语句，直接使用RequestsMethod类对象的request_all方法，发送请求
+            result = req_fix.request_all(req_method=case_req, req_url=case_url, req_mime=case_mime, case_data=case_data)
+            print("=" * 100)
+            print(result.text)
+
+            # 断言
+        try:
+            # 获取期望数据的key，期望数据的key对应的值是否和服务器返回的数据key对应的值相等，如果相等断言成功，否则断言失败
+            for key in expect_data.keys():
+                assert expect_data[key] == result.json().get(key)
+        except AssertionError:
+            logging.error("断言失败，用例数据为：" + str(case_data) + " 期望数据为：" + str(expect_data) + " 服务器返回的数据为：" + result.text)
+            raise AssertionError("断言失败")
+        else:
+            print("断言成功")
+
+if __name__ == '__main__':
+    pytest.main()
\ No newline at end of file