InterfaceAutoTest/core/creator.py

#!/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 pydantic import ValidationError
from commons.file_processors.yaml_processor import YamlProcessor as FileHandle, YamlLoadError
from core.models import RawSchema  # 导入之前定义的 Pydantic 模型
from typing import Any, List, Type, Generator, Union

logger = logging.getLogger(__name__)




@dataclass
class CaseEntity:
    """用例执行实体：解耦模型数据与执行上下文"""
    step_data: RawSchema
    row_context: dict[str, Any]


class TestTemplateBase:
    """
    具体的测试用例容器。
    此映射类不包含任何逻辑方法，仅用于承载由 Loader 挂载的 test_* 方法。
    """
    pass


class CaseDataLoader:
    """
    测试用例加载器
    职责：扫描文件系统 -> 载入 YAML -> 拆解参数化 -> 封装为 CaseInfo 模型
    """

    @staticmethod
    def fetch_yaml_files(cases_dir: Union[str, Path]) -> 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: List[CaseEntity] = []
        try:
            # 1. 使用重构后的 YamlProcessor 加载原始字典
            processor = FileHandle(file_path)
            raw_data = processor.load()

            if not raw_data:
                return []

            entities = cls._parse_parametrize(raw_data)

        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[CaseEntity]:
        """
        解析参数化逻辑：将 raw_data 中的 parametrize 展开为多个 CaseInfo 实例
        """
        entities = []
        parametrize_data = raw_data.pop("parametrize", None)

        # 2. 实例化唯一的模板对象 (Pydantic 校验)
        template_case = RawSchema.model_validate(raw_data)
        # template_case = CaseTemplate(**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))
                # 包装为实体，存入引用而非副本
                # 修正: 使用 model_copy() 避免多个用例共享同一个 Pydantic 模型实例，防止意外修改
                entities.append(CaseEntity(step_data=template_case.model_copy(), row_context=row_map))
        else:
            # 普通用例，上下文为空
            entities.append(CaseEntity(step_data=template_case.model_copy(), row_context={}))

        return entities

    @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


class CaseGenerator:
    """
    职责 2: 用例构造工厂
    负责将数据转化为 pytest 装饰的方法，并挂载到目标类
    """

    @classmethod
    def build_and_register(cls, target_cls: Type[TestTemplateBase], cases_dir: Union[str, Path]):
        # 1. 通过 Loader 获取数据
        all_cases = CaseDataLoader.get_all_cases(cases_dir)
        for index, case_info in enumerate(all_cases):
            case_title = case_info.row_context.get("title") or case_info.step_data.title

            dynamic_test_method = cls._create_case_method(title=case_title, entity=case_info)

            safe_title = "".join([c if c.isalnum() else "_" for c in case_title])[:50]
            method_name = f"test_{index:03d}_{safe_title}"
            print(method_name)
            setattr(target_cls, method_name, dynamic_test_method)
            print(target_cls.__dict__)
            logger.debug(f"Successfully registered: {method_name}")

    @staticmethod
    def _create_case_method(title, entity: CaseEntity):
        """封装具体的 pytest 执行节点"""
        case_template = entity.step_data
        context = entity.row_context

        def build_actual_case(instance: TestTemplateBase, api_env):
            # --- 1. 动态设置 Allure 报告属性 ---

            allure.dynamic.epic(case_template.epic or settings.allure_epic)
            allure.dynamic.feature(case_template.feature or settings.allure_feature)
            allure.dynamic.story(case_template.story or settings.allure_story)
            allure.dynamic.title(title)
            # 日志记录 (利用 instance 标注来源)
            logger.info(f"[Runner] Class: {instance.__class__.__name__} | Case: {title}")
            try:
                WorkflowExecutor.perform(case_template, api_env, context=context)

            except Exception as e:
                # 可以在这里记录更详细的运行上下文快照
                logger.error(f"Case 执行失败: {title} | 错误: {e}")
                raise

        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)