refactor(core/utils): 重构装饰器架构与日志追踪逻辑

- 优化将 trace_step 移至 decorators.py，并引入 ContextVar 实现日志层级缩进。 - 新增完善 StepTracer 和 action_screenshot 的 Docstrings，明确参数含义。 - 移除清理了 logger.py - 优化 main.py 中重复的目录创建逻辑及旧版冗余注释。 - 规范修正函数命名，提升代码在 BasePage 方案下的复用性。
2026-02-03 17:46:48 +08:00
parent 798b5a8142
commit 483a31793d
9 changed files with 299 additions and 163 deletions
--- 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")
--- 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

--- 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__":
--- a/test_cases/conftest.py
+++ b/test_cases/conftest.py
@@ -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}")
--- 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
+    assert driver.session_id is not None
--- 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
--- a/utils/dirs_manager.py
+++ 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
--- 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
--- a/utils/report_handler.py
+++ 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