增加读取内置函数，替换内置函数结果

.idea/api_project.iml

@@ -4,7 +4,7 @@
     <content url="file://$MODULE_DIR$">
       <excludeFolder url="file://$MODULE_DIR$/venv" />
     </content>
-    <orderEntry type="jdk" jdkName="Pipenv (apitest)" jdkType="Python SDK" />
+    <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="PackageRequirementsSettings">

.idea/misc.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
-  <component name="ProjectRootManager" version="2" project-jdk-name="Pipenv (apitest)" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Pipenv (api-test-project)" project-jdk-type="Python SDK" />
   <component name="PyCharmProfessionalAdvertiser">
     <option name="shown" value="true" />
   </component>

common/base_datas.py

@@ -14,12 +14,12 @@ class BaseDates:
 
     # 测试用例脚本目录
     # *****************************************************************
-    script = os.path.join(base_path, "script", "bgy")
+    script = os.path.join(base_path, "test_script", "auto_script")
     # *****************************************************************
 
     # 测试报告及 log 所在路径
     # *****************************************************************
-    test_report = os.path.join(base_path, "OutPut", "reports")
+    test_report = os.path.join(base_path, "OutPut", "Reports")
     log_path = os.path.join(base_path, "OutPut", "Log")
 
     wx_send_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key="

common/comparator/loaders.py

@@ -9,18 +9,19 @@
 # -------------------------------------------------------------------------------
 import types
 
-from common import functions
+from common import bif_functions
 from common.comparator import comparators
+from common.dependence import Dependence
 
 
 def load_built_in_functions():
     """
-    加载builtin包中的内建方法
+    加载bif_functions包中的内建方法
     Returns:
     """
     built_in_functions = {}
 
-    for name, item in vars(functions).items():
+    for name, item in vars(bif_functions).items():
         if isinstance(item, types.FunctionType):
             built_in_functions[name] = item
     return built_in_functions
@@ -48,7 +49,6 @@ def load_model_fun(model):
     """
     model_fun = {}
     for name, item in vars(model).items():
-        # print("-----", name, item)
         if isinstance(item, types.FunctionType):
             model_fun[name] = item
 
@@ -79,10 +79,22 @@ def load_model_fun(model):
 #     return ext_method_online_module, ext_methods_online
 
 
+def set_bif_fun():
+    """
+    将所有内置方法加载到依赖表中
+    Returns:
+
+    """
+    for k, v in load_built_in_functions().items():
+        Dependence.update_dep(f"{k}()", v)
+
+
 if __name__ == '__main__':
-    from common.functions import random_tools
+    from common.bif_functions import random_tools
 
     # func = load_model_fun(random_tools)
     # print(func)
     # print(load_built_in_functions())
     print(load_built_in_comparators())
+    print(set_bif_fun())
+    print(Dependence.get_dep())

common/comparator/validator.py

@@ -101,6 +101,8 @@ class Validator(object):
         Returns:返回校验结果
 
         """
+        if not validate_variables:
+            return ""
         self.uniform_validate(validate_variables)
         if not self.validate_variables_list:
             raise "uniform_validate 执行失败，无法进行 validate 校验"

common/dependence.py

@@ -10,6 +10,7 @@ class Dependence:
     dependence = {}  # 定义依赖表
     PATTERN = re.compile(r"{{(.*?)}}")  # 预编译正则表达式
     pattern = re.compile(r'({)')
+    pattern_fun = re.compile(r"{{(\w+\(\))}}")
 
     @classmethod
     def update_dep(cls, key, value):

common/do_sql/do_mysql.py

