张翔
156 lines
5.1 KiB
Python
156 lines
5.1 KiB
Python
"""
|
|
日志工具模块
|
|
"""
|
|
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
|
|
class TestLogger:
|
|
"""测试日志记录器"""
|
|
|
|
def __init__(self, name: str = "test", log_file: str = None, level: str = "INFO", console: bool = True):
|
|
"""
|
|
初始化日志记录器
|
|
|
|
Args:
|
|
name: 日志记录器名称
|
|
log_file: 日志文件路径
|
|
level: 日志级别
|
|
console: 是否输出到控制台
|
|
"""
|
|
self.logger = logging.getLogger(name)
|
|
self.logger.setLevel(getattr(logging, level.upper()))
|
|
|
|
# 清除现有的处理器
|
|
self.logger.handlers.clear()
|
|
|
|
# 创建格式化器
|
|
formatter = logging.Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
|
|
# 添加文件处理器
|
|
if log_file:
|
|
log_path = Path(log_file)
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
file_handler = logging.FileHandler(log_file, encoding='utf-8')
|
|
file_handler.setFormatter(formatter)
|
|
self.logger.addHandler(file_handler)
|
|
|
|
# 添加控制台处理器
|
|
if console:
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setFormatter(formatter)
|
|
self.logger.addHandler(console_handler)
|
|
|
|
def debug(self, message: str) -> None:
|
|
"""记录调试信息"""
|
|
self.logger.debug(message)
|
|
|
|
def info(self, message: str) -> None:
|
|
"""记录信息"""
|
|
self.logger.info(message)
|
|
|
|
def warning(self, message: str) -> None:
|
|
"""记录警告"""
|
|
self.logger.warning(message)
|
|
|
|
def error(self, message: str) -> None:
|
|
"""记录错误"""
|
|
self.logger.error(message)
|
|
|
|
def critical(self, message: str) -> None:
|
|
"""记录严重错误"""
|
|
self.logger.critical(message)
|
|
|
|
def exception(self, message: str) -> None:
|
|
"""记录异常信息"""
|
|
self.logger.exception(message)
|
|
|
|
def log_test_start(self, test_name: str) -> None:
|
|
"""记录测试开始"""
|
|
self.info(f"{'='*60}")
|
|
self.info(f"开始测试: {test_name}")
|
|
self.info(f"{'='*60}")
|
|
|
|
def log_test_end(self, test_name: str, passed: bool, duration: float = None) -> None:
|
|
"""记录测试结束"""
|
|
status = "✅ 通过" if passed else "❌ 失败"
|
|
duration_str = f" (耗时: {duration:.2f}ms)" if duration else ""
|
|
self.info(f"测试结束: {test_name} - {status}{duration_str}")
|
|
|
|
def log_request(self, method: str, url: str, data: dict = None, headers: dict = None) -> None:
|
|
"""记录请求信息"""
|
|
self.info(f"发送请求: {method} {url}")
|
|
if data:
|
|
self.info(f"请求数据: {data}")
|
|
if headers:
|
|
self.info(f"请求头: {headers}")
|
|
|
|
def log_response(self, status_code: int, response_time: float, data: dict = None) -> None:
|
|
"""记录响应信息"""
|
|
self.info(f"响应状态码: {status_code}")
|
|
self.info(f"响应时间: {response_time:.2f}ms")
|
|
if data:
|
|
self.info(f"响应数据: {data}")
|
|
|
|
def log_validation(self, rule_type: str, passed: bool, message: str = "") -> None:
|
|
"""记录验证结果"""
|
|
status = "✅" if passed else "❌"
|
|
self.info(f"{status} 验证规则 [{rule_type}]: {message or '通过'}")
|
|
|
|
def log_error(self, error: Exception) -> None:
|
|
"""记录错误详情"""
|
|
self.error(f"错误类型: {type(error).__name__}")
|
|
self.error(f"错误信息: {str(error)}")
|
|
self.exception("错误堆栈:")
|
|
|
|
def log_summary(self, total: int, passed: int, failed: int, skipped: int = 0, duration: float = None) -> None:
|
|
"""记录测试摘要"""
|
|
self.info(f"{'='*60}")
|
|
self.info(f"测试摘要:")
|
|
self.info(f" 总数: {total}")
|
|
self.info(f" 通过: {passed}")
|
|
self.info(f" 失败: {failed}")
|
|
self.info(f" 跳过: {skipped}")
|
|
if duration:
|
|
self.info(f" 总耗时: {duration:.2f}ms")
|
|
self.info(f" 通过率: {(passed/total*100):.2f}%")
|
|
self.info(f"{'='*60}")
|
|
|
|
|
|
class LoggerFactory:
|
|
"""日志记录器工厂"""
|
|
|
|
_loggers = {}
|
|
|
|
@classmethod
|
|
def get_logger(cls, name: str = "test", log_file: str = None, level: str = "INFO", console: bool = True) -> TestLogger:
|
|
"""
|
|
获取日志记录器实例
|
|
|
|
Args:
|
|
name: 日志记录器名称
|
|
log_file: 日志文件路径
|
|
level: 日志级别
|
|
console: 是否输出到控制台
|
|
|
|
Returns:
|
|
日志记录器实例
|
|
"""
|
|
key = f"{name}_{log_file}_{level}_{console}"
|
|
|
|
if key not in cls._loggers:
|
|
cls._loggers[key] = TestLogger(name, log_file, level, console)
|
|
|
|
return cls._loggers[key]
|
|
|
|
@classmethod
|
|
def clear_all(cls) -> None:
|
|
"""清除所有日志记录器"""
|
|
cls._loggers.clear() |