feat,fix(core,docs): 完善核心模块代码注释并添加架构改进文档

- 为 core 目录下主要模块 (models, context, creator, base_api, exchange, executor) 添加了详细的类和方法 Docstring。
   - 新增 docs/架构改进.md 文件。

commons/case_handler.py

@@ -1,80 +0,0 @@
-#!/usr/bin/env python
-# coding=utf-8
-
-"""
-@author: CNWei
-@Software: PyCharm
-@contact: t6i888@163.com
-@file: case_handler
-@date: 2025/5/26 22:13
-@desc: 
-"""
-import json
-
-import logging
-from dataclasses import dataclass, asdict
-from commons.models import TestCaseModel
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class TestCaseHandle(TestCaseModel):
-
-    @classmethod
-    def new(cls, testcase: dict) -> 'TestCaseHandle':
-
-        try:
-            instance = cls(**testcase)
-            return instance
-        except (TypeError, ValueError) as e:
-            logger.warning(f"解析错误：{e}")
-            raise e
-
-    def to_string(self) -> str:
-        """
-        将 字典 转换为 json 格式的字符串。
-
-        :return:
-            json 格式的字符串。
-        """
-        try:
-            res = json.dumps(asdict(self), ensure_ascii=False)
-            return res
-        except TypeError as e:
-            logger.error(f"将数据转换为 json 字符串时出错: {e}")
-            raise e
-
-    @staticmethod
-    def to_dict(json_str: str) -> dict:
-        """
-        将 json 格式的字符串转换为 字典.
-
-        :param
-            json_str: json 格式的字符串。
-        :return:
-        """
-        try:
-            res = json.loads(json_str)
-            return res
-        except json.JSONDecodeError as e:
-            logger.error(f"将 json 字符串转换为字典时出错: {e}")
-            raise e
-
-
-if __name__ == '__main__':
-    from pathlib import Path
-    from commons.file_processors import processor_factory
-
-    test_data = Path(r"E:\PyP\InterfaceAutoTest\TestCases\test_1_user.yaml")
-    yaml_data = processor_factory.get_processor_class(test_data)
-    case_info = TestCaseHandle.new(yaml_data.load())
-    print(case_info.to_string())
-    print(type(case_info.to_string()))
-    print(case_info.to_dict(case_info.to_string()))
-    print(type(case_info.to_dict(case_info.to_string())))
-    print(type(case_info))
-    print(case_info.parametrize)
-
-    for i in case_info.parametrize:
-        print(i)

commons/files.py

@@ -1,98 +0,0 @@
-#!/usr/bin/env python
-# coding=utf-8
-
-"""
-@author: chen wei
-@Software: PyCharm
-@contact: t6i888@163.com
-@file: files.py
-@date: 2024 2024/9/15 21:28
-@desc: 读取和保存yaml文件
-"""
-import logging
-from pathlib import Path
-import yaml
-
-logger = logging.getLogger(__name__)
-
-
-class YamlFile(dict):
-    def __init__(self, path=None, data=None):
-        super().__init__()
-        self.path = Path(path) if path else None
-
-        if data:
-            self.update(data)
-        elif self.path:
-            if self.path.is_dir():
-                raise IsADirectoryError(f"The path {self.path} is a directory, not a file.")
-            self.load()
-
-    def load(self):
-        if not self.path:
-            logger.warning("No path specified for YamlFile, cannot load.")
-            return self
-
-        if self.path.exists() and self.path.is_file():
-            with open(self.path, "r", encoding="utf-8") as f:
-                loaded_data = yaml.safe_load(f) or {}
-                self.clear()
-                self.update(loaded_data)
-        else:
-            logger.warning(f"File not found at {self.path}, YamlFile initialized as empty.")
-            self.clear()
-        return self
-
-    def to_yaml(self) -> str:
-        return yaml.safe_dump(
-            dict(self),
-            allow_unicode=True,
-            sort_keys=False
-        )
-
-    @classmethod
-    def by_yaml(cls, yaml_str):
-        data = yaml.safe_load(yaml_str) or {}
-        return cls(data=data)
-
-    def save(self):
-        if not self.path:
-            raise ValueError("Cannot save YamlFile instance without a specified path.")
-
-        # 确保父目录存在
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-
-        with open(self.path, "w", encoding="utf-8") as f:
-            yaml.safe_dump(
-                dict(self),
-                stream=f,
-                allow_unicode=True,
-                sort_keys=False
-            )
-        return self
-
-
-if __name__ == '__main__':
-    from core.models import CaseInfo
-    from core.settings import TEST_CASE_DIR
-
-    # 1. 创建一个用于测试的临时yaml文件
-    dummy_path = TEST_CASE_DIR / "test_model_demo.yaml"
-    dummy_data = {
-        "title": "Get user info",
-        "request": {"method": "GET", "url": "/users/1"},
-        "validate": [{"equals": ["status_code", 200]}]
-    }
-    YamlFile(path=dummy_path, data=dummy_data).save()
-    print(f"--- 已创建临时测试文件: {dummy_path}")
-
-    # 2. 加载文件并使用Pydantic模型进行校验
-    yaml_case = YamlFile(dummy_path)
-    print("\n--- 已加载YAML内容 ---\n", yaml_case.to_yaml())
-    case_model = CaseInfo(**yaml_case)
-    print("\n--- Pydantic模型校验成功 ---")
-    print(case_model.model_dump_json(indent=2, by_alias=True))
-
-    # 3. 清理临时文件
-    dummy_path.unlink()
-    print(f"\n--- 已清理临时文件: {dummy_path}")

