refactor(cases,yaml_processor): 优化测试用例加载功能以及文件加载功能

- 优化用例加载模块器
- 优化yaml文件读取模块

commons/cases.py

@@ -16,11 +16,11 @@ import allure
 import pytest
 
 from commons import settings
-from commons.files import YamlFile
+from commons.file_processors.yaml_processor import YamlFile
 from commons.models import CaseInfo
 from commons.session import Session
 from commons.exchange import Exchange
-
+from utils import data_driver
 
 logger = logging.getLogger(__name__)
 
@@ -45,28 +45,47 @@ class TestAPI:
             logger.info(f"加载文件：{yaml_path}")
 
             file = YamlFile(yaml_path)  # 自动读取yaml文件
+
             case_info = CaseInfo(**file)  # 校验yaml格式
 
             logger.info(f"case_info={case_info.to_yaml()}")  # 把case_info 转成字符串，然后记录日志
             # case_info = {yaml_path.stem:case_info}
             # logger.info(f"case_info_dict={case_info}")
-            case_func = cls.new_case(case_info)  # 从yaml格式转换为pytest格式
+            case_func = cls.new_case(yaml_path.stem, file)  # 从yaml格式转换为pytest格式
             print(yaml_path.stem)
             setattr(cls, f"{yaml_path.stem}", case_func)  # 把pytest格式添加到类中
 
     @classmethod