@@ -50,7 +50,7 @@ class DoMysql:
         :sql:
         :return: 返回操作结果，以字典形式返回
         """
-        print(f"type:{type(sql)}")
+        # print(f"type:{type(sql)}")
         if not sql:
             return
         result = None

common/extractor/data_extractor.py

@@ -22,22 +22,17 @@ REPLACE_DICT = {
     "false": False
 }
 
-dependence = Dependence()
+dependence = Dependence
 
 
 @singleton
 class DataExtractor:
 
     def __init__(self, response=None):
-        self.dependence = dependence.dependence
-        # self.dependence = getattr(Dependence, "dependence")
+        self.dependence = dependence.get_dep()
         self.response = response
         self.PATTERN = getattr(Dependence, "PATTERN")  # 预编译正则表达式
 
-    # def update_dependence(self, key, value):
-    #     # self.dependence[f"{{{{{key}}}}}"] = value
-    #     dependence.update_dependence(key, value)
-
     def substitute_data(self, regex=None, keys=None, deps=None, jp_dict=None):
         """
         方法接收一个正则表达式 regex 和一个关联参数表 deps，用于从接口返回的数据中提取关联参数。
@@ -52,6 +47,7 @@ class DataExtractor:
         Returns:
 
         """
+
         if not isinstance(self.response, (dict, str, list)):
             return {}
         if regex and keys:
@@ -63,7 +59,6 @@ class DataExtractor:
         if jp_dict:
             self.substitute_jsonpath(jp_dict)
         dependence.set_dep(self.dependence)
-        # setattr(Dependence, "dependence", self.dependence)
         return self.dependence
 
     def substitute_regex(self, regex, keys):
@@ -82,10 +77,8 @@ class DataExtractor:
         for i, key in enumerate(keys):
             try:
                 dependence.update_dep(key, groups[i])
-                # self.update_dependence(key, groups[i])
             except:
                 dependence.update_dep(key, None)
-                # self.update_dependence(key, None)
 
     def substitute_route(self, route_str):
         deps_list = re.sub(f"[\r\n]+", "", route_str).split(";")

common/extractor/dependent_parameter.py

@@ -8,11 +8,12 @@
 @desc:
 """
 import json
-import re
 
 from common.dependence import Dependence
 from common.tools.logger import MyLog
 
+logger = MyLog()
+
 
 class DependentParameter:
     """
@@ -20,14 +21,18 @@ class DependentParameter:
        然后，它将替换后的字符串转化为字典并返回。如果找不到需要替换的参数，则直接返回原始字符串。
     """
 
+    #  PATTERN = re.compile(r"{{(.*?)}}")  # 预编译正则表达式
+    #     pattern = re.compile(r'({)')
+    # S = re.compile(r"{{(.*?)/(/)}}")
     PATTERN = getattr(Dependence, "PATTERN")
     pattern = getattr(Dependence, "pattern")
+    pattern_fun = getattr(Dependence, "pattern_fun")
 
     def __init__(self):
-        self.dependence = getattr(Dependence, "dependence")
+        self.dependence = Dependence.get_dep()
 
     def get_dependent_value(self, key):