conftest.py

@@ -4,57 +4,136 @@
 """
 @desc: Pytest 配置文件，用于设置全局 Fixture 和钩子函数
 """
+import platform
+from typing import Any
 
 import pytest
 from pathlib import Path
 import logging
 
-from core import settings
-from commons.files import YamlFile
 from core.context import VariableStore, ExecutionEnv
-from core.executor import WorkflowExecutor
-from core.models import RawSchema
 from core.session import Session
 from core.exchange import Exchange
-from core.settings import EXTRACT_CACHE
+from core.settings import EXTRACT_CACHE,base_url
+
+
 logger = logging.getLogger(__name__)
 
 
-@pytest.fixture(scope="session")
-def api_env():
+# 注册命令行参数
+def pytest_addoption(parser: Any) -> None:
     """
-    工业级资源调度器
-    1. 保持全局单 Session (连接池复用)
-    2. 变量池 L2 内存镜像化 (减少 I/O)
+    注册自定义命令行参数。
+
+    允许用户通过命令行传递参数来控制测试执行的行为。
+
+    Args:
+        parser: Pytest 的命令行参数解析器对象。
+    """
+    parser.addoption("--test_dir", action="store", default=None, help="测试用例目录")
+    parser.addoption("--env", action="store", default="test", help="运行环境标识 (test/prod/dev)")
+
+@pytest.fixture(scope="session")
+def execution_context():
+    """
+    [Session级别 Fixture] 全局执行上下文环境。
+
+    职责：
+    1. 生命周期管理：初始化并管理全局唯一的 Session、变量存储 (Store) 和 变量交换器 (Exchanger)。
+    2. 资源复用：确保 HTTP 连接池复用，减少握手开销。
+    3. 数据持久化：在测试结束时自动将提取的变量回写到磁盘。
+
+    Yields:
+        ExecutionEnv: 包含 session, store, exchanger 的环境对象实例。
     """
     # Setup: 加载环境
-    store = VariableStore(settings.DATA_DIR / "extract.yaml")
+    store = VariableStore(EXTRACT_CACHE)
     exchanger = Exchange(variable_cache=store.store)
-    session = Session(settings.base_url)
-    executor = WorkflowExecutor()
+    session = Session(base_url)
 
-    env = ExecutionEnv(session, store, executor, exchanger)
+    env = ExecutionEnv(session, store, exchanger)
 
     yield env  # 注入到测试用例中
 
     # Teardown: 统一持久化与清理
     store.persist()
     session.close()
