diff --git a/pyproject.toml b/pyproject.toml index 3f39403..9f52bf8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,6 +6,7 @@ readme = "README.md" requires-python = ">=3.11" dependencies = [ "appium-python-client>=5.2.4", + "loguru>=0.7.3", ] [[tool.uv.index]] diff --git a/tests/test_logger.py b/tests/test_logger.py new file mode 100644 index 0000000..f9a57e7 --- /dev/null +++ b/tests/test_logger.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python +# coding=utf-8 + +""" +@author: CNWei,ChenWei +@Software: PyCharm +@contact: t6g888@163.com,chenwei@zygj.com +@file: test +@date: 2026/1/14 10:12 +@desc: +""" + +# import pytest +from enum import Enum +from typing import TypeVar +from utils.logger import logger, trace_step + + +# --- 模拟业务逻辑 --- + +class ServiceRole(Enum): + MANAGED = "受控模式" + EXTERNAL = "共享模式" + NULL = "无效模式" + + +class AppiumService: + def __init__(self, device_id: str, role: ServiceRole): + self.device_id = device_id + self.role = role + # 使用 bind 为该实例的所有日志绑定特定的设备 ID + self._log = logger.bind(source=f"Dev:{device_id}") + + @trace_step(step_desc="停止服务", source="Appium") + def stop(self, force=False): + """演示类方法追踪及逻辑分支""" + self._log.info(f"正在尝试停止服务，强制模式={force}") + + if self.role == ServiceRole.EXTERNAL: + self._log.warning("外部服务，跳过清理进程") + return "SKIPPED" + + if self.role == ServiceRole.MANAGED: + self._log.success("已发送 SIGTERM 信号清理进程") + return "SUCCESS" + + raise RuntimeError("无法停止处于未知状态的服务") + + @trace_step("简单打印") + def simple_log(self, msg: str): + self._log.info(f"消息回显: {msg}") + + +# --- 独立函数演示 --- + +@trace_step("执行数据计算", source="Calc") +def calculate_data(a: int, b: int): + if b == 0: + raise ZeroDivisionError("除数不能为 0") + return a / b + + +@trace_step("空值返回测试") +def return_none(): + return None + + +# --- 测试场景覆盖 --- + +def run_scenarios(): + print("\n" + "=" * 50) + print("🚀 开始执行全场景日志覆盖测试") + print("=" * 50 + "\n") + + # 1. 覆盖：正常类方法调用 (过滤 self) + logger.info(">>> 场景 1: 正常类方法 (MANAGED 模式)") + svc1 = AppiumService("emulator-5554", ServiceRole.MANAGED) + svc1.stop(force=True) + + # 2. 覆盖：类方法不同返回值 + logger.info(">>> 场景 2: 共享模式跳过清理") + svc2 = AppiumService("iPhone_15", ServiceRole.EXTERNAL) + svc2.stop() + + # 3. 覆盖：异常捕获 (自动记录错误日志并向上抛出) + logger.info(">>> 场景 3: 异常捕获测试") + try: + calculate_data(10, 0) + except ZeroDivisionError: + logger.warning("主流程已捕获预期的计算异常") + + # 4. 覆盖：复杂参数与 None 返回 + logger.info(">>> 场景 4: 复杂参数与 None 返回") + return_none() + + # 5. 覆盖：未定义状态导致的崩溃 + logger.info(">>> 场景 5: 业务逻辑崩溃测试") + svc3 = AppiumService("Unknown_Device", ServiceRole.NULL) + try: + svc3.stop() + except Exception as e: + logger.error(f"捕获到业务逻辑崩溃:{e}") + + # 6. 覆盖：原生 logger 与装饰器 logger 混合 + logger.info(">>> 场景 6: 验证自定义 source 标签") + # 这里会使用 setup_logger 中定义的默认 'System' 标签 + logger.debug("这是一条调试级别的原始日志") + + print("\n" + "=" * 50) + print("✅ 测试场景执行完毕，请检查 logs 文件夹中的 .log 文件") + print("=" * 50) + + +if __name__ == "__main__": + run_scenarios() \ No newline at end of file diff --git a/utils/logger.py b/utils/logger.py new file mode 100644 index 0000000..2dc3723 --- /dev/null +++ b/utils/logger.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python +# coding=utf-8 + +""" +@author: CNWei,ChenWei +@Software: PyCharm +@contact: t6g888@163.com,chenwei@zygj.com +@file: logger +@date: 2026/1/15 11:30 +@desc: +""" +import sys +import time +import functools +from pathlib import Path +import inspect +from loguru import logger + +# 1. 确定日志存储路径 +LOG_DIR = Path(__file__).parent.parent / "logs" +LOG_DIR.mkdir(exist_ok=True) + +# 2. 统一定义日志格式 (美化版) +# 等标签是控制台颜色，文件日志中会自动剥离颜色代码 +LOG_FORMAT = ( + "{time:YYYY-MM-DD HH:mm:ss.SSS} | " + "{level: <8} | " + "{extra[source]: <8} | " + "{module}:{function}:{line} - " + "{message}" +) + + +def setup_logger(): + """ + 只需在项目入口调用一次。 + 如果是简单的自动化脚本，甚至可以直接在模块内执行。 + """ + # 移除 Loguru 默认的控制台处理器（避免重复打印） + logger.remove() + + # 添加自定义控制台输出 + logger.add( + sys.stdout, + format=LOG_FORMAT, + level="INFO", + colorize=True, + # 默认给一个 'Global' 的 device 标签 + filter=lambda record: record["extra"].setdefault("source", "System") + ) + + # 添加按天滚动的日志文件 + logger.add( + str(LOG_DIR / "appium_{time:YYYY-MM-DD}.log"), + format=LOG_FORMAT, + level="DEBUG", + rotation="00:00", # 每天午夜滚动 + retention="10 days", # 保留最近10天 + compression="zip", # 旧日志自动压缩 + encoding="utf-8", + enqueue=True # 开启队列模式，确保多线程下日志不串行 + ) + + +# --- 核心特性 1：装饰器集成 --- +def trace_step(step_desc="", source: str = 'task'): + """ + 通用执行追踪装饰器： + 1. 智能识别并过滤 self/cls 参数 + 2. 记录入参、出参、耗时 + 3. 异常自动捕获并记录 + """ + + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + # --- 智能参数解析 --- + # 获取函数的签名 + sig = inspect.signature(func) + params = list(sig.parameters.values()) + + # 检查第一个参数名是否为 'self' 或 'cls' + # 这样既兼容了 PageObject 的实例方法，也兼容了纯函数 + if params and params[0].name in ('self', 'cls'): + display_args = args[1:] + else: + display_args = 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__}" + + # 使用 bind 临时改变这一步的 source 标签 + + _logger = logger.bind(source=source) + # 使用关联的上下文 logger + # logger.info(f"🚀 [步骤开始] {step_desc} | 执行方法: {func_name} | 参数: {display_args} {kwargs}") + _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.success(f"✅ [步骤成功] {step_desc} | 耗时: {duration:.2f}s | 返回: {result!r}") + return result + except Exception as e: + duration = time.perf_counter() - start_t + _logger.error( + f"❌ [步骤失败] {step_desc} | 耗时: {duration:.2f}s | 异常: {type(e).__name__}: {e}") + raise e + + return wrapper + + return decorator + + +# 初始化 +setup_logger() + +# 导出供外部使用 +__all__ = ["logger", "trace_step"] diff --git a/utils/日志模块使用指南.md b/utils/日志模块使用指南.md new file mode 100644 index 0000000..1240b56 --- /dev/null +++ b/utils/日志模块使用指南.md @@ -0,0 +1,81 @@ +日志与执行追踪模块使用指南 +本模块基于 Loguru 封装，专为自动化测试项目设计，提供工业级的日志记录、多线程安全保障以及业务步骤自动追踪能力。 + +🌟 核心特性 +全局一致性：统一日志格式，控制台带颜色显示，方便快速定位。 + +自动执行追踪：使用 @trace_step 装饰器，自动记录方法入参、出参、耗时及执行状态。 + +智能参数解析：装饰器自动识别并隐藏 self 和 cls 参数，保护日志整洁。 + +上下文透传：支持通过 .bind() 绑定设备 ID 或模块标识，解决多机并行日志混杂问题。 + +分层标识：默认区分系统日志 (System) 与业务任务日志 (task)。 + +异步安全：支持多进程/多线程安全写入，不影响 Appium 执行性能。 + +🚀 快速上手 +1. 基础日志记录 +在任何模块中直接导入 logger 即可使用。 +```python +from utils.logger import logger + +logger.info("这是一条普通信息") +logger.debug("调试模式下的详细信息") +logger.error("记录一个错误") +``` +2. 使用装饰器追踪业务步骤 (@trace_step) +在 PageObject 方法或任何业务函数上添加装饰器，即可获得全链路追踪。 +```python +from utils.logger import trace_step + +@trace_step("用户登录") +def login(username, password): + # 执行逻辑... + return True + +# 日志输出： +# 🚀 [START] 用户登录 -> login('admin', '****') +# ✅ [SUCCESS] 用户登录 | 耗时: 1.25s | 返回: True +``` +3. 多机并行：上下文透传 (.bind) +在 Appium 并行测试中，用于区分不同设备的日志流。 +```python +from utils.logger import logger + +class BasePage: + def __init__(self, driver, device_id): + self.driver = driver + # 绑定设备 ID 到当前实例的 logger + self.log = logger.bind(source=device_id) + + def click_element(self, loc): + self.log.info(f"点击元素: {loc}") + +# 日志输出： +# | INFO | Pixel_6 | base_page:click_element:15 - 点击元素: id=login_btn +# | INFO | S22 | base_page:click_element:15 - 点击元素: id=login_btn +``` +🛠️ 进阶配置 +标识符说明 (source 字段) +日志格式中包含一个 source 字段（占位符为 magenta 颜色），用于区分日志来源： + +System: (默认) 框架底层日志、驱动初始化等。 + +task: (装饰器默认) 具体的业务测试步骤。 + +自定义: 通过 @trace_step(source="SQL") 或 logger.bind(source="API") 自定义。 + +日志存储 +路径: 项目根目录 /logs/。 + +滚动: 每天午夜 00:00 自动切割。 + +保留: 默认保留最近 30 天 的日志。 + +⚠️ 注意事项 +不要在装饰器内手动接收返回值：@trace_step 已经自动处理了函数的返回值记录。 + +优先使用 self.log：在 PageObject 类中，请务必使用 self.log.info() 而非全局 logger.info()，以确保 bind 的上下文信息（如设备 ID）能正确显示。 + +希望这套日志系统能显著提升您的调试效率和项目质量！如有任何疑问，请随时查阅 utils/logger.py 源码。 \ No newline at end of file