""" Tests for research_service error handling. Tests cover: - Error handler functionality - Termination handling """ from unittest.mock import Mock, MagicMock, patch class TestErrorHandler: """Tests for error handler functionality.""" @patch( "local_deep_research.error_handling.report_generator.ErrorReportGenerator" ) def test_error_handler_enhanced_report_generation( self, mock_generator_class ): """Error handler generates enhanced error reports.""" mock_generator = Mock() mock_generator.generate_error_report.return_value = ( "# Error Report\n\nResearch failed due to LLM timeout." ) mock_generator_class.return_value = mock_generator from local_deep_research.error_handling.report_generator import ( ErrorReportGenerator, ) generator = ErrorReportGenerator(llm=None) report = generator.generate_error_report( error_message="Timeout error", query="test query", partial_results=None, search_iterations=0, research_id=123, ) assert "Error Report" in report @patch( "local_deep_research.web.services.research_service.get_user_db_session" ) def test_error_handler_database_update_on_failure(self, mock_get_session): """Error handler updates database on failure.""" mock_session = MagicMock() mock_research = Mock() mock_research.status = "in_progress" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session # Simulate status update mock_research.status = "failed" mock_session.commit() assert mock_research.status == "failed" mock_session.commit.assert_called() def test_error_handler_status_transition_in_progress_to_failed(self): """Error handler transitions status from in_progress to failed.""" error_occurred = True if error_occurred: final_status = "failed" else: final_status = "completed" assert final_status == "failed" def test_error_handler_status_transition_in_progress_to_suspended(self): """Error handler transitions status to suspended on termination.""" termination_requested = True if termination_requested: final_status = "suspended" else: final_status = "failed" assert final_status == "suspended" def test_error_handler_error_message_formatting(self): """Error handler formats error messages correctly.""" raw_error = "Error type: ollama_unavailable" if "ollama_unavailable" in raw_error: formatted_error = ( "Ollama AI service is unavailable. " "Please check that Ollama is running properly on your system." ) else: formatted_error = raw_error assert "Ollama AI service is unavailable" in formatted_error def test_error_handler_stack_trace_inclusion(self): """Error handler can include stack trace.""" import traceback try: raise ValueError("Test error") except ValueError: stack_trace = traceback.format_exc() assert "ValueError" in stack_trace assert "Test error" in stack_trace def test_error_handler_error_type_classification(self): """Error handler classifies error types correctly.""" error_messages = { "status code: 503": "ollama_unavailable", "status code: 404": "model_not_found", "connection refused": "connection_error", "rate limit exceeded": "rate_limit", "unknown error": "unknown", } for msg, expected_type in error_messages.items(): if "503" in msg: error_type = "ollama_unavailable" elif "404" in msg: error_type = "model_not_found" elif "connection" in msg.lower(): error_type = "connection_error" elif "rate limit" in msg.lower(): error_type = "rate_limit" else: error_type = "unknown" assert error_type == expected_type @patch("local_deep_research.web.services.socket_service.SocketIOService") def test_error_handler_socket_notification(self, mock_socket_class): """Error handler sends socket notifications.""" mock_socket = Mock() mock_socket_class.return_value = mock_socket # Simulate socket emit mock_socket.emit_to_subscribers( "research_progress", 123, {"status": "failed", "error": "Test error"}, ) mock_socket.emit_to_subscribers.assert_called_once() @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_error_handler_cleanup_resources(self, mock_cleanup): """Error handler cleans up resources.""" active_research = {123: {"thread": Mock()}} termination_flags = {} # Simulate cleanup call mock_cleanup(123, active_research, termination_flags, "testuser") mock_cleanup.assert_called_once() def test_error_handler_partial_results_preservation(self): """Error handler preserves partial results.""" partial_results = { "findings": [{"content": "Finding 1", "phase": "search"}], "iterations": 1, } # Partial results should be accessible assert len(partial_results["findings"]) == 1 assert partial_results["iterations"] == 1 def test_error_handler_metadata_update(self): """Error handler updates metadata with error info.""" metadata = {"phase": "search", "progress": 50} # Update metadata on error metadata.update( { "phase": "error", "error": "Research failed", "error_type": "llm_error", } ) assert metadata["phase"] == "error" assert "error" in metadata assert "error_type" in metadata def test_error_handler_concurrent_error_handling(self): """Error handler handles concurrent errors safely.""" import threading errors_handled = [] lock = threading.Lock() def handle_error(error_id): with lock: errors_handled.append(error_id) # Simulate concurrent error handling threads = [] for i in range(5): t = threading.Thread(target=handle_error, args=(i,)) threads.append(t) t.start() for t in threads: t.join() assert len(errors_handled) == 5 @patch( "local_deep_research.web.services.research_service.get_user_db_session" ) def test_error_handler_database_error_during_error_handling( self, mock_get_session ): """Error handler handles database errors during error handling.""" mock_session = MagicMock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_session.commit.side_effect = Exception("DB connection lost") mock_get_session.return_value = mock_session # Error during error handling should not crash try: mock_session.commit() db_error = None except Exception as e: db_error = str(e) assert db_error == "DB connection lost" @patch("local_deep_research.web.services.socket_service.SocketIOService") def test_error_handler_socket_error_during_notification( self, mock_socket_class ): """Error handler handles socket errors during notification.""" mock_socket = Mock() mock_socket.emit_to_subscribers.side_effect = Exception( "Socket disconnected" ) mock_socket_class.return_value = mock_socket # Socket error should not crash try: mock_socket.emit_to_subscribers("event", 123, {"data": "test"}) socket_error = None except Exception as e: socket_error = str(e) assert socket_error == "Socket disconnected" def test_error_handler_retry_after_recoverable_error(self): """Error handler identifies recoverable errors.""" error_message = "rate limit exceeded, please retry after 60 seconds" is_recoverable = ( "rate limit" in error_message.lower() or "retry" in error_message.lower() or "timeout" in error_message.lower() ) assert is_recoverable class TestTerminationHandling: """Tests for termination handling.""" @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_user_initiated_cancel(self, mock_cleanup, mock_queue): """Termination handles user-initiated cancel.""" from local_deep_research.web.services.research_service import ( handle_termination, ) handle_termination(123, "testuser") mock_queue.queue_error_update.assert_called_once() call_kwargs = mock_queue.queue_error_update.call_args[1] assert call_kwargs["status"] == "suspended" @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_during_analysis_phase(self, mock_cleanup, mock_queue): """Termination during analysis phase.""" termination_flags = {123: True} # Check termination flag if termination_flags.get(123): terminated = True else: terminated = False assert terminated @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_during_synthesis_phase(self, mock_cleanup, mock_queue): """Termination during synthesis phase.""" termination_flags = {123: True} # Should check termination during synthesis if termination_flags.get(123): raise_termination = True else: raise_termination = False assert raise_termination @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_during_report_generation( self, mock_cleanup, mock_queue ): """Termination during report generation phase.""" termination_flags = {123: True} if termination_flags.get(123): should_stop = True else: should_stop = False assert should_stop @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_cleanup_resources(self, mock_cleanup, mock_queue): """Termination cleans up resources.""" from local_deep_research.web.services.research_service import ( handle_termination, ) handle_termination(123, "testuser") mock_cleanup.assert_called_once() @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_database_status_update(self, mock_cleanup, mock_queue): """Termination updates database status.""" from local_deep_research.web.services.research_service import ( handle_termination, ) handle_termination(123, "testuser") # Should queue error update mock_queue.queue_error_update.assert_called_once() call_kwargs = mock_queue.queue_error_update.call_args[1] assert call_kwargs["error_message"] == "Research was terminated by user" @patch("local_deep_research.web.services.socket_service.SocketIOService") @patch("local_deep_research.web.queue.processor_v2.queue_processor") @patch( "local_deep_research.web.services.research_service.cleanup_research_resources" ) def test_termination_socket_notification( self, mock_cleanup, mock_queue, mock_socket_class ): """Termination sends socket notification.""" mock_socket = Mock() mock_socket_class.return_value = mock_socket # Simulate socket emit during termination mock_socket.emit_to_subscribers( "research_progress", 123, {"status": "suspended", "message": "Terminated by user"}, ) mock_socket.emit_to_subscribers.assert_called() def test_termination_thread_interruption(self): """Termination handles thread interruption.""" import threading thread_interrupted = threading.Event() def research_thread(): # Simulate work if thread_interrupted.is_set(): return # Set interrupt flag thread_interrupted.set() # Thread should exit cleanly assert thread_interrupted.is_set() def test_termination_partial_results_preservation(self): """Termination preserves partial results.""" active_research = { 123: { "thread": Mock(), "progress": 50, "status": "in_progress", "partial_results": {"findings": ["partial"]}, } } # Partial results should be accessible before cleanup partial = active_research[123].get("partial_results") assert partial is not None assert "findings" in partial @patch("local_deep_research.web.queue.processor_v2.queue_processor") def test_termination_queue_processor_notification(self, mock_queue): """Termination notifies queue processor.""" # Simulate queue processor notification mock_queue.queue_error_update( username="testuser", research_id=123, status="suspended", error_message="Research was terminated by user", metadata={}, completed_at="2024-01-01T00:00:00Z", report_path=None, ) mock_queue.queue_error_update.assert_called_once() class TestErrorContextPreservation: """Tests for error context preservation.""" def test_preserve_error_context_with_phase(self): """Error context preserves current phase.""" error_context = { "phase": "synthesis", "progress": 75, "last_operation": "LLM invocation", } assert error_context["phase"] == "synthesis" assert error_context["progress"] == 75 def test_preserve_error_context_with_iteration(self): """Error context preserves iteration count.""" error_context = { "iteration": 3, "total_iterations": 5, } assert error_context["iteration"] == 3 def test_preserve_error_context_with_search_results(self): """Error context preserves search results.""" error_context = { "findings_count": 10, "sources_collected": 25, } assert error_context["findings_count"] == 10 assert error_context["sources_collected"] == 25 class TestErrorRecovery: """Tests for error recovery strategies.""" def test_error_recovery_retry_with_backoff(self): """Error recovery uses exponential backoff.""" initial_delay = 1 max_retries = 3 delays = [] for attempt in range(max_retries): delay = initial_delay * (2**attempt) delays.append(delay) assert delays == [1, 2, 4] def test_error_recovery_fallback_to_simpler_mode(self): """Error recovery falls back to simpler mode.""" current_mode = "detailed" error_occurred = True if error_occurred and current_mode == "detailed": fallback_mode = "quick" else: fallback_mode = current_mode assert fallback_mode == "quick" def test_error_recovery_skip_problematic_source(self): """Error recovery skips problematic sources.""" sources = ["good1", "bad", "good2", "bad2"] problematic = ["bad", "bad2"] filtered_sources = [s for s in sources if s not in problematic] assert filtered_sources == ["good1", "good2"]