apiautotest-V2_2

APIAutoTest_V2_2/__init__.py

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

APIAutoTest_V2_2/commen/db.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/15 12:30
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pymysql
+
+from test_project.mine.APIAutoTest_V2_2.commen.read_ini import ReadIni
+
+
+class DB:
+    def __init__(self):
+        """连接数据库"""
+        ini = ReadIni()
+        self.conn = pymysql.connect(
+            host=ini.get_database_msg("host"),
+            port=int(ini.get_database_msg("port")),
+            user=ini.get_database_msg("user"),
+            password=ini.get_database_msg("pwd"),
+            database=ini.get_database_msg("database"),
+            charset="utf8"
+        )
+        # 创建游标对象
+        self.cur = self.conn.cursor()
+
+    def close(self):
+        """关闭数据库的方法"""
+        self.cur.close()
+        self.conn.close()
+
+    def delete(self, sql_sentence):
+        """删除语句"""
+        try:
+            self.cur.execute(sql_sentence)
+        except:
+            raise Exception("在执行数据库的删除时，给入的语句有误")
+        else:
+            self.conn.commit()
+
+    def select(self, sql_sentence):
+        """查询语句"""
+        try:
+            self.cur.execute(sql_sentence)
+        except:
+            raise Exception("在执行数据库的查询操作时，给入的语句有误")
+        else:
+            data = self.cur.fetchall()
+            return data
+
+

APIAutoTest_V2_2/commen/read_excel.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/15 12:30
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import openpyxl
+
+from test_project.mine.APIAutoTest_V2_2.commen.read_ini import ReadIni
+from test_project.mine.APIAutoTest_V2_2.commen.read_json import ReadJson
+
+
+class ReadExcel:
+    def __init__(self):
+        # 获取excel文件路径和工作表名称
+        self.ini = ReadIni()
+        excel_path = self.ini.get_file("excel")
+        table_name = self.ini.get_table("table")
+        # 创建openpyxl对象
+        wb = openpyxl.load_workbook(excel_path)
+        self.ws = wb[table_name]
+
+    def __get_cell_value(self, column, row):
+        """封装一个私有方法，根据坐标获取单元格的值"""
+        try:
+            Value = self.ws[column + str(row)].value
+            if Value:
+                return Value.strip()
+            else:
+                return None
+        except ValueError as v:
+            print("获取excel表格中的值时出现错误：")
+            raise v
+
+    def get_module_name(self, row):
+        """获取模块名称"""
+        try:
+            module_name = self.__get_cell_value("b", row)
+        except:
+            raise ValueError("获取模块名称时传入的row错误")
+        else:
+            return module_name
+
+    def get_api_name(self, row):
+        """获取接口名称"""
+        try:
+            api_name = self.__get_cell_value("c", row)
+        except:
+            raise ValueError("获取接口名称时传入的row错误")
+        else:
+            return api_name
+
+    def get_case_title(self, row):
+        """获取用例标题"""
+        try:
+            case_title = self.__get_cell_value("d", row)
+        except:
+            raise ValueError("获取用例标题时传入的row错误")
+        else:
+            return case_title
+
+    def get_request_method(self, row):
+        """获取请求方法"""
+        try:
+            request_method = self.__get_cell_value("f", row)
+        except:
+            raise ValueError("获取请求方法时传入的row错误")
+        else:
+            return request_method
+
+    def get_request_path(self, row):
+        """获取请求路径"""
+        try:
+            request_path = self.__get_cell_value("g", row)
+        except:
+            raise ValueError("获取请求路径时传入的row错误")
+        else:
+            if request_path:
+                host = self.ini.get_host("test_host")
+                return host + request_path
+
+    def get_mime(self, row):
+        """获取媒体类型"""
+        try:
+            mime = self.__get_cell_value("h", row)
+        except:
+            raise ValueError("获取媒体类型时传入的row错误")
+        else:
+            if mime:
+                return mime.strip()
+
+
+    def get_case_data(self, row):
+        """获取用例数据"""
+        try:
+            case = self.__get_cell_value("i", row)
+        except:
+            raise ValueError("获取用例数据时传入的row错误")
+        else:
+            if case:
+                case_data = ReadJson("case")[self.get_module_name(row)][self.get_api_name(row)][case]
+                return case_data
+
+    def get_expect_data(self, row):
+        """获取期望数据"""
+        try:
+            expect = self.__get_cell_value("j", row)
+        except:
+            raise ValueError("获取期望数据时传入的row错误")
+        else:
+            if expect:
+                expect_data = ReadJson("expect")[self.get_module_name(row)][self.get_api_name(row)][expect]
+                return expect_data
+
+    def get_sql_type(self, row):
+        """获取sql语句类型"""
+        try:
+            sql_type = self.__get_cell_value("k", row)
+        except:
+            raise ValueError("获取sql语句类型时传入的row错误")
+        else:
+            return sql_type
+
+    def get_sql_sentence(self, row):
+        """获取sql语句"""
+        try:
+            sql_data = self.__get_cell_value("l", row)
+        except:
+            raise ValueError("获取sql语句时时传入的row错误")
+        else:
+            if sql_data:
+                sql_sentence = ReadJson("sql")[self.get_module_name(row)][self.get_api_name(row)][sql_data]
+                if sql_sentence:
+                    return sql_sentence
+            else:
+                return None
+
+    def get_update_key(self, row):
+        """获取更新的key"""
+        try:
+            update_key = self.__get_cell_value("m", row)
+        except:
+            raise ValueError("获取更新的key时传入的row错误")
+        else:
+            return update_key
+
+    def get_data_all(self):
+        """将测试需要的数据封装进一个二维列表里面"""
+        data_list = []
+        for row in range(2, self.ws.max_row+1):
+            module = self.get_module_name(row)
+            api = self.get_api_name(row)
+            title = self.get_case_title(row)
+            req_method = self.get_request_method(row)
+            req_path = self.get_request_path(row)
+            req_mime = self.get_mime(row)
+            case_data = self.get_case_data(row)
+            expect = self.get_expect_data(row)
+            sql_type = self.get_sql_type(row)
+            sql_sentence = self.get_sql_sentence(row)
+            update_key = self.get_update_key(row)
+            list1 = [module, api, title, req_method, req_path, req_mime, case_data, expect, sql_type, sql_sentence, update_key]
+            if module is not None and api is not None:
+                data_list.append(list1)
+        return data_list
+
+
+
+
+
+if __name__ == "__main__":
+    re = ReadExcel()
+    print(re.get_data_all())
+    # print(re.get_case_data(10))
\ No newline at end of file

