接口自动化框架_V1.0

study_004/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 10:24
+# Description:
+# 
+# ---------------------------------------------------------------------------

study_004/case/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 10:24
+# Description:
+# 
+# ---------------------------------------------------------------------------

study_004/case/conftest.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      conftest.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:43
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pytest
+
+from Request_framework.study_004.common.db_004 import DB
+from Request_framework.study_004.method.requestmethod_004 import Requestmethod_004
+
+
+@pytest.fixture(scope='session')
+def resp_fix():
+    resp = Requestmethod_004()
+    yield resp
+
+@pytest.fixture(scope='session')
+def db4_fix():
+    db = DB()
+    yield db
+    db.close()
+
+def pytest_collection_modifyitems(items):
+    # item表示每个测试用例，解决用例名称中文显示问题
+    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

study_004/case/pytest.ini

+[pytest]
+;开启日志
+log_cli=true
+;设置日志的级别，如果不设置级别的话，可以设置为NOTSET，如果要设置级别，级别可以有debug，info，warning，error,致命
+log_level=NOTSET
+;设置日志显示的信息格式
+log_format=%(levelname)s--%(asctime)s--%(message)s
+;设置日志中时间显示的格式
+log_date_format=%Y-%m-%d %H:%M:%S
+;每个py文件运行的时候追加的命令
+;addopts=-vs
+
+;设置日志保存的文件
+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

