#!/usr/bin/env python # coding=utf-8 """ @author: CNWei,ChenWei @Software: PyCharm @contact: t6g888@163.com @file: creator @date: 2026/3/6 10:40 @desc: """ import logging import allure from pathlib import Path from dataclasses import dataclass from core import settings from core.executor import WorkflowExecutor from core.session import Session from core.exchange import Exchange from pydantic import ValidationError from commons.file_processors.yaml_processor import YamlProcessor as FileHandle,YamlLoadError from core.models import CaseInfo # 导入之前定义的 Pydantic 模型 from typing import Any, List, Type, Generator, Union logger = logging.getLogger(__name__) # 初始化全局组件 session = Session(settings.base_url) exchanger = Exchange(settings.DATA_DIR / "extract.yaml") # 指向 data/extract.yaml executor = WorkflowExecutor(session, exchanger) @dataclass class CaseEntity: """用例执行实体：解耦模型数据与执行上下文""" step_data: CaseInfo row_context: dict[str, Any] class TestTemplateBase: """ 具体的测试用例容器。 此映射类不包含任何逻辑方法，仅用于承载由 Loader 挂载的 test_* 方法。 """ pass class CaseDataLoader: """ 测试用例加载器 职责：扫描文件系统 -> 载入 YAML -> 拆解参数化 -> 封装为 CaseInfo 模型 """ @staticmethod def fetch_yaml_files(cases_dir: str) -> Generator[Path, None, None]: """扫描目录并迭代返回 (文件路径, 原始内容)""" base_path = Path(cases_dir) if not base_path.exists(): logger.error(f"📂 测试目录不存在: {base_path}") return # 匹配所有以 test_ 开头的 yaml 文件 yield from base_path.rglob("test_*.yaml") @classmethod def load_cases(cls, file_path: Path) -> List[CaseEntity]: """ 加载单个 YAML 文件并转化为 CaseInfo 列表 包含参数化数据的自动拆解逻辑 """ entities = [] try: # 1. 使用重构后的 YamlProcessor 加载原始字典 processor = FileHandle(file_path) raw_data = processor.load() if not raw_data: return [] # 1. 提取参数化数据 parametrize_data = raw_data.pop("parametrize", None) # 2. 实例化唯一的模板对象 (Pydantic 校验) # 此时占位符 ${var} 会被 SmartInt/SmartDict 校验器放行 template_case = CaseInfo(**raw_data) # 2. 检查是否存在参数化字段 if parametrize_data and isinstance(parametrize_data, list) and len(parametrize_data) >= 2: # 3. 参数化拆分 headers = parametrize_data[0] for row in parametrize_data[1:]: row_map = dict(zip(headers, row)) # 包装为实体，存入引用而非副本 entities.append(CaseEntity(step_data=template_case, row_context=row_map)) else: # 普通用例，上下文为空 entities.append(CaseEntity(step_data=template_case, row_context={})) except YamlLoadError: # YamlProcessor 已经记录了 error 日志，这里直接跳过 pass except ValidationError as e: logger.error(f"用例格式校验失败 [{file_path.name}]:\n{e.json()}") except Exception as e: logger.error(f"加载用例发生未知异常 [{file_path.name}]: {e}") return entities @staticmethod def _parse_parametrize(raw_data: dict[str, Any]) -> List[CaseInfo]: """ 解析参数化逻辑：将 raw_data 中的 parametrize 展开为多个 CaseInfo 实例 """ param_content = raw_data.pop("parametrize") if len(param_content) < 2: logger.warning(f"参数化数据不足（需包含 Header 和至少一行 Data）: {raw_data.get('title')}") return [CaseInfo(**raw_data)] # 第一行作为变量名 (Headers)，后续作为数据行 headers = param_content[0] data_rows = param_content[1:] case_list = [] for row in data_rows: # 将变量名和对应行数据打包成字典，例如 {"username": "user1", "title": "测试1"} row_map = dict(zip(headers, row)) # 深拷贝原始模板，避免多行数据互相干扰 case_tmp = raw_data.copy() # 关键优化：如果参数化里包含 'title'，自动更新顶层的 title 字段 # 注意：此处仅做初步合并，更复杂的 ${var} 替换由 WorkflowExecutor 的 Exchanger 完成 if "title" in row_map: case_tmp["title"] = row_map["title"] # 将当前行的数据注入到 CaseInfo 中（此处可以暂存在字段中，或由执行器处理） # 为了保持模型兼容，我们把 row_map 的信息合入 case_tmp case_list.append(CaseInfo(**case_tmp)) return case_list @classmethod def get_all_cases(cls, cases_dir: Union[str, Path]) -> List[CaseEntity]: """ 全量获取接口：供 CaseGenerator 调用 frank """ all_cases = [] for file in cls.fetch_yaml_files(cases_dir): all_cases.extend(cls.load_cases(file)) return all_cases # @staticmethod # def parse_parametrize(raw_data: Dict, default_name: str) -> Tuple[List[str], List[List[Any]], List[str]]: # """解析参数化结构，返回 (字段名列表, 数据值列表, ID列表)""" # if "parametrize" in raw_data: # fields = raw_data["parametrize"][0] # values = raw_data["parametrize"][1:] # ids = [f"{v[0]}" for v in values] # else: # fields = ["case_data"] # values = [[raw_data]] # ids = [raw_data.get("title", default_name)] # return fields, values, ids class CaseGenerator: """ 职责 2: 用例构造工厂 负责将数据转化为 pytest 装饰的方法，并挂载到目标类 """ @classmethod def build_and_register(cls, target_cls: Type[TestTemplateBase], cases_dir: str): # 1. 通过 Loader 获取数据 all_cases=CaseDataLoader.get_all_cases(cases_dir) for index, case_info in enumerate(all_cases): dynamic_test_method=cls._create_case_method(case_info) # for file_path, raw_data in CaseDataLoader.get_all_cases(cases_dir): # # 2. 解析参数化信息 # fields, values, ids = CaseDataLoader.parse_parametrize(raw_data, file_path.stem) # # # 3. 生成执行函数 (闭包) # dynamic_test_method = cls._create_case_method(raw_data, fields, values, ids) # # # 4. 挂载 # method_name = f"test_{file_path.stem}" case_title = case_info.row_context.get("title") or case_info.step_data.title method_name = f"test_case_{index}_{case_info.title[:20]}" safe_name = "".join([c if c.isalnum() else "_" for c in method_name]) # setattr(target_cls, method_name, dynamic_test_method) setattr(target_cls, safe_name, dynamic_test_method) logger.debug(f"Successfully registered: {method_name}") @staticmethod # def _create_case_method(case_template: Dict, fields: List[str], values: List[Any], ids: List[str]): def _create_case_method(entity: CaseEntity): """封装具体的 pytest 执行节点""" case_template = entity.step_data context = entity.row_context # 预取 Allure 层级信息 # epic = case_template.get("epic", settings.allure_epic) # feature = case_template.get("feature", settings.allure_feature) # story = case_template.get("story", settings.allure_story) epic = case_template.epic or settings.allure_epic feature = case_template.feature or settings.allure_feature story = case_template.story or settings.allure_story @allure.epic(epic) @allure.feature(feature) @allure.story(story) # @pytest.mark.parametrize("case_args", values, ids=ids) # def build_actual_case(instance: TestTemplateBase, case_args: List[Any]): def build_actual_case(instance: TestTemplateBase): # 数据组装 # current_params = dict(zip(fields, case_args)) # case_exec_data = {**case_template, **current_params} # case_title = current_params.get("title", "未命名用例") case_title = context.get("title") or case_template.title or "未命名用例" # 日志记录 (利用 instance 标注来源) logger.info(f"🚀 [Runner] Class: {instance.__class__.__name__} | Case: {case_title}") # 执行与断言 allure.dynamic.title(case_title) # executor.perform(case_exec_data) executor.perform(case_template,context=context) # 手动链路装饰 (Allure) # run_actual_case = allure.epic(epic)(run_actual_case) # run_actual_case = allure.feature(feature)(run_actual_case) # run_actual_case = allure.story(story)(run_actual_case) return build_actual_case if __name__ == '__main__': from settings import TEST_CASE_DIR print(CaseDataLoader.get_all_cases(TEST_CASE_DIR)) # --- 引导执行 --- # CaseGenerator.build_and_register(TestTemplateBase, settings.TEST_CASE_DIR)