-@pytest.fixture(scope="session")
-def session():
-    """全局共享的 Session Fixture"""
-    return Session(settings.base_url)
 
 
-@pytest.fixture(scope="session")
-def exchanger():
-    """全局共享的 Exchange Fixture"""
-    return Exchange(EXTRACT_CACHE)
+def pytest_exception_interact(node: Any, call: Any, report: Any) -> None:
+    """
+    [Hook] 异常交互钩子。
 
+    当测试用例执行失败（断言错误或代码异常）时触发。
+    主要用于捕获详细的错误堆栈信息，并将其格式化输出到日志中，
+    以便于在控制台或日志文件中快速定位问题。
 
-# @pytest.fixture(scope="session")
-# def case_engine(session, exchanger):
-#     """全局共享的 CaseEngine Fixture"""
-#     return CaseEngine(session, exchanger)
+    Args:
+        node: 发生异常的测试节点（Item 或 Collector）。
+        call: 测试调用信息（包含 excinfo 异常信息）。
+        report: 测试报告对象。
+    """
+    if report.failed:
+        # 获取详细的错误堆栈（包含 assert 的对比信息）
+        # long,short,no-locals
+        exc_info = call.excinfo.getrepr(style='short')
+        logger.error(f"\n{'=' * 40} TEST FAILED {'=' * 40}\n"
+                     f"Node: {node.name}\n"
+                     f"Error:\n{exc_info}"
+                     )
+        logger.error("=" * 93 + "\n")
         
+def pytest_sessionfinish(session: Any, exitstatus: int) -> None:
+    """
+    [Hook] 会话结束钩子。
 
+    在所有测试执行完毕后调用。主要完成以下工作：
+    1. 根据退出状态码记录不同级别的日志信息。
+    2. 收集测试环境信息（如 Base URL, Python 版本, 操作系统等）。
+    3. 生成 `environment.properties` 文件供 Allure 报告展示。
+
+    Args:
+        session: Pytest 会话对象。
+        exitstatus: 整体测试执行的退出状态码。
+    """
+    match exitstatus:
+        case pytest.ExitCode.OK:
+            logging.info("测试全部通过！")
+        case pytest.ExitCode.TESTS_FAILED:
+            logging.warning("部分测试用例执行失败，请检查报告。")
+        case pytest.ExitCode.INTERRUPTED:
+            logging.error("测试被用户手动中断（Ctrl+C）。")
+        case pytest.ExitCode.INTERNAL_ERROR:
+            logging.critical("Pytest 发生内部错误！")
+        case pytest.ExitCode.USAGE_ERROR:
+            logging.error("Pytest 命令行参数错误或用法不当。")
+        case pytest.ExitCode.NO_TESTS_COLLECTED:
+            logging.warning("未发现任何测试用例。")
+        case _:
+            logging.error(f"未知错误状态码: {exitstatus}")
+
+    report_dir = session.config.getoption("--alluredir")
+    if not report_dir:
+        return
+    report_path = Path(report_dir)
+    
+    # 收集环境信息 (适配接口自动化)
+    env_info = {
+        "Base URL": base_url,
+        "Environment": session.config.getoption("--env"),
+        "Python Version": platform.python_version(),
+        "OS System": platform.system(),
+        "Project": "Interface Auto Test"
+    }
+
+    try:
+        if not report_path.exists():
+            report_path.mkdir(parents=True, exist_ok=True)
+        # 生成 environment.properties 文件
+        env_file = report_path / "environment.properties"
+        with env_file.open("w", encoding="utf-8") as f:
+            for k, v in env_info.items():
+                f.write(f"{k}={v}\n")
+        logging.info("Allure 环境信息已生成。")
+    except Exception as e:
+        logging.error(f"无法写入环境属性: {e}")

core/base_api.py

@@ -6,7 +6,17 @@ from core.session import Session
 from core import settings
 
 class BaseApi:
+    """
+    所有 API 类的基类。
+    提供基础的 Session 管理和日志记录功能，供具体的业务 API 类继承。
+    """
     def __init__(self, session: Session = None):
