InterfaceAutoTest/core/executor.py

#!/usr/bin/env python
# coding=utf-8

"""
@desc: 核心测试用例执行引擎
"""

import logging
import importlib
from typing import Any, List, Optional
from collections import ChainMap
from pydantic import TypeAdapter

from core import settings
from core.context import ExecutionEnv
from core.models import RawSchema, ValidateItem, HttpAction, ApiActionModel
from core.session import Session
from core.exchange import Exchange
from utils.case_validator import CaseValidator

logger = logging.getLogger(__name__)

# 定义一个复用的适配器(减少初始化开销)
VALIDATE_LIST_ADAPTER = TypeAdapter(List[ValidateItem])


class WorkflowExecutor:
    """
    工作流执行器。
    作为测试执行的核心引擎，负责调度单个用例的完整生命周期：
    1. 上下文准备（变量池合并）。
    2. 动作路由与执行（HTTP 请求或 PO 方法反射调用）。
    3. 后处理（变量提取与断言校验）。
    """

    @classmethod
    def perform(cls, case_info: RawSchema, env: ExecutionEnv, context: Optional[dict[str, Any]] = None) -> Any:
        """执行单个用例：支持直接请求和PO模式调用"""
        context = context or {}
        # --- 重点 1：备份并切换上下文 ---
        # 保存 Exchange 当前的全局字典引用
        original_cache = env.exchanger.global_vars
        # 1. 建立优先级变量池 (参数化变量 > 全局提取变量)
        # ChainMap 是实现“局部覆盖全局”性能最好的方案
        combined_vars = ChainMap(context, original_cache)
        # 将 Exchange 的内部缓存临时指向这个合并池
        env.exchanger.global_vars = combined_vars

        resp = None  # 初始化 resp，避免异常时引用未定义
        try:
            # 2. 动态更新标题（如果 context 中包含 title）
            current_title = context.get("title") or case_info.title
            logger.info(f"🚀 执行用例: {current_title}")

            raw_action_dict = case_info.action.model_dump(by_alias=True, exclude_none=True)
            rendered_action_dict = env.exchanger.replace(raw_action_dict)

            # --- 2. 决定执行模式 ---
            if case_info.is_po_mode():
                # 重新校验以修复类型（如 params 里的 int）
                rendered_action = ApiActionModel.model_validate(rendered_action_dict)
                # PO 模式：反射调用
                resp = cls._execute_po_method(rendered_action, env)
            else:
                # 接口模式：直接请求
                rendered_request = HttpAction.model_validate(rendered_action_dict)
                request_kwargs = rendered_request.model_dump(by_alias=True, exclude_none=True)
                resp = env.session.request(**request_kwargs)

            # --- 3. 后处理：提取与断言 ---
            cls._post_process(resp, case_info, env, original_cache)

            return resp
        except Exception as e:
            logger.error(f"用例执行失败: {case_info.title} | 原因: {e}", exc_info=True)
            raise
        finally:
            # 兜底确保环境还原 (尽管 try 块中已经还原了一次，这里确保异常情况下也复位)
            env.exchanger.global_vars = original_cache

    @staticmethod
    def _execute_po_method(action: ApiActionModel, env: ExecutionEnv):
        """核心反射逻辑：根据字符串动态加载 api/ 目录下的类并执行方法"""
        class_name = action.module
        method_name = action.method
        params = action.params or {}
        # 1. 确定模块路径：优先级策略
        # 优先级 1: 显式映射 (API_MAP)
        module_name = settings.API_MAP.get(class_name)

        # 优先级 2: 规约命名 (UserAPI -> api.user_api)
        if not module_name:
            base_name = class_name.lower().replace('api', '')
            module_name = f"{settings.API_PACKAGE}.{base_name}_api"

        try:
            # 1. 动态导入模块（假设都在 api 目录下）

            module = importlib.import_module(module_name)

            # 2. 获取类并实例化
            cls = getattr(module, class_name)

            api_instance = cls(env.session)  # 传入 session 保持会话统一

            # 3. 调用方法并返回结果
            method = getattr(api_instance, method_name)
            logger.info(f"调用业务层: {class_name}.{method_name} 参数: {params}")
            return method(**params)
        except ImportError as e:
            logger.error(f"模块导入失败: 在 '{module_name}' 未找到对应文件。请检查文件名或 settings.API_MAP 配置。")
            raise e
        except AttributeError as e:
            logger.error(f"成员获取失败: 模块 '{module_name}' 中不存在类或方法 '{class_name}.{method_name}'。")
            raise e
        except Exception as e:
            logger.error(f"反射调用失败: {class_name}.{method_name} -> {e}")
            raise

    @classmethod
    def _post_process(cls, resp: Any, case_info: RawSchema, env: ExecutionEnv, original_cache: dict):
        """
        统一后处理逻辑：处理变量提取（写全局）和断言校验（读局部+全局）
        """
        # 记录当前的混合上下文 (ChainMap)，供断言使用
        combined_vars = env.exchanger.global_vars

        # 1. 变量提取 (Write Operation)
        if case_info.extract:
            try:
                # 必须切回 original_cache 才能持久化写入到全局变量池
                env.exchanger.global_vars = original_cache
                for var_name, extract_info in case_info.extract.items():
                    env.exchanger.extract(resp, var_name, *extract_info)
            finally:
                # 提取完成后，切回 combined_vars，防止后续逻辑（如断言）丢失局部变量上下文
                env.exchanger.global_vars = combined_vars

        # 2. 断言校验 (Read Operation)
        if case_info.validate_data:
            raw_validate_list = [
                item.model_dump(by_alias=True) if isinstance(item, ValidateItem) else item
                for item in case_info.validate_data
            ]

            rendered_validate_list = env.exchanger.replace(raw_validate_list)
            # 重新通过 Adapter 触发类型修复 (str -> int)
            final_validate_data = VALIDATE_LIST_ADAPTER.validate_python(rendered_validate_list)
            CaseValidator.validate(resp, final_validate_data)