167 lines
5.9 KiB
Python
167 lines
5.9 KiB
Python
import pytest
|
||
import logging
|
||
import tempfile
|
||
import os
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
from app.logger import LoggerManager, get_logger
|
||
from app.config import settings
|
||
|
||
|
||
class TestLogger:
|
||
"""日志系统测试用例"""
|
||
|
||
def setup_method(self):
|
||
"""每个测试方法执行前的设置"""
|
||
# 重置日志管理器状态
|
||
LoggerManager._initialized = False
|
||
LoggerManager._loggers.clear()
|
||
|
||
def test_logger_setup(self):
|
||
"""测试日志系统初始化"""
|
||
# 使用临时目录
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
temp_log_file = os.path.join(temp_dir, "test.log")
|
||
|
||
with patch.object(settings, 'log_file', temp_log_file):
|
||
# 初始化日志系统
|
||
LoggerManager.setup_logging()
|
||
|
||
# 验证日志文件是否创建
|
||
assert os.path.exists(temp_log_file)
|
||
|
||
# 验证根日志器配置
|
||
root_logger = logging.getLogger()
|
||
assert len(root_logger.handlers) > 0
|
||
|
||
def test_get_logger(self):
|
||
"""测试获取日志器"""
|
||
logger = get_logger("test")
|
||
|
||
assert isinstance(logger, logging.Logger)
|
||
assert logger.name == "test"
|
||
|
||
def test_get_logger_auto_name(self):
|
||
"""测试自动获取日志器名称"""
|
||
logger = get_logger()
|
||
|
||
assert isinstance(logger, logging.Logger)
|
||
# 应该获取到调用模块的名称
|
||
assert "test_logger" in logger.name
|
||
|
||
def test_multiple_loggers(self):
|
||
"""测试多个日志器"""
|
||
logger1 = get_logger("module1")
|
||
logger2 = get_logger("module2")
|
||
logger3 = get_logger("module1") # 重复获取
|
||
|
||
assert logger1.name == "module1"
|
||
assert logger2.name == "module2"
|
||
assert logger3.name == "module1"
|
||
assert logger1 is logger3 # 应该是同一个实例
|
||
|
||
def test_logger_levels(self):
|
||
"""测试不同日志级别"""
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
temp_log_file = os.path.join(temp_dir, "test.log")
|
||
|
||
with patch.object(settings, 'log_file', temp_log_file):
|
||
LoggerManager.setup_logging()
|
||
|
||
logger = get_logger("test_levels")
|
||
|
||
# 测试不同级别的日志
|
||
logger.debug("Debug message")
|
||
logger.info("Info message")
|
||
logger.warning("Warning message")
|
||
logger.error("Error message")
|
||
logger.critical("Critical message")
|
||
|
||
# 验证日志文件内容
|
||
with open(temp_log_file, 'r', encoding='utf-8') as f:
|
||
log_content = f.read()
|
||
|
||
assert "Debug message" in log_content
|
||
assert "Info message" in log_content
|
||
assert "Warning message" in log_content
|
||
assert "Error message" in log_content
|
||
assert "Critical message" in log_content
|
||
|
||
def test_log_rotation(self):
|
||
"""测试日志轮转功能"""
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
temp_log_file = os.path.join(temp_dir, "test.log")
|
||
|
||
# 设置很小的文件大小限制以触发轮转
|
||
with patch.object(settings, 'log_file', temp_log_file), \
|
||
patch.object(settings, 'log_max_bytes', 100):
|
||
|
||
LoggerManager.setup_logging()
|
||
|
||
logger = get_logger("test_rotation")
|
||
|
||
# 写入大量日志以触发轮转
|
||
for i in range(50):
|
||
logger.info(f"This is a long log message to trigger rotation {i}")
|
||
|
||
# 验证是否创建了备份文件
|
||
backup_file = f"{temp_log_file}.1"
|
||
# 注意：由于异步写入，可能需要等待一下
|
||
import time
|
||
time.sleep(0.1)
|
||
|
||
def test_colored_formatter(self):
|
||
"""测试彩色格式化器"""
|
||
from app.logger import ColoredFormatter
|
||
|
||
formatter = ColoredFormatter("%(levelname)s - %(message)s")
|
||
|
||
# 创建测试记录
|
||
record = logging.LogRecord(
|
||
name="test",
|
||
level=logging.INFO,
|
||
pathname="",
|
||
lineno=0,
|
||
msg="Test message",
|
||
args=(),
|
||
exc_info=None
|
||
)
|
||
|
||
formatted = formatter.format(record)
|
||
|
||
# 验证是否包含颜色代码
|
||
assert "\033[" in formatted # ANSI颜色代码
|
||
|
||
def test_predefined_loggers(self):
|
||
"""测试预定义的日志器"""
|
||
from app.logger import (
|
||
app_logger, api_logger, db_logger,
|
||
redis_logger, external_api_logger
|
||
)
|
||
|
||
assert app_logger.name == "app"
|
||
assert api_logger.name == "api"
|
||
assert db_logger.name == "database"
|
||
assert redis_logger.name == "redis"
|
||
assert external_api_logger.name == "external_api"
|
||
|
||
def test_logger_initialization_idempotency(self):
|
||
"""测试日志系统初始化的幂等性"""
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
temp_log_file = os.path.join(temp_dir, "test.log")
|
||
|
||
with patch.object(settings, 'log_file', temp_log_file):
|
||
# 多次初始化
|
||
LoggerManager.setup_logging()
|
||
LoggerManager.setup_logging()
|
||
LoggerManager.setup_logging()
|
||
|
||
# 验证处理器数量不会重复增加
|
||
root_logger = logging.getLogger()
|
||
# 应该有2个处理器（控制台+文件）
|
||
assert len(root_logger.handlers) == 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"]) |