优化处理用例数据方法（从全局变量替换指定关键字以及识别Python表达式），增加递归方式处理，避免层层套娃情况下，无法处理成功

case_utils/assert_handle.py

@@ -8,7 +8,7 @@
 
 from case_utils.allure_handle import allure_step
 from requests import Response
-from common_utils.data_handle import json_extractor, re_extract
+from case_utils.data_handle import json_extractor, re_extract
 from loguru import logger
 from case_utils.request_data_handle import response_type
 from common_utils.mysql_handle import MysqlServer
@@ -30,8 +30,7 @@ def assert_response(response: Response, expected: dict) -> None:
                 f"-------------Start：响应断言--------------------\n")
     if expected is None:
         logger.info("当前用例无响应断言！")
-        allure_step(step_title='判断是否存在响应断言',
+        allure_step(step_title='判断是否存在响应断言---->当前用例无响应断言！')
-                    content='当前用例无响应断言！')
         return
     logger.debug(f"响应断言预期结果：{expected}, {type(expected)}")
     index = 0
@@ -49,8 +48,7 @@ def assert_response(response: Response, expected: dict) -> None:
                     actual = re_extract(response.text, _k)
             index += 1
             logger.info(f'第{index}个响应断言 -|- 预期结果: {_k}: {_v}, {type(_v)}   {k}   实际结果: {actual}, {type(actual)}')
-            allure_step(step_title=f'第{index}个响应断言数据',
+            allure_step(step_title=f'第{index}个响应断言数据---->预期结果: {_k}: {_v}, {type(_v)}   {k}   实际结果: {actual}, {type(actual)}')
-                        content=f'预期结果: {_k}: {_v}, {type(_v)}   {k}   实际结果: {actual}, {type(actual)}')
             try:
                 if k == "eq":  # 预期结果 = 实际结果
                     assert _v == actual
@@ -73,8 +71,7 @@ def assert_response(response: Response, expected: dict) -> None:
                                 content='目前仅支持如下关键字：eq, in, gt, lt, not')
             except AssertionError:
                 logger.error(f"第{index}个响应断言失败 -|- 预期结果: {_k}: {_v}, {type(_v)}   {k}   实际结果: {actual}, {type(actual)}")
-                allure_step(step_title=f'第{index}个响应断言失败',
+                allure_step(step_title=f'第{index}个响应断言失败---->预期结果: {_k}: {_v}, {type(_v)}   {k}   实际结果: {actual}, {type(actual)}')
-                            content=f'预期结果: {_k}: {_v}, {type(_v)}   {k}   实际结果: {actual}, {type(actual)}')
                 logger.info('\n-------------End：响应断言--------------------\n' \
                             "=====================================================")
                 raise AssertionError(
@@ -100,14 +97,12 @@ def assert_sql(db_info, expected: dict):
                 f"-------------Start：数据库断言--------------------")
     if expected is None:
         logger.info("当前用例无数据库断言！")
-        allure_step(step_title='判断是否存在数据库断言',
+        allure_step(step_title='判断是否存在数据库断言---->当前用例无数据库断言')
-                    content='当前用例无数据库断言')
         return
     if not db_info:
         logger.error("当前环境无数据库配置，跳过数据库断言！")
