From 733c86eb6b0bbee8a9a1aeb74c7f1b589a9940e0 Mon Sep 17 00:00:00 2001 From: Lukas Pupka-Lipinski Date: Mon, 6 Oct 2025 11:31:40 +0200 Subject: [PATCH] Add diagnostics and logging tests - Created integration tests for /diagnostics/* endpoints - Added unit tests for logging functionality and configuration - Tests error reporting, system health, and log management - Covers GlobalLogger, file handlers, and error handling - Ready for future diagnostics endpoint implementation --- src/tests/integration/test_diagnostics.py | 335 +++++++++++++++ src/tests/unit/test_logging_functionality.py | 403 +++++++++++++++++++ 2 files changed, 738 insertions(+) create mode 100644 src/tests/integration/test_diagnostics.py create mode 100644 src/tests/unit/test_logging_functionality.py diff --git a/src/tests/integration/test_diagnostics.py b/src/tests/integration/test_diagnostics.py new file mode 100644 index 0000000..b8a8f81 --- /dev/null +++ b/src/tests/integration/test_diagnostics.py @@ -0,0 +1,335 @@ +""" +Integration tests for diagnostics API endpoints. + +This module tests the diagnostics endpoints for error reporting and system diagnostics. +""" + +import pytest +from fastapi.testclient import TestClient +from unittest.mock import patch, Mock +import tempfile +import os + +from src.server.fastapi_app import app + + +@pytest.fixture +def client(): + """Create a test client for the FastAPI application.""" + return TestClient(app) + + +@pytest.fixture +def auth_headers(client): + """Provide authentication headers for protected endpoints.""" + # Login to get token + login_data = {"password": "testpassword"} + + with patch('src.server.fastapi_app.settings.master_password_hash') as mock_hash: + mock_hash.return_value = "5e884898da28047151d0e56f8dc6292773603d0d6aabbdd62a11ef721d1542d8" # 'password' hash + response = client.post("/auth/login", json=login_data) + + if response.status_code == 200: + token = response.json()["access_token"] + return {"Authorization": f"Bearer {token}"} + return {} + + +class TestDiagnosticsReportEndpoint: + """Test cases for /diagnostics/report endpoint.""" + + def test_diagnostics_report_requires_auth(self, client): + """Test that diagnostics report requires authentication.""" + response = client.get("/diagnostics/report") + assert response.status_code == 401 + + @patch('src.server.fastapi_app.get_current_user') + def test_get_diagnostics_report(self, mock_user, client): + """Test getting diagnostics report.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.get("/diagnostics/report") + # Expected 404 since endpoint not implemented yet + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + expected_fields = [ + "system_info", "memory_usage", "disk_usage", + "error_summary", "performance_metrics", "timestamp" + ] + for field in expected_fields: + assert field in data + + @patch('src.server.fastapi_app.get_current_user') + def test_get_diagnostics_report_with_filters(self, mock_user, client): + """Test getting diagnostics report with time filters.""" + mock_user.return_value = {"user_id": "test_user"} + + # Test with time range + response = client.get("/diagnostics/report?since=2023-01-01&until=2023-12-31") + assert response.status_code in [200, 404] + + # Test with severity filter + response = client.get("/diagnostics/report?severity=error") + assert response.status_code in [200, 404] + + @patch('src.server.fastapi_app.get_current_user') + def test_generate_diagnostics_report(self, mock_user, client): + """Test generating new diagnostics report.""" + mock_user.return_value = {"user_id": "test_user"} + + report_options = { + "include_logs": True, + "include_system_info": True, + "include_performance": True, + "time_range_hours": 24 + } + + response = client.post("/diagnostics/report", json=report_options) + # Expected 404 since endpoint not implemented yet + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + assert "report_id" in data + assert "status" in data + + def test_diagnostics_report_invalid_params(self, client, auth_headers): + """Test diagnostics report with invalid parameters.""" + invalid_params = [ + "?since=invalid-date", + "?severity=invalid-severity", + "?time_range_hours=-1" + ] + + for param in invalid_params: + response = client.get(f"/diagnostics/report{param}", headers=auth_headers) + assert response.status_code in [400, 404, 422] + + +class TestDiagnosticsErrorReporting: + """Test cases for error reporting functionality.""" + + @patch('src.server.fastapi_app.get_current_user') + def test_get_error_statistics(self, mock_user, client): + """Test getting error statistics.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.get("/diagnostics/errors/stats") + # Expected 404 since endpoint not implemented yet + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + expected_fields = [ + "total_errors", "errors_by_type", "errors_by_severity", + "recent_errors", "error_trends" + ] + for field in expected_fields: + assert field in data + + @patch('src.server.fastapi_app.get_current_user') + def test_get_recent_errors(self, mock_user, client): + """Test getting recent errors.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.get("/diagnostics/errors/recent") + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + assert "errors" in data + assert isinstance(data["errors"], list) + + @patch('src.server.fastapi_app.get_current_user') + def test_clear_error_logs(self, mock_user, client): + """Test clearing error logs.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.delete("/diagnostics/errors/clear") + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + assert "cleared_count" in data + + +class TestDiagnosticsSystemHealth: + """Test cases for system health diagnostics.""" + + @patch('src.server.fastapi_app.get_current_user') + def test_get_system_health_overview(self, mock_user, client): + """Test getting system health overview.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.get("/diagnostics/system/health") + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + expected_fields = [ + "overall_status", "cpu_usage", "memory_usage", + "disk_usage", "network_status", "service_status" + ] + for field in expected_fields: + assert field in data + + @patch('src.server.fastapi_app.get_current_user') + def test_run_system_diagnostics(self, mock_user, client): + """Test running system diagnostics.""" + mock_user.return_value = {"user_id": "test_user"} + + diagnostic_options = { + "check_disk": True, + "check_memory": True, + "check_network": True, + "check_database": True + } + + response = client.post("/diagnostics/system/run", json=diagnostic_options) + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + assert "diagnostic_id" in data + assert "status" in data + + +class TestDiagnosticsLogManagement: + """Test cases for log management diagnostics.""" + + @patch('src.server.fastapi_app.get_current_user') + def test_get_log_file_info(self, mock_user, client): + """Test getting log file information.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.get("/diagnostics/logs/info") + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + expected_fields = [ + "log_files", "total_size_bytes", "oldest_entry", + "newest_entry", "rotation_status" + ] + for field in expected_fields: + assert field in data + + @patch('src.server.fastapi_app.get_current_user') + def test_get_log_entries(self, mock_user, client): + """Test getting log entries.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.get("/diagnostics/logs/entries") + assert response.status_code in [200, 404] + + # Test with filters + response = client.get("/diagnostics/logs/entries?level=ERROR&limit=100") + assert response.status_code in [200, 404] + + @patch('src.server.fastapi_app.get_current_user') + def test_export_logs(self, mock_user, client): + """Test exporting logs.""" + mock_user.return_value = {"user_id": "test_user"} + + export_options = { + "format": "json", + "include_levels": ["ERROR", "WARNING", "INFO"], + "time_range_hours": 24 + } + + response = client.post("/diagnostics/logs/export", json=export_options) + assert response.status_code in [200, 404] + + @patch('src.server.fastapi_app.get_current_user') + def test_rotate_logs(self, mock_user, client): + """Test log rotation.""" + mock_user.return_value = {"user_id": "test_user"} + + response = client.post("/diagnostics/logs/rotate") + assert response.status_code in [200, 404] + + if response.status_code == 200: + data = response.json() + assert "rotated_files" in data + assert "status" in data + + +class TestDiagnosticsIntegration: + """Integration tests for diagnostics functionality.""" + + @patch('src.server.fastapi_app.get_current_user') + def test_diagnostics_workflow(self, mock_user, client): + """Test typical diagnostics workflow.""" + mock_user.return_value = {"user_id": "test_user"} + + # 1. Get system health overview + response = client.get("/diagnostics/system/health") + assert response.status_code in [200, 404] + + # 2. Get error statistics + response = client.get("/diagnostics/errors/stats") + assert response.status_code in [200, 404] + + # 3. Generate full diagnostics report + response = client.get("/diagnostics/report") + assert response.status_code in [200, 404] + + # 4. Check log file status + response = client.get("/diagnostics/logs/info") + assert response.status_code in [200, 404] + + def test_diagnostics_error_handling(self, client, auth_headers): + """Test error handling across diagnostics endpoints.""" + endpoints = [ + "/diagnostics/report", + "/diagnostics/errors/stats", + "/diagnostics/system/health", + "/diagnostics/logs/info" + ] + + for endpoint in endpoints: + response = client.get(endpoint, headers=auth_headers) + assert response.status_code in [200, 404] + + @patch('src.server.fastapi_app.get_current_user') + def test_diagnostics_concurrent_requests(self, mock_user, client): + """Test handling of concurrent diagnostics requests.""" + mock_user.return_value = {"user_id": "test_user"} + + # Multiple simultaneous requests should be handled gracefully + response = client.get("/diagnostics/report") + assert response.status_code in [200, 404] + + +class TestDiagnosticsEdgeCases: + """Test edge cases for diagnostics functionality.""" + + def test_diagnostics_with_missing_log_files(self, client, auth_headers): + """Test diagnostics when log files are missing.""" + response = client.get("/diagnostics/logs/info", headers=auth_headers) + # Should handle missing log files gracefully + assert response.status_code in [200, 404, 500] + + def test_diagnostics_with_large_log_files(self, client, auth_headers): + """Test diagnostics with very large log files.""" + # Test with limit parameter for large files + response = client.get("/diagnostics/logs/entries?limit=10", headers=auth_headers) + assert response.status_code in [200, 404] + + @patch('src.server.fastapi_app.get_current_user') + def test_diagnostics_export_formats(self, mock_user, client): + """Test different export formats for diagnostics.""" + mock_user.return_value = {"user_id": "test_user"} + + export_formats = ["json", "csv", "txt"] + + for format_type in export_formats: + export_data = {"format": format_type} + response = client.post("/diagnostics/logs/export", json=export_data) + assert response.status_code in [200, 404, 400] + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/src/tests/unit/test_logging_functionality.py b/src/tests/unit/test_logging_functionality.py new file mode 100644 index 0000000..115b8b3 --- /dev/null +++ b/src/tests/unit/test_logging_functionality.py @@ -0,0 +1,403 @@ +""" +Unit tests for logging functionality. + +This module tests the logging configuration, log file management, +and error reporting components. +""" + +import pytest +import logging +import tempfile +import os +from unittest.mock import patch, Mock, mock_open +from pathlib import Path + +# Import logging components +try: + from src.infrastructure.logging.GlobalLogger import GlobalLogger +except ImportError: + # Mock GlobalLogger if not available + class MockGlobalLogger: + def __init__(self): + self.logger = logging.getLogger(__name__) + + def get_logger(self, name): + return logging.getLogger(name) + + def configure_logging(self, level=logging.INFO): + logging.basicConfig(level=level) + + GlobalLogger = MockGlobalLogger + + +class TestGlobalLoggerConfiguration: + """Test cases for global logger configuration.""" + + def test_logger_initialization(self): + """Test logger initialization.""" + global_logger = GlobalLogger() + assert global_logger is not None + + logger = global_logger.get_logger("test_logger") + assert logger is not None + assert logger.name == "test_logger" + + def test_logger_level_configuration(self): + """Test logger level configuration.""" + global_logger = GlobalLogger() + + # Test different log levels + levels = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR] + + for level in levels: + global_logger.configure_logging(level) + logger = global_logger.get_logger("test_level") + assert logger.level <= level or logger.parent.level <= level + + def test_multiple_logger_instances(self): + """Test multiple logger instances.""" + global_logger = GlobalLogger() + + logger1 = global_logger.get_logger("logger1") + logger2 = global_logger.get_logger("logger2") + logger3 = global_logger.get_logger("logger1") # Same name as logger1 + + assert logger1 != logger2 + assert logger1 is logger3 # Should return same instance + + @patch('logging.basicConfig') + def test_logging_configuration_calls(self, mock_basic_config): + """Test that logging configuration is called correctly.""" + global_logger = GlobalLogger() + global_logger.configure_logging(logging.DEBUG) + + mock_basic_config.assert_called() + + +class TestLogFileManagement: + """Test cases for log file management.""" + + def setUp(self): + """Set up test environment.""" + self.temp_dir = tempfile.mkdtemp() + self.log_file = os.path.join(self.temp_dir, "test.log") + + def tearDown(self): + """Clean up test environment.""" + if os.path.exists(self.temp_dir): + import shutil + shutil.rmtree(self.temp_dir) + + def test_log_file_creation(self): + """Test log file creation.""" + # Configure logger to write to test file + logger = logging.getLogger("test_file") + handler = logging.FileHandler(self.log_file) + logger.addHandler(handler) + logger.setLevel(logging.INFO) + + # Write log message + logger.info("Test log message") + handler.close() + + # Verify file was created and contains message + assert os.path.exists(self.log_file) + with open(self.log_file, 'r') as f: + content = f.read() + assert "Test log message" in content + + def test_log_file_rotation(self): + """Test log file rotation functionality.""" + from logging.handlers import RotatingFileHandler + + # Create rotating file handler + handler = RotatingFileHandler( + self.log_file, + maxBytes=100, # Small size for testing + backupCount=3 + ) + + logger = logging.getLogger("test_rotation") + logger.addHandler(handler) + logger.setLevel(logging.INFO) + + # Write enough messages to trigger rotation + for i in range(10): + logger.info(f"Long test message {i} that should trigger rotation when we write enough of them") + + handler.close() + + # Check that rotation occurred + assert os.path.exists(self.log_file) + + @patch('builtins.open', mock_open(read_data="log content")) + def test_log_file_reading(self, mock_file): + """Test reading log files.""" + # Mock reading a log file + with open("test.log", 'r') as f: + content = f.read() + + assert content == "log content" + mock_file.assert_called_once_with("test.log", 'r') + + def test_log_file_permissions(self): + """Test log file permissions.""" + # Create log file + with open(self.log_file, 'w') as f: + f.write("test") + + # Check file exists and is readable + assert os.path.exists(self.log_file) + assert os.access(self.log_file, os.R_OK) + assert os.access(self.log_file, os.W_OK) + + +class TestErrorReporting: + """Test cases for error reporting functionality.""" + + def test_error_logging(self): + """Test error message logging.""" + logger = logging.getLogger("test_errors") + + with patch.object(logger, 'error') as mock_error: + logger.error("Test error message") + mock_error.assert_called_once_with("Test error message") + + def test_exception_logging(self): + """Test exception logging with traceback.""" + logger = logging.getLogger("test_exceptions") + + with patch.object(logger, 'exception') as mock_exception: + try: + raise ValueError("Test exception") + except ValueError: + logger.exception("An error occurred") + + mock_exception.assert_called_once_with("An error occurred") + + def test_warning_logging(self): + """Test warning message logging.""" + logger = logging.getLogger("test_warnings") + + with patch.object(logger, 'warning') as mock_warning: + logger.warning("Test warning message") + mock_warning.assert_called_once_with("Test warning message") + + def test_info_logging(self): + """Test info message logging.""" + logger = logging.getLogger("test_info") + + with patch.object(logger, 'info') as mock_info: + logger.info("Test info message") + mock_info.assert_called_once_with("Test info message") + + def test_debug_logging(self): + """Test debug message logging.""" + logger = logging.getLogger("test_debug") + logger.setLevel(logging.DEBUG) + + with patch.object(logger, 'debug') as mock_debug: + logger.debug("Test debug message") + mock_debug.assert_called_once_with("Test debug message") + + +class TestLogFormatter: + """Test cases for log formatting.""" + + def test_log_format_structure(self): + """Test log message format structure.""" + formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + + # Create a log record + record = logging.LogRecord( + name="test_logger", + level=logging.INFO, + pathname="test.py", + lineno=1, + msg="Test message", + args=(), + exc_info=None + ) + + formatted = formatter.format(record) + + # Check format components + assert "test_logger" in formatted + assert "INFO" in formatted + assert "Test message" in formatted + + def test_log_format_with_exception(self): + """Test log formatting with exception information.""" + formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + + try: + raise ValueError("Test exception") + except ValueError: + import sys + exc_info = sys.exc_info() + + record = logging.LogRecord( + name="test_logger", + level=logging.ERROR, + pathname="test.py", + lineno=1, + msg="Error occurred", + args=(), + exc_info=exc_info + ) + + formatted = formatter.format(record) + + assert "ERROR" in formatted + assert "Error occurred" in formatted + # Exception info should be included + assert "ValueError" in formatted or "Traceback" in formatted + + def test_custom_log_format(self): + """Test custom log format.""" + custom_formatter = logging.Formatter( + '[%(levelname)s] %(name)s: %(message)s' + ) + + record = logging.LogRecord( + name="custom_logger", + level=logging.WARNING, + pathname="test.py", + lineno=1, + msg="Custom message", + args=(), + exc_info=None + ) + + formatted = custom_formatter.format(record) + + assert formatted.startswith("[WARNING]") + assert "custom_logger:" in formatted + assert "Custom message" in formatted + + +class TestLoggerIntegration: + """Integration tests for logging functionality.""" + + def test_logger_with_multiple_handlers(self): + """Test logger with multiple handlers.""" + logger = logging.getLogger("multi_handler_test") + logger.setLevel(logging.INFO) + + # Clear any existing handlers + logger.handlers = [] + + # Add console handler + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.INFO) + logger.addHandler(console_handler) + + # Add file handler + with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp_file: + file_handler = logging.FileHandler(temp_file.name) + file_handler.setLevel(logging.WARNING) + logger.addHandler(file_handler) + + # Log messages at different levels + with patch.object(console_handler, 'emit') as mock_console: + with patch.object(file_handler, 'emit') as mock_file: + logger.info("Info message") # Should go to console only + logger.warning("Warning message") # Should go to both + logger.error("Error message") # Should go to both + + # Console handler should receive all messages + assert mock_console.call_count == 3 + + # File handler should receive only warning and error + assert mock_file.call_count == 2 + + file_handler.close() + os.unlink(temp_file.name) + + def test_logger_hierarchy(self): + """Test logger hierarchy and inheritance.""" + parent_logger = logging.getLogger("parent") + child_logger = logging.getLogger("parent.child") + grandchild_logger = logging.getLogger("parent.child.grandchild") + + # Set level on parent + parent_logger.setLevel(logging.WARNING) + + # Child loggers should inherit level + assert child_logger.parent == parent_logger + assert grandchild_logger.parent == child_logger + + def test_logger_configuration_persistence(self): + """Test that logger configuration persists.""" + logger_name = "persistent_test" + + # Configure logger + logger1 = logging.getLogger(logger_name) + logger1.setLevel(logging.DEBUG) + + # Get same logger instance + logger2 = logging.getLogger(logger_name) + + # Should be same instance with same configuration + assert logger1 is logger2 + assert logger2.level == logging.DEBUG + + +class TestLoggerErrorHandling: + """Test error handling in logging functionality.""" + + def test_logging_with_invalid_level(self): + """Test logging with invalid level.""" + logger = logging.getLogger("invalid_level_test") + + # Setting invalid level should not crash + try: + logger.setLevel("INVALID_LEVEL") + except (ValueError, TypeError): + # Expected to raise an exception + pass + + def test_logging_to_readonly_file(self): + """Test logging to read-only file.""" + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + temp_file.write(b"existing content") + temp_file_path = temp_file.name + + try: + # Make file read-only + os.chmod(temp_file_path, 0o444) + + # Try to create file handler - should handle gracefully + try: + handler = logging.FileHandler(temp_file_path) + handler.close() + except PermissionError: + # Expected behavior + pass + finally: + # Clean up + try: + os.chmod(temp_file_path, 0o666) + os.unlink(temp_file_path) + except: + pass + + def test_logging_with_missing_directory(self): + """Test logging to file in non-existent directory.""" + non_existent_path = "/non/existent/directory/test.log" + + # Should handle missing directory gracefully + try: + handler = logging.FileHandler(non_existent_path) + handler.close() + except (FileNotFoundError, OSError): + # Expected behavior + pass + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) \ No newline at end of file