APIAutoTest_V2_2/commen/read_ini.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/15 12:30
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import configparser
+import os.path
+
+
+class ReadIni:
+    def __init__(self):
+        """获取ini文件路径"""
+        self.path = os.path.dirname(os.path.dirname(__file__)) + "\data_config\\"
+        ini_path = self.path + r"config.ini"
+        # 创建读取对象
+        self.con = configparser.ConfigParser()
+        # 读取ini文件
+        self.con.read(ini_path, encoding="utf-8")
+
+    def get_file(self, key):
+        """根据key返回文件的绝对路径"""
+        try:
+            file_name = self.con.get("file", key)
+        except Exception as e:
+            print("获取文件的名称时出现错误：")
+            raise e
+        else:
+            return self.path + file_name
+
+    def get_host(self, key):
+        """根据key返回被测系统的域名"""
+        try:
+            host = self.con.get("host", key)
+        except Exception as e:
+            print("获取被测系统域名时出现错误：")
+            raise e
+        else:
+            return host
+
+    def get_table(self, key):
+        """根据Key返回工作表名称"""
+        try:
+            table = self.con.get("table_name", key)
+        except Exception as e:
+            print("获取工作表名称时出现错误：")
+            raise e
+        else:
+            return table
+
+    def get_database_msg(self, key):
+        """根据key返回数据库连接信息"""
+        try:
+            msg = self.con.get("sql", key)
+        except Exception as e:
+            print("获取数据库连接信息时出现错误，错误为：")
+            raise e
+        else:
+            return msg
+
+if __name__ == "__main__":
+    ri = ReadIni()
+    print(ri.get_host("test_host"))
+    print(ri.get_file("expect"))
+    print(ri.get_database_msg("user"))
\ No newline at end of file

