"""Unit tests for Error Tracking utilities. Tests cover: - Error tracking and history management - Error statistics calculation - Request context management - Context stack operations - Global singleton instances - Error deduplication and cleanup """ from datetime import datetime, timezone from typing import Any, Dict import pytest from src.server.utils.error_tracking import ( ErrorTracker, RequestContextManager, get_context_manager, get_error_tracker, reset_error_tracker, ) @pytest.fixture def error_tracker(): """Create error tracker instance.""" return ErrorTracker() @pytest.fixture def context_manager(): """Create request context manager instance.""" return RequestContextManager() class TestErrorTrackerInitialization: """Tests for ErrorTracker initialization.""" def test_initialization(self, error_tracker): """Test error tracker initialization.""" assert error_tracker.error_history == [] assert error_tracker.max_history_size == 1000 def test_default_max_history_size(self, error_tracker): """Test default max history size.""" assert error_tracker.max_history_size == 1000 class TestTrackError: """Tests for error tracking functionality.""" def test_track_error_basic(self, error_tracker): """Test basic error tracking.""" error_id = error_tracker.track_error( error_type="ValueError", message="Test error", request_path="/api/test", request_method="GET", ) assert error_id is not None assert len(error_tracker.error_history) == 1 error_entry = error_tracker.error_history[0] assert error_entry["id"] == error_id assert error_entry["type"] == "ValueError" assert error_entry["message"] == "Test error" assert error_entry["request_path"] == "/api/test" assert error_entry["request_method"] == "GET" assert error_entry["status_code"] == 500 # default def test_track_error_with_user_id(self, error_tracker): """Test error tracking with user ID.""" error_id = error_tracker.track_error( error_type="AuthError", message="Unauthorized access", request_path="/api/protected", request_method="POST", user_id="user123", ) error_entry = error_tracker.error_history[0] assert error_entry["user_id"] == "user123" def test_track_error_with_custom_status_code(self, error_tracker): """Test error tracking with custom status code.""" error_id = error_tracker.track_error( error_type="NotFoundError", message="Resource not found", request_path="/api/resource/123", request_method="GET", status_code=404, ) error_entry = error_tracker.error_history[0] assert error_entry["status_code"] == 404 def test_track_error_with_details(self, error_tracker): """Test error tracking with additional details.""" details = { "stack_trace": "line 1\nline 2", "user_agent": "Mozilla/5.0", } error_id = error_tracker.track_error( error_type="RuntimeError", message="Runtime error occurred", request_path="/api/action", request_method="PUT", details=details, ) error_entry = error_tracker.error_history[0] assert error_entry["details"] == details def test_track_error_with_request_id(self, error_tracker): """Test error tracking with request ID for correlation.""" error_id = error_tracker.track_error( error_type="DatabaseError", message="Connection failed", request_path="/api/data", request_method="GET", request_id="req-12345", ) error_entry = error_tracker.error_history[0] assert error_entry["request_id"] == "req-12345" def test_track_multiple_errors(self, error_tracker): """Test tracking multiple errors.""" error_id1 = error_tracker.track_error( error_type="Error1", message="First error", request_path="/api/1", request_method="GET", ) error_id2 = error_tracker.track_error( error_type="Error2", message="Second error", request_path="/api/2", request_method="POST", ) assert len(error_tracker.error_history) == 2 assert error_id1 != error_id2 assert error_tracker.error_history[0]["id"] == error_id1 assert error_tracker.error_history[1]["id"] == error_id2 def test_error_has_timestamp(self, error_tracker): """Test that errors have ISO formatted timestamps.""" error_id = error_tracker.track_error( error_type="TestError", message="Test", request_path="/test", request_method="GET", ) error_entry = error_tracker.error_history[0] timestamp = error_entry["timestamp"] # Should be valid ISO format with timezone parsed = datetime.fromisoformat(timestamp) assert parsed.tzinfo is not None class TestErrorHistoryManagement: """Tests for error history management.""" def test_history_size_limit(self, error_tracker): """Test that history size is limited to max_history_size.""" error_tracker.max_history_size = 5 # Track 10 errors for i in range(10): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", ) # Only last 5 should remain assert len(error_tracker.error_history) == 5 # Should be errors 5-9 assert error_tracker.error_history[0]["type"] == "Error5" assert error_tracker.error_history[-1]["type"] == "Error9" def test_clear_history(self, error_tracker): """Test clearing error history.""" # Track some errors for i in range(3): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", ) assert len(error_tracker.error_history) == 3 error_tracker.clear_history() assert len(error_tracker.error_history) == 0 def test_get_recent_errors(self, error_tracker): """Test getting recent errors.""" # Track 5 errors for i in range(5): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", ) # Get last 3 recent = error_tracker.get_recent_errors(limit=3) assert len(recent) == 3 assert recent[0]["type"] == "Error2" assert recent[1]["type"] == "Error3" assert recent[2]["type"] == "Error4" def test_get_recent_errors_with_empty_history(self, error_tracker): """Test get_recent_errors with empty history.""" recent = error_tracker.get_recent_errors() assert recent == [] def test_get_recent_errors_default_limit(self, error_tracker): """Test get_recent_errors default limit is 10.""" # Track 15 errors for i in range(15): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", ) # Default limit is 10 recent = error_tracker.get_recent_errors() assert len(recent) == 10 # Should be errors 5-14 assert recent[0]["type"] == "Error5" assert recent[-1]["type"] == "Error14" def test_get_recent_errors_limit_exceeds_history(self, error_tracker): """Test get_recent_errors when limit exceeds history size.""" # Track 3 errors for i in range(3): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", ) # Request more than available recent = error_tracker.get_recent_errors(limit=10) assert len(recent) == 3 class TestErrorStatistics: """Tests for error statistics calculation.""" def test_get_error_stats_empty_history(self, error_tracker): """Test error stats with empty history.""" stats = error_tracker.get_error_stats() assert stats["total_errors"] == 0 assert stats["error_types"] == {} def test_get_error_stats_single_error(self, error_tracker): """Test error stats with single error.""" error_tracker.track_error( error_type="ValueError", message="Test error", request_path="/api/test", request_method="GET", status_code=400, ) stats = error_tracker.get_error_stats() assert stats["total_errors"] == 1 assert stats["error_types"] == {"ValueError": 1} assert stats["status_codes"] == {400: 1} assert stats["last_error"]["type"] == "ValueError" def test_get_error_stats_multiple_error_types(self, error_tracker): """Test error stats with multiple error types.""" error_tracker.track_error( error_type="ValueError", message="Error 1", request_path="/api/1", request_method="GET", status_code=400, ) error_tracker.track_error( error_type="ValueError", message="Error 2", request_path="/api/2", request_method="GET", status_code=400, ) error_tracker.track_error( error_type="RuntimeError", message="Error 3", request_path="/api/3", request_method="POST", status_code=500, ) stats = error_tracker.get_error_stats() assert stats["total_errors"] == 3 assert stats["error_types"] == {"ValueError": 2, "RuntimeError": 1} assert stats["status_codes"] == {400: 2, 500: 1} def test_get_error_stats_multiple_status_codes(self, error_tracker): """Test error stats with multiple status codes.""" status_codes = [400, 404, 500, 400, 404] for i, code in enumerate(status_codes): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", status_code=code, ) stats = error_tracker.get_error_stats() assert stats["status_codes"] == {400: 2, 404: 2, 500: 1} def test_get_error_stats_last_error(self, error_tracker): """Test that last_error contains most recent error.""" error_tracker.track_error( error_type="FirstError", message="First", request_path="/api/1", request_method="GET", ) error_tracker.track_error( error_type="LastError", message="Last", request_path="/api/2", request_method="GET", ) stats = error_tracker.get_error_stats() assert stats["last_error"]["type"] == "LastError" assert stats["last_error"]["message"] == "Last" class TestRequestContextManager: """Tests for RequestContextManager.""" def test_initialization(self, context_manager): """Test context manager initialization.""" assert context_manager.context_stack == [] def test_push_context(self, context_manager): """Test pushing context onto stack.""" context_manager.push_context( request_id="req-123", request_path="/api/test", request_method="GET", ) assert len(context_manager.context_stack) == 1 context = context_manager.context_stack[0] assert context["request_id"] == "req-123" assert context["request_path"] == "/api/test" assert context["request_method"] == "GET" assert context["user_id"] is None assert "timestamp" in context def test_push_context_with_user_id(self, context_manager): """Test pushing context with user ID.""" context_manager.push_context( request_id="req-123", request_path="/api/protected", request_method="POST", user_id="user456", ) context = context_manager.context_stack[0] assert context["user_id"] == "user456" def test_push_multiple_contexts(self, context_manager): """Test pushing multiple contexts.""" context_manager.push_context( request_id="req-1", request_path="/api/1", request_method="GET", ) context_manager.push_context( request_id="req-2", request_path="/api/2", request_method="POST", ) assert len(context_manager.context_stack) == 2 assert context_manager.context_stack[0]["request_id"] == "req-1" assert context_manager.context_stack[1]["request_id"] == "req-2" def test_pop_context(self, context_manager): """Test popping context from stack.""" context_manager.push_context( request_id="req-123", request_path="/api/test", request_method="GET", ) popped = context_manager.pop_context() assert popped is not None assert popped["request_id"] == "req-123" assert len(context_manager.context_stack) == 0 def test_pop_context_empty_stack(self, context_manager): """Test popping from empty stack returns None.""" popped = context_manager.pop_context() assert popped is None def test_pop_context_order(self, context_manager): """Test that pop_context follows LIFO order.""" context_manager.push_context( request_id="req-1", request_path="/api/1", request_method="GET", ) context_manager.push_context( request_id="req-2", request_path="/api/2", request_method="POST", ) # Pop should return last pushed popped1 = context_manager.pop_context() assert popped1["request_id"] == "req-2" popped2 = context_manager.pop_context() assert popped2["request_id"] == "req-1" # Stack should be empty assert len(context_manager.context_stack) == 0 def test_get_current_context(self, context_manager): """Test getting current context without popping.""" context_manager.push_context( request_id="req-123", request_path="/api/test", request_method="GET", ) current = context_manager.get_current_context() assert current is not None assert current["request_id"] == "req-123" # Stack should still have the context assert len(context_manager.context_stack) == 1 def test_get_current_context_empty_stack(self, context_manager): """Test getting current context from empty stack.""" current = context_manager.get_current_context() assert current is None def test_get_current_context_returns_last(self, context_manager): """Test that get_current_context returns most recent.""" context_manager.push_context( request_id="req-1", request_path="/api/1", request_method="GET", ) context_manager.push_context( request_id="req-2", request_path="/api/2", request_method="POST", ) current = context_manager.get_current_context() assert current["request_id"] == "req-2" def test_context_has_timestamp(self, context_manager): """Test that contexts have timestamps.""" context_manager.push_context( request_id="req-123", request_path="/api/test", request_method="GET", ) context = context_manager.get_current_context() timestamp = context["timestamp"] # Should be valid ISO format with timezone parsed = datetime.fromisoformat(timestamp) assert parsed.tzinfo is not None class TestGlobalInstances: """Tests for global singleton instances.""" def test_get_error_tracker_singleton(self): """Test that get_error_tracker returns singleton.""" reset_error_tracker() tracker1 = get_error_tracker() tracker2 = get_error_tracker() assert tracker1 is tracker2 def test_reset_error_tracker(self): """Test reset_error_tracker creates new instance.""" tracker1 = get_error_tracker() reset_error_tracker() tracker2 = get_error_tracker() assert tracker1 is not tracker2 def test_get_context_manager_singleton(self): """Test that get_context_manager returns singleton.""" manager1 = get_context_manager() manager2 = get_context_manager() assert manager1 is manager2 def test_error_tracker_state_persists(self): """Test that error tracker state persists across calls.""" reset_error_tracker() tracker1 = get_error_tracker() tracker1.track_error( error_type="TestError", message="Test", request_path="/test", request_method="GET", ) tracker2 = get_error_tracker() assert len(tracker2.error_history) == 1 assert tracker2.error_history[0]["type"] == "TestError" class TestErrorTrackerEdgeCases: """Tests for edge cases and error conditions.""" def test_track_error_without_details(self, error_tracker): """Test that details default to empty dict.""" error_id = error_tracker.track_error( error_type="Error", message="Test", request_path="/test", request_method="GET", ) error_entry = error_tracker.error_history[0] assert error_entry["details"] == {} def test_track_error_with_none_user_id(self, error_tracker): """Test that user_id can be None.""" error_id = error_tracker.track_error( error_type="Error", message="Test", request_path="/test", request_method="GET", user_id=None, ) error_entry = error_tracker.error_history[0] assert error_entry["user_id"] is None def test_unique_error_ids(self, error_tracker): """Test that each error gets unique ID.""" ids = set() for i in range(100): error_id = error_tracker.track_error( error_type="Error", message="Test", request_path="/test", request_method="GET", ) ids.add(error_id) # All IDs should be unique assert len(ids) == 100 def test_history_trimming_preserves_recent(self, error_tracker): """Test that trimming preserves most recent errors.""" error_tracker.max_history_size = 3 # Track errors with unique types for i in range(5): error_tracker.track_error( error_type=f"Error{i}", message=f"Error {i}", request_path=f"/api/{i}", request_method="GET", ) # Should keep last 3 (errors 2, 3, 4) assert len(error_tracker.error_history) == 3 types = [e["type"] for e in error_tracker.error_history] assert types == ["Error2", "Error3", "Error4"]