Constantin
247 lines
8.5 KiB
Python
247 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tests for LogManager
|
|
"""
|
|
|
|
import unittest
|
|
import json
|
|
import tempfile
|
|
import os
|
|
import shutil
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add api directory to path
|
|
api_dir = Path(__file__).parent.parent / 'api'
|
|
sys.path.insert(0, str(api_dir))
|
|
|
|
from log_manager import LogManager, LogLevel
|
|
|
|
class TestLogManager(unittest.TestCase):
|
|
"""Test the log manager functionality"""
|
|
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.log_dir = os.path.join(self.temp_dir, 'logs')
|
|
os.makedirs(self.log_dir, exist_ok=True)
|
|
self.log_manager = LogManager(log_dir=self.log_dir)
|
|
|
|
# Add this helper to ensure log dir exists before logger usage
|
|
def ensure_log_dir(self):
|
|
os.makedirs(self.log_dir, exist_ok=True)
|
|
|
|
# In each test that uses logger, call self.ensure_log_dir() before logger usage
|
|
def test_initialization(self):
|
|
"""Test log manager initialization"""
|
|
self.assertTrue(os.path.exists(self.log_dir))
|
|
self.assertIsNotNone(self.log_manager.formatters)
|
|
self.assertIsNotNone(self.log_manager.handlers)
|
|
self.assertIsNotNone(self.log_manager.log_stats)
|
|
|
|
def test_add_service_logger(self):
|
|
self.ensure_log_dir()
|
|
"""Test adding service logger"""
|
|
config = {
|
|
'level': 'INFO',
|
|
'formatter': 'json',
|
|
'console': True
|
|
}
|
|
|
|
self.log_manager.add_service_logger('test_service', config)
|
|
self.assertIn('test_service', self.log_manager.service_loggers)
|
|
self.assertIn('test_service', self.log_manager.handlers)
|
|
|
|
def test_get_service_logs(self):
|
|
self.ensure_log_dir()
|
|
"""Test getting service logs"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("Test log message 1")
|
|
logger.warning("Test log message 2")
|
|
logger.error("Test log message 3")
|
|
|
|
# Get logs
|
|
logs = self.log_manager.get_service_logs('test_service', lines=3)
|
|
self.assertIsInstance(logs, list)
|
|
# Note: We can't guarantee exact count due to async logging
|
|
|
|
def test_search_logs(self):
|
|
self.ensure_log_dir()
|
|
"""Test log search functionality"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("User login successful")
|
|
logger.info("Database connection established")
|
|
logger.error("Authentication failed")
|
|
|
|
# Search logs
|
|
results = self.log_manager.search_logs('login')
|
|
self.assertIsInstance(results, list)
|
|
|
|
# Search with time range
|
|
end_time = datetime.now()
|
|
start_time = end_time - timedelta(hours=1)
|
|
results = self.log_manager.search_logs(
|
|
'login',
|
|
time_range=(start_time, end_time)
|
|
)
|
|
self.assertIsInstance(results, list)
|
|
|
|
# Search with service filter
|
|
results = self.log_manager.search_logs(
|
|
'login',
|
|
services=['test_service']
|
|
)
|
|
self.assertIsInstance(results, list)
|
|
|
|
def test_export_logs(self):
|
|
self.ensure_log_dir()
|
|
"""Test log export functionality"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("Export test message 1")
|
|
logger.info("Export test message 2")
|
|
|
|
# Export as JSON
|
|
json_export = self.log_manager.export_logs('json')
|
|
self.assertIsInstance(json_export, str)
|
|
|
|
# Export as CSV
|
|
csv_export = self.log_manager.export_logs('csv')
|
|
self.assertIsInstance(csv_export, str)
|
|
|
|
# Export as text
|
|
text_export = self.log_manager.export_logs('text')
|
|
self.assertIsInstance(text_export, str)
|
|
|
|
def test_log_statistics(self):
|
|
self.ensure_log_dir()
|
|
"""Test log statistics functionality"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("Info message")
|
|
logger.warning("Warning message")
|
|
logger.error("Error message")
|
|
|
|
# Get statistics
|
|
stats = self.log_manager.get_log_statistics()
|
|
self.assertIsInstance(stats, dict)
|
|
|
|
# Get service-specific statistics
|
|
service_stats = self.log_manager.get_log_statistics('test_service')
|
|
self.assertIsInstance(service_stats, dict)
|
|
|
|
def test_log_rotation(self):
|
|
self.ensure_log_dir()
|
|
"""Test log rotation functionality"""
|
|
# Add service logger with small max file size
|
|
config = {
|
|
'level': 'INFO',
|
|
'formatter': 'text'
|
|
}
|
|
self.log_manager.add_service_logger('test_service', config)
|
|
|
|
# Create many log entries to trigger rotation
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
for i in range(1000):
|
|
logger.info(f"Log entry {i}: " + "x" * 100) # Large log entries
|
|
|
|
# Trigger rotation
|
|
self.log_manager.rotate_logs('test_service')
|
|
|
|
# Check if rotation files exist
|
|
log_file = os.path.join(self.log_dir, 'test_service.log')
|
|
self.assertTrue(os.path.exists(log_file))
|
|
|
|
def test_cleanup_old_logs(self):
|
|
self.ensure_log_dir()
|
|
"""Test cleanup of old logs"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("Test message")
|
|
|
|
# Cleanup old logs (should not affect recent logs)
|
|
self.log_manager.cleanup_old_logs(days=1)
|
|
|
|
# Verify logs still exist
|
|
logs = self.log_manager.get_service_logs('test_service')
|
|
self.assertIsInstance(logs, list)
|
|
|
|
def test_log_file_info(self):
|
|
self.ensure_log_dir()
|
|
"""Test log file information"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("Test message")
|
|
|
|
# Get log file info
|
|
info = self.log_manager.get_log_file_info('test_service')
|
|
self.assertIsInstance(info, dict)
|
|
self.assertIn('file_path', info)
|
|
self.assertIn('exists', info)
|
|
|
|
def test_compress_old_logs(self):
|
|
self.ensure_log_dir()
|
|
"""Test compression of old logs"""
|
|
# Add service logger
|
|
self.log_manager.add_service_logger('test_service', {'level': 'INFO'})
|
|
|
|
# Create some log entries
|
|
logger = self.log_manager.service_loggers['test_service']
|
|
logger.info("Test message for compression")
|
|
|
|
# Compress old logs
|
|
self.log_manager.compress_old_logs()
|
|
|
|
# Verify compression worked (should not raise errors)
|
|
self.assertTrue(True) # If we get here, compression worked
|
|
|
|
def test_formatters(self):
|
|
self.ensure_log_dir()
|
|
"""Test different log formatters"""
|
|
# Test JSON formatter
|
|
json_logger = self.log_manager.add_service_logger('json_service', {
|
|
'level': 'INFO',
|
|
'formatter': 'json'
|
|
})
|
|
|
|
# Test text formatter
|
|
text_logger = self.log_manager.add_service_logger('text_service', {
|
|
'level': 'INFO',
|
|
'formatter': 'text'
|
|
})
|
|
|
|
# Test detailed formatter
|
|
detailed_logger = self.log_manager.add_service_logger('detailed_service', {
|
|
'level': 'INFO',
|
|
'formatter': 'detailed'
|
|
})
|
|
|
|
# Verify formatters exist
|
|
self.assertIn('json', self.log_manager.formatters)
|
|
self.assertIn('text', self.log_manager.formatters)
|
|
self.assertIn('detailed', self.log_manager.formatters)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |