From 914b0301ba4489bd2363aaddde780a094c21d2bf Mon Sep 17 00:00:00 2001 From: CNWei Date: Fri, 7 Mar 2025 17:28:41 +0800 Subject: [PATCH] =?UTF-8?q?feat,fix():=20=E4=BC=98=E5=8C=96=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 优化yaml_processor（优化文件类型转换逻辑） - 修复bug --- commons/cases.py | 26 +-- commons/exchange.py | 35 ++-- commons/file_processors/base.py | 6 +- commons/file_processors/file_handle.py | 46 +++++ commons/file_processors/json_processor.py | 211 ++++++++++++++++++++++ commons/file_processors/yaml_processor.py | 60 ++++-- commons/models.py | 61 +++---- utils/case_validator.py | 3 +- utils/data_driver.py | 80 ++++---- 9 files changed, 419 insertions(+), 109 deletions(-) create mode 100644 commons/file_processors/file_handle.py create mode 100644 commons/file_processors/json_processor.py diff --git a/commons/cases.py b/commons/cases.py index c47b1be..460b495 100644 --- a/commons/cases.py +++ b/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]'] - titles = ['查询用户信息', '查询用户信息', '查询用户信息', '查询用户信息'] + # keys_list = ['test_1_user[0]', 'test_1_user[1]', 'test_1_user[2]', 'test_1_user[3]'] + # 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) - for var_name, extract_info in new_case_info.extract.items(): - # logger.info(f"保存变量：{var_name}{extract_info}") + # new_case_info = CaseInfo(**new_case_info) + for var_name, extract_info in new_case_info.get("extract").items(): + 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 diff --git a/commons/exchange.py b/commons/exchange.py index ed3366c..51915ab 100644 --- a/commons/exchange.py +++ b/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( - title="单元测试", - request={ + # mock_case_info = CaseInfo( + # title="单元测试", + # 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={}, - validate={} - ) + "extract": {}, + "validate": {} + } + new_mock_case_info = exchanger.replace(mock_case_info) print(new_mock_case_info) diff --git a/commons/file_processors/base.py b/commons/file_processors/base.py index 0cf65de..28e9583 100644 --- a/commons/file_processors/base.py +++ b/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 diff --git a/commons/file_processors/file_handle.py b/commons/file_processors/file_handle.py new file mode 100644 index 0000000..03d6e27 --- /dev/null +++ b/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 diff --git a/commons/file_processors/json_processor.py b/commons/file_processors/json_processor.py new file mode 100644 index 0000000..fceae06 --- /dev/null +++ b/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() diff --git a/commons/file_processors/yaml_processor.py b/commons/file_processors/yaml_processor.py index 4b94833..f9b3166 100644 --- a/commons/file_processors/yaml_processor.py +++ b/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() - self.update(loaded_data) # 清空并更新 + return 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 - def to_string(cls, data: dict) -> str: + # @classmethod + # 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 - def to_dict(cls, data: str) -> Union[None, dict]: + # @classmethod + # 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)) diff --git a/commons/models.py b/commons/models.py index 444ac42..0cdf0f9 100644 --- a/commons/models.py +++ b/commons/models.py @@ -47,15 +47,14 @@ class CaseInfo: obj = cls(**yaml.safe_load(yaml_str)) return obj - @allure.step("断言") - def assert_all(self): - _validator = case_validator.CaseValidator() - # print(case_validator.VALIDATORS) - - if not self.validate: - return - - _validator.assert_all(self.validate) + # @allure.step("断言") + # def assert_all(self): + # _validator = case_validator.CaseValidator() + # # print(case_validator.VALIDATORS) + # + # + # + # _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 - case_list = [] - if not self.parametrize: # 没有使用数据驱动测试 - 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) # 加入到返回值 - return case_list + # def ddt(self) -> list: # 返回一个列表，列表中应该包含N个注入了变量的caseInfo + # case_list = [] + # if not self.parametrize: # 没有使用数据驱动测试 + # 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) # 加入到返回值 + # return case_list if __name__ == '__main__': diff --git a/utils/case_validator.py b/utils/case_validator.py index d5c2a40..7ad6389 100644 --- a/utils/case_validator.py +++ b/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) diff --git a/utils/data_driver.py b/utils/data_driver.py index 801cd8c..c0d1a20 100644 --- a/utils/data_driver.py +++ b/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) - # print(file_path.stem) + file_obj = YamlProcessor(file_path) + print(file_path.stem) file_name = file_path.stem - mock_case_info = { - "case_info0": { - "feature": "页面状态", - "story": "状态", - "title": "查询状态信息", - "request": "", - "extract": "", - "validate": "", - "parametrize": [["title", "username", "password", "msg"], ["测试1", "user1", "pass1", "200"], - ["测试2", "user2", "pass2", "300"]] - }, - "case_info1": { - "feature": "页面状态", - "story": "状态", - "title": "查询状态信息", - "request": "", - "extract": "", - "validate": "", - "parametrize": [1, 2, 3] - }, - "case_info2": { - "feature": "页面状态", - "story": "状态", - "title": "查询状态信息", - "request": "", - "extract": "", - "validate": "", - "parametrize": [1, 2, 3] - } - - } + # mock_case_info = { + # "case_info0": { + # "feature": "页面状态", + # "story": "状态", + # "title": "查询状态信息", + # "request": "", + # "extract": "", + # "validate": "", + # "parametrize": [["title", "username", "password", "msg"], ["测试1", "user1", "pass1", "200"], + # ["测试2", "user2", "pass2", "300"]] + # }, + # "case_info1": { + # "feature": "页面状态", + # "story": "状态", + # "title": "查询状态信息", + # "request": "", + # "extract": "", + # "validate": "", + # "parametrize": [1, 2, 3] + # }, + # "case_info2": { + # "feature": "页面状态", + # "story": "状态", + # "title": "查询状态信息", + # "request": "", + # "extract": "", + # "validate": "", + # "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)