diff --git a/.gitignore b/.gitignore index 29eada9..8873681 100644 --- a/.gitignore +++ b/.gitignore @@ -2,7 +2,26 @@ .idea/ .venv/ poetry.lock -.pytest_cache/ -report/ + +logs/ + +# --- 依赖与环境 --- +.venv +venv/ +node_modules/ +uv.lock + +# --- 屏蔽outputs --- +outputs/ + +# --- Allure 报告 --- temp/ -logs/ \ No newline at end of file +reports/ +.allure/ + +# --- pytest缓存 --- +.pytest_cache/ +.allure_cache/ + +# --- 配置文件 --- +.env \ No newline at end of file diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..2c07333 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.11 diff --git a/core/creator.py b/core/creator.py new file mode 100644 index 0000000..a38fb80 --- /dev/null +++ b/core/creator.py @@ -0,0 +1,124 @@ +#!/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 +import pytest +from pathlib import Path +from core import settings +from core.executor import WorkflowExecutor +from core.session import Session +from commons.exchange import Exchange +from commons.file_processors.yaml_processor import YamlProcessor as FileHandle +from typing import Any, Dict, List, Type, Generator, Tuple + +logger = logging.getLogger(__name__) + +# 初始化全局组件 +session = Session(settings.base_url) +exchanger = Exchange(settings.DATA_DIR / "extract.yaml") # 指向 data/extract.yaml +executor = WorkflowExecutor(session, exchanger) + + +class TestTemplateBase: + """ + 具体的测试用例容器。 + 此映射类不包含任何逻辑方法，仅用于承载由 Loader 挂载的 test_* 方法。 + """ + pass + + +class CaseDataLoader: + """ + 职责 1: 数据加载器 + 负责与底层存储（YAML文件）打交道，输出标准化的原始数据对象 + """ + + @staticmethod + def fetch_yaml_cases(cases_dir: str) -> Generator[Tuple[Path, Dict], None, None]: + """扫描目录并迭代返回 (文件路径, 原始内容)""" + yaml_files = Path(cases_dir).glob("**/test_*.yaml") + for file_path in yaml_files: + yield file_path, FileHandle(file_path) + + @staticmethod + def parse_parametrize(raw_data: Dict, default_name: str) -> Tuple[List[str], List[List[Any]], List[str]]: + """解析参数化结构，返回 (字段名列表, 数据值列表, ID列表)""" + if "parametrize" in raw_data: + fields = raw_data["parametrize"][0] + values = raw_data["parametrize"][1:] + ids = [f"{v[0]}" for v in values] + else: + fields = ["case_data"] + values = [[raw_data]] + ids = [raw_data.get("title", default_name)] + return fields, values, ids + + +class CaseGenerator: + """ + 职责 2: 用例构造工厂 + 负责将数据转化为 pytest 装饰的方法，并挂载到目标类 + """ + + @classmethod + def build_and_register(cls, target_cls: Type[TestTemplateBase], cases_dir: str): + # 1. 通过 Loader 获取数据 + for file_path, raw_data in CaseDataLoader.fetch_yaml_cases(cases_dir): + # 2. 解析参数化信息 + fields, values, ids = CaseDataLoader.parse_parametrize(raw_data, file_path.stem) + + # 3. 生成执行函数 (闭包) + dynamic_test_method = cls._create_case_method(raw_data, fields, values, ids) + + # 4. 挂载 + method_name = f"test_{file_path.stem}" + setattr(target_cls, method_name, dynamic_test_method) + logger.debug(f"Successfully registered: {method_name}") + + @staticmethod + def _create_case_method(case_template: Dict, fields: List[str], values: List[Any], ids: List[str]): + """封装具体的 pytest 执行节点""" + + # 预取 Allure 层级信息 + epic = case_template.get("epic", settings.allure_epic) + feature = case_template.get("feature", settings.allure_feature) + story = case_template.get("story", settings.allure_story) + + @allure.epic(epic) + @allure.feature(feature) + @allure.story(story) + @pytest.mark.parametrize("case_args", values, ids=ids) + def build_actual_case(instance: TestTemplateBase, case_args: List[Any]): + # 数据组装 + current_params = dict(zip(fields, case_args)) + case_exec_data = {**case_template, **current_params} + case_title = current_params.get("title", "未命名用例") + + # 日志记录 (利用 instance 标注来源) + logger.info(f"🚀 [Runner] Class: {instance.__class__.__name__} | Case: {case_title}") + + # 执行与断言 + allure.dynamic.title(case_title) + executor.perform(case_exec_data) + + # 手动链路装饰 (Allure) + # run_actual_case = allure.epic(epic)(run_actual_case) + # run_actual_case = allure.feature(feature)(run_actual_case) + # run_actual_case = allure.story(story)(run_actual_case) + + return build_actual_case + + +if __name__ == '__main__': + # --- 引导执行 --- + CaseGenerator.build_and_register(TestTemplateBase, settings.TEST_CASE_DIR) diff --git a/core/executor.py b/core/executor.py new file mode 100644 index 0000000..fd20113 --- /dev/null +++ b/core/executor.py @@ -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 \ No newline at end of file diff --git a/commons/session.py b/core/session.py similarity index 100% rename from commons/session.py rename to core/session.py diff --git a/commons/settings.py b/core/settings.py similarity index 59% rename from commons/settings.py rename to core/settings.py index 3b55e0c..c9285d7 100644 --- a/commons/settings.py +++ b/core/settings.py @@ -9,21 +9,31 @@ @date: 2025/2/23 21:34 @desc: """ -from pathlib import Path import os +from pathlib import Path + from dotenv import load_dotenv +BASE_DIR = (Path(__file__)).resolve().parents[1] + load_dotenv() -root_path = (Path(__file__)).resolve().parents[1] +# --- 目录配置 --- +TEST_CASE_DIR = BASE_DIR / "test_cases" -base_url = os.getenv("BASE_URL") -cases_dir = rf"{root_path}\TestCases\answer" -exchanger = rf"{root_path}\extract.yaml" -id_path = rf"{root_path}\id.yaml" +OUTPUT_DIR = BASE_DIR / "outputs" +SCREENSHOT_DIR = OUTPUT_DIR / "screenshots" +LOG_DIR = OUTPUT_DIR / "logs" +LOG_BACKUP_DIR = LOG_DIR / "backups" + +ALLURE_TEMP = BASE_DIR / "temp" +REPORT_DIR = BASE_DIR / "reports" +CONFIG_DIR = BASE_DIR / "config" +DATA_DIR = BASE_DIR / "data" test_suffix = "yaml" +base_url = os.getenv("BASE_URL") db_host = os.getenv("DB_HOST") # ip db_port = os.getenv("DB_PORT") # 端口 db_user = os.getenv("DB_USER") # 用户名 @@ -38,5 +48,5 @@ rsa_public = "" rsa_private = "" if __name__ == '__main__': - print(root_path) - print(base_url,db_host,db_port,db_user,db_password,db_database) + print(BASE_DIR) + print(BASE_DIR, db_host, db_port, db_user, db_password, db_database) diff --git a/pyproject.toml b/pyproject.toml index 949a5c7..382c99b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,29 +1,24 @@ -[tool.poetry] +[project] name = "interfaceautotest" version = "0.1.0" description = "" -authors = ["NianJiu "] readme = "README.md" - -[tool.poetry.dependencies] -python = "^3.10" +requires-python = ">=3.11" +dependencies = [ + "requests>=2.32.3", + "pyyaml>=6.0.1", + "pytest>=8.3.3", + "jsonpath>=0.82.2", + "pymysql>=1.1.1", + "pytest-result-log>=1.2.2", + "allure-pytest>=2.13.5", + "cryptography>=44.0.2", + "python-dotenv>=0.9.9", + "pydantic>=2.12.5", + "lxml>=6.0.2", +] -requests = "^2.32.3" -pyyaml = "^6.0.2" -pytest = "^8.3.3" -jsonpath = "^0.82.2" -pymysql = "^1.1.1" -pytest-result-log = "^1.2.2" -allure-pytest = "^2.13.5" -cryptography = "^44.0.2" -dotenv = "^0.9.9" -[build-system] -requires = ["poetry-core"] -build-backend = "poetry.core.masonry.api" - - -[[tool.poetry.source]] -name = "tsinghua" +[[tool.uv.index]] url = "https://pypi.tuna.tsinghua.edu.cn/simple" -priority = "primary" \ No newline at end of file +default = true diff --git a/pytest.ini b/pytest.ini index 556fa84..f6d23d8 100644 --- a/pytest.ini +++ b/pytest.ini @@ -5,6 +5,6 @@ addopts = -q --show-capture=no log_file = logs/pytest.log log_file_level = debug log_file_format = %(asctime)s [%(name)s] %(levelname)s %(module)s.%(funcName)s:%(lineno)d - %(message)s -log_file_date_format = %m/%d/%Y %I:%M:%S %p +log_file_date_format = %m/%d/%Y %H:%M:%S %p disable_test_id_escaping_and_forfeit_all_rights_to_community_support = true \ No newline at end of file