+        """
+        初始化 BaseApi。
+
+        Args:
+            session: HTTP 会话对象。如果未提供，将使用默认配置创建一个新的 Session。
+        """
         self.session = session or Session(base_url=settings.base_url)
         self.logger = logging.getLogger(self.__class__.__name__)
 

core/context.py

@@ -10,7 +10,7 @@
 @desc: 
 """
 from dataclasses import dataclass
-from typing import Dict, Any
+from typing import Any
 from pathlib import Path
 
 from core.exchange import Exchange
@@ -22,10 +22,16 @@ class VariableStore:
     """内存变量仓库：负责 L2 缓存与磁盘的唯一交互"""
 
     def __init__(self, seed_file: Path):
+        """
+        初始化变量仓库。
+
+        Args:
+            seed_file: 初始变量文件路径（YAML格式），用于加载种子数据。
+        """
         self.seed_file = seed_file
         self.processor = YamlProcessor(seed_file)
         # 启动时仅加载一次
-        self.store: Dict[str, Any] = self.processor.load() or {}
+        self.store: dict[str, Any] = self.processor.load() or {}
 
     def persist(self):
         """测试结束时统一写盘"""

core/creator.py

@@ -14,6 +14,8 @@ import logging
 import allure
 from pathlib import Path
 from dataclasses import dataclass
+
+from conftest import execution_context
 from core import settings
 from core.executor import WorkflowExecutor
 from pydantic import ValidationError
@@ -131,6 +133,15 @@ class CaseGenerator:
 
     @classmethod
     def build_and_register(cls, target_cls: Type[TestTemplateBase], cases_dir: Union[str, Path]):
+        """
+        构建测试用例并注册到目标测试类中。
+
+        遍历指定目录下的用例文件，解析数据，生成测试方法并动态绑定到 target_cls 上。
+
+        Args:
+            target_cls: 目标测试类（通常继承自 TestTemplateBase）。
+            cases_dir: 测试用例文件所在的目录路径。
+        """
         # 1. 通过 Loader 获取数据
         all_cases = CaseDataLoader.get_all_cases(cases_dir)
         for index, case_info in enumerate(all_cases):
@@ -142,16 +153,26 @@ class CaseGenerator:
             method_name = f"test_{index:03d}_{safe_title}"
             print(method_name)
             setattr(target_cls, method_name, dynamic_test_method)
-            print(target_cls.__dict__)
+            # print(target_cls.__dict__)
             logger.debug(f"Successfully registered: {method_name}")
 
     @staticmethod
     def _create_case_method(title, entity: CaseEntity):
-        """封装具体的 pytest 执行节点"""
+        """
+        封装具体的 pytest 执行节点。
+
+        创建并返回一个闭包函数，该函数包含完整的测试执行逻辑（Allure 设置、日志、执行器调用）。
+
+        Args:
+            title: 测试用例标题。
+            entity: 包含用例数据和上下文的实体对象。
+
+        Returns:
+            function: 可被 pytest 识别和执行的测试方法。
+        """
         case_template = entity.step_data
         context = entity.row_context
-
-        def build_actual_case(instance: TestTemplateBase, api_env):
+        def build_actual_case(instance: TestTemplateBase, execution_context):
             # --- 1. 动态设置 Allure 报告属性 ---
 
             allure.dynamic.epic(case_template.epic or settings.allure_epic)
@@ -161,7 +182,7 @@ class CaseGenerator:
             # 日志记录 (利用 instance 标注来源)
             logger.info(f"[Runner] Class: {instance.__class__.__name__} | Case: {title}")
             try:
-                WorkflowExecutor.perform(case_template, api_env, context=context)
+                WorkflowExecutor.perform(case_template, execution_context, context=context)
 
             except Exception as e:
                 # 可以在这里记录更详细的运行上下文快照

core/exchange.py

@@ -24,17 +24,31 @@ T = TypeVar("T", bound=Union[dict, list, str, Any])
 
 
 class Exchange:
+    """
+    变量交换器类。
+    负责管理全局变量缓存，核心职能包括：
+    1. Extract: 从响应结果中提取变量。
+    2. Replace: 将数据中的变量占位符替换为实际值。
+    """
     def __init__(self, variable_cache: dict[str, Any]):
+        """
+        初始化交换器。
+
+        Args:
+            variable_cache: 初始变量缓存字典（引用传递，修改会影响源数据）。
+        """
         self._cache = variable_cache
         # 匹配标准变量 ${var}，排除函数调用 ${func()}
         self.var_only_pattern = re.compile(r"^\$\{([a-zA-Z_]\w*)}$")
 
     @property
     def global_vars(self) -> dict:
+        """获取当前全局变量缓存。"""
         return self._cache
 
     @global_vars.setter
     def global_vars(self, global_vars: dict) -> None:
+        """设置全局变量缓存（通常用于上下文切换，如 ChainMap 合并）。"""
         self._cache = global_vars
 
     def extract(self, resp: Any, var_name: str, attr: str, expr: str, index: int = 0):

core/executor.py

@@ -25,6 +25,13 @@ 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:
@@ -74,7 +81,7 @@ class WorkflowExecutor:
     @staticmethod
     def _execute_po_method(action: ApiActionModel, env: ExecutionEnv):
         """核心反射逻辑：根据字符串动态加载 api/ 目录下的类并执行方法"""
-        class_name = action.api_class
+        class_name = action.module
         method_name = action.method
         params = action.params or {}
         # 1. 确定模块路径：优先级策略

core/models.py

@@ -18,6 +18,10 @@ logger = logging.getLogger(__name__)
 
 
 class HttpAction(BaseModel):
+    """
+    HTTP 请求动作模型。
+    定义了发起 HTTP 请求所需的所有参数，包括方法、URL、头信息、参数、请求体等。
+    """
     method: str = Field(..., description="HTTP 请求方法: get, post, etc.")
     url: str = Field(..., description="接口路径或完整 URL")
     headers: dict[str, Any] | None = Field(default=None, description="HTTP 请求头")
@@ -31,6 +35,10 @@ class HttpAction(BaseModel):
 
 
 class ApiActionModel(BaseModel):
+    """
+    PO (Page Object) 模式动作模型。
+    定义了调用封装在 API 类中的方法所需的信息，通过反射机制动态执行。
+    """
     module: str = Field(..., alias="class", description="要调用的 API 类名")
     method: str = Field(..., description="类中的方法名")
     params: dict[str, Any] = Field(default_factory=dict, description="传给方法的参数")
@@ -39,6 +47,10 @@ class ApiActionModel(BaseModel):
 
 
 class ValidateItem(BaseModel):
+    """
+    断言项模型。
+    定义了测试用例执行后的校验规则，包括检查字段、断言方法和期望值。
+    """
     check: str = Field(..., description="要检查的字段或表达式")
     assert_method: str = Field(alias="assert", default="equals")
     expect: Any = Field(..., description="期望值")
@@ -48,6 +60,10 @@ class ValidateItem(BaseModel):
 
 
 class RawSchema(BaseModel):
+    """
+    测试用例原始数据模型。
+    对应 YAML 用例文件的结构，包含元数据、动作定义、变量提取和断言规则。
+    """
     title: str = Field(..., description="用例标题")
     epic: str | None = None
     feature: str | None = None

core/session.py

@@ -6,7 +6,7 @@
 @Software: PyCharm
 @contact: t6i888@163.com
 @file: session.py
-@date: 2024 2024/9/12 21:56
+@date: 2024/9/12 21:56
 @desc: 
 """
 import logging
