张翔
f14002559e
refactor(components): 调整头部和页脚布局样式 style(hero-section): 更新徽章动画效果 docs: 添加测试框架README文档 test: 实现首页、导航和联系表单的测试用例 ci: 添加CI测试脚本和配置
273 lines
8.9 KiB
Python
273 lines
8.9 KiB
Python
"""
|
|
日志工具模块
|
|
提供测试过程中的日志记录功能
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional, Union
|
|
from functools import wraps
|
|
import traceback
|
|
|
|
from config.settings import get_settings
|
|
|
|
|
|
class ColoredFormatter(logging.Formatter):
|
|
"""彩色日志格式化器"""
|
|
|
|
# ANSI颜色代码
|
|
COLORS = {
|
|
'DEBUG': '\033[36m', # 青色
|
|
'INFO': '\033[32m', # 绿色
|
|
'WARNING': '\033[33m', # 黄色
|
|
'ERROR': '\033[31m', # 红色
|
|
'CRITICAL': '\033[35m', # 紫色
|
|
'RESET': '\033[0m', # 重置
|
|
}
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
# 获取颜色
|
|
color = self.COLORS.get(record.levelname, self.COLORS['RESET'])
|
|
|
|
# 格式化消息
|
|
message = super().format(record)
|
|
|
|
# 添加颜色（如果不是纯文本输出）
|
|
if sys.stdout.isatty():
|
|
return f"{color}{message}{self.COLORS['RESET']}"
|
|
return message
|
|
|
|
|
|
class TestLogger:
|
|
"""测试日志管理器"""
|
|
|
|
_instance: Optional['TestLogger'] = None
|
|
_initialized: bool = False
|
|
|
|
def __new__(cls) -> 'TestLogger':
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if not TestLogger._initialized:
|
|
self._setup_logging()
|
|
TestLogger._initialized = True
|
|
|
|
def _setup_logging(self) -> None:
|
|
"""设置日志配置"""
|
|
self.settings = get_settings()
|
|
self.logger = logging.getLogger("e2e_tests")
|
|
self.logger.setLevel(getattr(logging, self.settings.log_level))
|
|
|
|
# 清除现有处理器
|
|
self.logger.handlers.clear()
|
|
|
|
# 创建日志目录
|
|
log_dir = Path(self.settings.log_file).parent
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 文件处理器
|
|
file_handler = logging.FileHandler(
|
|
self.settings.log_file,
|
|
encoding='utf-8',
|
|
mode='a'
|
|
)
|
|
file_handler.setLevel(logging.DEBUG)
|
|
|
|
# 控制台处理器
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setLevel(getattr(logging, self.settings.log_level))
|
|
|
|
# 设置格式化器
|
|
file_format = logging.Formatter(
|
|
'%(asctime)s | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
|
|
console_format_str = '%(asctime)s | %(levelname)-8s | %(message)s'
|
|
|
|
file_handler.setFormatter(file_format)
|
|
|
|
if sys.stdout.isatty():
|
|
console_handler.setFormatter(ColoredFormatter(console_format_str, datefmt='%H:%M:%S'))
|
|
else:
|
|
console_handler.setFormatter(logging.Formatter(console_format_str, datefmt='%H:%M:%S'))
|
|
|
|
self.logger.addHandler(file_handler)
|
|
self.logger.addHandler(console_handler)
|
|
|
|
def debug(self, message: str, **kwargs) -> None:
|
|
"""记录DEBUG级别日志"""
|
|
self.logger.debug(self._format_message(message, **kwargs))
|
|
|
|
def info(self, message: str, **kwargs) -> None:
|
|
"""记录INFO级别日志"""
|
|
self.logger.info(self._format_message(message, **kwargs))
|
|
|
|
def warning(self, message: str, **kwargs) -> None:
|
|
"""记录WARNING级别日志"""
|
|
self.logger.warning(self._format_message(message, **kwargs))
|
|
|
|
def error(self, message: str, exc_info: bool = True, **kwargs) -> None:
|
|
"""记录ERROR级别日志"""
|
|
self.logger.error(
|
|
self._format_message(message, **kwargs),
|
|
exc_info=exc_info
|
|
)
|
|
|
|
def critical(self, message: str, exc_info: bool = True, **kwargs) -> None:
|
|
"""记录CRITICAL级别日志"""
|
|
self.logger.critical(
|
|
self._format_message(message, **kwargs),
|
|
exc_info=exc_info
|
|
)
|
|
|
|
def exception(self, message: str, **kwargs) -> None:
|
|
"""记录异常日志（自动包含堆栈信息）"""
|
|
self.error(message, exc_info=True, **kwargs)
|
|
|
|
def _format_message(self, message: str, **kwargs) -> str:
|
|
"""格式化日志消息"""
|
|
if kwargs:
|
|
extra_info = " | ".join(f"{k}={v}" for k, v in kwargs.items())
|
|
return f"{message} | {extra_info}"
|
|
return message
|
|
|
|
def log_test_start(self, test_name: str, **extra_info) -> None:
|
|
"""记录测试开始"""
|
|
self.info(f"🧪 测试开始: {test_name}", **extra_info)
|
|
|
|
def log_test_end(self, test_name: str, status: str, duration: float, **extra_info) -> None:
|
|
"""记录测试结束"""
|
|
emoji = "✅" if status == "PASSED" else "❌" if status == "FAILED" else "⏭️"
|
|
self.info(f"{emoji} 测试结束: {test_name} | 状态: {status} | 耗时: {duration:.2f}s", **extra_info)
|
|
|
|
def log_step(self, step_name: str, **extra_info) -> None:
|
|
"""记录测试步骤"""
|
|
self.info(f"📋 步骤: {step_name}", **extra_info)
|
|
|
|
def log_action(self, action: str, **extra_info) -> None:
|
|
"""记录用户操作"""
|
|
self.info(f"👆 操作: {action}", **extra_info)
|
|
|
|
def log_assertion(self, assertion: str, result: bool, **extra_info) -> None:
|
|
"""记录断言结果"""
|
|
status = "✅ 通过" if result else "❌ 失败"
|
|
self.info(f"🔍 断言: {assertion} | {status}", **extra_info)
|
|
|
|
def log_performance(self, metric: str, value: float, threshold: Optional[float] = None, **extra_info) -> None:
|
|
"""记录性能指标"""
|
|
if threshold and value > threshold:
|
|
self.warning(f"📊 性能指标 - {metric}: {value:.2f}ms (阈值: {threshold:.2f}ms)", **extra_info)
|
|
else:
|
|
self.info(f"📊 性能指标 - {metric}: {value:.2f}ms", **extra_info)
|
|
|
|
def log_error_context(self, context: str, error: Exception, **extra_info) -> None:
|
|
"""记录错误上下文"""
|
|
self.error(f"🚨 错误上下文: {context}", exc_info=False, **extra_info)
|
|
self.error(f"错误信息: {str(error)}", exc_info=False)
|
|
self.debug(f"堆栈跟踪:\n{traceback.format_exc()}")
|
|
|
|
def section(self, title: str) -> None:
|
|
"""记录分段标题"""
|
|
separator = "=" * 60
|
|
self.info(f"\n{separator}")
|
|
self.info(f" {title}")
|
|
self.info(f"{separator}\n")
|
|
|
|
def divider(self, char: str = "-", length: int = 40) -> None:
|
|
"""记录分隔线"""
|
|
self.info(char * length)
|
|
|
|
|
|
def get_logger() -> TestLogger:
|
|
"""获取日志管理器实例"""
|
|
return TestLogger()
|
|
|
|
|
|
def log_decorator(func):
|
|
"""函数日志装饰器"""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
logger = get_logger()
|
|
func_name = func.__name__
|
|
|
|
logger.log_test_start(func_name)
|
|
logger.divider()
|
|
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
logger.log_test_end(func_name, "PASSED", 0)
|
|
return result
|
|
except Exception as e:
|
|
logger.log_test_end(func_name, "FAILED", 0)
|
|
logger.log_error_context(func_name, e)
|
|
raise
|
|
|
|
return wrapper
|
|
|
|
|
|
class PerformanceTimer:
|
|
"""性能计时器"""
|
|
|
|
def __init__(self, logger: Optional[TestLogger] = None):
|
|
self.logger = logger or get_logger()
|
|
self.start_time: Optional[float] = None
|
|
self.end_time: Optional[float] = None
|
|
self.elapsed: Optional[float] = None
|
|
|
|
def start(self) -> 'PerformanceTimer':
|
|
"""开始计时"""
|
|
self.start_time = self._time_ms()
|
|
return self
|
|
|
|
def stop(self) -> 'PerformanceTimer':
|
|
"""停止计时"""
|
|
self.end_time = self._time_ms()
|
|
self.elapsed = self.end_time - self.start_time
|
|
return self
|
|
|
|
def reset(self) -> 'PerformanceTimer':
|
|
"""重置计时器"""
|
|
self.start_time = None
|
|
self.end_time = None
|
|
self.elapsed = None
|
|
return self
|
|
|
|
@staticmethod
|
|
def _time_ms() -> float:
|
|
"""获取当前时间（毫秒）"""
|
|
import time
|
|
return time.time() * 1000
|
|
|
|
@property
|
|
def seconds(self) -> float:
|
|
"""获取经过时间（秒）"""
|
|
return self.elapsed / 1000 if self.elapsed else 0
|
|
|
|
@property
|
|
def milliseconds(self) -> float:
|
|
"""获取经过时间（毫秒）"""
|
|
return self.elapsed if self.elapsed else 0
|
|
|
|
def log(self, operation: str, threshold: Optional[float] = None) -> None:
|
|
"""记录操作耗时"""
|
|
self.logger.log_performance(
|
|
operation,
|
|
self.milliseconds,
|
|
threshold
|
|
)
|
|
|
|
def __enter__(self) -> 'PerformanceTimer':
|
|
"""上下文管理器入口"""
|
|
self.start()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
|
|
"""上下文管理器出口"""
|
|
self.stop()
|