-        return str(self.dependence[key]).replace("null", "None")
+        return self.dependence[key]
 
     def replace_dependent_parameter(self, jst):
         """
@@ -39,19 +44,29 @@ class DependentParameter:
         if not jst:
             return jst
         jst = json.dumps(jst) if isinstance(jst, (dict, list)) else jst
+        # 替换
         while self.PATTERN.search(jst):
-            key = self.PATTERN.search(jst).group()
-            if key in self.dependence.keys():
-                jst = jst.replace(key, self.get_dependent_value(key))
+            if self.pattern_fun.search(jst):
+                # 函数替换
+                key = self.pattern_fun.search(jst).group()
+                if key in self.dependence.keys():
+                    value_ = self.get_dependent_value(key)()
+                    jst = jst.replace(key, str(value_))
             else:
-                MyLog().my_log(f"key:{key},在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常\n")
-                break
+                key = self.PATTERN.search(jst).group()
+                # 字符串替换
+                if key in self.dependence.keys():
+                    jst = jst.replace(key, str(self.get_dependent_value(key)))
+                    logger.my_log(f"key:{key},替换成功：{self.get_dependent_value(key)}")
+                else:
+                    logger.my_log(f"key:{key},在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常\n")
+                    break
             jst = jst.replace("True", "true").replace("False", "false")
         if self.pattern.search(jst):
             try:
                 jst = json.loads(jst)
             except json.JSONDecodeError as e:
-                MyLog().my_log(f"{jst}:{e}")
+                logger.my_log(f"JSONDecodeError:{jst}:{e}")
 
         return jst
 
@@ -63,15 +78,20 @@ if __name__ == '__main__':
         "{{var_d}}": None,
         "{{var_e_1}}": True,
         "{{var_e_2}}": "bar",
-        # "{{var_f}}": ["baz", False],
-        # "{{var_g}}": {'g': 'gg', 'g1': 'gg', 'g2': 'gg2'}
+        "{{var_f}}": ["baz", False],
+        "{{var_g}}": {'g': 'gg', 'g1': 'gg', 'g2': 'gg2'}
     }
-    setattr(Dependence, "dependence", dps)
+    from common.comparator import loaders
+
+    Dependence.set_dep(dps)
+    loaders.set_bif_fun()
+    print(Dependence.get_dep())
     dat = {
         "a": "{{var_a}}",
         "b": {"c": "{{var_c}}", "d": "{{var_d}}", "e": ["{{var_e_1}}", "{{var_e_2}}"]},
         "f": "{{var_f}}",
-        "g": "{{var_g}}"
+        "g": "{{var_g}}",
+        "t": "{{get_timestamp()}}"
     }
-    t = DependentParameter().replace_dependent_parameter("")
+    t = DependentParameter().replace_dependent_parameter(dat)
     print(t)

common/tools/logger.py

@@ -90,7 +90,4 @@ if __name__ == '__main__':
     #     time.sleep(0.04)
     @MyLog().decorator_log
     def add():
-        raise
-
-
-    print(add())
+        raise

common/tools/req.py

@@ -25,16 +25,16 @@ def req(hosts, methods, url, **kwargs):
     # 关闭 https 警告信息
 
     urllib3.disable_warnings()
-    if re.match(r"https?", url):
-        url = url
-    else:
-        url = hosts + url
+    if not url:
+        raise "URL 不能为None"
+    url = url if re.match(r"https?", url) else hosts + url
     res = None
     data = kwargs.get("data", None)
     headers = kwargs.get("headers")
     if methods.lower() == 'post':
         try:
             if "application/x-www-form-urlencoded" in headers.values():
+
                 res = requests.post(url=url, data=data, headers=headers, verify=False, timeout=30)
             else:
                 res = requests.post(url=url, json=data, headers=headers, verify=False, timeout=30)

debug/debug.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python
+# encoding: utf-8
+"""
+@author: kira
+@contact: 262667641@qq.com
+@file: debug.py
+@time: 2023/3/29 11:40
+@desc:
+"""
+import re
+from common.dependence import Dependence
+from common.comparator import loaders
+
+lo = loaders.set_bif_fun()
+pa = Dependence.PATTERN
+di = Dependence.get_dep()
+patten = re.compile(r"\{\{(\w+\(\))\}\}")
+jst = '{{get_current_time()}}'
+ret = patten.search(jst)
+print(ret)
+if ret:
+    if jst in di.keys():
+        print(di.get(jst))
+        jst = jst.replace(jst, di.get(jst)())
+print(jst)

test_main.py

@@ -14,14 +14,13 @@ sys.path.append("./common")
 sys.path.append("./")
 
 from common.base_datas import BaseDates
-# from common.decorator_send_info import decorator_send_info
 from HTMLTestRunnerNew import HTMLTestRunner
 
 
 # @decorator_send_info()
 def run_test_case(case_name, url_key=None):
-    # test_report = BaseDates.test_report
-    test_report = r"E:\Program Project\api-test-project\scrit\apifox-reports"
+    test_report = BaseDates.test_report
+    print(test_report, BaseDates.script, case_name)
     t_case = unittest.defaultTestLoader.discover(BaseDates.script, pattern=f"{case_name}.py")
     with open(BaseDates.test_report + f"/{case_name} 测试报告.html", "wb") as fb:
         runner = HTMLTestRunner(stream=fb, verbosity=2, title=f"{case_name} 接口自动化测试报告",
@@ -29,22 +28,23 @@ def run_test_case(case_name, url_key=None):
         # 失败重跑
         # runner.run(test_case, 0, False)
         runner.run(t_case)
-        data = [url_key, test_report]
-    return data
+        # data = [url_key, test_report]
+    # return data
 
 
 def run():
-    keys = {"test_api": "8b1647d4-dc32-447c-b524-548acf18a938"  # 企業微信key
+    keys = {"test_": "8b1647d4-dc32-447c-b524-548acf18a938"  # 企業微信key
             }
     # 获取测试用例脚本文件夹下所有文件
     test_case_names = os.listdir(BaseDates.script)
+    print(test_case_names)
     for name in test_case_names:
-        if re.match(r"test_api.+?py", name):
-            test_case = re.match(r"test.+?py", name).group()
+        if re.match(r"test_.+?py", name):
+            test_case = re.match(r"test_.+?py", name).group()
             case = test_case.split(".")[0]
             for key, value in keys.items():
-                print(f"當前運行的用例：{key}; 對應的機器人key： {value}")
                 if key in case:
+                    print(f"當前運行的用例：{case}; 對應的機器人key： {value}")
                     run_test_case(case, value)
                 else:
                     print("No script to be executed")
@@ -52,4 +52,3 @@ def run():
 
 if __name__ == '__main__':
     run()
-

test_script/auto_script/get_init.py

@@ -22,8 +22,8 @@ def get_init():
     excel_handle = DoExcel(test_file)  # 实例化对象
     try:
         excel_handle.clear_date()  # 清空 excel 中实际结果列的数据
-        test_case = excel_handle.do_excel()  # 获取 excel 中的测试用例
-        # test_case = excel_handle.do_excel_yield()  # 获取 excel 中的测试用例
+        # test_case = excel_handle.do_excel()  # 获取 excel 中的测试用例
+        test_case = excel_handle.do_excel_yield()  # 获取 excel 中的测试用例
         init_data = excel_handle.get_excel_init()  # 获取初始化基本数据
         MyLog().my_log(f"如下是初始化得到得数据：{init_data}", "info")
     except Exception as e:

test_script/auto_script/login.py

@@ -13,9 +13,10 @@ def login(host, username, password):
     headers = {"Content-Type": "application/json"}
     try:
         res = req(host, 'POST', url, data=data, headers=headers).json()
-        MyLog().my_log(f"登录成功:{res}", "info")
-        token = res.get("token")
-        return {**headers, **{"token": token}}
+        MyLog().my_log(f"======================登录成功======================", "info")
+        token = res.get("data").get("bspToken")
+        current_tenant_id = res.get("data").get("currentTenantId")
+        return {**headers, **{"BSP_TOKEN": token, "BSP_USER_TENANT": current_tenant_id}}
     except:
         raise
 

test_script/auto_script/test_standard.py

@@ -8,15 +8,15 @@ from ddt import ddt, data
 
 sys.path.append("../../")
 sys.path.append("../../common")
-from .login import login
-from .get_init import get_init
+from test_script.auto_script.login import login
+from test_script.auto_script.get_init import get_init
 from common.extractor.dependent_parameter import DependentParameter
 from common.extractor.data_extractor import DataExtractor
 from common.encryption.do_encryption import do_encrypt
 from common.do_sql.do_mysql import DoMysql
 from common.tools.req import req
 from common.tools.logger import MyLog
-from common.comparator.loaders import load_built_in_functions
+from common.comparator import loaders
 from common.dependence import Dependence
 from common.comparator.validator import Validator
 
@@ -39,34 +39,32 @@ class TestProjectApi(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls) -> None:
+        loaders.set_bif_fun()  # 加载内置方法
         # 获取初始化基础数据
         cls.host = init_data.get('host')
         cls.path = init_data.get("path")
         username = dep.get_dep("{{account}}")
-        password = dep.get_dep("{{password}}")
+        password = dep.get_dep("{{passwd}}")
         cls.headers = login(cls.host + cls.path, username, password)
         dep.update_dep("headers", cls.headers)
         # 加载内置方法
-        for k, v in load_built_in_functions().items():
-            dep.update_dep(k, v)
-        print("----->", dep.get_dep())
+        logger.my_log(f"内置方法：{dep.get_dep()}", "info")
 
     def setUp(self) -> None:
         logger.my_log("-----------------------------------start_test_api-----------------------------------", "info")
 
-    # @logger.decorator_log()
     @data(*test_case)  # {"":""}
+    @logger.decorator_log()
     def test_api(self, item):  # item = {測試用例}
-        sheet = item.get("sheet_name")
+        sheet = item.get("sheet")
         item_id = item.get("Id")
         name = item.get("name")
         description = item.get("description")
         host = self.host
         path = self.path
         headers = self.headers
-        url = item.get("url")
+        url = item.get("Url")
         run = item.get("Run")
-        print("123111", run)
         method = item.get("Method")
         sql_variable = item.get("sql变量")
         sqlps = item.get("SQL")
@@ -85,19 +83,22 @@ class TestProjectApi(unittest.TestCase):
         if method == "TIME":
             try:
                 time.sleep(int(url))
+                logger.my_log(f"暂存成功:{url}", "info")
                 return
             except Exception as e:
                 MyLog().my_log(f'暂停时间必须是数字')
                 raise e
         # 首先执行sql替换,将sql替换为正确的sql语句
         sql = dep_par.replace_dependent_parameter(sqlps)
-        print("2334241234", sql)
         if method == "SQL":
             try:
                 execute_sql_results = mysql.do_mysql(sql)
+                logger.my_log(f'sql执行成功:{execute_sql_results}', "info")
                 if execute_sql_results and sql_variable:
                     # 执行sql数据提取
                     DataExtractor(execute_sql_results).substitute_data(jp_dict=sql_variable)
+                    logger.my_log(f'sql 提取成功', "info")
+                    return
             except Exception as e:
                 logger.my_log(f'sql:{sql},异常:{e}')
                 raise e
@@ -114,38 +115,38 @@ class TestProjectApi(unittest.TestCase):
         expected = dep_par.replace_dependent_parameter(expect)
         # 提取请求参数信息
         DataExtractor(parameters).substitute_data(deps=parameters_key)
-        # 执行参数替换
-        # 替换预期结果
         # 判断是否执行加密
         if encryption:
             parameters = do_encrypt(encryption, parameters)  # 数据加密：MD5 ｏｒ　ｓｈａ１
-        print("--URL-- ", url)
-        print("--HEADER-- ", headers)
-        print("--BODY-- ", parameters)
-        print("--SQL-- ", sql)
-        print("--SQL_RESULT-- ", sql_res)
-        print("--EXPECTED-- ", expected)
+        logger.my_log(f"--sheet-- {sheet}", "info")
+        logger.my_log(f"--URL--{host + path + url}", "info")
+        logger.my_log(f"--HEADER-- {headers}", "info")
+        logger.my_log(f"--BODY-- {parameters}", "info")
+        logger.my_log(f"--SQL-- {sql}", "info")
+        logger.my_log(f"--SQL_RESULT-- {sql_res}", "info")
+        logger.my_log(f"--EXPECTED-- {expected}", "info")
         try:
             # 执行请求操作
             response = req(host + path, method, url, data=parameters, headers=headers)
-            result_tuple = validator.run_validate(expected, response.json())  # 执行断言 返回结果元组
-            print("--RESPONSE-- ", response.text)
+            logger.my_log(f"--RESPONSE-- {response.text}", "info")
         except Exception as e:
             result = "失败"
-            logger.my_log(f'{item_id}-->{name}_{description},异常:{e}')
+            logger.my_log(f'{result}:{item_id}-->{name}_{description},异常:{e}')
             raise e
-        else:
-            if "FAIL" in result_tuple:
-                result = "失败"
-                raise
-            else:
-                result = "通过"
+
+        result_tuple = validator.run_validate(expected, response.json())  # 执行断言 返回结果元组
+        result = "PASS"
+        try:
+            self.assertNotIn("FAIL", result_tuple)
+        except:
+            result = "FAIL"
+            raise
         finally:
             # 响应结果及测试结果回写 excel
             excel_handle.write_back(
                 sheet_name=sheet,
                 i=item_id,
-                response_value=response,
+                response_value=response.text,
                 test_result=result,
                 assert_log=result_tuple
             )