@@ -19,22 +19,69 @@ import allure
 
 # logger = logging.getLogger("requests.session")
 logger = logging.getLogger(__name__)
-logging.basicConfig(level=logging.INFO)
+# logging.basicConfig(level=logging.INFO)
 
 
 class Session(requests.Session):
+    """
+    自定义会话管理类，继承自 requests.Session。
+
+    增强功能：
+    1. Base URL 管理：支持相对路径自动拼接。
+    2. Allure 集成：自动将请求操作包装为 Allure 步骤。
+    3. 日志记录：详细记录请求和响应的 头部、正文、状态码等信息。
+    """
+
     def __init__(self, base_url=None):
+        """
+        初始化会话。
+
+        Args:
+            base_url: 基础 URL，用于拼接相对路径请求。
+        """
         super().__init__()  # 先执行父类的初始化
         self.base_url = base_url  # 在执行子类的初始化操作
 
     @allure.step("发送请求")
     def request(self, method, url: str, *args, **kwargs) -> Response:
+        """
+        发送 HTTP 请求（重写）。
+
+        逻辑：
+        1. 如果 url 是相对路径，自动拼接 base_url。
+        2. 记录 Allure 步骤。
+
+        Args:
+            method: 请求方法 (GET, POST, etc.)
+            url: 请求 URL (支持相对路径)
+            *args: 透传给 requests.Session.request 的位置参数
+            **kwargs: 透传给 requests.Session.request 的关键字参数
+
+        Returns:
+            Response: 响应对象
+        """
         if not url.startswith("http"):
             # 自动添加baseurl
             url = urljoin(self.base_url, url)
         return super().request(method, url, *args, **kwargs)  # 按照原有方式执行
 
     def send(self, request: PreparedRequest, *args, **kwargs) -> Response:
+        """
+        发送底层 PreparedRequest（重写）。
+
+        逻辑：
+        1. 记录请求详细日志 (URL, Headers, Body)。
+        2. 执行真实网络请求。
+        3. 记录响应详细日志 (Status, Headers, Body)。
+
+        Args:
+            request: 已准备好的请求对象
+            *args: 透传参数
+            **kwargs: 透传参数
+
+        Returns:
+            Response: 响应对象
+        """
         logger.info(f"发送请求>>>>>> 接口地址 = {request.method} {request.url}")
         logger.info(f"发送请求>>>>>> 请求头 = {request.headers}")
         logger.info(f"发送请求>>>>>> 请求正文 = {request.body} ")

docs/架构改进.md

@@ -0,0 +1,80 @@
+# 自动化测试框架架构改进建议
+
+本文档基于对当前 `InterfaceAutoTest` 项目代码的深度分析，整理了针对框架稳定性、扩展性和易用性的架构改进建议。
+
+## 1. 并发执行支持 (Concurrency Support)
+
+### 现状问题
+当前 `VariableStore` 使用简单的文件读写 (`extract.yaml`) 来存储全局变量。
+- 在使用 `pytest-xdist` 进行多进程并发测试时，每个进程会加载独立的内存变量副本。
+- 测试结束写回文件时，不同进程会相互覆盖，导致变量提取丢失或数据不一致。
+
+### 改进方案
+1.  **引入分布式缓存 (推荐)**:
+    -   使用 **Redis** 作为变量存储后端。
+    -   Redis 天然支持原子操作和并发读写，能完美解决多进程数据共享问题。
+2.  **文件锁机制 (轻量级)**:
+    -   如果不引入 Redis，需在 `VariableStore` 的读写操作中增加 **文件锁 (File Lock)** (如使用 `filelock` 库)。
+    -   这会降低并发性能，但能保证数据一致性。
+
+## 2. 配置管理增强 (Configuration Management)
+
+### 现状问题
+`settings.py` 中存在大量硬编码配置（如 API 映射、日志路径），且缺乏对多环境（Dev/Test/Prod）的动态切换支持。
+
+### 改进方案
+1.  **多环境配置文件**:
+    -   建立 `config/` 目录，分离 `base_config.yaml`, `dev.yaml`, `prod.yaml`。
+    -   运行时通过环境变量 `ENV=prod` 加载对应配置并合并。
+2.  **环境变量集成**:
+    -   使用 `.env` 文件管理敏感信息和基础路径。
+    -   利用 `python-dotenv` 在项目启动时加载环境变量。
+
+## 3. 扩展性与钩子机制 (Extensibility & Hooks)
+
+### 现状问题
+`WorkflowExecutor` 的执行逻辑（准备 -> 请求 -> 后处理）是固定的。如果需要添加自定义逻辑（如请求签名加密、复杂的响应解密），目前很难插入。
+
+### 改进方案
+在执行器中引入 **Hooks (钩子)** 机制，允许注册回调函数：
+-   `before_request(request_data)`: 请求发出前调用，用于修改 Header、计算签名。
+-   `after_response(response)`: 收到响应后调用，用于全局解密、统一错误码判断。
+-   `before_case(context)` / `after_case(result)`: 用例级别的 setup/teardown。
+
+## 4. 安全性管理 (Security)
+
+### 现状问题
+敏感数据（如密码、SecretKey）可能明文写在 YAML 用例中。
+
+### 改进方案
+扩展 `Exchange` 类的变量替换逻辑，增加对环境变量的读取支持：
+-   **语法示例**: `password: ${ENV:DB_PASSWORD}`
+-   在运行时从系统环境变量中读取，避免将其提交到代码仓库。
+
+## 5. 可观测性增强 (Observability)
+
+### 现状问题
+虽然 `Session` 类中有日志记录，但在高并发或海量日志场景下，难以串联单个用例的完整执行链路。
+
+### 改进方案
+1.  **全链路 Trace ID**:
+    -   在用例开始执行时生成唯一的 `trace_id`。
+    -   将其注入到 `logging` 的 `Extra` 信息中，使其出现在每一行日志里。
+    -   同时将 `trace_id` 添加到 HTTP 请求头中（如 `X-Trace-Id`），便于服务端排查。
+2.  **结构化日志**:
+    -   考虑使用 JSON 格式输出日志，便于接入 ELK 等日志分析系统。
+
+## 6. 代码健壮性 (Robustness)
+
+### 修复建议
+-   **属性一致性**: 检查 `core/executor.py` 中的 PO 模式反射逻辑，确保属性访问与 `core/models.py` 定义一致。
+    -   `ApiActionModel` 定义了 `module` (alias=`class`)。
+    -   确保执行器中使用 `action.module` 而非 `action.api_class`，防止 `AttributeError`。
+
+---
+
+**实施路线图建议**:
+1.  优先修复代码健壮性问题（属性一致性）。
+2.  实施配置管理增强，便于环境隔离。
+3.  引入 Redis 或文件锁解决并发问题。
+4.  逐步完善 Hooks 和 Trace ID。

