feat,fix(): 优化项目

- 优化yaml_processor（优化文件类型转换逻辑）
- 修复bug

commons/cases.py

@@ -16,11 +16,11 @@ import allure
 import pytest
 
 from commons import settings
-from commons.file_processors.yaml_processor import YamlFile
+from commons.file_processors.yaml_processor import YamlProcessor
 from commons.models import CaseInfo
 from commons.session import Session
 from commons.exchange import Exchange
-from utils import data_driver
+from utils import data_driver, case_validator
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +44,7 @@ class TestAPI:
         for yaml_path in yaml_path_list:
             logger.info(f"加载文件：{yaml_path}")
 
-            file = YamlFile(yaml_path)  # 自动读取yaml文件
+            file = YamlProcessor(yaml_path)  # 自动读取yaml文件
 
             case_info = CaseInfo(**file)  # 校验yaml格式
 
@@ -59,8 +59,10 @@ class TestAPI:
     def new_case(cls,file_name, case_info: dict):
         case_data = data_driver.DataDriver().generate_cases(file_name,case_info)
         # ddt_data = case_info.ddt()
-        keys_list = ['test_1_user[0]', 'test_1_user[1]', 'test_1_user[2]', 'test_1_user[3]']
+        # keys_list = ['test_1_user[0]', 'test_1_user[1]', 'test_1_user[2]', 'test_1_user[3]']
-        titles = ['查询用户信息', '查询用户信息', '查询用户信息', '查询用户信息']
+        # titles = ['查询用户信息', '查询用户信息', '查询用户信息', '查询用户信息']
+        keys_list = []
+        titles = []
         for item in case_data:
             # 遍历列表中的每个字典
             for key, value in item.items():
@@ -70,7 +72,7 @@ class TestAPI:
                 #     # 遍历内层字典（这里内层字典其实只有一个键值对）
                 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']]}}
+        # 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}")
@@ -99,17 +101,19 @@ class TestAPI:
 
             logger.info(f"3，正在提取变量...")
             # 2，保存变量(接口关联)
-            new_case_info = CaseInfo(**new_case_info)
+            # new_case_info = CaseInfo(**new_case_info)
-            for var_name, extract_info in new_case_info.extract.items():
+            for var_name, extract_info in new_case_info.get("extract").items():
-                # logger.info(f"保存变量：{var_name}{extract_info}")
+                logger.info(f"保存变量：{var_name}{extract_info}")
                 exchanger.extract(resp, var_name, *extract_info)
             # 3，断言
             logger.info(f"4，正在断言...")
             assert_case_info = exchanger.replace(case_info)  # 为断言加载变量
             # logger.info(f"替换变量后：{assert_case_info}")
-            assert_case_info.assert_all()  # 执行断言
+            # assert_case_info.assert_all()  # 执行断言
+            _validator = case_validator.CaseValidator()
+            _validator.assert_all(assert_case_info.get("validate"))
 
-            logger.info(f"用例执行结束：{case_info.title}".center(80, "="))
+            logger.info(f"用例执行结束：{case_info.get('title')}".center(80, "="))
 
         return test_func
 

commons/exchange.py

@@ -19,7 +19,7 @@ import allure
 from commons.templates import Template
 import jsonpath
 
-from commons.file_processors.yaml_processor import YamlFile,StringOrDict
+from commons.file_processors.yaml_processor import YamlProcessor, StringOrDict
 from commons.models import CaseInfo
 
 logger = logging.getLogger(__name__)
@@ -27,7 +27,7 @@ logger = logging.getLogger(__name__)
 
 class Exchange:
     def __init__(self, path):
-        self.file = YamlFile(path)
+        self.file = YamlProcessor(path)
 
     @allure.step("提取变量")
     def extract(self, resp, var_name, attr, expr: str, index):
@@ -72,18 +72,19 @@ class Exchange:
     #     new_case_info = case_info.by_yaml(case_info_str)
     #     return new_case_info
     @allure.step("替换变量")
-    def replace(self, case_info: dict):
+    def replace(self, case_info: dict) -> dict:
         logger.info(case_info)
         # 1，将case_info转换为字符串
-        case_info_str = StringOrDict().to_string(case_info)
+        case_info_str = YamlProcessor.to_string(case_info)
         print(f"{case_info_str=}")
         # 2，替换字符串
         case_info_str = Template(case_info_str).render(self.file)
         print(f"{case_info_str=}")
         # 3，将字符串转换成case_info
