#!/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()