study_004/case/test_case_004.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      test_case_004.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:49
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import logging
+import re
+
+import allure
+import pytest
+
+from Request_framework.study_004.common.readexcel_004 import Read_excel_004
+
+
+class Test_bpm_004:
+
+    @allure.epic('BPM系统')
+    @pytest.mark.parametrize("module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence,"
+                             " update_key", Read_excel_004().into_list())
+    def test_004(self, resp_fix, db4_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":
+            db4_fix.delete(sql_sentence)
+        elif sql_type == "select":
+            select_result = db4_fix.select(sql_sentence)
+            data[update_key] = select_result
+        elif sql_type == "delete|select" or sql_type == "select|delete":
+            db4_fix.delete(sql_sentence["delete"])
+            select_result = db4_fix.select(sql_sentence["select"])
+            data[update_key] = select_result
+        result = resp_fix.request_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

study_004/common/db_004.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      db_004.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:36
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import pymysql
+
+from Request_framework.study_004.common.readini_004 import Read_ini_004
+from Request_framework.study_004.config.settings import *
+
+
+class DB:
+    def __init__(self):
+        read_ini_004 = Read_ini_004()
+        self.conn = pymysql.connect(
+            host=read_ini_004.get_sql(SQLHOST),
+            port=int(read_ini_004.get_sql(PORT)),
+            user=read_ini_004.get_sql(USER),
+            password=read_ini_004.get_sql(PWD),
+            database=read_ini_004.get_sql(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

study_004/common/readexcel_004.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      readexcel_004.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:20
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import openpyxl
+import os
+
+from Request_framework.study_004.common.readini_004 import Read_ini_004
+from Request_framework.study_004.common.readjson_004 import read_json
+from Request_framework.study_004.config.settings import *
+
+
+class Read_excel_004:
+    def __init__(self):
+        self.readini_004 = Read_ini_004()
+
+        case_path = self.readini_004.get_file_name(CASE)
+        expect_path = self.readini_004.get_file_name(EXPECT)
+        sql_path = self.readini_004.get_file_name(SQL)
+
+        self.case_data_dict = read_json(case_path)
+        self.expect_data_dict = read_json(expect_path)
+        self.sql_data_dict = read_json(sql_path)
+
+        excel_path = self.readini_004.get_file_name(EXCEL)
+        wb = openpyxl.load_workbook(excel_path)
+        self.ws = wb[self.readini_004.get_table_name(TABLE)]
+
+    def get_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_value(MODULE, row)
+
+    def interface_name(self, row):
+        return self.get_value(API, row)
+
+    def case_title(self, row):
+        return self.get_value(TITLE, row)
+
+    def case_level(self, row):
+        return self.get_value(LEVEL, row)
+
+    def case_method(self, row):
+        return self.get_value(METHOD, row)
+
+    def case_url(self, row):
+        host = self.readini_004.get_host(HOST)
+
+        return host+self.get_value(PATH, row)
+
+    def case_mime(self, row):
+        value = self.get_value(MIME, row)
+
+        if value:
+            return value.lower()
+
+    def case_data(self, row):
+        case_data_key = self.get_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_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_value(SQLTYPE, row)
+        if value:
+            return value.lower()
+
+    def sql_sentence(self, row):
+        sql_data_key = self.get_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_value(UPDATEKEY, row)
+
+    def into_list(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_004 = Read_excel_004()
+    print(read_excel_004.into_list())
\ No newline at end of file

study_004/common/readini_004.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      readini_004.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:05
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import configparser
+import os
+
+
+
+class Read_ini_004:
+
+    def __init__(self):
+        self.data_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'config')
+        ini_path = os.path.join(self.data_path, 'config.ini')
+        self.con = configparser.ConfigParser()
+        self.con.read(ini_path, encoding='utf-8')
+
+    def get_file_name(self, key):
+        file_name = self.con.get('file', key)
+        return os.path.join(self.data_path, file_name)
+
+    def get_host(self, key):
+        return self.con.get('host', key)
+
+    def get_table_name(self, key):
+        return self.con.get('table', key)
+
+    def get_sql(self, key):
+        return self.con.get('sql', key)
+
+if __name__ == '__main__':
+    read_ini_004 = Read_ini_004()
+    print(read_ini_004.get_file_name('case'))
\ No newline at end of file

study_004/common/readjson_004.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      readjson_004.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:19
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import json
+
+
+def read_json(filename):
+    with open(filename, mode="r", encoding="utf-8") as fp:
+        return json.load(fp)

study_004/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": "test56_dem",
+                "description": "测试56期的维度",
+                "isDefault": 0,
+                "name": "测试56期的维度"
+            }
+        },
+        "根据维度编码删除维度": {
+            "DeleteDemSuccess": {"ids": "不知道"}
+        }
+    },
+    "组织管理": {
+        "添加组织": {
+            "AddOrgSuccess": {
+                "code": "test56_org",
+                "demId": "不知道",
+                "exceedLimitNum": 0,
+                "grade": "",
+                "limitNum": 0,
+                "name": "测试的组织",
+                "nowNum": 0,
+                "orderNo": 0,
+                "parentId": "0"
+            }
+        },
+        "删除组织": {
+            "DeleteOrgSuccess": "test56_org"
+        },
+        "保存组织参数": {
+            "SaveOrgParamSuccess": {
+                "query": {"orgCode": "test56_org"},
+                "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

study_004/config/config.ini

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

study_004/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": "保存组织参数成功！"}
+        }
+    }
+}
+

study_004/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 = "G"
+METHOD = "F"
+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"""
+SQLHOST = "host"
+PORT = "port"
+USER = "user"
+PWD = "pwd"
+DATABASE = "database"
\ No newline at end of file

study_004/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

study_004/method/__init__.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      __init__.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 10:24
+# Description:
+# 
+# ---------------------------------------------------------------------------

study_004/method/requestmethod_004.py

+# -*-coding:utf-8 -*- #
+# ---------------------------------------------------------------------------
+# ProjectName:   Study_Python
+# FileName:      requestmethod_004.py
+# Author:       xxxxxxx
+# Datetime:     2023/8/7 19:40
+# Description:
+# 
+# ---------------------------------------------------------------------------
+import base64
+
+import requests
+
+from Request_framework.study_004.common.readini_004 import Read_ini_004
+
+
+class Requestmethod_004:
+    def __init__(self):
+        """关联token"""
+        url = Read_ini_004().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 request_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("传入的媒体类型的名称错误")
\ No newline at end of file