diff --git a/commons/case_handler.py b/commons/case_handler.py deleted file mode 100644 index b04c065..0000000 --- a/commons/case_handler.py +++ /dev/null @@ -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) diff --git a/commons/files.py b/commons/files.py deleted file mode 100644 index 31c6cfa..0000000 --- a/commons/files.py +++ /dev/null @@ -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}") diff --git a/conftest.py b/conftest.py index 5d48d96..45d97ab 100644 --- a/conftest.py +++ b/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}") diff --git a/core/context.py b/core/context.py index a64ca13..1998455 100644 --- a/core/context.py +++ b/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 @@ -25,7 +25,7 @@ class VariableStore: 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): """测试结束时统一写盘""" diff --git a/core/creator.py b/core/creator.py index ec88ee5..78c66b4 100644 --- a/core/creator.py +++ b/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 @@ -142,7 +144,7 @@ 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 @@ -150,8 +152,7 @@ class CaseGenerator: """封装具体的 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 +162,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: # 可以在这里记录更详细的运行上下文快照 @@ -176,4 +177,4 @@ if __name__ == '__main__': # print(CaseDataLoader.get_all_cases(TEST_CASE_DIR)) # --- 引导执行 --- - CaseGenerator.build_and_register(TestTemplateBase, settings.TEST_CASE_DIR) + CaseGenerator.build_and_register(TestTemplateBase, settings.TEST_CASE_DIR) \ No newline at end of file diff --git a/core/executor.py b/core/executor.py index 0c75d4c..8aafc26 100644 --- a/core/executor.py +++ b/core/executor.py @@ -139,4 +139,4 @@ class WorkflowExecutor: 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) + CaseValidator.validate(resp, final_validate_data) \ No newline at end of file diff --git a/test_cases/test_collector.py b/test_cases/test_collector.py index 53c166a..38998c1 100644 --- a/test_cases/test_collector.py +++ b/test_cases/test_collector.py @@ -27,4 +27,4 @@ try: except Exception as e: logger.critical(f"--- [Collector] 动态生成测试用例时发生致命错误，测试执行中止 ---", exc_info=True) # 抛出异常，让 pytest 捕获并报告为收集错误 (Collection Error) - raise RuntimeError("测试用例收集失败，请检查日志中的详细错误信息。") from e \ No newline at end of file + raise RuntimeError("测试用例收集失败，请检查日志中的详细错误信息。") from e diff --git a/utils/dirs_manager.py b/utils/dirs_manager.py new file mode 100644 index 0000000..f0c308f --- /dev/null +++ b/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 diff --git a/utils/report_handler.py b/utils/report_handler.py new file mode 100644 index 0000000..843ead8 --- /dev/null +++ b/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