-        allure_step(step_title='判断是否存在数据库配置',
+        allure_step(step_title='判断是否存在数据库配置---->当前环境无数据库配置，跳过数据库断言！')
-                    content='当前环境无数据库配置，跳过数据库断言！}')
+        logger.info('\n-------------End：数据库断言--------------------\n'
-        logger.info('\n-------------End：数据库断言--------------------\n' \
                     "=====================================================")
         return
     for k, v in expected.items():
@@ -118,13 +113,13 @@ def assert_sql(db_info, expected: dict):
                     # 查询数据库，获取查询结果
                     sql_result = MysqlServer(**db_info).query_one(_v)
                     logger.info(f'数据库响应断言 -|- SQL：{_v} || 查询结果：{sql_result}')
-                    allure_step(step_title=f'数据库断言',
+                    allure_step(step_title=f'数据库断言---->SQL：{_v} ',
-                                content=f'SQL：{_v} || 查询结果：{sql_result}')
+                                content=f'查询结果：{sql_result}')
                 except Exception as e:
                     logger.error(f'数据库服务报错：{e}')
                     allure_step(step_title=f'数据库服务报错',
                                 content=f'{e} ')
-                    logger.info('\n-------------End：数据库断言--------------------\n' \
+                    logger.info('\n-------------End：数据库断言--------------------\n'
                                 "=====================================================")
                     raise AssertionError(f"数据库服务报错：{e}")
 
@@ -133,19 +128,16 @@ def assert_sql(db_info, expected: dict):
                     if _k == "len":
                         assert _v == len(sql_result)
                         logger.info(f"预期结果: {_v} ==  实际结果: {len(sql_result)}, 断言成功！")
-                        allure_step(step_title=f'---->数据库断言结果',
+                        allure_step(step_title=f'数据库断言结果---->预期结果: {_v} ==  实际结果: {len(sql_result)}, 断言成功！')
-                                    content=f'预期结果: {_v} ==  实际结果: {len(sql_result)}, 断言成功！')
                     # 如果时$.开头，则从数据库查询结果中提取相应的值作为实际结果
                     elif _k.startswith("$."):
                         actual = json_extractor(sql_result, _k)
                         assert _v == actual
                         logger.info(f"预期结果: {_v} ==  实际结果: {actual}, 断言成功！")
-                        allure_step(step_title=f'---->数据库断言结果',
+                        allure_step(step_title=f'数据库断言结果---->预期结果: {_v} ==  实际结果: {actual}, 断言成功！')
-                                    content=f'预期结果: {_v} ==  实际结果: {actual}, 断言成功！')
             except AssertionError:
                 logger.error(f"数据库断言失败 -|- 预期结果:{_v}  {k}   实际结果: {sql_result})")
-                allure_step(step_title=f'数据库断言失败',
+                allure_step(step_title=f'数据库断言失败---->预期结果:{_v}  {k}   实际结果: {sql_result}')
-                            content=f'预期结果:{_v}  {k}   实际结果: {sql_result}')
                 logger.info('\n-------------End：数据库断言--------------------\n' \
                             "=====================================================")
                 raise AssertionError(f"数据库断言失败 -|-预期结果: {_v}  {k}  实际结果: {sql_result}")

case_utils/data_handle.py

@@ -5,114 +5,42 @@
 # @File    : data_handle.py
 # @Software: PyCharm
 # @Desc: 数据处理
-from jsonpath import jsonpath
+
 import re
-from faker import Faker
 from string import Template
+from jsonpath import jsonpath
 from loguru import logger
+from faker import Faker
 
 faker = Faker()
 
 
-def data_replace(content, source):
+def data_handle(obj, source):
     """
-    用例数据替换的方法
+    递归处理字典、列表中的字符串，将${}占位符替换成source中的值
-    :param content: 原始的字符串内容
-    :param source: 需要替换字符串的来源
-    return content： 替换表达式后的字符串
     """
-    try:
+    obj = eval_data_process(obj)
-        if content is None:
+    if isinstance(obj, str):
-            return None
+        # 寻找${}， 在source中找到对应的关键字进行替换，如obj=${user_id}, 去寻找source中对应键user_id的值（假设user_id=1），使得obj=1
-        logger.debug("\n======================================================\n" \
+        obj = Template(obj).safe_substitute(source)
-                     "-------------Start：数据替换--------------------\n"
+        # 寻找${python表达式}， 将Python表达式eval得出其具体值
-                     f"替换源: {source}\n" \
+        for func in re.findall('\\${(.*?)}', obj):
-                     f"初始字符串: {content}\n" \
+            obj = obj.replace('${%s}' % func, eval_data_process(func))
-                     "=====================================================")
+            obj = eval_data_process(obj)
-        print(f"-----Start-----数据替换: 初始字符串为：{content}")
+            return obj
-        if len(content) != 0:
+        return eval_data_process(obj)
-            # safe_substitute() 方法会保留没有被替换的占位符，不会抛出 KeyError 异常。
+    elif isinstance(obj, list):
-            # 所以，如果 content 中不存在占位符，使用 safe_substitute() 方法进行替换后，得到的结果和原始字符串是一样的。
+        for index, item in enumerate(obj):
-            content = Template(str(content)).safe_substitute(source)
+            obj[index] = data_handle(item, source)
-            for func in re.findall('\\${(.*?)}', content):
+        return obj
-                content = content.replace('${%s}' % func, exec_func(func))
-                try:
-                    content = content.replace('${%s}' % func, exec_func(func))
-                except Exception as e:
-                    logger.error(f"\n======================================================\n" \
-                                 "-------------End：数据替换--------------------\n"
-                                 f"替换数据时出现了异常：{e}" \
-                                 "=====================================================")
-                    print(f"-----END-----替换数据时出现了异常：{e}")
-                    return e
-            logger.debug("\n======================================================\n" \
-                         "-------------End：数据替换--------------------\n"
-                         f"新字符串：{content}\n" \
-                         "=====================================================")
-            print(f"-----End-----数据替换完成: 新字符串为：{type(content)} || {content}\n")
-            return content
-    except Exception as e:
-        logger.error(f"进行数据替换时报错：{e}")
-        print(f"进行数据替换时报错：{e}")
 
+    elif isinstance(obj, dict):
+        for key, value in obj.items():
+            obj[key] = data_handle(value, source)
+        return obj
 
-def json_extractor(obj: dict, expr: str = '.'):
+    else:
-    """
+        return obj
-    :param obj :json/dict类型数据
-    :param expr: 表达式, . 提取字典所有内容， $.test_api_case 提取一级字典case， $.test_api_case.data 提取case字典下的data
-    :return result: 提取的结果，未提取到返回 None
-    """
-    try:
-        result = jsonpath(obj, expr)[0]
-        logger.debug("\n======================================================\n" \
-                     "-------------Start：json_extractor--------------------\n"
-                     f"提取表达式为： {expr} 提取值为 {result}\n" \
-                     "=====================================================")
-        print("提取响应内容成功，提取表达式为： {} 提取值为 {}".format(expr, result))
-    except Exception as e:
-        logger.debug("\n======================================================\n" \
-                     "-------------End：json_extractor--------------------\n"
-                     f"提取表达式为： {expr} 提取数据为 {obj}\n， 错误信息为：{e}\n" \
-                     "=====================================================")
-        print(f'未提取到内容，请检查表达式是否错误！提取表达式为：{expr} 提取数据为 {obj}， 错误信息为：{e}')
-        result = None
-    return result
-
-
-def re_extract(obj: str, expr: str = '.'):
-    """
-    :param obj : 字符串数据
-    :param expr: 正则表达式
-    :return result: 提取的结果，未提取到返回 None
-    """
-    try:
-        result = re.findall(expr, obj)[0]
-        logger.debug("\n======================================================\n" \
-                     "-------------Start：re_extract--------------------\n"
-                     f"提取表达式为： {expr} 提取值为： {result}\n" \
-                     "=====================================================")
-        print("提取响应内容成功，提取表达式为： {} 提取值为： {}".format(expr, result))
-    except Exception as e:
-        logger.debug(f"\n======================================================\n" \
-                     "-------------End：re_extract--------------------\n"
-                     f"提取表达式为： {expr} 提取数据为：{obj}\n， 错误信息为：{e}\n" \
-                     "=====================================================")
-        print(f'未提取到内容，请检查表达式是否错误！提取表达式为：{expr} 提取数据为 {obj}， 错误信息为：{e}')
-        result = None
-    return result
-
-
-def exec_func(func) -> str:
-    """
-    :params func 字符的形式调用函数
-    : return 返回的转换成函数执行的结果,已字符串格式返回
-    """
-    result = eval(func)
-    logger.debug("\n======================================================\n" \
-                 "-------------exec_func--------------------\n"
-                 f"函数： {func} 执行结果： {result}\n" \
-                 "=====================================================")
-    return str(result)
 
 
 # 将"[1,2,3]" 或者"{'k':'v'}" -> [1,2,3], {'k':'v'}
@@ -145,3 +73,55 @@ def eval_data_process(data):
         for key, value in data.items():
             data[key] = eval_data_process(eval_data(value))
     return data
+
+
+def json_extractor(obj: dict, expr: str = '.'):
+    """
+    :param obj :json/dict类型数据
+    :param expr: 表达式, . 提取字典所有内容， $.test_api_case 提取一级字典case， $.test_api_case.data 提取case字典下的data
+    :return result: 提取的结果，未提取到返回 None
+    """
+    try:
+        result = jsonpath(obj, expr)[0]
+        logger.debug("\n======================================================\n" \
+                     "-------------Start：json_extractor--------------------\n"
+                     f"提取表达式为： {expr} \n"
+                     f"提取值为： {result}\n"
+                     "=====================================================")
+        print("提取响应内容成功，提取表达式为： {} 提取值为 {}".format(expr, result))
+    except Exception as e:
+        logger.debug("\n======================================================\n" \
+                     "-------------End：json_extractor--------------------\n"
+                     f"提取表达式为： {expr}\n"
+                     f"提取数据为： {obj}\n"
+                     f"错误信息为：{e}\n"
+                     "=====================================================")
+        print(f'未提取到内容，请检查表达式是否错误！提取表达式为：{expr} 提取数据为 {obj}， 错误信息为：{e}')
+        result = None
+    return result
+
+
+def re_extract(obj: str, expr: str = '.'):
+    """
+    :param obj : 字符串数据
+    :param expr: 正则表达式
+    :return result: 提取的结果，未提取到返回 None
+    """
+    try:
+        result = re.findall(expr, obj)[0]
+        logger.debug("\n======================================================\n" \
+                     "-------------Start：re_extract--------------------\n"
+                     f"提取表达式为： {expr}\n" \
+                     f"提取值为： {result}\n" \
+                     "=====================================================")
+        print("提取响应内容成功，提取表达式为： {} 提取值为： {}".format(expr, result))
+    except Exception as e:
+        logger.debug(f"\n======================================================\n" \
+                     "-------------End：re_extract--------------------\n"
+                     f"提取表达式为： {expr}\n" \
+                     f"提取数据为： {obj}\n" \
+                     f"错误信息为：{e}\n" \
+                     "=====================================================")
+        print(f'未提取到内容，请检查表达式是否错误！提取表达式为：{expr} 提取数据为 {obj}， 错误信息为：{e}')
+        result = None
+    return result

case_utils/request_data_handle.py

@@ -8,11 +8,11 @@
 import json
 import os.path
 from common_utils.files_handle import get_file_field
-from common_utils.data_handle import eval_data_process, data_replace
+from case_utils.data_handle import eval_data_process, data_handle
 from config.global_vars import GLOBAL_VARS
 from requests import Response
 from loguru import logger
-from common_utils.data_handle import json_extractor, re_extract
+from case_utils.data_handle import json_extractor, re_extract
 from case_utils.allure_handle import allure_step
 from common_utils.base_request import BaseRequest
 from config.path_config import FILES_DIR
@@ -28,17 +28,17 @@ class RequestPreDataHandle:
     def __init__(self, request_data):
         logger.debug(f"\n======================================================\n" \
                      "-------------Start：处理用例数据前--------------------\n"
-                     f"用例标题: {type(request_data.get('title', None))} || {request_data.get('title', None)}\n" \
+                     f"用例标题(title): {type(request_data.get('title', None))} || {request_data.get('title', None)}\n" \
-                     f"请求路径: {type(request_data.get('url', None))} || {request_data.get('url', None)}\n" \
+                     f"请求路径(url): {type(request_data.get('url', None))} || {request_data.get('url', None)}\n" \
-                     f"请求方式: {type(request_data.get('method', None))} || {request_data.get('method', None)}\n" \
+                     f"请求方式(method): {type(request_data.get('method', None))} || {request_data.get('method', None)}\n" \
-                     f"请求头:   {type(request_data.get('headers', None))} || {request_data.get('headers', None)}\n" \
+                     f"请求头(headers):   {type(request_data.get('headers', None))} || {request_data.get('headers', None)}\n" \
                      f"请求cookies: {type(request_data.get('cookies', None))} || {request_data.get('cookies', None)}\n" \
-                     f"请求类型: {type(request_data.get('request_type', None))} || {request_data.get('request_type', None)}\n" \
+                     f"请求类型(request_type): {type(request_data.get('request_type', None))} || {request_data.get('request_type', None)}\n" \
-                     f"请求内容: {type(request_data.get('payload', None))} || {request_data.get('payload', None)}\n" \
+                     f"请求参数(payload): {type(request_data.get('payload', None))} || {request_data.get('payload', None)}\n" \
-                     f"请求文件: {type(request_data.get('files', None))} || {request_data.get('files', None)}\n" \
+                     f"请求文件(files): {type(request_data.get('files', None))} || {request_data.get('files', None)}\n" \
-                     f"后置提取参数: {type(request_data.get('extract', None))} || {request_data.get('extract', None)}\n" \
+                     f"后置提取参数(extract): {type(request_data.get('extract', None))} || {request_data.get('extract', None)}\n" \
-                     f"响应断言: {type(request_data.get('assert_response', None))} || {request_data.get('assert_response', None)}\n" \
+                     f"响应断言(assert_response): {type(request_data.get('assert_response', None))} || {request_data.get('assert_response', None)}\n" \
-                     f"数据库断言: {type(request_data.get('assert_sql', None))} || {request_data.get('assert_sql', None)}\n" \
+                     f"数据库断言(assert_sql): {type(request_data.get('assert_sql', None))} || {request_data.get('assert_sql', None)}\n" \
                      "=====================================================")
         self.request_data = request_data
 
@@ -56,17 +56,17 @@ class RequestPreDataHandle:
         self.assert_handle()
         logger.debug(f"\n======================================================\n" \
                      "-------------End：处理用例数据后--------------------\n"
-                     f"用例标题:  {type(self.request_data.get('title', None))} || {self.request_data.get('title', None)}\n" \
+                     f"用例标题(title):  {type(self.request_data.get('title', None))} || {self.request_data.get('title', None)}\n" \
-                     f"请求路径: {type(self.request_data.get('url', None))} || {self.request_data.get('url', None)}\n" \
+                     f"请求路径(url): {type(self.request_data.get('url', None))} || {self.request_data.get('url', None)}\n" \
-                     f"请求方式: {type(self.request_data.get('method', None))} || {self.request_data.get('method', None)}\n" \
+                     f"请求方式(method): {type(self.request_data.get('method', None))} || {self.request_data.get('method', None)}\n" \
-                     f"请求头:   {type(self.request_data.get('headers', None))} || {self.request_data.get('headers', None)}\n" \
+                     f"请求头(headers):   {type(self.request_data.get('headers', None))} || {self.request_data.get('headers', None)}\n" \
                      f"请求cookies: {type(self.request_data.get('cookies', None))} || {self.request_data.get('cookies', None)}\n" \
-                     f"请求类型: {type(self.request_data.get('request_type', None))} || {self.request_data.get('request_type', None)}\n" \
+                     f"请求类型(request_type): {type(self.request_data.get('request_type', None))} || {self.request_data.get('request_type', None)}\n" \
-                     f"请求内容: {type(self.request_data.get('payload', None))} || {self.request_data.get('payload', None)}\n" \
+                     f"请求参数(payload): {type(self.request_data.get('payload', None))} || {self.request_data.get('payload', None)}\n" \
-                     f"请求文件: {type(self.request_data.get('files', None))} || {self.request_data.get('files', None)}\n" \
+                     f"请求文件(files): {type(self.request_data.get('files', None))} || {self.request_data.get('files', None)}\n" \
-                     f"后置提取参数: {type(self.request_data.get('extract', None))} || {self.request_data.get('extract', None)}\n" \
+                     f"后置提取参数(extract): {type(self.request_data.get('extract', None))} || {self.request_data.get('extract', None)}\n" \
-                     f"响应断言: {type(self.request_data.get('assert_response', None))} || {self.request_data.get('assert_response', None)}\n" \
+                     f"响应断言(assert_response): {type(self.request_data.get('assert_response', None))} || {self.request_data.get('assert_response', None)}\n" \
-                     f"数据库断言: {type(self.request_data.get('assert_sql', None))} || {self.request_data.get('assert_sql', None)}\n" \
+                     f"数据库断言(assert_sql): {type(self.request_data.get('assert_sql', None))} || {self.request_data.get('assert_sql', None)}\n" \
                      "=====================================================")
         return self.request_data
 
@@ -76,7 +76,7 @@ class RequestPreDataHandle:
             用例数据中获取到的url(一般是不带host的，个别特殊的带有host，则不进行处理)
             """
             # 检测url中是否存在需要替换的参数，如果存在则进行替换
-            data_replace(content=self.request_data.get("url", None), source=GLOBAL_VARS)
+            data_handle(obj=self.request_data.get("url", None), source=GLOBAL_VARS)
             # 进行url处理，最终得到full_url
             host = GLOBAL_VARS.get("host", "")
             url = self.request_data.get("url", "")
@@ -115,7 +115,7 @@ class RequestPreDataHandle:
             # 从用例数据中获取cookies， 处理cookies
             if cookies:
                 # 通过全局变量替换cookies，得到的是一个str类型
-                cookies = data_replace(content=cookies, source=GLOBAL_VARS)
+                cookies = data_handle(obj=cookies, source=GLOBAL_VARS)
                 if isinstance(cookies, str):
                     # 如果是字符串类型，就转成字典
                     self.request_data["cookies"] = json.loads(cookies)
@@ -133,8 +133,7 @@ class RequestPreDataHandle:
         try:
             # 从用例数据中获取header， 处理header
             if headers:
-                self.request_data["headers"] = eval_data_process(
+                self.request_data["headers"] = data_handle(obj=headers, source=GLOBAL_VARS)
-                    data_replace(content=headers, source=GLOBAL_VARS))
                 # 如果请求头中有cookies，需要进行单独处理
                 if self.request_data["headers"].get("cookies", None):
                     cookies = self.request_data["headers"]["cookies"]
@@ -152,8 +151,7 @@ class RequestPreDataHandle:
         payload = self.request_data.get("payload", None)
         try:
             if payload:
-                self.request_data["payload"] = eval_data_process(
+                self.request_data["payload"] = data_handle(obj=payload, source=GLOBAL_VARS)
-                    data_replace(content=payload, source=GLOBAL_VARS))
         except Exception as e:
             logger.error(f"处理{payload}报错了：{e}")
             print(f"处理{payload}报错了：{e}")
@@ -207,8 +205,7 @@ class RequestPreDataHandle:
         assert_response = self.request_data.get("assert_response", None)
         try:
             if assert_response:
-                self.request_data["assert_response"] = eval_data_process(
+                self.request_data["assert_response"] = data_handle(obj=assert_response, source=GLOBAL_VARS)
-                    data_replace(content=assert_response, source=GLOBAL_VARS))
             # 由于数据库断言里面的变量需要请求响应后进行提取，因此目前不进行处理
         except Exception as e:
             logger.error(f"处理{assert_response}报错了：{e}")
@@ -231,30 +228,43 @@ class RequestHandle:
         response = BaseRequest.send_request(self.case_data)
         # 处理数据库断言 - 从全局变量中获取最新值，替换数据库断言中的参数
         if self.case_data.get('assert_sql', None):
-            self.case_data["assert_sql"] = eval_data_process(
+            self.case_data["assert_sql"] = data_handle(obj=self.case_data["assert_sql"], source=GLOBAL_VARS)
-                data_replace(content=self.case_data["assert_sql"], source=GLOBAL_VARS))
+        logger.info(f"\n======================================================\n"
-        logger.info(f"\n======================================================\n" \
+                    "-------------执行请求获取响应数据--------------------\n"
-                    "-------------请求数据--------------------\n"
+                    f"用例标题(title): {type(self.case_data.get('title', None))} || {self.case_data.get('title', None)}\n"
-                    f"用例标题: {type(self.case_data.get('title', None))} || {self.case_data.get('title', None)}\n" \
+                    f"请求路径(url): {type(self.case_data.get('url', None))} || {self.case_data.get('url', None)}\n"
-                    f"请求路径: {type(self.case_data.get('url', None))} || {self.case_data.get('url', None)}\n" \
+                    f"请求方式(method): {type(self.case_data.get('method', None))} || {self.case_data.get('method', None)}\n"
-                    f"请求方式: {type(self.case_data.get('method', None))} || {self.case_data.get('method', None)}\n" \
+                    f"请求头(headers):   {type(self.case_data.get('headers', None))} || {self.case_data.get('headers', None)}\n"
-                    f"请求头:   {type(self.case_data.get('headers', None))} || {self.case_data.get('headers', None)}\n" \
+                    f"请求cookies: {type(self.case_data.get('cookies', None))} || {self.case_data.get('cookies', None)}\n"
-                    f"请求cookies: {type(self.case_data.get('cookies', None))} || {self.case_data.get('cookies', None)}\n" \
+                    f"请求类型(request_type): {type(self.case_data.get('request_type', None))} || {self.case_data.get('request_type', None)}\n"
-                    f"请求类型: {type(self.case_data.get('request_type', None))} || {self.case_data.get('request_type', None)}\n" \
+                    f"请求参数(payload): {type(self.case_data.get('payload', None))} || {self.case_data.get('payload', None)}\n")
-                    f"请求内容: {type(self.case_data.get('payload', None))} || {self.case_data.get('payload', None)}\n" \
+        allure_step(step_title=f"请求地址(url):{self.case_data['url']}")
-                    f"请求文件: {type(self.case_data.get('files', None))} || {self.case_data.get('files', None)}\n" \
+        allure_step(step_title=f"请求方式(method)：{self.case_data['method']}")
-                    f"请求响应数据: {response.text}\n" \
+        allure_step(step_title="请求头(headers)", content=self.case_data['headers'])
-                    f"请求响应码: {response.status_code}\n" \
-                    f"响应耗时: {round(response.elapsed.total_seconds(), 2)} s || {round(response.elapsed.total_seconds() * 1000, 2)} ms\n" \
-                    "=====================================================")
-        allure_step(step_title=f"请求地址:{self.case_data['url']}")
-        allure_step(step_title=f"请求方式：{self.case_data['method']}")
-        allure_step(step_title="请求头", content=self.case_data['headers'])
         allure_step(step_title="请求Cookies", content=str(self.case_data['cookies']))
-        allure_step(step_title="请求参数", content=self.case_data['payload'])
+        allure_step(step_title="请求参数(payload)", content=self.case_data['payload'])
-        allure_step(step_title="请求文件", content=self.case_data['files'])
+        # 处理请求里面的files，使得日志以及allure中写入的是文件，而不是文件二进制内容
-        allure_step(step_title="请求响应数据", content=response.text)
+        if self.case_data.get('files', None):
-        allure_step(step_title=f"请求响应码:{response.status_code}")
+            files = self.case_data["files"]
+            if isinstance(files, list):
+                for file in files:
+                    _file = os.path.join(FILES_DIR, file[1][0])
+                    logger.info(
+                        f"\n请求文件(files): {type(_file)} || {_file}\n")
+                    allure_step(step_title="请求文件(files)", source=_file)
+            elif isinstance(files, dict):
+                dict_values = list(files.values())[0]
+                _file = os.path.join(FILES_DIR, dict_values[0])
+                logger.info(
+                    f"\n请求文件(files): {type(_file)} || {_file}\n")
+                allure_step(step_title="请求文件(files)", source=_file)
+        logger.info(
+            f"\n请求响应数据(response): {response.text}\n"
+            f"请求响应码(code): {response.status_code}\n"
+            f"响应耗时: {round(response.elapsed.total_seconds(), 2)} s || {round(response.elapsed.total_seconds() * 1000, 2)} ms\n"
+            "=====================================================")
+        allure_step(step_title="请求响应数据(response)", content=response.text)
+        allure_step(step_title=f"请求响应码(code):{response.status_code}")
         allure_step(
             step_title=f"响应耗时:{round(response.elapsed.total_seconds(), 2)} s || {round(response.elapsed.total_seconds() * 1000, 2)} ms")
         return response
@@ -308,7 +318,3 @@ def response_type(response: Response) -> str:
         return "json"
     except:
         return "str"
-
-if __name__ == '__main__':
-    test = RequestPreDataHandle(request_data={"files": {"file": ["导入TOC订单.xls", "toc.xls"]}})
-    test.files_handle()

case_utils/send_result_handle.py

@@ -12,7 +12,7 @@ from config.settings import SEND_RESULT_TYPE, email, ding_talk, wechat, email_su
     ding_talk_content, wechat_content
 from common_utils.dingding_handle import DingTalkBot
 from common_utils.wechat_handle import WechatBot
-from common_utils.data_handle import data_replace
+from case_utils.data_handle import data_handle
 from case_utils.get_results_handle import get_test_results_from_pytest_html_report, \
     get_test_results_from_from_allure_report
 
@@ -125,7 +125,7 @@ def send_result(report_path, report_type="allure", attachment_path=None):
     if SEND_RESULT_TYPE in notification_mappings:
         notification = notification_mappings[SEND_RESULT_TYPE]
         # 获取消息内容并替换
-        notification['sender_args']['content'] = data_replace(notification['sender_args']['content'],
+        notification['sender_args']['content'] = data_handle(notification['sender_args']['content'],
                                                               source=results)
         # 获取消息发送函数
         sender = notification['sender']
@@ -138,7 +138,7 @@ def send_result(report_path, report_type="allure", attachment_path=None):
         # 遍历所有消息发送方式
         for notification in notification_mappings.values():
             # 获取消息内容并替换
-            notification['sender_args']['content'] = data_replace(notification['sender_args']['content'],
+            notification['sender_args']['content'] = data_handle(notification['sender_args']['content'],
                                                                   source=results)
             # 获取消息发送函数
             sender = notification['sender']

data/gitlink/login_demo.yaml

@@ -7,7 +7,7 @@ case_common:
 case_login_demo_01:
   feature: 登录
   title: 用户名密码正确，登录成功(不校验数据库)
-  run: True
+  run: False
   url: /api/accounts/login.json
   method: POST
   headers: {"Content-Type": "application/json; charset=utf-8;"}

data/gitlink/test_new_project_demo.yaml

@@ -34,7 +34,7 @@ case_new_project_demo_01:
 case_new_project_demo_02:
   feature: 新建项目
   title: 正确输入各项必填参数，新建项目成功（不校验数据库，单独传cookies）
-  run: True
+  run: False
   url: /api/projects.json
   method: POST
   headers:

data/gitlink/test_upload_files.yaml

@@ -1,6 +1,6 @@
 # 公共参数
 case_common:
-  allure_epic: GitLink接口（手动编写用例）
+  allure_epic: GitLink接口(自动生成用例)
   allure_feature: 上传文件模块
   allure_story: 上传文件
 

pytest.ini

@@ -1,6 +1,6 @@
 [pytest]
 # cmd params
-addopts = -s --cache-clear --capture=sys --self-contained-html --reruns=0 --reruns-delay=5
+addopts = -s --cache-clear --capture=sys --self-contained-html --reruns=0 --reruns-delay=0
 disable_test_id_escaping_and_forfeit_all_rights_to_community_support = True
 markers =
     test_login_demo: login

test_case/test_manual_case/test_login_demo.py

@@ -21,7 +21,7 @@ from config.settings import db_info
 from config.global_vars import GLOBAL_VARS
 
 # 读取用例数据
-yaml_data = YamlHandle(filename=os.path.join(DATA_DIR, "login_demo.yaml")).read_yaml
+yaml_data = YamlHandle(filename=os.path.join(DATA_DIR, "gitlink", "login_demo.yaml")).read_yaml
 case_common = yaml_data["case_common"]
 cases = []
 for k, v in yaml_data.items():