From 483a31793d8173f8b82eb984c1bfbea5d33e84a7 Mon Sep 17 00:00:00 2001 From: CNWei Date: Tue, 3 Feb 2026 17:46:48 +0800 Subject: [PATCH] =?UTF-8?q?refactor(core/utils):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E8=A3=85=E9=A5=B0=E5=99=A8=E6=9E=B6=E6=9E=84=E4=B8=8E=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E8=BF=BD=E8=B8=AA=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 优化 将 trace_step 移至 decorators.py，并引入 ContextVar 实现日志层级缩进。 - 新增 完善 StepTracer 和 action_screenshot 的 Docstrings，明确参数含义。 - 移除 清理了 logger.py - 优化 main.py 中重复的目录创建逻辑及旧版冗余注释。 - 规范 修正函数命名，提升代码在 BasePage 方案下的复用性。 --- conftest.py | 29 +++++--- core/settings.py | 20 +++--- main.py | 69 ++++++++++++------ test_cases/conftest.py | 52 -------------- test_cases/test_settings.py | 15 ++-- utils/decorators.py | 139 +++++++++++++++++++++++++++++++++--- utils/dirs_manager.py | 32 +++++++++ utils/logger.py | 58 --------------- utils/report_handler.py | 48 +++++++++++++ 9 files changed, 299 insertions(+), 163 deletions(-) delete mode 100644 test_cases/conftest.py create mode 100644 utils/dirs_manager.py create mode 100644 utils/report_handler.py diff --git a/conftest.py b/conftest.py index 2c7fdc8..59ff4bd 100644 --- a/conftest.py +++ b/conftest.py @@ -16,14 +16,22 @@ import allure import pytest from core.run_appium import start_appium_service, stop_appium_service from core.driver import CoreDriver -from core.settings import ANDROID_CAPS +from utils.dirs_manager import ensure_dirs_ok + + +def pytest_configure(config): + """ + Pytest 钩子函数：在测试收集开始前执行 + """ + # ensure_dirs_ok() + ... @pytest.fixture(scope="session") def app_server(): """ 第一层：管理 Appium Server 进程。 - 利用你原本 start_appium_service 里的 40 次轮询和 sys.exit(1) 逻辑。 + 利用start_appium_service 里的 40 次轮询和 sys.exit(1) 逻辑。 """ # 启动服务 service = start_appium_service() @@ -67,15 +75,17 @@ def pytest_exception_interact(node, call, report): 我们在这里手动把错误信息喂给 logging。 """ # 获取名为 'Error' 的记录器，它会遵循 pytest.ini 中的 log_file 配置 - logger = logging.getLogger("Error") + logger = logging.getLogger("pytest") if report.failed: # 获取详细的错误堆栈（包含 assert 的对比信息） - exc_info = call.excinfo.getrepr(style='no-locals') - name = f"异常截图_{secrets.token_hex(8)}" + # long,short,no-locals + exc_info = call.excinfo.getrepr(style='short') + screenshot_name = f"异常截图_{secrets.token_hex(4)}" - logger.error(f"TEST FAILED: {node.nodeid}") - logger.error(f"截图名称: {name}") + logger.error("\n" + "=" * 40 + " TEST FAILED " + "=" * 40) + logger.error(f"Node ID: {node.nodeid}") + logger.error(f"截图名称: {screenshot_name}") logger.error(f"详细错误信息如下:\n{exc_info}") # 3. 自动截图：尝试从 fixture 中获取 driver @@ -83,10 +93,13 @@ def pytest_exception_interact(node, call, report): driver = node.funcargs.get("driver") if driver: try: + # 同时保存到 Allure 和本地（如果你需要本地有一份备份） + screenshot_data = driver.get_screenshot_as_png() allure.attach( - driver.get_screenshot_as_png(), + screenshot_data, name=name, attachment_type=allure.attachment_type.PNG ) except Exception as e: logger.error(f"执行异常截图失败: {e}") + logger.error("=" * 93 + "\n") diff --git a/core/settings.py b/core/settings.py index 7172038..1f28678 100644 --- a/core/settings.py +++ b/core/settings.py @@ -9,35 +9,37 @@ @date: 2026/1/19 16:54 @desc: """ -import os -# core/settings.py + from pathlib import Path # 项目根目录 (core 的上一级) # BASE_DIR = Path(__file__).parent.parent BASE_DIR = Path(__file__).resolve().parents[1] # 获取根路径（绝对路径） -# print(BASE_DIR) + # --- 目录配置 --- 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 / "report" -SCREENSHOT_DIR = OUTPUT_DIR / "screenshots" -# 确保必要的目录存在 -for folder in [LOG_DIR, LOG_BACKUP_DIR, ALLURE_TEMP, SCREENSHOT_DIR]: - folder.mkdir(parents=True, exist_ok=True) + + +# 需要初始化的目录列表 (将在入口文件(mani.py)或 conftest.py 中被调用创建) +REQUIRED_DIRS = [LOG_DIR, LOG_BACKUP_DIR, ALLURE_TEMP, SCREENSHOT_DIR] # --- 文件路径 --- LOG_SOURCE = LOG_DIR / "pytest.log" # --- 启动 Appium 最大尝试次数 --- MAX_RETRIES = 40 + # --- 业务常量 (可选) --- +APPIUM_SERVER = "http://127.0.0.1:4723" + +# --- 核心配置 --- IMPLICIT_WAIT_TIMEOUT = 10 EXPLICIT_WAIT_TIMEOUT = 10 -APPIUM_SERVER = "http://127.0.0.1:4723" -# --- 核心配置 --- APPIUM_HOST = "127.0.0.1" APPIUM_PORT = 4723 diff --git a/main.py b/main.py index e53d6c3..c82534c 100644 --- a/main.py +++ b/main.py @@ -9,53 +9,80 @@ @date: 2026/1/13 16:54 @desc: """ -import os import shutil -import subprocess import datetime from pathlib import Path import pytest -from core.settings import LOG_SOURCE, LOG_BACKUP_DIR, ALLURE_TEMP, REPORT_DIR +from core.settings import LOG_SOURCE, LOG_BACKUP_DIR, ALLURE_TEMP +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=os.path.getmtime) + 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: - os.remove(file_to_remove) + 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: - # 2. 执行 Pytest - # 建议保留你之前配置的 -s -v 等参数 + # 1. 创建目录 + ensure_dirs_ok() + + # 2. 处理日志 + _archive_logs() + + # 3. 执行 Pytest # 注意：-x 表示遇到错误立即停止，如果是全量回归建议去掉 -x pytest.main(["test_cases", "-x", "-v", f"--alluredir={ALLURE_TEMP}"]) - # 3. 生成报告 - if ALLURE_TEMP.exists(): - # 使用 subprocess 替代 os.system，更安全且跨平台兼容性更好 - subprocess.run(f'allure generate {ALLURE_TEMP} -o {REPORT_DIR} --clean', shell=True, check=False) + # 4. 生成报告 + generate_allure_report() except Exception as e: print(f"自动化测试执行过程中发生异常: {e}") finally: - # 4. 备份日志 (无论测试是否崩溃都执行) - if LOG_SOURCE.exists(): - now = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') - backup_path = LOG_BACKUP_DIR / f"pytest_{now}.log" - shutil.copy2(LOG_SOURCE, backup_path) - print(f"日志已备份至: {backup_path}") - _clean_old_logs(LOG_BACKUP_DIR) - else: - print("未找到原始日志文件，跳过备份。") + print("Time-of-check to Time-of-use") if __name__ == "__main__": diff --git a/test_cases/conftest.py b/test_cases/conftest.py deleted file mode 100644 index 6d1a552..0000000 --- a/test_cases/conftest.py +++ /dev/null @@ -1,52 +0,0 @@ -#!/usr/bin/env python -# coding=utf-8 - -""" -@author: CNWei,ChenWei -@Software: PyCharm -@contact: t6g888@163.com -@file: conftest -@date: 2026/1/19 14:08 -@desc: -""" -import pytest -import allure -from pathlib import Path - - -@pytest.hookimpl(tryfirst=True, hookwrapper=True) -def pytest_runtest_makereport(item, call): - """ - 本钩子函数会在每个测试阶段（setup, call, teardown）执行后被调用。 - item: 测试用例对象 - call: 测试执行阶段的信息 - """ - # 1. 先执行常规的用例报告生成逻辑 - outcome = yield - report = outcome.get_result() - - # 2. 我们只关注测试执行阶段 ("call") - # 如果该阶段失败了（failed），则触发截图 - if report.when == "call" and report.failed: - # 3. 从测试用例中获取 driver 实例 - # 假设你在 fixture 中注入的参数名为 'driver' - driver_instance = item.funcargs.get("driver") - - if driver_instance: - try: - # 4. 调用你在 CoreDriver 中实现的底层截图方法 - # 这里的 name 我们可以动态取测试用例的名字 - case_name = item.name - file_path = driver_instance.full_screen_screenshot(name=f"CRASH_{case_name}") - - # 5. 如果路径存在，将其关联到 Allure 报告 - if file_path: - p = Path(file_path) - if p.exists(): - allure.attach.file( - source=p, - name="【故障现场自动截图】", - attachment_type=allure.attachment_type.PNG - ) - except Exception as e: - print(f"故障自动截图执行失败: {e}") \ No newline at end of file diff --git a/test_cases/test_settings.py b/test_cases/test_settings.py index ce01f0b..80c5e17 100644 --- a/test_cases/test_settings.py +++ b/test_cases/test_settings.py @@ -11,20 +11,23 @@ """ import logging -from utils.logger import trace_step +import allure + +from utils.decorators import step_trace logger = logging.getLogger(__name__) - -@trace_step("验证失败",) +@allure.epic("中银国际移动端重构项目") +@allure.feature("登认证模块") +@step_trace("验证失败",) def test_settings_page_display(driver): """ 测试设置页面是否成功加载 """ # 此时 driver 已经通过 fixture 完成了初始化 - current_act = driver.driver.current_activity + current_act = driver.current_activity logger.info(f"捕获到当前 Activity: {current_act}") - assert ".unihome.UniHomeLauncher" in current_act + assert ".app.main.launcher.LauncherActivity" in current_act def test_wifi_entry_exists(driver): @@ -34,4 +37,4 @@ def test_wifi_entry_exists(driver): # 这里的 driver 就是 appium.webdriver.Remote 实例 # 假设我们要查找“网络”相关的 ID # el = driver.find_element(by='id', value='android:id/title') - assert driver.driver.session_id is not None \ No newline at end of file + assert driver.session_id is not None \ No newline at end of file diff --git a/utils/decorators.py b/utils/decorators.py index c23fbe7..634b88a 100644 --- a/utils/decorators.py +++ b/utils/decorators.py @@ -1,12 +1,85 @@ +#!/usr/bin/env python +# coding=utf-8 + +""" +@author: CNWei,ChenWei +@Software: PyCharm +@contact: t6g888@163.com +@file: decorators +@date: 2026/1/15 11:30 +@desc: +""" + import logging +import time +import inspect from functools import wraps from typing import Union, Callable +from contextvars import ContextVar +from contextlib import ContextDecorator from core.custom_expected_conditions import get_condition logger = logging.getLogger(__name__) +# 定义一个上下文变量，初始值为 0 +indent_var = ContextVar("indent_level", default=0) + + +class StepTracer(ContextDecorator): + """ + 既是装饰器也是上下文管理器 + 职责：负责计时、日志格式化和异常记录 + """ + + def __init__(self, step_desc, source='wrapper', func_info=None): + self.step_desc = step_desc + self.logger = logging.getLogger(source) + self.func_info = func_info + self.start_t = None + + def __enter__(self): + # 1. 获取当前层级并计算前缀 + level = indent_var.get() + # 使用 " " (空格) 或 "│ " 作为缩进符号 + self.prefix = "│ " * level + + self.start_t = time.perf_counter() + info = f" | 方法: {self.func_info}" if self.func_info else "" + # self.logger.info(f"[步骤开始] | {self.step_desc}{info}") + self.logger.info(f"{self.prefix}┌── [步骤开始] | {self.step_desc}{info}") + + # 2. 进入下一层，层级 +1 + indent_var.set(level + 1) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + # 3. 恢复层级，层级 -1 + level = indent_var.get() - 1 + indent_var.set(level) + + duration = time.perf_counter() - self.start_t + + prefix = "│ " * level + + if exc_type: + # 异常发生 + # self.logger.error( + # f"[步骤失败] {self.step_desc} | 耗时: {duration:.2f}s | 异常: {exc_type.__name__}" + # ) + self.logger.error( + f"{prefix}└── [步骤失败] {self.step_desc} | 耗时: {duration:.2f}s | 异常: {exc_type.__name__}" + ) + else: + # 执行成功 + # self.logger.info(f"[步骤成功] {self.step_desc} | 耗时: {duration:.2f}s") + self.logger.info( + f"{prefix}└── [步骤成功] {self.step_desc} | 耗时: {duration:.2f}s" + ) + # return False 确保异常继续向上抛出，不拦截异常 + return False + def resolve_wait_method(func): """ @@ -38,17 +111,65 @@ def resolve_wait_method(func): return wrapper -def exception_capture(func): +def action_screenshot(func): """ - 仅在原子动作失败时，触发 BasePage 层的业务截图逻辑 + 显式截图装饰器：在方法执行前（或后）立即触发 BasePage 的截图逻辑。 + 用于记录关键操作后的页面状态。 """ + @wraps(func) def wrapper(self, *args, **kwargs): - try: - return func(self, *args, **kwargs) - except Exception as e: - # 自动捕获：调用 BasePage 层的 log_screenshot - if hasattr(self, "log_screenshot"): - self.log_screenshot(f"自动异常捕获{func.__name__}") - raise e + # 1. 正常执行原方法 + result = func(self, *args, **kwargs) + + # 2. 执行成功后立即截图（如果你希望在操作后的状态截图） + if hasattr(self, "attach_screenshot_bytes"): + try: + class_name = self.__class__.__name__ + func_name = func.__name__ + msg = f"操作记录_{class_name}_{func_name}" + # 传入当前方法名作为截图备注 + self.attach_screenshot_bytes(msg) + except Exception as e: + logger.warning(f"装饰器执行截图失败: {e}") + + return result + return wrapper + + +def _format_params(func, *args, **kwargs): + """辅助函数：专门处理参数过滤和格式化""" + sig = inspect.signature(func) + params = list(sig.parameters.values()) + display_args = args[1:] if params and params[0].name in ('self', 'cls') else args + + # 格式化参数显示，方便阅读 + args_repr = [repr(a) for a in display_args] + kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()] + all_params = ", ".join(args_repr + kwargs_repr) + return all_params + + +def step_trace(step_desc="", source='wrapper'): + """ + 通用执行追踪装饰器： + 1. 智能识别并过滤 self/cls 参数 + 2. 记录入参、出参、耗时 + 3. 异常自动捕获并记录 + """ + + def decorator(func): + @wraps(func) + def wrapper(*args, **kwargs): + # 1. 提取参数显示逻辑 (抽离出来可以作为工具函数) + all_params = _format_params(func, *args, **kwargs) + func_name = f"{func.__module__}.{func.__name__}" + + # 2. 使用上下文管理器 + with StepTracer(step_desc, source, f"{func_name}({all_params})"): + return func(*args, **kwargs) + + return wrapper + + return decorator diff --git a/utils/dirs_manager.py b/utils/dirs_manager.py new file mode 100644 index 0000000..c73d11f --- /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: path_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/logger.py b/utils/logger.py index bc9b919..555cce3 100644 --- a/utils/logger.py +++ b/utils/logger.py @@ -9,62 +9,4 @@ @date: 2026/1/15 11:30 @desc: """ -import time -import functools -import inspect -import logging -from core.settings import LOG_DIR - -# 1. 确定日志存储路径 -LOG_DIR.mkdir(parents=True, exist_ok=True) - -logger = logging.getLogger(__name__) - - -# --- 核心特性：装饰器集成 --- -def trace_step(step_desc="", source='wrapper'): - """ - 通用执行追踪装饰器： - 1. 智能识别并过滤 self/cls 参数 - 2. 记录入参、出参、耗时 - 3. 异常自动捕获并记录 - """ - - def decorator(func): - @functools.wraps(func) - def wrapper(*args, **kwargs): - # 获取对应的 logger - _logger = logging.getLogger(source) - # _logger = logging.getLogger(f"{source}.{func.__name__}") - - # 参数解析 (跳过 self/cls) - sig = inspect.signature(func) - params = list(sig.parameters.values()) - display_args = args[1:] if params and params[0].name in ('self', 'cls') else args - - # 格式化参数显示，方便阅读 - args_repr = [repr(a) for a in display_args] - kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()] - all_params = ", ".join(args_repr + kwargs_repr) - - func_name = f"{func.__module__}.{func.__name__}" - - _logger.info(f"[步骤开始] | {step_desc} | 方法: {func_name}({all_params})") - - start_t = time.perf_counter() - try: - result = func(*args, **kwargs) - duration = time.perf_counter() - start_t - _logger.info(f"[步骤成功] {step_desc} | 耗时: {duration:.2f}s | 返回: {result!r}") - return result - except Exception as e: - duration = time.perf_counter() - start_t - # logging.exception 关键点：它会自动把详细的堆栈信息写入日志文件 - _logger.error(f"[步骤失败] {step_desc}| 耗时: {duration:.2f}s|异常: {type(e).__name__}") - # f"[步骤失败] {step_desc} | 耗时: {duration:.2f}s | 异常: {type(e).__name__}: {e}") - raise e - - return wrapper - - return decorator 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