feat: 新增日志系统与执行追踪装饰器

- 基于 Loguru 重新封装,支持异步写入和多线程安全。
- 实现 @trace_step 装饰器,自动记录步骤名、参数及执行耗时。
- 引入 source 标签区分框架系统(System)与业务任务(task)日志。
- 新增 logger 模块测试用例 test_logger.py
This commit is contained in:
2026-01-15 18:00:05 +08:00
parent eabce16972
commit 9355a576a9
4 changed files with 321 additions and 0 deletions

View File

@@ -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]]

115
tests/test_logger.py Normal file
View File

@@ -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()

124
utils/logger.py Normal file
View File

@@ -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. 统一定义日志格式 (美化版)
# <green> 等标签是控制台颜色,文件日志中会自动剥离颜色代码
LOG_FORMAT = (
"<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
"<level>{level: <8}</level> | "
"<magenta>{extra[source]: <8}</magenta> | "
"<cyan>{module}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
"<level>{message}</level>"
)
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"]

View File

@@ -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 源码。