From 2116016a0d36fc8378f048c493a071c865249dd4 Mon Sep 17 00:00:00 2001 From: CNWei Date: Wed, 11 Mar 2026 17:11:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(executor):=20=E9=87=8D=E6=9E=84=E7=94=A8?= =?UTF-8?q?=E4=BE=8B=E5=8A=A0=E8=BD=BD=E4=B8=8E=E6=89=A7=E8=A1=8C=E9=80=BB?= =?UTF-8?q?=E8=BE=91=EF=BC=8C=E6=94=AF=E6=8C=81=E5=8F=82=E6=95=B0=E5=8C=96?= =?UTF-8?q?=E5=8F=98=E9=87=8F=E4=BC=98=E5=85=88=E7=BA=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 引入 CaseEntity 包装器，实现数据模型与执行上下文解耦。 - 移除加载阶段的 deepcopy，优化大规模参数化用例的内存占用。 - 实现 perform 阶段的局部变量注入，确保参数化数据优先级高于全局缓存。 --- core/creator.py | 47 +++++++++--- core/executor.py | 19 ++++- core/settings.py | 10 ++- main.py | 108 ++++++++++++++++++++++----- pytest.ini | 25 +++++-- test_cases/answer/test_1_status.yaml | 30 ++++---- test_cases/request_model_test.yaml | 31 ++++---- 7 files changed, 201 insertions(+), 69 deletions(-) diff --git a/core/creator.py b/core/creator.py index c30cc2a..19adfff 100644 --- a/core/creator.py +++ b/core/creator.py @@ -13,6 +13,7 @@ import logging import allure from pathlib import Path +from dataclasses import dataclass from core import settings from core.executor import WorkflowExecutor from core.session import Session @@ -30,6 +31,12 @@ session = Session(settings.base_url) exchanger = Exchange(settings.DATA_DIR / "extract.yaml") # 指向 data/extract.yaml executor = WorkflowExecutor(session, exchanger) +@dataclass +class CaseEntity: + """用例执行实体：解耦模型数据与执行上下文""" + step_data: CaseInfo + row_context: dict[str, Any] + class TestTemplateBase: """ @@ -56,12 +63,12 @@ class CaseDataLoader: yield from base_path.rglob("test_*.yaml") @classmethod - def load_cases(cls, file_path: Path) -> List[CaseInfo]: + def load_cases(cls, file_path: Path) -> List[CaseEntity]: """ 加载单个 YAML 文件并转化为 CaseInfo 列表 包含参数化数据的自动拆解逻辑 """ - cases = [] + entities = [] try: # 1. 使用重构后的 YamlProcessor 加载原始字典 processor = FileHandle(file_path) @@ -69,13 +76,24 @@ class CaseDataLoader: if not raw_data: return [] + # 1. 提取参数化数据 + parametrize_data = raw_data.pop("parametrize", None) + + # 2. 实例化唯一的模板对象 (Pydantic 校验) + # 此时占位符 ${var} 会被 SmartInt/SmartDict 校验器放行 + template_case = CaseInfo(**raw_data) # 2. 检查是否存在参数化字段 - if "parametrize" in raw_data and isinstance(raw_data["parametrize"], list): - cases.extend(cls._parse_parametrize(raw_data)) + if parametrize_data and isinstance(parametrize_data, list) and len(parametrize_data) >= 2: + # 3. 参数化拆分 + headers = parametrize_data[0] + for row in parametrize_data[1:]: + row_map = dict(zip(headers, row)) + # 包装为实体，存入引用而非副本 + entities.append(CaseEntity(step_data=template_case, row_context=row_map)) else: - # 3. 普通单条用例封装 - cases.append(CaseInfo(**raw_data)) + # 普通用例，上下文为空 + entities.append(CaseEntity(step_data=template_case, row_context={})) except YamlLoadError: # YamlProcessor 已经记录了 error 日志，这里直接跳过 @@ -85,7 +103,7 @@ class CaseDataLoader: except Exception as e: logger.error(f"加载用例发生未知异常 [{file_path.name}]: {e}") - return cases + return entities @staticmethod def _parse_parametrize(raw_data: dict[str, Any]) -> List[CaseInfo]: @@ -121,9 +139,10 @@ class CaseDataLoader: return case_list @classmethod - def get_all_cases(cls, cases_dir: Union[str, Path]) -> List[CaseInfo]: + def get_all_cases(cls, cases_dir: Union[str, Path]) -> List[CaseEntity]: """ - 全量获取接口：供 CaseGenerator 调用 + 全量获取接口：供 CaseGenerator 调用 frank + """ all_cases = [] for file in cls.fetch_yaml_files(cases_dir): @@ -165,6 +184,8 @@ class CaseGenerator: # # # 4. 挂载 # method_name = f"test_{file_path.stem}" + case_title = case_info.row_context.get("title") or case_info.step_data.title + method_name = f"test_case_{index}_{case_info.title[:20]}" safe_name = "".join([c if c.isalnum() else "_" for c in method_name]) # setattr(target_cls, method_name, dynamic_test_method) @@ -173,8 +194,10 @@ class CaseGenerator: @staticmethod # def _create_case_method(case_template: Dict, fields: List[str], values: List[Any], ids: List[str]): - def _create_case_method(case_template: CaseInfo): + def _create_case_method(entity: CaseEntity): """封装具体的 pytest 执行节点""" + case_template = entity.step_data + context = entity.row_context # 预取 Allure 层级信息 # epic = case_template.get("epic", settings.allure_epic) @@ -194,7 +217,7 @@ class CaseGenerator: # current_params = dict(zip(fields, case_args)) # case_exec_data = {**case_template, **current_params} # case_title = current_params.get("title", "未命名用例") - case_title = case_template.title or "未命名用例" + case_title = context.get("title") or case_template.title or "未命名用例" # 日志记录 (利用 instance 标注来源) logger.info(f"🚀 [Runner] Class: {instance.__class__.__name__} | Case: {case_title}") @@ -202,7 +225,7 @@ class CaseGenerator: # 执行与断言 allure.dynamic.title(case_title) # executor.perform(case_exec_data) - executor.perform(case_template) + executor.perform(case_template,context=context) # 手动链路装饰 (Allure) # run_actual_case = allure.epic(epic)(run_actual_case) diff --git a/core/executor.py b/core/executor.py index b52a77a..27d27b2 100644 --- a/core/executor.py +++ b/core/executor.py @@ -7,7 +7,7 @@ import logging import importlib -from typing import Any, List +from typing import Any, List, Optional from pydantic import TypeAdapter @@ -28,9 +28,20 @@ class WorkflowExecutor: self.session = session self.exchanger = exchanger - def perform(self, case_info: CaseInfo) -> Any: + def perform(self, case_info: CaseInfo,context: Optional[dict[str, Any]] = None) -> Any: """执行单个用例：支持直接请求和PO模式调用""" + context = context or {} + + # 1. 局部变量优先级注入 + # 备份全局缓存，将当前行数据合并进去 + old_cache = self.exchanger._variable_cache.copy() + self.exchanger._variable_cache.update(context) + try: + # 2. 动态更新标题（如果 context 中包含 title） + current_title = context.get("title") or case_info.title + logger.info(f"🚀 执行用例: {current_title}") + # raw_data = case_info.model_dump(by_alias=True, exclude_none=True) # 1. 变量替换（将 ${var} 替换为真实值） # rendered_dict = self.exchanger.replace(raw_data) @@ -62,6 +73,10 @@ class WorkflowExecutor: except Exception as e: logger.error(f"用例执行失败: {case_info.title} | 原因: {e}", exc_info=True) raise + finally: + # 4. 关键：清理现场，还原全局变量池 + self.exchanger._variable_cache = old_cache + def _execute_po_method(self, action: ApiActionModel): """核心反射逻辑：根据字符串动态加载 api/ 目录下的类并执行方法""" diff --git a/core/settings.py b/core/settings.py index b798255..8e7f7c7 100644 --- a/core/settings.py +++ b/core/settings.py @@ -21,7 +21,7 @@ load_dotenv() # --- 目录配置 --- TEST_CASE_DIR = BASE_DIR / "test_cases" -EXTRACT_CACHE = BASE_DIR / "data/extract.yaml" + OUTPUT_DIR = BASE_DIR / "outputs" SCREENSHOT_DIR = OUTPUT_DIR / "screenshots" @@ -33,8 +33,14 @@ REPORT_DIR = BASE_DIR / "reports" CONFIG_DIR = BASE_DIR / "config" DATA_DIR = BASE_DIR / "data" + +# 需要初始化的目录列表 +REQUIRED_DIRS = [LOG_DIR, LOG_BACKUP_DIR, ALLURE_TEMP, SCREENSHOT_DIR] + # 核心 API 目录路径 API_PACKAGE = "api" +LOG_SOURCE = LOG_DIR / "pytest.log" +EXTRACT_CACHE = BASE_DIR / "data/extract.yaml" # 可选：显式映射（类名 -> 完整模块路径），解决文件名不规则的问题 API_MAP = { @@ -55,4 +61,4 @@ rsa_private = "" if __name__ == '__main__': print(BASE_DIR) - print(BASE_DIR, db_host, db_port, db_user, db_password, db_database) + diff --git a/main.py b/main.py index be3fb54..cbd89a0 100644 --- a/main.py +++ b/main.py @@ -1,22 +1,90 @@ -import os +import shutil +import datetime +from pathlib import Path + import pytest -if __name__ == '__main__': - # 定义报告和临时文件目录 - reports_dir = "reports" - temp_dir = os.path.join(reports_dir, "temp") - html_dir = os.path.join(reports_dir, "html") - - # 1. 执行 Pytest 测试 - # -v: 输出详细信息 - # --alluredir: 指定 Allure 临时数据目录 - # -c: 指定 pytest.ini 配置文件路径 - pytest.main([ - "-v", - "tests/", # 明确指定测试目录 - f"--alluredir={temp_dir}", - "-c=config/pytest.ini" - ]) - - # 2. 生成 Allure HTML 报告 - os.system(f'allure generate {temp_dir} -o {html_dir} --clean') +from core.settings import LOG_SOURCE, LOG_BACKUP_DIR, ALLURE_TEMP +# from core.enums import AppPlatform +from utils.dirs_manager import ensure_dirs_ok +from utils.report_handler import generate_allure_report + + +# netstat -ano | findstr :4723 +# taskkill /PID 12345 /F + +def _archive_logs(): + """ + 在测试开始前，归档上一次运行的日志文件。 + 此时没有任何句柄占用，move 操作是 100% 安全的。 + """ + # 4. 备份日志 (无论测试是否崩溃都执行) + if LOG_SOURCE.exists() and LOG_SOURCE.stat().st_size > 0: + now = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') + backup_path = LOG_BACKUP_DIR / f"pytest_{now}.log" + try: + # 移动并重命名 + shutil.move(str(LOG_SOURCE), str(backup_path)) + print(f"已自动归档上次运行的日志: {backup_path}") + # shutil.copy2(LOG_SOURCE, backup_path) + # print(f"日志已备份至: {backup_path}") + _clean_old_logs(LOG_BACKUP_DIR) + except Exception as e: + print(f"归档旧日志失败 (可能被外部编辑器打开): {e}") + else: + print("未找到原始日志文件，跳过备份。") + + +# 日志清理 +def _clean_old_logs(backup_dir, keep_count=10): + files = sorted(Path(backup_dir).glob("pytest_*.log"), key=lambda p: p.stat().st_mtime) + while len(files) > keep_count: + file_to_remove = files.pop(0) + try: + file_to_remove.unlink(missing_ok=True) + except OSError as e: + print(f"清理旧日志失败 {file_to_remove}: {e}") + + +def _clean_temp_dirs(): + """ + 可选：如果你想在测试前清理掉旧的临时文件 + """ + if ALLURE_TEMP.exists(): + shutil.rmtree(ALLURE_TEMP) + # 加上 ignore_errors 是为了防止文件被占用导致整个测试无法启动 + shutil.rmtree(ALLURE_TEMP, ignore_errors=True) + ALLURE_TEMP.mkdir(parents=True, exist_ok=True) + + +def main(): + try: + # 1. 创建目录 + ensure_dirs_ok() + + # 2. 处理日志 + _archive_logs() + + # 3. 执行 Pytest + + args = [ + "test_cases", + "-x", # 注意：-x 表示遇到错误立即停止，如果是全量回归建议去掉 -x + "-v", + f"--alluredir={ALLURE_TEMP}", + # f"--platform={AppPlatform.ANDROID.value}", + # "--caps_name=wan_android" + ] + pytest.main(args) + + # 4. 生成报告 + generate_allure_report() + except Exception as e: + print(f"自动化测试执行过程中发生异常: {e}") + + finally: + print("Time-of-check to Time-of-use") + + +if __name__ == "__main__": + main() diff --git a/pytest.ini b/pytest.ini index 57ff75e..96b642f 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,11 +1,22 @@ [pytest] -testpaths = tests -addopts = -q --show-capture=no +addopts = -q --show-capture=no --reruns 2 --reruns-delay 1 +# 1. 开启实时控制台日志 +log_cli = True +log_cli_level = INFO +log_cli_format = %(asctime)s %(levelname)-5s [%(name)s] - %(message)s +log_cli_date_format = %H:%M:%S -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 %H:%M:%S %p +# 2. 开启日志文件记录 +log_file = outputs/logs/pytest.log +log_file_level = INFO +log_file_format = %(asctime)s %(levelname)-5s [%(name)s] %(module)s.%(funcName)s:%(lineno)d - %(message)s +log_file_date_format = %Y-%m-%d %H:%M:%S -disable_test_id_escaping_and_forfeit_all_rights_to_community_support = true \ No newline at end of file +# 3. 基础配置 +# 解决中文测试用例显示为乱码（Unicode）的问题 +disable_test_id_escaping_and_forfeit_all_rights_to_community_support = True + +# 限制 Pytest 搜索范围，提升启动速度 +testpaths = test_cases +python_files = test_*.py \ No newline at end of file diff --git a/test_cases/answer/test_1_status.yaml b/test_cases/answer/test_1_status.yaml index 2d297d4..7af0400 100644 --- a/test_cases/answer/test_1_status.yaml +++ b/test_cases/answer/test_1_status.yaml @@ -12,6 +12,9 @@ request: User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36 Edg/131.0.0.0 Referer: http://119.91.19.171:40065/users/login Accept-Encoding: gzip, deflate + json_body: + username: "${username}" + password: "${password}" extract: # 提取变量 msg: - "json" @@ -19,20 +22,21 @@ extract: # 提取变量 - 0 validate: - - check: status_code # 检查的对象（或是变量名） - assert: equals # 断言方法 - expect: 200 # 期望值 - msg: "校验接口状态码" # 描述（可选） + - check: status_code + assert: ${status_assert} # <--- 动态断言方法 + expect: ${status_expect} # <--- 动态期望值 + msg: "校验接口状态码" - - check: $.msg - assert: contains - expect: "Success" + - check: message + assert: ${msg_assert} # <--- 动态断言方法 + expect: ${msg_expect} # <--- 动态期望值 msg: "检查返回消息" parametrize: # 数据驱动测试 - - [ "title","username","password","msg" ] # 变量名 - - [ "测试1","user1","pass1","200" ] # 变量值 - - [ "测试2","user2","pass2","300" ] # 变量值 - - [ "测试3","user3","pass3","200" ] # 变量值 - - [ "测试4","user4","pass4","200" ] # 变量值 - - [ "测试5","user5","pass5","200" ] # 变量值 \ No newline at end of file + # 定义列名，包括了断言方法和期望值 + - [ "title", "username", "password", "status_assert", "status_expect", "msg_assert", "msg_expect" ] + # 定义每一行的数据，现在可以为每次运行指定不同的断言逻辑 + - [ "场景1: 成功-状态码相等-消息包含Success", "user1", "pass1", "equals", 200, "contains", "Success" ] + - [ "场景2: 失败-状态码不相等-消息不包含Error", "user2", "pass2", "not_equals", 200, "not_contains", "Error" ] + - [ "场景3: 成功-状态码大于199-消息相等", "user3", "pass3", "greater_than", 199, "equals", "Success" ] + - [ "场景4: 失败-状态码小于500-消息为空", "user4", "pass4", "less_than", 500, "is_empty", "" ] \ No newline at end of file diff --git a/test_cases/request_model_test.yaml b/test_cases/request_model_test.yaml index bfd3718..b4caeed 100644 --- a/test_cases/request_model_test.yaml +++ b/test_cases/request_model_test.yaml @@ -12,25 +12,30 @@ request: User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36 Edg/131.0.0.0 Referer: http://119.91.19.171:40065/users/login Accept-Encoding: gzip, deflate + json_body: { username:${ username },password:${ password } } extract: # 提取变量 msg: - "json" - "$.msg" - 0 -validate: - - check: status_code # 检查的对象（或是变量名） - assert: equals # 断言方法 - expect: 200 # 期望值 - msg: "校验接口状态码" # 描述（可选） - - check: $.msg - assert: contains - expect: "Success" +validate: + - check: status_code + assert: ${status_assert} # <--- 动态断言方法 + expect: ${status_expect} # <--- 动态期望值 + msg: "校验接口状态码" + + - check: message + assert: ${msg_assert} # <--- 动态断言方法 + expect: ${msg_expect} # <--- 动态期望值 msg: "检查返回消息" + parametrize: # 数据驱动测试 - - [ "title","username","password","msg" ] # 变量名 - - [ "测试1","user1","pass1","200" ] # 变量值 - - [ "测试2","user2","pass2","300" ] # 变量值 - - [ "测试3","user3","pass3","200" ] # 变量值 - - [ "测试4","user4","pass4","200" ] # 变量值 \ No newline at end of file + # 定义列名，包括了断言方法和期望值 + - [ "title", "username", "password", "status_assert", "status_expect", "msg_assert", "msg_expect" ] + # 定义每一行的数据，现在可以为每次运行指定不同的断言逻辑 + - [ "场景1: 成功-状态码相等-消息包含Success", "user1", "pass1", "equals", 200, "contains", "Success" ] + - [ "场景2: 失败-状态码不相等-消息不包含Error", "user2", "pass2", "not_equals", 200, "not_contains", "Error" ] + - [ "场景3: 成功-状态码大于199-消息相等", "user3", "pass3", "greater_than", 199, "equals", "Success" ] + - [ "场景4: 失败-状态码小于500-消息为空", "user4", "pass4", "less_than", 500, "is_empty", "" ] \ No newline at end of file