-    def new_case(cls, case_info: CaseInfo):
+    def new_case(cls,file_name, case_info: dict):
-        ddt_data = case_info.ddt()
+        case_data = data_driver.DataDriver().generate_cases(file_name,case_info)
-        print(f"测试数据：{ddt_data}")
+        # ddt_data = case_info.ddt()
-
+        keys_list = ['test_1_user[0]', 'test_1_user[1]', 'test_1_user[2]', 'test_1_user[3]']
-        ddt_title = [data.title for data in ddt_data]
+        titles = ['查询用户信息', '查询用户信息', '查询用户信息', '查询用户信息']
-        logger.info(f"{ddt_title=}")
+        for item in case_data:
-
+            # 遍历列表中的每个字典
-        @allure.feature(case_info.feature)
+            for key, value in item.items():
-        @allure.story(case_info.story)
+                print(f"key:{key}")
-        @pytest.mark.parametrize("case_info", ddt_data, ids=ddt_title)
+                keys_list.append(key)
-        def test_func(self, case_info: CaseInfo):
+                print(f"value:{value}")
-            allure.dynamic.title(case_info.title)
+                #     # 遍历内层字典（这里内层字典其实只有一个键值对）
+                titles.append(value['title'])
+        print(f"测试数据：{case_data}")
+        item={'test_1_user[0]': {'feature': '特征', 'story': '事件', 'title': '查询用户信息', 'request': {'method': 'get', 'url': 'http://119.91.19.171:40065/answer/api/v1/connector/info', 'headers': {'Accept-Encoding': 'gzip, deflate', 'Accept-Language': 'zh_CN', 'Content-Type': 'application/json', 'Cookie': 'psession=33c6c2de-7e5d-40e2-9bbc-3c637a690c3f; lang=zh-CN; 3x-ui=MTcyNjU2NDcwOHxEWDhFQVFMX2dBQUJFQUVRQUFCMV80QUFBUVp6ZEhKcGJtY01EQUFLVEU5SFNVNWZWVk5GVWhoNExYVnBMMlJoZEdGaVlYTmxMMjF2WkdWc0xsVnpaWExfZ1FNQkFRUlZjMlZ5QWYtQ0FBRUVBUUpKWkFFRUFBRUlWWE5sY201aGJXVUJEQUFCQ0ZCaGMzTjNiM0prQVF3QUFRdE1iMmRwYmxObFkzSmxkQUVNQUFBQUdQLUNGUUVDQVFkNGRXa3lNREkwQVFkNGRXa3lNREkwQUE9PXwLOhLRIDjzvQ3oI-UF-GhkMheEENkxRJ8GkAZ79eFHvg==', 'Host': '119.91.19.171:40065', 'Origin': 'http://119.91.19.171:40065', 'Referer': 'http://119.91.19.171:40065/users/login', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML,like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0'}}, 'extract': {'code': ['json', '$.code', 0], 'msg': ['json', '$.msg', 0]}, 'validate': {'equals': {'状态码等于200': [200, 'code1']}, 'not_equals': {'状态码不等于404': [404, 'code1']}, 'contains': {'包含关系': [404, 'code1']}, 'not_contains': {'不包含关系': [404, 'code1']}}, 'parametrize': [['title', 'username', 'password', 'code'], ['测试1', 'user1', 'pass1', 'code1'], ['测试2', 'user2', 'pass2', 'code2'], ['测试3', 'user3', 'pass3', 'code3'], ['测试4', 'user4', 'pass4', 'code4']]}}
+        # ddt_title = [data.title for data in ddt_data]
+        # logger.info(f"{ddt_title=}")
+        logger.info(f"keys_list={keys_list}")
+        logger.info(f"titles={titles}")
+        logger.info(f"feature={case_info.get('feature')}")
+        logger.info(f"story={case_info.get('story')}")
+        @allure.feature(case_info.get("feature"))
+        @allure.story(case_info.get("story"))
+        @pytest.mark.parametrize("case_key", keys_list, ids=titles)
+        def test_func(self, case_key):
+            logger.info(f"case_key={case_key}")
+            item.get(case_key)
+            logger.info(f"========:{item}")
+            logger.info(f"========:{item.get(case_key)}")
+            # allure.dynamic.title(case_info.title)
 
             logger.info(f"用例开始执行：{case_info.title}".center(80, "="))
 

commons/exchange.py

@@ -19,7 +19,7 @@ import allure
 from commons.templates import Template
 import jsonpath
 
-from commons.files import YamlFile
+from commons.file_processors.yaml_processor import YamlFile
 from commons.models import CaseInfo
 
 logger = logging.getLogger(__name__)
@@ -85,7 +85,7 @@ if __name__ == '__main__':
 
     # print(mock_resp.text)
     # print(mock_resp.json())
-    exchanger = Exchange(r"D:\CNWei\CNW\InterfaceAutoTest\extract.yaml")
+    exchanger = Exchange(r"E:\PyP\InterfaceAutoTest\extract.yaml")
     exchanger.extract(mock_resp, "name", "json", '$.name', 0)
     exchanger.extract(mock_resp, "age", "json", '$.age', 0)
     exchanger.extract(mock_resp, "data", "json", '$.data', 0)

commons/file_processors/yaml_processor.py

@@ -14,7 +14,7 @@ from typing import Union
 from dataclasses import dataclass, asdict, field
 from pathlib import Path
 import yaml
-from base import BaseFileProcessor
+from commons.file_processors.base import BaseFileProcessor
 
 logger = logging.getLogger(__name__)
 
@@ -115,6 +115,41 @@ class YamlFile(BaseFileProcessor,dict):
             logger.error(f"保存 YAML 文件 {filepath} 时出错: {e}")
 
 
+class StringOrDict:
+    @classmethod
+    def to_string(cls, data: dict) -> str:
+        """
+        将字典 (自身) 转换为 YAML 格式的字符串。
+
+        Returns:
+            YAML 格式的字符串。
+        """
+        try:
+            return yaml.safe_dump(
+                dict(data),  # 使用dict转换为标准的字典
+                allow_unicode=True,
+                sort_keys=False,
+                default_flow_style=False
+            )
+        except TypeError as e:
+            logger.error(f"将数据转换为 YAML 字符串时出错: {e}")
+            return ""
+
+    @classmethod
+    def to_dict(cls, data: str) -> None:
+        """
+        将 YAML 格式的字符串转换为字典，并更新当前字典的内容.
+
+        Args:
+            data: YAML 格式的字符串。
+        """
+        try:
+            loaded_data = yaml.safe_load(data) or {}
+            return loaded_data
+        except yaml.YAMLError as e:
+            logger.error(f"将 YAML 字符串转换为字典时出错: {e}")
+
+
 if __name__ == '__main__':
     # 示例用法
     yaml_path = r'D:\CNWei\CNW\InterfaceAutoTest\TestCases\test_1_user.yaml'  # 你的 YAML 文件路径

utils/data_driver.py

@@ -9,29 +9,41 @@
 @date: 2025/3/3 10:56
 @desc: 
 """
+from pathlib import Path
 
 from commons.models import CaseInfo
 from commons.templates import Template
+from commons.file_processors.yaml_processor import StringOrDict
 
 
 class DataDriver:
     @staticmethod
-    def generate_cases(case_info) -> list:
+    def generate_cases(file_name, case_info) -> list:
-        if not case_info.get("parametrize"):
-            return [case_info]
 
-        cases = []
+        if not case_info.get("parametrize"):
+            return {file_name + "[--]": case_info}
+
+        # cases = []
         args_names = case_info.get("parametrize")[0]
-        for args_values in case_info.get("parametrize")[1:]:
+        for i, args_values in enumerate(case_info.get("parametrize")[1:]):
-            print(args_values)
+            # print(args_values)
             context = dict(zip(args_names, args_values))
-            print(context)
+            # print(context)
-            rendered = Template(CaseInfo(**case_info).to_yaml()).render(context)
+            # rendered = Template(CaseInfo(**case_info).to_yaml()).render(context)
-            cases.append({args_names[0]:CaseInfo(**case_info).by_yaml(rendered)})
+            rendered = Template(StringOrDict.to_string(case_info)).render(context)
-        return cases
+            # cases.append({file_name + "[" + str(i) + "]": StringOrDict.to_dict(rendered)})
+            yield {file_name + "[" + str(i) + "]": StringOrDict.to_dict(rendered)}
+        # return cases
 
 
 if __name__ == '__main__':
+    from commons.file_processors.yaml_processor import YamlFile
+
+    file_path = Path(r"E:\PyP\InterfaceAutoTest\TestCases\test_1_user.yaml")
+
+    file_obj = YamlFile(file_path)
+    # print(file_path.stem)
+    file_name = file_path.stem
     mock_case_info = {
         "case_info0": {
             "feature": "页面状态",
@@ -40,7 +52,8 @@ if __name__ == '__main__':
             "request": "",
             "extract": "",
             "validate": "",
-            "parametrize": [[ "title","username","password","msg" ] ,[ "测试1","user1","pass1","200" ] ,[ "测试2","user2","pass2","300" ]]
+            "parametrize": [["title", "username", "password", "msg"], ["测试1", "user1", "pass1", "200"],
+                            ["测试2", "user2", "pass2", "300"]]
         },
         "case_info1": {
             "feature": "页面状态",
@@ -64,5 +77,25 @@ if __name__ == '__main__':
     }
 
     dd = DataDriver()
-    cases = dd.generate_cases(mock_case_info.get("case_info0"))
+    # cases = dd.generate_cases(mock_case_info.get("case_info0"))
-    print(cases)
+    cases = dd.generate_cases(file_name, file_obj)
+    # print(cases)
+    # print(len(cases))
+    keys_list = []
+    titles = []
+    for item in cases:
+        print(item)
+        # 遍历列表中的每个字典
+        for key, value in item.items():
+            print(f"key:{key}")
+            keys_list.append(key)
+            print(f"value:{value}")
+        #     # 遍历内层字典（这里内层字典其实只有一个键值对）
+            titles.append(value['title'])
+        print(item)
+
+    print(keys_list)
+    print(titles)
+
+    # ddt_title = [data.title for data in ddt_data]
+    # logger.info(f"{ddt_title=}")