refactor(): 重构动态用例生成逻辑并解耦核心组件

- 将 `CaseGenerator` 拆分为 `CaseDataLoader`(数据加载)和 `CaseGenerator`(用例构造),实现单一职责原则。
- 引入 `TestTemplateBase` 作为纯净的方法挂载容器,避免逻辑代码污染测试用例。
- 优化 YAML 解析流程,将文件扫描、参数化解析与 pytest 方法构建逻辑完全分离。
- 改进装饰器写法,使用更直观的 @ 语法糖处理 Allure 和 pytest.mark.parametrize。
- 增强执行日志,通过类型注解和实例引用记录更详细的运行上下文。
This commit is contained in:
2026-03-06 15:06:54 +08:00
parent 300b5a92d4
commit 69a96a0060
8 changed files with 258 additions and 34 deletions

75
core/executor.py Normal file
View File

@@ -0,0 +1,75 @@
#!/usr/bin/env python
# coding=utf-8
"""
@desc: 核心测试用例执行引擎
"""
import logging
import importlib
from typing import Any
from commons.models.case_model import CaseInfo
from core.session import Session
from commons.exchange import Exchange
from commons.asserts import Asserts
logger = logging.getLogger(__name__)
class WorkflowExecutor:
def __init__(self, session: Session, exchanger: Exchange):
self.session = session
self.exchanger = exchanger
def perform(self, case_data: dict) -> Any:
"""执行单个用例支持直接请求和PO模式调用"""
# 1. 变量替换(将 ${var} 替换为真实值)
rendered_case = self.exchanger.replace(case_data)
# 2. 决定执行模式
if "api_action" in rendered_case:
# --- PO 模式:反射调用业务层 ---
action = rendered_case["api_action"]
resp = self._execute_po_method(
class_name=action["class"],
method_name=action["method"],
params=action.get("params", {})
)
else:
# --- 数据驱动模式:直接发送请求 ---
# 使用 Pydantic 校验 request 结构
case_info = CaseInfo(**rendered_case)
request_info = case_info.request.model_dump(by_alias=True, exclude_none=True)
resp = self.session.request(**request_info)
# 3. 提取变量 (接口关联)
if rendered_case.get("extract"):
for var_name, extract_info in rendered_case["extract"].items():
self.exchanger.extract(resp, var_name, *extract_info)
# 4. 断言校验
if rendered_case.get("validate"):
Asserts.validate(resp, rendered_case["validate"])
return resp
def _execute_po_method(self, class_name: str, method_name: str, params: dict):
"""核心反射逻辑:根据字符串动态加载 api/ 目录下的类并执行方法"""
try:
# 1. 动态导入模块(假设都在 api 目录下)
# 例如 class_name 是 UserAPI则尝试从 api.user 导入
# 这里简单处理,你可以根据你的文件名约定进一步优化逻辑
module_name = f"api.{class_name.lower().replace('api', '')}"
module = importlib.import_module(module_name)
# 2. 获取类并实例化
cls = getattr(module, class_name)
api_instance = cls(self.session) # 传入 session 保持会话统一
# 3. 调用方法并返回结果
method = getattr(api_instance, method_name)
logger.info(f"🚀 调用业务层: {class_name}.{method_name} 参数: {params}")
return method(**params)
except Exception as e:
logger.error(f"反射调用失败: {class_name}.{method_name} -> {e}")
raise