APIAutoTest_V2_2/commen/read_json.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      read_ini.py
+# Author:       liulei
+# Datetime:     2024/7/15 12:30
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import json
+import os.path
+
+from test_project.mine.APIAutoTest_V2_2.commen.read_ini import ReadIni
+
+
+def ReadJson(key):
+    # 获取json文件路径
+    path = ReadIni().get_file(key)
+
+    # 打开json文件
+    if os.path.isfile(path) and path.endswith(".json"):
+        try:
+            with open(path, 'r', encoding='utf-8') as f:
+                json_data = json.loads(f.read())
+        except Exception as e:
+            print("根据传入的key获取json文件内容时出现错误：")
+            raise e
+        else:
+            return json_data
+    else:
+        print("传入的key有错误")
+
+
+if __name__ == "__main__":
+    rj = ReadJson("case")
+    print(type(rj))
+
+

APIAutoTest_V2_2/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_2/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_2/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_2/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_2/request_method/request_method.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      request_method.py
+# Author:       liulei
+# Datetime:     2024/7/15 13:38
+# Description:
+# 
+# ---------------------------------------------------------------------------
+
+import requests
+
+from test_project.mine.APIAutoTest_V2_2.commen.read_ini import ReadIni
+
+
+class RequestMethod:
+    def __init__(self):
+        """配置登录成功的token"""
+        login_url = ReadIni().get_host("test_host") + r"/auth"
+        login_data = {"username": "admin", "password": "F4/DVgPS/NEruLxVVRqHktsb1R2fVpw81t5VuGfFjwp0G7U4k6spHPr/ejPlw8XxIVilJ+SyIH0G5FbQStFEd/94mmI7+2Dw2c7MXXIERYKjd3XNe4gZR4ANJclCJHNGfE+mtnX5voprYwEo9m6ponCdmmXTMx9cWVEJ4K/nbR8="}
+        # 创建session对象
+        self.session = requests.sessions.Session()
+        token = self.session.request(method='post', url=login_url, json=login_data).json().get("token")
+        # 配置session的头部
+        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_2/test_case/__init__.py

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

APIAutoTest_V2_2/test_case/conftest.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      conftest.py
+# Author:       liulei
+# Datetime:     2024/7/15 13:46
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pytest
+
+from test_project.mine.APIAutoTest_V2_2.commen.db import DB
+from test_project.mine.APIAutoTest_V2_2.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()
\ No newline at end of file

APIAutoTest_V2_2/test_case/test_bpm.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   pythonProject
+# FileName:      test_bpm.py
+# Author:       liulei
+# Datetime:     2024/7/15 13:48
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pytest
+
+from test_project.mine.APIAutoTest_V2_2.commen.read_excel import ReadExcel
+
+
+class TestBpm:
+    @pytest.mark.parametrize("module, api, title, req_method, req_path, req_mime, case_data, expect, sql_type, sql_sentence, update_key", ReadExcel().get_data_all())
+    def test_bpm(self, req_fix, db_fix, module, api, title, req_method, req_path, req_mime, case_data, expect, sql_type, sql_sentence, update_key):
+        print(module, api, title, req_method, req_path, req_mime, case_data, expect, sql_type, sql_sentence, update_key)
+
+        if sql_type == "delete":
+            db_fix.delete(sql_sentence)
+
+            res = req_fix.request_all(req_method=req_method, req_url=req_path, req_mime=req_mime, case_data=case_data)
+            print(res.json())
+
+        elif sql_type == "select":
+            sql_res = db_fix.select(sql_sentence)
+            print("sql语句为：", sql_sentence)
+            print("更新的用例为：", update_key)
+            case_data[update_key] = sql_res
+            res = req_fix.request_all(req_method=req_method, req_url=req_path, req_mime=req_mime, case_data=case_data)
+            print(res.json())
+
+        else:
+            res = req_fix.request_all(req_method=req_method, req_url=req_path, req_mime=req_mime, case_data=case_data)
+            print(res.json())
+
+        try:
+            for key in expect.keys():
+                assert res.json().get(key) == expect[key]
+        except:
+            raise AssertionError("断言失败")
+        else:
+            print("断言成功")
+