docs/重构总结.md

@@ -0,0 +1,22 @@
+本次重构核心总结：升级为“模型驱动+混合模式”的自动化测试框架我们本次重构的目标是将现有框架从基于字典（dict）的松散操作，升级为一个结构严谨、易于扩展的现代化测试框架。其核心包含以下四大支柱：1.
+核心驱动力：Pydantic 模型层•目标：用强类型、带校验的模型对象取代脆弱的字典操作。•实现：创建 commons/models/case_model.py
+文件，并定义 CaseInfo 类。•关键收益：•健壮性：在执行测试前，通过模型实例化，对 YAML
+文件中的字段、类型、结构进行严格校验，提前发现拼写错误或格式问题。•可维护性：代码中不再出现 case.get("request")
+这类“魔法字符串”，而是通过 case.request 这样的属性访问，IDE 可以提供智能提示和补全，代码更清晰、更安全。•灵活性：支持使用
+alias，让 YAML 中的字段名（如 validate）与模型属性名（如 validate_data）解耦，使模型设计更符合 Python 规范。2.
+执行模式：支持混合模式（Hybrid Mode）•目标：让框架同时适应简单的数据驱动测试和复杂的业务流测试。•实现：•YAML 驱动模式：保留并优化
+TestAPI 类。它负责扫描 tests/features/ 目录下的 test_*.yaml 文件，并动态生成 pytest 用例。此模式非常适合单接口、多场景的数据验证。•手动脚本模式：允许在
+tests/flows/ 目录下直接编写 test_*.py 脚本。开发者可以像写普通 pytest 用例一样，通过导入业务方法来编排复杂的、跨多个接口的业务流程。3.
+架构设计：清晰的三层分离•目标：遵循最佳实践，分离关注点，让框架结构清晰，避免混乱。•实现：•数据层 (YAML + Pydantic Model)
+：定义测试的输入数据和预期结果（是什么）。•业务层/服务层 (api/*.py)：将原始的 HTTP 请求封装成具有业务含义的方法，如
+api.auth.login()。它定义了如何执行具体业务操作（怎么做）。•测试层 (TestAPI 或 test_*.py)
+：作为“导演”，负责编排测试流程。它从数据层获取数据，调用业务层的方法执行动作，并进行最终断言（测什么）。4.
+上下文与状态：统一的会话与变量池•目标：打通 YAML 驱动和手动脚本之间的数据壁垒，实现真正的端到端流程测试。•实现：•所有测试（无论来源）共享同一个
+core.session.Session 实例，确保 Cookie、Header 等会话状态的连续性。•所有测试共享同一个 commons.exchange.Exchange
+实例（变量交换器）。•关键收益：手动脚本（.py）中通过登录获取的 token，可以被无缝地注入到后续的 YAML 用例中；反之，YAML 用例提取的
+ID 也能被后续的 .py 脚本使用。重构后的标准执行流程（以 YAML 为例）：1.加载：TestAPI 扫描并加载 test_*.yaml
+文件。2.数据驱动：DataDriver 将 YAML 文件内容解析为多个独立的、参数化的测试用例。3.执行：在 pytest 的 test_func 内部： a.
+变量替换：exchanger.replace() 将用例中的 ${variable} 替换为实际值。 b. 模型校验：CaseInfo(**replaced_case_data) 将替换后的字典实例化为
+CaseInfo 模型对象，完成数据校验。（这是与旧流程最核心的区别） c. 请求发送：使用模型对象的数据发送请求 session.request(**
+case.request.model_dump())。 d. 变量提取：exchanger.extract() 从响应中提取数据，并存入全局变量池。 e.
+断言：validator.assert_all(case.validate_data) 使用模型中的断言数据进行校验。

utils/dirs_manager.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: dirs_manager
+@date: 2026/2/3 10:52
+@desc: 
+"""
+
+from pathlib import Path
+
+from core.settings import REQUIRED_DIRS
+
+
+def ensure_dirs_ok():
+    """
+    统一管理项目目录的创建逻辑
+    """
+    for folder in REQUIRED_DIRS:
+        # 使用 exist_ok=True 避免并发冲突
+        folder.mkdir(parents=True, exist_ok=True)
+
+
+def ensure_dir(path: Path) -> Path:
+    """确保路径存在并返回路径本身"""
+    if not isinstance(path, Path):
+        path = Path(path)
+    path.mkdir(parents=True, exist_ok=True)
+    return path

utils/report_handler.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: report_handler
+@date: 2026/2/3 13:51
+@desc: 
+"""
+import logging
+import subprocess
+import shutil
+
+from core.settings import ALLURE_TEMP, REPORT_DIR
+
+logger = logging.getLogger(__name__)
+
+
+def generate_allure_report() -> bool:
+    """
+    将 JSON 原始数据转换为 HTML 报告
+    """
+    if not ALLURE_TEMP.exists() or not any(ALLURE_TEMP.iterdir()):
+        logger.warning("未发现 Allure 测试数据，跳过报告生成。")
+        return False
+
+    # 检查环境是否有 allure 命令行工具
+    if not shutil.which("allure"):
+        logger.error("系统未安装 Allure 命令行工具，请先安装：https://allurereport.org/docs/")
+        return False
+
+    try:
+        logger.info("正在生成 Allure HTML 报告...")
+        # --clean 会清理掉 REPORT_DIR 里的旧报告
+        subprocess.run(
+            f'allure generate "{ALLURE_TEMP}" -o "{REPORT_DIR}" --clean',
+            shell=True,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+        logger.info(f"Allure 报告已生成至: {REPORT_DIR}")
+        return True
+    except subprocess.CalledProcessError as e:
+        logger.error(f"Allure 报告生成失败: {e.stderr}")
+        return False