-        new_case_info =  StringOrDict().to_dict(case_info_str)
+        new_case_info = YamlProcessor.to_dict(case_info_str)
         return new_case_info
 
+
 if __name__ == '__main__':
     class MockResponse:
         text = '{"name":"张三","age":"18","data":[3,4,5],"aaa":null}'
@@ -96,19 +97,29 @@ if __name__ == '__main__':
 
     # print(mock_resp.text)
     # print(mock_resp.json())
-    exchanger = Exchange(r"E:\PyP\InterfaceAutoTest\extract.yaml")
+    exchanger = Exchange(r"D:\CNWei\CNW\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)
     exchanger.extract(mock_resp, "aaa", "json", '$.aaa', 0)
-    mock_case_info = CaseInfo(
+    # mock_case_info = CaseInfo(
-        title="单元测试",
+    #     title="单元测试",
-        request={
+    #     request={
+    #         "data":
+    #             {"name": "${name}", "age": "${str(age)}", "time": "${add(1,2)}"}
+    #     },
+    #     extract={},
+    #     validate={}
+    # )
+    mock_case_info = {
+        "title": "单元测试",
+        "request": {
             "data":
                 {"name": "${name}", "age": "${str(age)}", "time": "${add(1,2)}"}
         },
-        extract={},
+        "extract": {},
-        validate={}
+        "validate": {}
-    )
+    }
+
     new_mock_case_info = exchanger.replace(mock_case_info)
     print(new_mock_case_info)

commons/file_processors/base.py

@@ -23,13 +23,15 @@ class BaseFileProcessor(abc.ABC):  # 使用 abc 模块定义抽象基类
         """加载."""
         pass
 
+    @staticmethod
     @abc.abstractmethod
-    def to_string(self) -> str:
+    def to_string(data: dict) -> str:
         """将文件内容转换为字符串。"""
         pass
 
+    @staticmethod
     @abc.abstractmethod
-    def to_dict(self, data: str) -> dict:
+    def to_dict(data: str) -> dict:
         """将文件内容转换为字典。"""
         pass
 

commons/file_processors/file_handle.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei
+@Software: PyCharm
+@contact: t6i888@163.com
+@file: file_handle
+@date: 2025/3/7 09:31
+@desc: 
+"""
+from pathlib import Path
+from typing import Union
+
+from yaml_processor import YamlProcessor
+from json_processor import JsonProcessor
+
+class FileHandle:
+    def __init__(self, filepath: Union[str, Path], data: Union[dict, None] = None):
+        # self.filepath: Path = Path(filepath)  # 确保 filepath 是 Path 对象
+        # self.data: Union[dict, None] = data
+        self.processor = get_processor(filepath, data)
+
+    def load(self) -> None:
+        self.processor.load()
+
+    def to_string(self) -> str:
+        return self.processor.to_string()
+
+    def to_dict(self, data: str) -> None:
+        self.processor.to_dict(data)
+
+    def save(self, new_filepath: Union[str, Path, None] = None):
+        self.processor.save(new_filepath)
+
+
+def get_processor(filepath, data):
+    ext = Path(filepath).suffix.lower()[1:]  # 获取后缀名，如 'json'
+    processors = {
+        'yaml': YamlProcessor,
+        'yml': YamlProcessor,
+        'json': JsonProcessor,
+
+    }
+    agent_model = processors.get(ext, YamlProcessor)  # 代理模式
+    return agent_model(filepath, data)  # 默认回退到 Yaml

commons/file_processors/json_processor.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei
+@Software: PyCharm
+@contact: t6i888@163.com
+@file: yaml_processor
+@date: 2025/3/4 17:28
+@desc: 
+"""
+import logging
+from typing import Union
+from dataclasses import dataclass, asdict, field
+from pathlib import Path
+import yaml
+from commons.file_processors.base import BaseFileProcessor
+
+logger = logging.getLogger(__name__)
+
+
+class JsonProcessor(BaseFileProcessor, dict):
+    """
+    用于处理 YAML 文件的类，继承自 dict。
+    提供了从文件加载、保存到文件、转换为字符串和从字符串转换的功能,
+    并可以直接像字典一样访问 YAML 数据。
+    """
+
+    def __init__(self, filepath: Union[str, Path], data: Union[dict, None] = None):
+        """
+        初始化 YamlFile 对象。
+
+        Args:
+            filepath: YAML 文件的路径 (可以是字符串或 pathlib.Path 对象).
+            data: 可选的初始数据字典。如果提供，则用该字典初始化 YamlFile。
+                  如果不提供，则尝试从 filepath 加载数据。
+        """
+        super().__init__()  # 初始化父类 dict
+        self.filepath: Path = Path(filepath)  # 确保 filepath 是 Path 对象
+        if data is not None:
+            self.update(data)  # 如果提供了初始数据，则更新字典
+        else:
+            self.load()  # 否则，尝试从文件加载
+
+    def load(self) -> None:
+        """
+        从 YAML 文件加载数据并更新字典。
+        如果文件不存在或加载失败，则清空字典并记录警告/错误。
+        """
+        self.clear()  # 清空现有数据
+        if self.filepath.exists():
+            try:
+                with open(self.filepath, "r", encoding="utf-8") as f:
+                    loaded_data = yaml.safe_load(f) or {}
+                    self.update(loaded_data)  # 使用加载的数据更新字典
+            except yaml.YAMLError as e:
+                logger.error(f"加载 YAML 文件 {self.filepath} 时出错: {e}")
+                # 保持字典为空 (已在开头 clear)
+        else:
+            logger.warning(f"文件 {self.filepath} 不存在, 字典保持为空.")
+            # 保持字典为空 (已在开头 clear)
+
+    def to_string(self) -> str:
+        """
+        将字典 (自身) 转换为 YAML 格式的字符串。
+
+        Returns:
+            YAML 格式的字符串。
+        """
+        try:
+            return yaml.safe_dump(
+                dict(self),  # 使用dict转换为标准的字典
+                allow_unicode=True,
+                sort_keys=False,
+                default_flow_style=False
+            )
+        except TypeError as e:
+            logger.error(f"将数据转换为 YAML 字符串时出错: {e}")
+            return ""
+
+    def to_dict(self, data: str) -> None:
+        """
+        将 YAML 格式的字符串转换为字典，并更新当前字典的内容.
+
+        Args:
+            data: YAML 格式的字符串。
+        """
+        try:
+            loaded_data = yaml.safe_load(data) or {}
+            self.clear()
+            self.update(loaded_data)  # 清空并更新
+        except yaml.YAMLError as e:
+            logger.error(f"将 YAML 字符串转换为字典时出错: {e}")
+            self.clear()  # 出错时也清空
+
+    def save(self, new_filepath: Union[str, Path, None] = None):
+        """
+        将字典数据 (自身) 保存到 YAML 文件。
+
+        Args:
+            new_filepath: 可选参数，指定新的文件路径。如果为 None，则覆盖原文件。
+        """
+        filepath = Path(new_filepath) if new_filepath else self.filepath
+
+        try:
+            with open(filepath, "w", encoding="utf-8") as f:
+                yaml.safe_dump(
+                    dict(self),  # 使用dict转换为标准的字典
+                    stream=f,
+                    allow_unicode=True,
+                    sort_keys=False,
+                    default_flow_style=False
+                )
+        except (TypeError, OSError) as e:
+            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) -> Union[None, dict]:
+        """
+        将 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 文件路径
+    yaml_file = YamlFile(yaml_path)
+    print(yaml_file)
+    print(type(yaml_file))
+
+    # # 直接像字典一样访问数据
+    # print("加载的数据:", yaml_file)  # 直接打印对象，就是打印字典内容
+    # print("title:", yaml_file.get("title"))  # 使用 get 方法
+    # if "title" in yaml_file:  # 使用 in 检查键
+    #     print("原始title:", yaml_file["title"])  # 使用方括号访问
+    #     yaml_file["title"] = "新的标题"  # 使用方括号修改
+    #     print("修改后的title:", yaml_file["title"])
+    # #
+    # yaml_file["new_key"] = "new_value"  # 添加新的键值对
+    #
+    # # 将字典转换为 YAML 字符串
+    # yaml_string = yaml_file.to_string()
+    # print("\nYAML 字符串:", yaml_string)
+    # #
+    # # 将 YAML 字符串转换回字典 (并更新 yaml_file)
+    # yaml_file.to_dict(yaml_string)
+    # print("\n从字符串加载的数据:", yaml_file)
+    #
+    # # 保存修改后的数据 (覆盖原文件)
+    # yaml_file.save()
+    #
+    # # 保存到新文件
+    # new_yaml_path = r'D:\CNWei\CNW\InterfaceAutoTest\TestCases\test_1_user_new.yaml'
+    # yaml_file.save(new_filepath=new_yaml_path)
+
+    # 测试从字符串初始化
+    # yaml_string2 = """
+    # name: Test User
+    # age: 30
+    # """
+
+    # yaml_file2 = YamlFile("test2.yaml", data=yaml.safe_load(yaml_string2))  # 从字符串初始化
+    # print("\n从字符串初始化的 YamlFile:", yaml_file2)
+    # yaml_file2.save()  # 保存到 test2.yaml
+    #
+    # 测试文件不存在的情形
+    # non_existent_file = YamlFile("non_existent_file.yaml")
+    # print("\n加载不存在的文件:", non_existent_file)  # 应该打印空字典 {}
+    # non_existent_file['a'] = 1  # 可以直接添加
+    # print("\n加载不存在的文件:", non_existent_file)
+
+# if __name__ == '__main__':
+#     from commons.models import CaseInfo
+#
+#     yaml_path = r'D:\CNWei\CNW\InterfaceAutoTest\TestCases\test_1_user.yaml'
+#     yaml_file = YamlFile(yaml_path)
+#     print(yaml_file.load())
+#     # yaml_file.load()
+#     # case_info = CaseInfo(**yaml_file)
+#     # print(case_info)
+#     # yaml_file["title"] = "查询用户信息"
+#     # yaml_file.save()

commons/file_processors/yaml_processor.py

@@ -19,7 +19,7 @@ from commons.file_processors.base import BaseFileProcessor
 logger = logging.getLogger(__name__)
 
 
-class YamlFile(BaseFileProcessor, dict):
+class YamlProcessor(BaseFileProcessor, dict):
     """
     用于处理 YAML 文件的类，继承自 dict。
     提供了从文件加载、保存到文件、转换为字符串和从字符串转换的功能,
@@ -60,7 +60,40 @@ class YamlFile(BaseFileProcessor, dict):
             logger.warning(f"文件 {self.filepath} 不存在, 字典保持为空.")
             # 保持字典为空 (已在开头 clear)
 
-    def to_string(self) -> str:
+    # def to_string(self) -> str:
+    #     """
+    #     将字典 (自身) 转换为 YAML 格式的字符串。
+    #
+    #     Returns:
+    #         YAML 格式的字符串。
+    #     """
+    #     try:
+    #         return yaml.safe_dump(
+    #             dict(self),  # 使用dict转换为标准的字典
+    #             allow_unicode=True,
+    #             sort_keys=False,
+    #             default_flow_style=False
+    #         )
+    #     except TypeError as e:
+    #         logger.error(f"将数据转换为 YAML 字符串时出错: {e}")
+    #         return ""
+
+    # def to_dict(self, data: str) -> None:
+    #     """
+    #     将 YAML 格式的字符串转换为字典，并更新当前字典的内容.
+    #
+    #     Args:
+    #         data: YAML 格式的字符串。
+    #     """
+    #     try:
+    #         loaded_data = yaml.safe_load(data) or {}
+    #         self.clear()
+    #         self.update(loaded_data)  # 清空并更新
+    #     except yaml.YAMLError as e:
+    #         logger.error(f"将 YAML 字符串转换为字典时出错: {e}")
+    #         self.clear()  # 出错时也清空
+    @staticmethod
+    def to_string(data: dict) -> str:
         """
         将字典 (自身) 转换为 YAML 格式的字符串。
 
@@ -69,7 +102,7 @@ class YamlFile(BaseFileProcessor, dict):
         """
         try:
             return yaml.safe_dump(
-                dict(self),  # 使用dict转换为标准的字典
+                dict(data),  # 使用dict转换为标准的字典
                 allow_unicode=True,
                 sort_keys=False,
                 default_flow_style=False
@@ -78,7 +111,8 @@ class YamlFile(BaseFileProcessor, dict):
             logger.error(f"将数据转换为 YAML 字符串时出错: {e}")
             return ""
 
-    def to_dict(self, data: str) -> None:
+    @staticmethod
+    def to_dict(data: str) -> Union[None, dict]:
         """
         将 YAML 格式的字符串转换为字典，并更新当前字典的内容.
 
@@ -87,11 +121,9 @@ class YamlFile(BaseFileProcessor, dict):
         """
         try:
             loaded_data = yaml.safe_load(data) or {}
-            self.clear()
+            return loaded_data
-            self.update(loaded_data)  # 清空并更新
         except yaml.YAMLError as e:
             logger.error(f"将 YAML 字符串转换为字典时出错: {e}")
-            self.clear()  # 出错时也清空
 
     def save(self, new_filepath: Union[str, Path, None] = None):
         """
@@ -116,8 +148,10 @@ class YamlFile(BaseFileProcessor, dict):
 
 
 class StringOrDict:
-    @classmethod
+    # @classmethod
-    def to_string(cls, data: dict) -> str:
+    # def to_string(cls, data: dict) -> str:
+    @staticmethod
+    def to_string(data: dict) -> str:
         """
         将字典 (自身) 转换为 YAML 格式的字符串。
 
@@ -135,8 +169,10 @@ class StringOrDict:
             logger.error(f"将数据转换为 YAML 字符串时出错: {e}")
             return ""
 
-    @classmethod
+    # @classmethod
-    def to_dict(cls, data: str) -> Union[None, dict]:
+    # def to_dict(cls, data: str) -> Union[None, dict]:
+    @staticmethod
+    def to_dict(data: str) -> Union[None, dict]:
         """
         将 YAML 格式的字符串转换为字典，并更新当前字典的内容.
 
@@ -153,7 +189,7 @@ class StringOrDict:
 if __name__ == '__main__':
     # 示例用法
     yaml_path = r'D:\CNWei\CNW\InterfaceAutoTest\TestCases\test_1_user.yaml'  # 你的 YAML 文件路径
-    yaml_file = YamlFile(yaml_path)
+    yaml_file = YamlProcessor(yaml_path)
     print(yaml_file)
     print(type(yaml_file))
 

commons/models.py

@@ -47,15 +47,14 @@ class CaseInfo:
         obj = cls(**yaml.safe_load(yaml_str))
         return obj
 
-    @allure.step("断言")
+    # @allure.step("断言")
-    def assert_all(self):
+    # def assert_all(self):
-        _validator = case_validator.CaseValidator()
+    #     _validator = case_validator.CaseValidator()
-        # print(case_validator.VALIDATORS)
+    #     # print(case_validator.VALIDATORS)
-
+    #
-        if not self.validate:
+    #
-            return
+    #
-
+    #     _validator.assert_all(self.validate)
-        _validator.assert_all(self.validate)
         # for assert_type, assert_value in self.validate.items():
         #     for msg, data in assert_value.items():
         #         a, b = data[0], data[1]
@@ -75,28 +74,28 @@ class CaseInfo:
         #                 assert a not in b, msg
         # case "xxxxx
 
-    def ddt(self) -> list:  # 返回一个列表，列表中应该包含N个注入了变量的caseInfo
+    # def ddt(self) -> list:  # 返回一个列表，列表中应该包含N个注入了变量的caseInfo
-        case_list = []
+    #     case_list = []
-        if not self.parametrize:  # 没有使用数据驱动测试
+    #     if not self.parametrize:  # 没有使用数据驱动测试
-            logger.info("1，执行这一步")
+    #         logger.info("1，执行这一步")
+    #         # case_info_str = self.to_yaml()  # 转字符串
+    #         # case_info_str = Template(case_info_str).render(d)  # 输入变量
+    #         # case_info = self.by_yaml(case_info_str)  # 转成类
+    #         # case_list.append(case_info)
+    #         case_list.append(self)
+    #     else:  # 使用数据驱动测试
+    #         args_name = self.parametrize[0]
+    #         args_value_list = self.parametrize[1:]
+    #         for args_value in args_value_list:
+    #             d = dict(zip(args_name, args_value))
+    #             print(f"D的值：{d}")
+    #             # d 就是数据驱动测试的变量，应输入到用例中
     #             case_info_str = self.to_yaml()  # 转字符串
     #             case_info_str = Template(case_info_str).render(d)  # 输入变量
     #             case_info = self.by_yaml(case_info_str)  # 转成类
-            # case_list.append(case_info)
+    #
-            case_list.append(self)
+    #             case_list.append(case_info)  # 加入到返回值
-        else:  # 使用数据驱动测试
+    #     return case_list
-            args_name = self.parametrize[0]
-            args_value_list = self.parametrize[1:]
-            for args_value in args_value_list:
-                d = dict(zip(args_name, args_value))
-                print(f"D的值：{d}")
-                # d 就是数据驱动测试的变量，应输入到用例中
-                case_info_str = self.to_yaml()  # 转字符串
-                case_info_str = Template(case_info_str).render(d)  # 输入变量
-                case_info = self.by_yaml(case_info_str)  # 转成类
-
-                case_list.append(case_info)  # 加入到返回值
-        return case_list
 
 
 if __name__ == '__main__':

utils/case_validator.py

@@ -27,7 +27,8 @@ class CaseValidator:
 
     @classmethod
     def assert_all(cls, validate: dict):
-
+        if not validate:
+            return
         for assert_type, cases in validate.items():
             logger.info(f"键：{assert_type}，值：{cases}")
             validator = cls.VALIDATORS.get(assert_type)

utils/data_driver.py

@@ -37,62 +37,62 @@ class DataDriver:
 
 
 if __name__ == '__main__':
-    from commons.file_processors.yaml_processor import YamlFile
+    from commons.file_processors.yaml_processor import YamlProcessor
 
-    file_path = Path(r"E:\PyP\InterfaceAutoTest\TestCases\test_1_user.yaml")
+    file_path = Path(r"D:\CNWei\CNW\InterfaceAutoTest\TestCases\test_1_user.yaml")
 
-    file_obj = YamlFile(file_path)
+    file_obj = YamlProcessor(file_path)
-    # print(file_path.stem)
+    print(file_path.stem)
     file_name = file_path.stem
-    mock_case_info = {
+    # mock_case_info = {
-        "case_info0": {
+    #     "case_info0": {
-            "feature": "页面状态",
+    #         "feature": "页面状态",
-            "story": "状态",
+    #         "story": "状态",
-            "title": "查询状态信息",
+    #         "title": "查询状态信息",
-            "request": "",
+    #         "request": "",
-            "extract": "",
+    #         "extract": "",
-            "validate": "",
+    #         "validate": "",
-            "parametrize": [["title", "username", "password", "msg"], ["测试1", "user1", "pass1", "200"],
+    #         "parametrize": [["title", "username", "password", "msg"], ["测试1", "user1", "pass1", "200"],
-                            ["测试2", "user2", "pass2", "300"]]
+    #                         ["测试2", "user2", "pass2", "300"]]
-        },
+    #     },
-        "case_info1": {
+    #     "case_info1": {
-            "feature": "页面状态",
+    #         "feature": "页面状态",
-            "story": "状态",
+    #         "story": "状态",
-            "title": "查询状态信息",
+    #         "title": "查询状态信息",
-            "request": "",
+    #         "request": "",
-            "extract": "",
+    #         "extract": "",
-            "validate": "",
+    #         "validate": "",
-            "parametrize": [1, 2, 3]
+    #         "parametrize": [1, 2, 3]
-        },
+    #     },
-        "case_info2": {
+    #     "case_info2": {
-            "feature": "页面状态",
+    #         "feature": "页面状态",
-            "story": "状态",
+    #         "story": "状态",
-            "title": "查询状态信息",
+    #         "title": "查询状态信息",
-            "request": "",
+    #         "request": "",
-            "extract": "",
+    #         "extract": "",
-            "validate": "",
+    #         "validate": "",
-            "parametrize": [1, 2, 3]
+    #         "parametrize": [1, 2, 3]
-        }
+    #     }
-
+    #
-    }
+    # }
 
     dd = DataDriver()
     # cases = dd.generate_cases(mock_case_info.get("case_info0"))
     cases = dd.generate_cases(file_name, file_obj)
-    # print(cases)
+    print(cases)
     # print(len(cases))
     keys_list = []
     titles = []
     for item in cases:
-        print(item)
+        # print(item)
         # 遍历列表中的每个字典
         for key, value in item.items():
-            print(f"key:{key}")
+            # print(f"key:{key}")
             keys_list.append(key)
-            print(f"value:{value}")
+            # print(f"value:{value}")
         #     # 遍历内层字典（这里内层字典其实只有一个键值对）
             titles.append(value['title'])
-        print(item)
+        # print(item)
 
     print(keys_list)
     print(titles)