"""Comprehensive coverage tests for benchmarks/web_api/benchmark_service.py. Focuses on uncovered code paths: complex branching in _run_benchmark_thread, _process_benchmark_task, _sync_results_to_database, get_benchmark_status, sync_pending_results, cancel_benchmark, update_benchmark_status, and more. """ import json from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, Mock, patch import pytest # --------------------------------------------------------------------------- # Helpers to import from the module under test # --------------------------------------------------------------------------- MODULE = "local_deep_research.benchmarks.web_api.benchmark_service" SETTINGS_CTX_MODULE = "local_deep_research.config.thread_settings" def _import_service(): """Import BenchmarkService with SocketIOService mocked.""" with patch(f"{MODULE}.SocketIOService"): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) return BenchmarkService def _make_service(socket=None): """Create a BenchmarkService with a mock socket service.""" cls = _import_service() svc = cls(socket_service=socket or MagicMock()) return svc # ============================================================ # BenchmarkQueueTracker – deeper coverage # ============================================================ class TestQueueTrackerCleanup: """Cover cleanup_completed_tasks edge cases.""" def test_cleanup_removes_old_completed(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("old1", "user") tracker.update_task_status("old1", BenchmarkTaskStatus.COMPLETED) # Manually set updated_at to the past tracker.tasks["old1"]["updated_at"] = datetime.now(UTC) - timedelta( seconds=7200 ) tracker.cleanup_completed_tasks(max_age_seconds=3600) assert "old1" not in tracker.tasks def test_cleanup_keeps_recent_completed(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("new1", "user") tracker.update_task_status("new1", BenchmarkTaskStatus.COMPLETED) tracker.cleanup_completed_tasks(max_age_seconds=3600) assert "new1" in tracker.tasks def test_cleanup_removes_old_failed(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("fail1", "user") tracker.update_task_status("fail1", BenchmarkTaskStatus.FAILED) tracker.tasks["fail1"]["updated_at"] = datetime.now(UTC) - timedelta( seconds=7200 ) tracker.cleanup_completed_tasks(max_age_seconds=3600) assert "fail1" not in tracker.tasks def test_cleanup_removes_old_cancelled(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("canc1", "user") tracker.update_task_status("canc1", BenchmarkTaskStatus.CANCELLED) tracker.tasks["canc1"]["updated_at"] = datetime.now(UTC) - timedelta( seconds=7200 ) tracker.cleanup_completed_tasks(max_age_seconds=3600) assert "canc1" not in tracker.tasks def test_cleanup_keeps_processing_tasks(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("proc1", "user") tracker.update_task_status("proc1", BenchmarkTaskStatus.PROCESSING) tracker.tasks["proc1"]["updated_at"] = datetime.now(UTC) - timedelta( seconds=7200 ) tracker.cleanup_completed_tasks(max_age_seconds=3600) assert "proc1" in tracker.tasks def test_cleanup_uses_created_at_when_no_updated_at(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("t1", "user") # Set status directly without going through update (no updated_at) tracker.tasks["t1"]["status"] = BenchmarkTaskStatus.COMPLETED.value tracker.tasks["t1"]["created_at"] = datetime.now(UTC) - timedelta( seconds=7200 ) if "updated_at" in tracker.tasks["t1"]: del tracker.tasks["t1"]["updated_at"] tracker.cleanup_completed_tasks(max_age_seconds=3600) assert "t1" not in tracker.tasks def test_update_nonexistent_task_logs_warning(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() # Should not raise; logs a warning tracker.update_task_status("nonexistent", BenchmarkTaskStatus.COMPLETED) def test_remove_task(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, ) tracker = BenchmarkQueueTracker() tracker.add_task("rm1", "user") tracker.remove_task("rm1") assert tracker.get_task_status("rm1") is None def test_remove_nonexistent_task(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, ) tracker = BenchmarkQueueTracker() # Should not raise tracker.remove_task("doesnotexist") # ============================================================ # BenchmarkService – initialization # ============================================================ class TestBenchmarkServiceInit: def test_init_with_provided_socket(self): mock_socket = MagicMock() svc = _make_service(socket=mock_socket) assert svc.socket_service is mock_socket def test_init_fallback_socket(self): """When SocketIOService raises, a MockSocketService is created.""" with patch( f"{MODULE}.SocketIOService", side_effect=RuntimeError("no app") ): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) svc = BenchmarkService() # The mock socket should have emit_to_room method svc.socket_service.emit_to_room("room", "event", {}) # ============================================================ # generate_config_hash / generate_query_hash # ============================================================ class TestHashGeneration: def test_config_hash_deterministic(self): svc = _make_service() cfg = {"iterations": 5, "search_tool": "searxng", "model_name": "gpt-4"} h1 = svc.generate_config_hash(cfg) h2 = svc.generate_config_hash(cfg) assert h1 == h2 assert len(h1) == 8 def test_config_hash_ignores_none_values(self): svc = _make_service() cfg1 = {"iterations": 5, "search_tool": None} cfg2 = {"iterations": 5} assert svc.generate_config_hash(cfg1) == svc.generate_config_hash(cfg2) def test_config_hash_differs_for_different_configs(self): svc = _make_service() h1 = svc.generate_config_hash({"iterations": 5}) h2 = svc.generate_config_hash({"iterations": 10}) assert h1 != h2 def test_query_hash_deterministic(self): svc = _make_service() h1 = svc.generate_query_hash("What is AI?", "simpleqa") h2 = svc.generate_query_hash("What is AI?", "simpleqa") assert h1 == h2 def test_query_hash_strips_whitespace(self): svc = _make_service() h1 = svc.generate_query_hash(" What is AI? ", "simpleqa") h2 = svc.generate_query_hash("What is AI?", "simpleqa") assert h1 == h2 def test_query_hash_case_insensitive_dataset(self): svc = _make_service() h1 = svc.generate_query_hash("What is AI?", "SimpleQA") h2 = svc.generate_query_hash("What is AI?", "simpleqa") assert h1 == h2 # ============================================================ # create_benchmark_run # ============================================================ class TestCreateBenchmarkRun: @patch(f"{MODULE}.BenchmarkRun") def test_create_benchmark_run_success(self, mock_run_cls): svc = _make_service() mock_session = MagicMock() mock_run_instance = MagicMock() mock_run_instance.id = 42 mock_run_cls.return_value = mock_run_instance with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) result = svc.create_benchmark_run( run_name="test", search_config={"iterations": 5}, evaluation_config={}, datasets_config={"simpleqa": {"count": 10}}, username="user1", ) assert result == 42 mock_session.add.assert_called_once_with(mock_run_instance) mock_session.commit.assert_called_once() @patch(f"{MODULE}.BenchmarkRun") def test_create_benchmark_run_db_error(self, mock_run_cls): svc = _make_service() mock_session = MagicMock() mock_session.commit.side_effect = RuntimeError("db error") mock_run_cls.return_value = MagicMock() with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) with pytest.raises(RuntimeError, match="db error"): svc.create_benchmark_run( run_name="test", search_config={}, evaluation_config={}, datasets_config={}, ) mock_session.rollback.assert_called_once() # ============================================================ # _create_task_queue # ============================================================ class TestCreateTaskQueue: @patch(f"{MODULE}.load_dataset") def test_creates_tasks_from_dataset(self, mock_load): svc = _make_service() mock_load.return_value = [ {"id": "ex1", "problem": "Q1?", "answer": "A1"}, {"id": "ex2", "problem": "Q2?", "answer": "A2"}, ] tasks = svc._create_task_queue( {"simpleqa": {"count": 2}}, benchmark_run_id=1, ) assert len(tasks) == 2 assert tasks[0]["question"] == "Q1?" assert tasks[0]["dataset_type"] == "simpleqa" @patch(f"{MODULE}.load_dataset") def test_passes_seed_from_dataset_config(self, mock_load): """A "seed" in the dataset config drives reproducible sampling.""" svc = _make_service() mock_load.return_value = [ {"id": "ex1", "problem": "Q1?", "answer": "A1"}, ] svc._create_task_queue( {"simpleqa": {"count": 1, "seed": 7}}, benchmark_run_id=1, ) mock_load.assert_called_once_with( dataset_type="simpleqa", num_examples=1, seed=7 ) @patch(f"{MODULE}.load_dataset") def test_no_seed_means_random_sampling(self, mock_load): """Without a configured seed the sample stays random (seed=None).""" svc = _make_service() mock_load.return_value = [ {"id": "ex1", "problem": "Q1?", "answer": "A1"}, ] svc._create_task_queue( {"simpleqa": {"count": 1}}, benchmark_run_id=1, ) mock_load.assert_called_once_with( dataset_type="simpleqa", num_examples=1, seed=None ) @patch(f"{MODULE}.load_dataset") def test_queues_every_sampled_question(self, mock_load): """Regression test for #4498: every sampled question becomes a task. The removed cross-run reuse feature silently skipped questions that had results in previous compatible runs, which broke the completed/total accounting (#4451). A run must always process its full sample, even if identical questions were answered before.""" svc = _make_service() examples = [ {"id": f"ex{i}", "problem": f"Q{i}?", "answer": f"A{i}"} for i in range(5) ] mock_load.return_value = examples tasks = svc._create_task_queue( {"simpleqa": {"count": 5}}, benchmark_run_id=1, ) assert len(tasks) == len(examples) assert [t["question"] for t in tasks] == [ e["problem"] for e in examples ] assert [t["task_index"] for t in tasks] == list(range(5)) @patch(f"{MODULE}.load_dataset") def test_skips_zero_count_datasets(self, mock_load): svc = _make_service() tasks = svc._create_task_queue( {"simpleqa": {"count": 0}}, benchmark_run_id=1, ) assert len(tasks) == 0 mock_load.assert_not_called() @patch(f"{MODULE}.load_dataset") def test_browsecomp_dataset_type(self, mock_load): svc = _make_service() mock_load.return_value = [ {"id": "b1", "problem": "Browse Q?", "answer": "Browse A"}, ] tasks = svc._create_task_queue( {"browsecomp": {"count": 1}}, benchmark_run_id=1, ) assert len(tasks) == 1 assert tasks[0]["dataset_type"] == "browsecomp" @patch(f"{MODULE}.load_dataset") def test_example_without_id_gets_default(self, mock_load): svc = _make_service() mock_load.return_value = [ {"problem": "Q?", "answer": "A"}, # no id ] tasks = svc._create_task_queue( {"simpleqa": {"count": 1}}, benchmark_run_id=1, ) assert tasks[0]["example_id"] == "example_0" # ============================================================ # _process_benchmark_task # ============================================================ class TestProcessBenchmarkTask: def _make_task(self, **overrides): task = { "benchmark_run_id": 1, "example_id": "ex1", "dataset_type": "simpleqa", "question": "What is 2+2?", "correct_answer": "4", "query_hash": "abc123", "task_index": 0, "username": "user1", "user_password": None, } task.update(overrides) return task @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_successful_processing_with_grading( self, mock_format, mock_summary, mock_extract, mock_grade ): svc = _make_service() mock_format.return_value = "formatted query" mock_summary.return_value = { "summary": "The answer is 4", "sources": [{"url": "http://example.com"}], } mock_extract.return_value = { "extracted_answer": "4", "confidence": "95", } mock_grade.return_value = { "is_correct": True, "graded_confidence": "98", "grader_response": "Correct answer", } mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task( self._make_task(), {"iterations": 5}, {}, ) assert result["is_correct"] is True assert result["extracted_answer"] == "4" assert result["response"] == "The answer is 4" @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_grading_error_in_result( self, mock_format, mock_summary, mock_extract, mock_grade ): svc = _make_service() mock_format.return_value = "q" mock_summary.return_value = {"summary": "resp", "sources": []} mock_extract.return_value = { "extracted_answer": "ans", "confidence": "50", } mock_grade.return_value = {"grading_error": "model unavailable"} mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task(self._make_task(), {}, {}) assert result["is_correct"] is None assert "model unavailable" in result["grader_response"] @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_grading_returns_none( self, mock_format, mock_summary, mock_extract, mock_grade ): svc = _make_service() mock_format.return_value = "q" mock_summary.return_value = {"summary": "resp", "sources": []} mock_extract.return_value = {"extracted_answer": "ans"} mock_grade.return_value = None mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task(self._make_task(), {}, {}) assert result["is_correct"] is None assert "No evaluation results returned" in result["grader_response"] @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_grading_exception( self, mock_format, mock_summary, mock_extract, mock_grade ): svc = _make_service() mock_format.return_value = "q" mock_summary.return_value = {"summary": "resp", "sources": []} mock_extract.return_value = {"extracted_answer": "ans"} mock_grade.side_effect = ValueError("grade fail") mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task(self._make_task(), {}, {}) assert result["is_correct"] is None assert "grade fail" in result["grader_response"] @patch(f"{MODULE}.format_query") def test_research_error(self, mock_format): svc = _make_service() mock_format.side_effect = RuntimeError("research crashed") mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task(self._make_task(), {}, {}) assert "research_error" in result assert "research crashed" in result["research_error"] @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_extract_returns_string( self, mock_format, mock_summary, mock_extract, mock_grade ): """When extract_answer_from_response returns a string instead of dict.""" svc = _make_service() mock_format.return_value = "q" mock_summary.return_value = {"summary": "resp", "sources": []} mock_extract.return_value = "plain string answer" mock_grade.return_value = { "is_correct": False, "graded_confidence": "10", "grader_response": "Wrong", } mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task(self._make_task(), {}, {}) assert result["extracted_answer"] == "plain string answer" assert result["confidence"] == "100" @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_sources_from_all_links( self, mock_format, mock_summary, mock_extract, mock_grade ): """When sources is empty but all_links_of_system is present.""" svc = _make_service() mock_format.return_value = "q" mock_summary.return_value = { "summary": "resp", "sources": [], "all_links_of_system": ["http://link1.com"], } mock_extract.return_value = {"extracted_answer": "ans"} mock_grade.return_value = { "is_correct": True, "graded_confidence": "90", "grader_response": "ok", } mock_settings = MagicMock() mock_settings.snapshot = {} with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): result = svc._process_benchmark_task(self._make_task(), {}, {}) sources = json.loads(result["sources"]) assert "http://link1.com" in sources # ============================================================ # _send_progress_update # ============================================================ class TestSendProgressUpdate: def test_sends_progress_via_socket(self): mock_socket = MagicMock() svc = _make_service(socket=mock_socket) svc._send_progress_update(1, 5, 10) mock_socket.emit_to_subscribers.assert_called_once() call_args = mock_socket.emit_to_subscribers.call_args assert call_args[0][0] == "research_progress" assert call_args[0][2]["progress"] == 50.0 def test_sends_zero_progress_when_total_zero(self): mock_socket = MagicMock() svc = _make_service(socket=mock_socket) svc._send_progress_update(1, 0, 0) call_args = mock_socket.emit_to_subscribers.call_args assert call_args[0][2]["progress"] == 0 def test_exception_in_send_is_caught(self): mock_socket = MagicMock() mock_socket.emit_to_subscribers.side_effect = RuntimeError( "socket error" ) svc = _make_service(socket=mock_socket) # Should not raise svc._send_progress_update(1, 5, 10) # ============================================================ # sync_pending_results # ============================================================ class TestSyncPendingResults: def test_returns_zero_when_no_active_run(self): svc = _make_service() assert svc.sync_pending_results(999) == 0 def test_saves_new_results(self): svc = _make_service() svc.active_runs[1] = { "data": {"username": "user1", "user_password": None}, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "task_index": 0, }, ], } mock_session = MagicMock() # No rows persisted for this run yet mock_session.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) count = svc.sync_pending_results(1, "user1") assert count == 1 mock_session.add.assert_called_once() mock_session.commit.assert_called_once() def test_skips_already_saved_indices(self): svc = _make_service() svc.active_runs[1] = { "data": {"username": "user1", "user_password": None}, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "task_index": 0, }, ], "saved_indices": {0}, } mock_session = MagicMock() # No rows persisted for this run yet; idx 0 is skipped via saved_indices. mock_session.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) count = svc.sync_pending_results(1) assert count == 0 def test_skips_existing_db_result(self): svc = _make_service() svc.active_runs[1] = { "data": {"username": "user1", "user_password": None}, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "task_index": 0, }, ], } mock_session = MagicMock() # query_hash "h1" already persisted for this run mock_session.query.return_value.filter.return_value.all.return_value = [ ("h1",) ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) count = svc.sync_pending_results(1) assert count == 0 mock_session.add.assert_not_called() def test_handles_db_error(self): svc = _make_service() svc.active_runs[1] = { "data": {"username": "user1", "user_password": None}, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "task_index": 0, }, ], } mock_session = MagicMock() mock_session.query.side_effect = RuntimeError("db error") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) count = svc.sync_pending_results(1) assert count == 0 def test_uses_username_from_run_data(self): svc = _make_service() svc.active_runs[1] = { "data": {"username": "fromdata", "user_password": "pw"}, "results": [], } with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = MagicMock() mock_session.query.return_value.filter.return_value.all.return_value = [] mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc.sync_pending_results(1) mock_get_session.assert_called_once_with("fromdata", "pw") # ============================================================ # _sync_results_to_database # ============================================================ class TestSyncResultsToDatabase: def test_returns_early_if_no_active_run(self): svc = _make_service() # Should not raise svc._sync_results_to_database(999) def test_returns_early_if_thread_not_complete(self): svc = _make_service() svc.active_runs[1] = {"thread_complete": False} svc._sync_results_to_database(1) # No DB calls expected def test_syncs_completed_run(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() svc.active_runs[1] = { "thread_complete": True, "data": {"username": "user1", "user_password": None}, "completion_info": { "status": BenchmarkStatus.COMPLETED, "end_time": datetime.now(UTC), "completed_examples": 2, "failed_examples": 0, }, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "is_correct": True, "processing_time": 10.0, "task_index": 0, }, { "example_id": "ex2", "query_hash": "h2", "dataset_type": "simpleqa", "question": "Q2?", "correct_answer": "A2", "is_correct": False, "processing_time": 20.0, "task_index": 1, }, ], } mock_session = MagicMock() mock_benchmark_run = MagicMock() mock_benchmark_run.status = BenchmarkStatus.COMPLETED mock_session.query.return_value.filter.return_value.first.return_value = mock_benchmark_run # _persist_unsaved_results reads existing query_hashes for the run. mock_session.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc._sync_results_to_database(1) mock_session.commit.assert_called_once() # Active run should be cleaned up assert 1 not in svc.active_runs def test_calculates_accuracy_correctly(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() svc.active_runs[1] = { "thread_complete": True, "data": {"username": "user1", "user_password": None}, "completion_info": { "status": BenchmarkStatus.COMPLETED, "completed_examples": 2, "failed_examples": 0, }, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "is_correct": True, "processing_time": 10.0, "task_index": 0, }, { "example_id": "ex2", "query_hash": "h2", "dataset_type": "simpleqa", "question": "Q2?", "correct_answer": "A2", "is_correct": True, "processing_time": 20.0, "task_index": 1, }, ], } mock_session = MagicMock() mock_benchmark_run = MagicMock() mock_benchmark_run.status = BenchmarkStatus.COMPLETED mock_session.query.return_value.filter.return_value.first.return_value = mock_benchmark_run # _persist_unsaved_results reads existing query_hashes for the run. mock_session.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc._sync_results_to_database(1) # 2 correct out of 2 = 100% assert mock_benchmark_run.overall_accuracy == 100.0 def test_skips_already_saved_results(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() svc.active_runs[1] = { "thread_complete": True, "data": {"username": "user1", "user_password": None}, "completion_info": { "status": BenchmarkStatus.FAILED, "error_message": "something failed", }, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "task_index": 0, }, ], "saved_indices": {0}, } mock_session = MagicMock() mock_benchmark_run = MagicMock() mock_session.query.return_value.filter.return_value.first.return_value = mock_benchmark_run # _persist_unsaved_results reads existing query_hashes for the run. mock_session.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc._sync_results_to_database(1) # Only the run status update, no result adds assert mock_session.add.call_count == 0 def test_handles_db_exception(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() svc.active_runs[1] = { "thread_complete": True, "data": {"username": "user1", "user_password": None}, "completion_info": { "status": BenchmarkStatus.COMPLETED, }, "results": [], } with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( side_effect=RuntimeError("db down") ) mock_get_session.return_value.__exit__ = Mock(return_value=False) # Should not raise svc._sync_results_to_database(1) def test_no_accuracy_for_non_completed_status(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() svc.active_runs[1] = { "thread_complete": True, "data": {"username": "user1", "user_password": None}, "completion_info": { "status": BenchmarkStatus.FAILED, "error_message": "crashed", }, "results": [ { "example_id": "ex1", "query_hash": "h1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "is_correct": True, "processing_time": 10.0, "task_index": 0, }, ], } mock_session = MagicMock() mock_benchmark_run = MagicMock() mock_benchmark_run.status = BenchmarkStatus.FAILED mock_session.query.return_value.filter.return_value.first.return_value = mock_benchmark_run # _persist_unsaved_results reads existing query_hashes for the run. mock_session.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc._sync_results_to_database(1) # Should NOT set overall_accuracy for failed runs # (the code only calculates accuracy when status == COMPLETED) # The mock starts with whatever default it has; we just verify # the condition was checked by checking the status was set to FAILED assert mock_benchmark_run.status == BenchmarkStatus.FAILED # ============================================================ # _calculate_final_accuracy # ============================================================ class TestCalculateFinalAccuracy: def test_calculates_accuracy(self): svc = _make_service() mock_session = MagicMock() mock_r1 = MagicMock() mock_r1.is_correct = True mock_r1.processing_time = 30.0 mock_r2 = MagicMock() mock_r2.is_correct = False mock_r2.processing_time = 60.0 mock_session.query.return_value.filter.return_value.filter.return_value.all.return_value = [ mock_r1, mock_r2, ] mock_run = MagicMock() mock_session.query.return_value.filter.return_value.first.return_value = mock_run with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc._calculate_final_accuracy(1, "user1") assert mock_run.overall_accuracy == 50.0 def test_no_results(self): svc = _make_service() mock_session = MagicMock() mock_session.query.return_value.filter.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) # Should not raise svc._calculate_final_accuracy(1) def test_handles_exception(self): svc = _make_service() mock_session = MagicMock() mock_session.query.side_effect = RuntimeError("db error") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) # Should not raise svc._calculate_final_accuracy(1) # ============================================================ # update_benchmark_status # ============================================================ class TestUpdateBenchmarkStatus: def test_update_status_in_progress(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() mock_session = MagicMock() mock_run = MagicMock() mock_run.start_time = None mock_run.end_time = None mock_session.query.return_value.filter.return_value.first.return_value = mock_run with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc.update_benchmark_status(1, BenchmarkStatus.IN_PROGRESS) assert mock_run.status == BenchmarkStatus.IN_PROGRESS assert mock_run.start_time is not None def test_update_status_completed(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() mock_session = MagicMock() mock_run = MagicMock() mock_run.start_time = datetime.now(UTC) mock_run.end_time = None mock_session.query.return_value.filter.return_value.first.return_value = mock_run with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc.update_benchmark_status(1, BenchmarkStatus.COMPLETED) assert mock_run.end_time is not None def test_update_status_failed_sets_end_time(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() mock_session = MagicMock() mock_run = MagicMock() mock_run.start_time = datetime.now(UTC) mock_run.end_time = None mock_session.query.return_value.filter.return_value.first.return_value = mock_run with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) svc.update_benchmark_status( 1, BenchmarkStatus.FAILED, error_message="oops" ) assert mock_run.end_time is not None assert mock_run.error_message == "oops" def test_update_status_not_found(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() mock_session = MagicMock() mock_session.query.return_value.filter.return_value.first.return_value = None with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) # Should not raise svc.update_benchmark_status(1, BenchmarkStatus.COMPLETED) def test_update_status_db_error(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() mock_session = MagicMock() mock_session.query.side_effect = RuntimeError("db error") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) # Should not raise svc.update_benchmark_status(1, BenchmarkStatus.COMPLETED) mock_session.rollback.assert_called_once() # ============================================================ # cancel_benchmark # ============================================================ class TestCancelBenchmark: def test_cancel_active_run(self): svc = _make_service() svc.active_runs[1] = {"status": "running"} with patch.object(svc, "update_benchmark_status"): result = svc.cancel_benchmark(1, "user1") assert result is True assert svc.active_runs[1]["status"] == "cancelled" def test_cancel_nonexistent_run(self): svc = _make_service() with patch.object(svc, "update_benchmark_status"): result = svc.cancel_benchmark(999, "user1") assert result is True def test_cancel_exception(self): svc = _make_service() with patch.object( svc, "update_benchmark_status", side_effect=RuntimeError("err") ): result = svc.cancel_benchmark(1) assert result is False # ============================================================ # get_benchmark_status # ============================================================ class TestGetBenchmarkStatus: def test_returns_none_when_not_found(self): svc = _make_service() mock_session = MagicMock() mock_session.query.return_value.filter.return_value.first.return_value = None with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) result = svc.get_benchmark_status(999) assert result is None def test_returns_status_with_running_accuracy(self): from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) svc = _make_service() mock_session = MagicMock() mock_run = MagicMock() mock_run.id = 1 mock_run.run_name = "test run" mock_run.status = BenchmarkStatus.IN_PROGRESS mock_run.completed_examples = 2 mock_run.total_examples = 10 mock_run.failed_examples = 0 mock_run.overall_accuracy = None mock_run.processing_rate = None mock_run.config_hash = "abc123" mock_run.created_at = datetime.now(UTC) mock_run.start_time = datetime.now(UTC) - timedelta(seconds=60) mock_run.end_time = None mock_run.error_message = None mock_r1 = MagicMock() mock_r1.is_correct = True mock_r1.dataset_type = MagicMock() mock_r1.dataset_type.value = "simpleqa" mock_r2 = MagicMock() mock_r2.is_correct = False mock_r2.dataset_type = MagicMock() mock_r2.dataset_type.value = "simpleqa" # Mock the chain of queries # We need to handle multiple query() calls differently call_count = {"n": 0} def side_effect_query(*args): call_count["n"] += 1 m = MagicMock() if call_count["n"] == 1: # benchmark_run query m.filter.return_value.first.return_value = mock_run elif call_count["n"] == 2: # current_results query m.filter.return_value.filter.return_value.all.return_value = [ mock_r1, mock_r2, ] elif call_count["n"] == 3: # all_results_for_timing query m.filter.return_value.all.return_value = [mock_r1, mock_r2] return m mock_session.query.side_effect = side_effect_query with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) result = svc.get_benchmark_status(1, "user1") assert result is not None assert result["running_accuracy"] == 50.0 assert "simpleqa_accuracy" in result def test_handles_exception(self): svc = _make_service() mock_session = MagicMock() mock_session.query.side_effect = RuntimeError("db error") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) result = svc.get_benchmark_status(1) assert result is None # ============================================================ # _run_benchmark_thread # ============================================================ class TestRunBenchmarkThread: @patch(f"{MODULE}._global_research_semaphore") @patch(f"{SETTINGS_CTX_MODULE}.set_settings_context") def test_thread_runs_tasks(self, mock_set_ctx, mock_semaphore): svc = _make_service() svc.active_runs[1] = { "data": { "username": "user1", "user_password": None, "config_hash": "abc", "datasets_config": {"simpleqa": {"count": 1}}, "search_config": {"iterations": 1}, "evaluation_config": {}, "settings_snapshot": {}, }, "results": [], } mock_task_result = {"example_id": "ex1", "is_correct": True} with ( patch.object( svc, "_create_task_queue", return_value=[ { "benchmark_run_id": 1, "example_id": "ex1", "task_index": 0, } ], ), patch.object( svc, "_process_benchmark_task", return_value=mock_task_result ), patch.object(svc, "_send_progress_update"), patch.object(svc, "_sync_results_to_database"), ): svc._run_benchmark_thread(1) assert len(svc.active_runs[1]["results"]) == 1 assert svc.active_runs[1]["thread_complete"] is True @patch(f"{MODULE}._global_research_semaphore") @patch(f"{SETTINGS_CTX_MODULE}.set_settings_context") def test_thread_handles_cancelled_run(self, mock_set_ctx, mock_semaphore): svc = _make_service() svc.active_runs[1] = { "data": { "username": "user1", "user_password": None, "config_hash": "abc", "datasets_config": {}, "search_config": {}, "evaluation_config": {}, "settings_snapshot": {}, }, "status": "cancelled", "results": [], } with ( patch.object( svc, "_create_task_queue", return_value=[ { "benchmark_run_id": 1, "example_id": "ex1", "task_index": 0, } ], ), patch.object(svc, "_sync_results_to_database"), ): svc._run_benchmark_thread(1) info = svc.active_runs[1]["completion_info"] from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) assert info["status"] == BenchmarkStatus.CANCELLED @patch(f"{MODULE}._global_research_semaphore") @patch(f"{SETTINGS_CTX_MODULE}.set_settings_context") def test_thread_handles_task_error_rate_limit( self, mock_set_ctx, mock_semaphore ): svc = _make_service() svc.active_runs[1] = { "data": { "username": "user1", "user_password": None, "config_hash": "abc", "datasets_config": {}, "search_config": {}, "evaluation_config": {}, "settings_snapshot": {}, }, "results": [], } with ( patch.object( svc, "_create_task_queue", return_value=[ { "benchmark_run_id": 1, "example_id": "ex1", "task_index": 0, } ], ), patch.object( svc, "_process_benchmark_task", side_effect=RuntimeError("403 Forbidden rate limit"), ), patch.object(svc, "_sync_results_to_database"), ): svc._run_benchmark_thread(1) assert svc.rate_limit_detected.get(1) is True @patch(f"{MODULE}._global_research_semaphore") @patch(f"{SETTINGS_CTX_MODULE}.set_settings_context") def test_thread_handles_missing_data(self, mock_set_ctx, mock_semaphore): svc = _make_service() svc.active_runs[1] = {} # No "data" key with patch.object(svc, "_sync_results_to_database"): svc._run_benchmark_thread(1) info = svc.active_runs[1]["completion_info"] from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkStatus, ) assert info["status"] == BenchmarkStatus.FAILED @patch(f"{MODULE}._global_research_semaphore") @patch(f"{SETTINGS_CTX_MODULE}.set_settings_context") def test_thread_settings_context_values(self, mock_set_ctx, mock_semaphore): """Test that SettingsContext correctly extracts values from setting objects.""" svc = _make_service() svc.active_runs[1] = { "data": { "username": "user1", "user_password": None, "config_hash": "abc", "datasets_config": {}, "search_config": {}, "evaluation_config": {}, "settings_snapshot": { "key1": {"value": "val1"}, "key2": "direct_val", }, }, "results": [], } captured_ctx = {} def capture_ctx(ctx): captured_ctx["ctx"] = ctx mock_set_ctx.side_effect = capture_ctx with ( patch.object(svc, "_create_task_queue", return_value=[]), patch.object(svc, "_sync_results_to_database"), ): svc._run_benchmark_thread(1) ctx = captured_ctx["ctx"] assert ctx.get_setting("key1") == "val1" assert ctx.get_setting("key2") == "direct_val" assert ctx.get_setting("nonexistent", "default") == "default" @patch(f"{MODULE}._global_research_semaphore") @patch(f"{SETTINGS_CTX_MODULE}.set_settings_context") def test_thread_total_zero_examples_progress( self, mock_set_ctx, mock_semaphore ): """Cover branch where total_examples is 0 for progress calculation.""" svc = _make_service() svc.active_runs[1] = { "data": { "username": "user1", "user_password": None, "config_hash": "abc", "datasets_config": {}, "search_config": {}, "evaluation_config": {}, "settings_snapshot": {}, }, "results": [], } with ( patch.object(svc, "_create_task_queue", return_value=[]), patch.object(svc, "_sync_results_to_database"), ): svc._run_benchmark_thread(1) # Completion should still happen (progress = 0 since total = 0) assert svc.active_runs[1]["thread_complete"] is True # ============================================================ # start_benchmark # ============================================================ class TestStartBenchmark: def test_start_benchmark_run_not_found(self): svc = _make_service() mock_session = MagicMock() mock_session.query.return_value.filter.return_value.first.return_value = None with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_get_session.return_value.__enter__ = Mock( return_value=mock_session ) mock_get_session.return_value.__exit__ = Mock(return_value=False) result = svc.start_benchmark(999, "user1") assert result is False # ============================================================ # Progress callback in _process_benchmark_task # ============================================================ class TestProgressCallback: @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_callback_milestone_types( self, mock_format, mock_summary, mock_extract, mock_grade ): """Test that the progress callback categorizes log types correctly.""" mock_socket = MagicMock() svc = _make_service(socket=mock_socket) def capture_quick_summary(**kwargs): cb = kwargs.get("progress_callback") if cb: # Test various status strings cb("Starting phase", 10, {"phase": "init"}) cb("Completed search", 50, {}) cb("Error occurred", 80, {}) cb("Processing data", 60, {}) return {"summary": "result", "sources": []} mock_format.return_value = "q" mock_summary.side_effect = capture_quick_summary mock_extract.return_value = {"extracted_answer": "ans"} mock_grade.return_value = { "is_correct": True, "graded_confidence": "90", "grader_response": "ok", } mock_settings = MagicMock() mock_settings.snapshot = {} task = { "benchmark_run_id": 1, "example_id": "ex1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "query_hash": "h1", "task_index": 0, "username": "user1", "user_password": None, } with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): svc._process_benchmark_task(task, {}, {}) # Verify socket was called multiple times (from callback + regular calls) assert mock_socket.emit_to_subscribers.call_count >= 4 @patch(f"{MODULE}.grade_single_result") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.format_query") def test_callback_handles_exception( self, mock_format, mock_summary, mock_extract, mock_grade ): """Test that exceptions in the progress callback are caught.""" mock_socket = MagicMock() mock_socket.emit_to_subscribers.side_effect = RuntimeError( "socket dead" ) svc = _make_service(socket=mock_socket) def trigger_callback(**kwargs): cb = kwargs.get("progress_callback") if cb: cb("test status", 50, {}) # Should not raise return {"summary": "result", "sources": []} mock_format.return_value = "q" mock_summary.side_effect = trigger_callback mock_extract.return_value = {"extracted_answer": "ans"} mock_grade.return_value = { "is_correct": True, "graded_confidence": "90", "grader_response": "ok", } mock_settings = MagicMock() mock_settings.snapshot = {} task = { "benchmark_run_id": 1, "example_id": "ex1", "dataset_type": "simpleqa", "question": "Q?", "correct_answer": "A", "query_hash": "h1", "task_index": 0, "username": "user1", "user_password": None, } with patch( f"{SETTINGS_CTX_MODULE}.get_settings_context", return_value=mock_settings, ): # Should not raise despite socket error in callback result = svc._process_benchmark_task(task, {}, {}) assert result is not None