""" Tests for benchmarks/web_api/benchmark_service.py Tests cover: - BenchmarkQueueTracker functionality - BenchmarkService initialization and methods - Config hash generation """ from datetime import datetime, timedelta, UTC from unittest.mock import Mock, patch class TestBenchmarkTaskStatus: """Tests for BenchmarkTaskStatus enum.""" def test_status_values(self): """Test that status enum has expected values.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkTaskStatus, ) assert BenchmarkTaskStatus.QUEUED.value == "queued" assert BenchmarkTaskStatus.PROCESSING.value == "processing" assert BenchmarkTaskStatus.COMPLETED.value == "completed" assert BenchmarkTaskStatus.FAILED.value == "failed" assert BenchmarkTaskStatus.CANCELLED.value == "cancelled" class TestBenchmarkQueueTracker: """Tests for BenchmarkQueueTracker class.""" def test_add_task(self): """Test adding a task to the tracker.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-123", "testuser", "benchmark") task = tracker.get_task_status("task-123") assert task is not None assert task["username"] == "testuser" assert task["task_type"] == "benchmark" assert task["status"] == BenchmarkTaskStatus.QUEUED.value def test_update_task_status(self): """Test updating task status.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-123", "testuser") tracker.update_task_status("task-123", BenchmarkTaskStatus.PROCESSING) task = tracker.get_task_status("task-123") assert task["status"] == BenchmarkTaskStatus.PROCESSING.value def test_update_nonexistent_task(self): """Test updating status of non-existent task logs warning.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() # Should not raise, just log warning tracker.update_task_status("nonexistent", BenchmarkTaskStatus.COMPLETED) task = tracker.get_task_status("nonexistent") assert task is None def test_remove_task(self): """Test removing a task from tracker.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-123", "testuser") tracker.remove_task("task-123") task = tracker.get_task_status("task-123") assert task is None def test_remove_nonexistent_task(self): """Test removing non-existent task doesn't raise.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, ) tracker = BenchmarkQueueTracker() # Should not raise tracker.remove_task("nonexistent") def test_cleanup_completed_tasks(self): """Test cleanup of old completed tasks.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-old", "testuser") tracker.update_task_status("task-old", BenchmarkTaskStatus.COMPLETED) # Manually set updated_at to be old tracker.tasks["task-old"]["updated_at"] = datetime.now(UTC) - timedelta( hours=2 ) tracker.add_task("task-new", "testuser") tracker.update_task_status("task-new", BenchmarkTaskStatus.COMPLETED) # Cleanup with 1 hour max age tracker.cleanup_completed_tasks(max_age_seconds=3600) # Old task should be removed assert tracker.get_task_status("task-old") is None # New task should remain assert tracker.get_task_status("task-new") is not None def test_cleanup_skips_processing_tasks(self): """Test that cleanup doesn't remove processing tasks.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-processing", "testuser") tracker.update_task_status( "task-processing", BenchmarkTaskStatus.PROCESSING ) # Set old timestamp tracker.tasks["task-processing"]["updated_at"] = datetime.now( UTC ) - timedelta(hours=2) tracker.cleanup_completed_tasks(max_age_seconds=3600) # Processing task should remain assert tracker.get_task_status("task-processing") is not None class TestBenchmarkServiceInit: """Tests for BenchmarkService initialization.""" def test_init_default(self): """Test default initialization.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) # Mock socket service with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() assert service.active_runs == {} assert service.socket_service is not None assert service.queue_tracker is not None def test_init_with_custom_socket(self): """Test initialization with custom socket service.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) assert service.socket_service is mock_socket def test_init_socket_service_fallback(self): """Test fallback to mock socket when Flask unavailable.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService", side_effect=Exception("No Flask app"), ): service = BenchmarkService() # Should have fallback mock socket assert service.socket_service is not None # Mock should have emit_to_room method assert hasattr(service.socket_service, "emit_to_room") class TestBenchmarkServiceConfigHash: """Tests for config hash generation.""" def test_generate_config_hash(self): """Test generating config hash.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() config = { "iterations": 2, "questions_per_iteration": 3, "search_strategy": "iterdrag", } hash1 = service.generate_config_hash(config) hash2 = service.generate_config_hash(config) # Same config should produce same hash assert hash1 == hash2 assert len(hash1) == 8 # Short hash def test_different_configs_different_hashes(self): """Test that different configs produce different hashes.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() config1 = {"iterations": 2} config2 = {"iterations": 3} hash1 = service.generate_config_hash(config1) hash2 = service.generate_config_hash(config2) assert hash1 != hash2 class TestBenchmarkServiceActiveRuns: """Tests for active run tracking.""" def test_track_active_run(self): """Test tracking an active benchmark run.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() service.active_runs[1] = { "run_id": 1, "status": "running", "progress": 50, } assert 1 in service.active_runs assert service.active_runs[1]["progress"] == 50 def test_rate_limit_tracking(self): """Test rate limit tracking per run.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() service.rate_limit_detected[1] = True service.rate_limit_detected[2] = False assert service.rate_limit_detected[1] is True assert service.rate_limit_detected[2] is False class TestBenchmarkQueueTrackerThreadSafety: """Tests for thread safety of BenchmarkQueueTracker.""" def test_concurrent_add_and_update(self): """Test concurrent access to tracker.""" import threading from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() errors = [] def add_tasks(): try: for i in range(100): tracker.add_task( f"task-{threading.current_thread().name}-{i}", "user" ) except Exception as e: errors.append(e) def update_tasks(): try: for i in range(100): tracker.update_task_status( f"task-{threading.current_thread().name}-{i}", BenchmarkTaskStatus.COMPLETED, ) except Exception as e: errors.append(e) threads = [] for i in range(4): t = threading.Thread(target=add_tasks, name=f"add-{i}") threads.append(t) t = threading.Thread(target=update_tasks, name=f"update-{i}") threads.append(t) for t in threads: t.start() for t in threads: t.join() # No errors should have occurred assert len(errors) == 0 class TestBenchmarkServiceQueueTrackerIntegration: """Tests for queue tracker integration with service.""" def test_service_uses_queue_tracker(self): """Test that service uses queue tracker.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, BenchmarkTaskStatus, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() # Use the tracker through the service service.queue_tracker.add_task("test-task", "testuser", "benchmark") task = service.queue_tracker.get_task_status("test-task") assert task is not None assert task["status"] == BenchmarkTaskStatus.QUEUED.value class TestBenchmarkServiceQueryHash: """Tests for query hash generation.""" def test_generate_query_hash(self): """Test generating query hash.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() hash1 = service.generate_query_hash("What is AI?", "simpleqa") hash2 = service.generate_query_hash("What is AI?", "simpleqa") # Same question and dataset should produce same hash assert hash1 == hash2 assert len(hash1) == 32 # MD5 hex digest def test_different_questions_different_hashes(self): """Test that different questions produce different hashes.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() hash1 = service.generate_query_hash("Question 1", "simpleqa") hash2 = service.generate_query_hash("Question 2", "simpleqa") assert hash1 != hash2 def test_different_datasets_different_hashes(self): """Test that different datasets produce different hashes.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() hash1 = service.generate_query_hash("Same question", "simpleqa") hash2 = service.generate_query_hash("Same question", "browsecomp") assert hash1 != hash2 def test_query_hash_strips_whitespace(self): """Test that query hash strips whitespace from question.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch( "local_deep_research.benchmarks.web_api.benchmark_service.SocketIOService" ) as mock_socket: mock_socket.return_value = Mock() service = BenchmarkService() hash1 = service.generate_query_hash(" Question ", "simpleqa") hash2 = service.generate_query_hash("Question", "simpleqa") assert hash1 == hash2 class TestBenchmarkServiceProgressUpdate: """Tests for progress update functionality.""" def test_send_progress_update(self): """Test sending progress update.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) service._send_progress_update( benchmark_run_id=1, completed=5, total=10, ) # Should have called emit_to_subscribers 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][1] == 1 def test_send_progress_update_handles_zero_total(self): """Test progress update handles zero total gracefully.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) # Should not raise division by zero service._send_progress_update( benchmark_run_id=1, completed=0, total=0, ) mock_socket.emit_to_subscribers.assert_called_once() class TestBenchmarkQueueTrackerCleanup: """Additional tests for queue tracker cleanup.""" def test_cleanup_failed_tasks(self): """Test cleanup of old failed tasks.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-failed", "testuser") tracker.update_task_status("task-failed", BenchmarkTaskStatus.FAILED) # Set old timestamp tracker.tasks["task-failed"]["updated_at"] = datetime.now( UTC ) - timedelta(hours=2) tracker.cleanup_completed_tasks(max_age_seconds=3600) # Failed task should be removed assert tracker.get_task_status("task-failed") is None def test_cleanup_cancelled_tasks(self): """Test cleanup of old cancelled tasks.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-cancelled", "testuser") tracker.update_task_status( "task-cancelled", BenchmarkTaskStatus.CANCELLED ) # Set old timestamp tracker.tasks["task-cancelled"]["updated_at"] = datetime.now( UTC ) - timedelta(hours=2) tracker.cleanup_completed_tasks(max_age_seconds=3600) # Cancelled task should be removed assert tracker.get_task_status("task-cancelled") is None def test_cleanup_skips_queued_tasks(self): """Test that cleanup doesn't remove queued tasks.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-queued", "testuser") # Keep as QUEUED (default) # Set old timestamp tracker.tasks["task-queued"]["created_at"] = datetime.now( UTC ) - timedelta(hours=2) tracker.cleanup_completed_tasks(max_age_seconds=3600) # Queued task should remain assert tracker.get_task_status("task-queued") is not None def test_cleanup_uses_created_at_if_no_updated_at(self): """Test that cleanup uses created_at if updated_at is missing.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkQueueTracker, BenchmarkTaskStatus, ) tracker = BenchmarkQueueTracker() tracker.add_task("task-old", "testuser") tracker.update_task_status("task-old", BenchmarkTaskStatus.COMPLETED) # Remove updated_at and set old created_at if "updated_at" in tracker.tasks["task-old"]: del tracker.tasks["task-old"]["updated_at"] tracker.tasks["task-old"]["created_at"] = datetime.now(UTC) - timedelta( hours=2 ) tracker.cleanup_completed_tasks(max_age_seconds=3600) # Task should be removed based on created_at assert tracker.get_task_status("task-old") is None class TestBenchmarkServiceCancelBenchmark: """Tests for cancel benchmark functionality.""" def test_cancel_sets_status_cancelled(self): """Test that cancel sets status to cancelled.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) # Add an active run service.active_runs[1] = {"status": "running"} with patch.object(service, "update_benchmark_status"): service.cancel_benchmark(1, username="testuser") assert service.active_runs[1]["status"] == "cancelled" def test_cancel_returns_true_on_success(self): """Test that cancel returns True on success.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) service.active_runs[1] = {"status": "running"} with patch.object(service, "update_benchmark_status"): result = service.cancel_benchmark(1, username="testuser") assert result is True class TestBenchmarkServiceSyncResults: """Tests for sync pending results functionality.""" def test_sync_pending_results_returns_zero_for_unknown_run(self): """Test that sync returns 0 for unknown benchmark run.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) result = service.sync_pending_results(99999, username="testuser") assert result == 0 class TestBenchmarkServiceCreateBenchmarkRun: """Tests for create_benchmark_run functionality.""" def test_create_benchmark_run_success(self): """Test creating a benchmark run in the database.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) # Mock the database session - patch at the source module with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session # Mock the BenchmarkRun model to capture the created object created_run = Mock() created_run.id = 1 def add_side_effect(run): run.id = 1 mock_session.add.side_effect = add_side_effect mock_session.commit = Mock() search_config = {"iterations": 2, "search_strategy": "iterdrag"} evaluation_config = {"model_name": "test-model"} datasets_config = {"simpleqa": {"count": 10}} run_id = service.create_benchmark_run( run_name="Test Run", search_config=search_config, evaluation_config=evaluation_config, datasets_config=datasets_config, username="testuser", ) assert run_id == 1 mock_session.add.assert_called_once() mock_session.commit.assert_called_once() def test_create_benchmark_run_generates_config_hash(self): """Test that create_benchmark_run generates config hash.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session def capture_run(run): run.id = 1 assert run.config_hash is not None # Config hash is first 8 chars of MD5 hexdigest (see generate_config_hash) assert len(run.config_hash) == 8 mock_session.add.side_effect = capture_run service.create_benchmark_run( run_name="Test", search_config={"iterations": 2}, evaluation_config={}, datasets_config={"simpleqa": {"count": 5}}, ) def test_create_benchmark_run_calculates_total_examples(self): """Test that total_examples is calculated correctly.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session def capture_run(run): run.id = 1 assert run.total_examples == 25 # 10 + 15 mock_session.add.side_effect = capture_run service.create_benchmark_run( run_name="Test", search_config={}, evaluation_config={}, datasets_config={ "simpleqa": {"count": 10}, "browsecomp": {"count": 15}, }, ) def test_create_benchmark_run_handles_db_error(self): """Test that create_benchmark_run handles database errors.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) import pytest mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_session.commit.side_effect = Exception("Database error") mock_get_session.return_value = mock_session with pytest.raises(Exception, match="Database error"): service.create_benchmark_run( run_name="Test", search_config={}, evaluation_config={}, datasets_config={"simpleqa": {"count": 5}}, ) mock_session.rollback.assert_called_once() class TestBenchmarkServiceStartBenchmark: """Tests for start_benchmark functionality.""" def test_start_benchmark_creates_thread(self): """Test that start_benchmark creates a background thread.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session # Mock the benchmark run query mock_run = Mock() mock_run.id = 1 mock_run.config_hash = "abc12345" mock_run.datasets_config = {"simpleqa": {"count": 2}} mock_run.search_config = {} mock_run.evaluation_config = {} mock_session.query.return_value.filter.return_value.first.return_value = mock_run # Mock SettingsManager with patch( "local_deep_research.settings.SettingsManager" ) as mock_settings_mgr: mock_settings_mgr.return_value.get_all_settings.return_value = {} # Mock flask session with patch( "flask.session", {"session_id": "test-session"}, ): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_password_store: mock_password_store.get_session_password.return_value = "test-password" # Mock the thread execution with patch.object( service, "_run_benchmark_thread", return_value=None, ): result = service.start_benchmark( 1, username="testuser", user_password="test" ) assert result is True assert 1 in service.active_runs assert service.active_runs[1]["status"] == "running" def test_start_benchmark_stores_data_in_memory(self): """Test that start_benchmark stores benchmark data in memory.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_run = Mock() mock_run.id = 1 mock_run.config_hash = "abc12345" mock_run.datasets_config = {"simpleqa": {"count": 2}} mock_run.search_config = {"iterations": 2} mock_run.evaluation_config = {"model_name": "test"} mock_session.query.return_value.filter.return_value.first.return_value = mock_run with patch( "local_deep_research.settings.SettingsManager" ) as mock_settings_mgr: mock_settings_mgr.return_value.get_all_settings.return_value = { "key": "value" } with patch( "flask.session", {"session_id": "test-session"}, ): with patch( "local_deep_research.database.session_passwords.session_password_store" ): with patch.object( service, "_run_benchmark_thread", return_value=None, ): service.start_benchmark(1, username="testuser") assert "data" in service.active_runs[1] assert ( service.active_runs[1]["data"][ "benchmark_run_id" ] == 1 ) assert ( service.active_runs[1]["data"]["username"] == "testuser" ) def test_start_benchmark_handles_not_found(self): """Test that start_benchmark handles benchmark not found.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session # Return None for the benchmark run mock_session.query.return_value.filter.return_value.first.return_value = None result = service.start_benchmark(999, username="testuser") assert result is False def test_start_benchmark_cleans_up_active_runs_on_spawn_failure(self): """If Thread.start raises after active_runs was populated, the stale entry must be removed so later cancel/status calls don't see a thread-less zombie run.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) from local_deep_research.database.models import BenchmarkStatus mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_run = Mock() mock_run.id = 42 mock_run.config_hash = "hash42" mock_run.datasets_config = {} mock_run.search_config = {} mock_run.evaluation_config = {} mock_session.query.return_value.filter.return_value.first.return_value = mock_run with patch("local_deep_research.settings.SettingsManager") as sm: sm.return_value.get_all_settings.return_value = {} with ( patch("flask.session", {"session_id": "sid"}), patch( "local_deep_research.database.session_passwords.session_password_store" ), patch("threading.Thread") as mock_thread_cls, ): mock_thread = Mock() mock_thread.start.side_effect = RuntimeError( "thread spawn failed" ) mock_thread_cls.return_value = mock_thread result = service.start_benchmark( 42, username="testuser", user_password="pw" ) assert result is False # active_runs must not retain the zombie entry. assert 42 not in service.active_runs # BenchmarkRun.status was flipped to FAILED by the error handler. assert mock_run.status == BenchmarkStatus.FAILED class TestBenchmarkServiceProcessTask: """Tests for _process_benchmark_task functionality.""" def test_process_benchmark_task_success(self): """Test successful processing of a benchmark task.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) task = { "benchmark_run_id": 1, "example_id": "ex1", "dataset_type": "simpleqa", "question": "What is 2+2?", "correct_answer": "4", "query_hash": "hash123", "task_index": 0, } search_config = {"iterations": 1} evaluation_config = {} with patch( "local_deep_research.config.thread_settings.get_settings_context" ) as mock_get_ctx: mock_ctx = Mock() mock_ctx.snapshot = {} mock_get_ctx.return_value = mock_ctx with patch( "local_deep_research.benchmarks.web_api.benchmark_service.format_query" ) as mock_format: mock_format.return_value = "formatted query" with patch( "local_deep_research.benchmarks.web_api.benchmark_service.quick_summary" ) as mock_summary: mock_summary.return_value = { "summary": "The answer is 4.", "sources": [], } with patch( "local_deep_research.benchmarks.web_api.benchmark_service.extract_answer_from_response" ) as mock_extract: mock_extract.return_value = { "extracted_answer": "4", "confidence": "100", } with patch( "local_deep_research.benchmarks.web_api.benchmark_service.grade_single_result" ) as mock_grade: mock_grade.return_value = { "is_correct": True, "graded_confidence": "100", "grader_response": "Correct!", } result = service._process_benchmark_task( task, search_config, evaluation_config ) assert result["response"] == "The answer is 4." assert result["is_correct"] is True assert result["query_hash"] == "hash123" def test_process_benchmark_task_handles_research_error(self): """Test handling of research errors in task processing.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) task = { "benchmark_run_id": 1, "example_id": "ex1", "dataset_type": "simpleqa", "question": "What is 2+2?", "correct_answer": "4", "query_hash": "hash123", "task_index": 0, } with patch( "local_deep_research.config.thread_settings.get_settings_context" ) as mock_get_ctx: mock_ctx = Mock() mock_ctx.snapshot = {} mock_get_ctx.return_value = mock_ctx with patch( "local_deep_research.benchmarks.web_api.benchmark_service.format_query" ) as mock_format: mock_format.return_value = "formatted query" with patch( "local_deep_research.benchmarks.web_api.benchmark_service.quick_summary" ) as mock_summary: mock_summary.side_effect = Exception("Research failed") result = service._process_benchmark_task(task, {}, {}) assert "research_error" in result assert "Research failed" in result["research_error"] def test_process_benchmark_task_handles_evaluation_error(self): """Test handling of evaluation errors in task processing.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) task = { "benchmark_run_id": 1, "example_id": "ex1", "dataset_type": "simpleqa", "question": "What is 2+2?", "correct_answer": "4", "query_hash": "hash123", "task_index": 0, } with patch( "local_deep_research.config.thread_settings.get_settings_context" ) as mock_get_ctx: mock_ctx = Mock() mock_ctx.snapshot = {} mock_get_ctx.return_value = mock_ctx with patch( "local_deep_research.benchmarks.web_api.benchmark_service.format_query" ) as mock_format: mock_format.return_value = "formatted query" with patch( "local_deep_research.benchmarks.web_api.benchmark_service.quick_summary" ) as mock_summary: mock_summary.return_value = { "summary": "Answer", "sources": [], } with patch( "local_deep_research.benchmarks.web_api.benchmark_service.extract_answer_from_response" ) as mock_extract: mock_extract.return_value = { "extracted_answer": "4", "confidence": "100", } with patch( "local_deep_research.benchmarks.web_api.benchmark_service.grade_single_result" ) as mock_grade: mock_grade.side_effect = Exception("Grading failed") result = service._process_benchmark_task( task, {}, {} ) assert result["is_correct"] is None assert "evaluation_error" in result class TestBenchmarkServiceGetBenchmarkStatus: """Tests for get_benchmark_status functionality.""" def test_get_benchmark_status_returns_none_for_unknown(self): """Test that get_benchmark_status returns None for unknown run.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_session.query.return_value.filter.return_value.first.return_value = None result = service.get_benchmark_status(999) assert result is None def test_get_benchmark_status_method_exists(self): """Test that get_benchmark_status method exists and is callable.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) assert hasattr(service, "get_benchmark_status") assert callable(service.get_benchmark_status) def test_get_benchmark_status_returns_dict_or_none(self): """Test that get_benchmark_status returns dict or None.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) import inspect mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) # Verify function signature sig = inspect.signature(service.get_benchmark_status) params = list(sig.parameters.keys()) assert "benchmark_run_id" in params def test_get_benchmark_status_does_not_leak_raw_error(self): """A failed run's raw error_message (which can carry server-level LLM endpoints / filesystem paths) must NOT be returned by get_benchmark_status — it is replaced with a generic message (CWE-209). """ from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, BenchmarkStatus, _GENERIC_BENCHMARK_ERROR, ) secret = "Connection refused to http://internal-llm:8000/v1 /srv/secret" mock_run = Mock() mock_run.id = 1 mock_run.run_name = "run" mock_run.status = BenchmarkStatus.FAILED mock_run.completed_examples = 0 mock_run.total_examples = 10 mock_run.failed_examples = 1 mock_run.overall_accuracy = None mock_run.processing_rate = None mock_run.created_at = None mock_run.start_time = None # skips the timing-calculation block mock_run.end_time = None mock_run.error_message = secret service = BenchmarkService(socket_service=Mock()) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session q = mock_session.query.return_value q.filter.return_value.first.return_value = mock_run q.filter.return_value.all.return_value = [] q.filter.return_value.filter.return_value.all.return_value = [] result = service.get_benchmark_status(1) assert result is not None assert result["error_message"] == _GENERIC_BENCHMARK_ERROR assert secret not in str(result) assert "internal-llm" not in str(result) class TestBenchmarkServiceTaskQueue: """Tests for task queue creation.""" def test_create_task_queue_creates_tasks(self): """Test that _create_task_queue creates tasks correctly.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) datasets_config = {"simpleqa": {"count": 3}} # Mock load_dataset with patch( "local_deep_research.benchmarks.web_api.benchmark_service.load_dataset" ) as mock_load: mock_load.return_value = [ {"id": "1", "problem": "Q1", "answer": "A1"}, {"id": "2", "problem": "Q2", "answer": "A2"}, {"id": "3", "problem": "Q3", "answer": "A3"}, ] tasks = service._create_task_queue( datasets_config=datasets_config, benchmark_run_id=1, ) assert len(tasks) == 3 assert tasks[0]["question"] == "Q1" assert tasks[0]["correct_answer"] == "A1" assert tasks[0]["benchmark_run_id"] == 1 class TestBenchmarkServiceUpdateStatus: """Tests for update_benchmark_status functionality.""" def test_update_benchmark_status_updates_db(self): """Test that update_benchmark_status updates the database.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) from local_deep_research.database.models.benchmark import ( BenchmarkStatus, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_run = Mock() mock_run.status = BenchmarkStatus.PENDING mock_run.start_time = None mock_run.end_time = None mock_session.query.return_value.filter.return_value.first.return_value = mock_run service.update_benchmark_status( 1, BenchmarkStatus.IN_PROGRESS, username="testuser" ) assert mock_run.status == BenchmarkStatus.IN_PROGRESS mock_session.commit.assert_called_once() def test_update_benchmark_status_sets_start_time(self): """Test that start_time is set when transitioning to IN_PROGRESS.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) from local_deep_research.database.models.benchmark import ( BenchmarkStatus, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_run = Mock() mock_run.status = BenchmarkStatus.PENDING mock_run.start_time = None mock_run.end_time = None mock_session.query.return_value.filter.return_value.first.return_value = mock_run service.update_benchmark_status(1, BenchmarkStatus.IN_PROGRESS) assert mock_run.start_time is not None def test_update_benchmark_status_sets_end_time_on_completion(self): """Test that end_time is set when transitioning to COMPLETED.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) from local_deep_research.database.models.benchmark import ( BenchmarkStatus, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_run = Mock() mock_run.status = BenchmarkStatus.IN_PROGRESS 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 service.update_benchmark_status(1, BenchmarkStatus.COMPLETED) assert mock_run.end_time is not None def test_update_benchmark_status_stores_error_message(self): """Test that error message is stored when provided.""" from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) from local_deep_research.database.models.benchmark import ( BenchmarkStatus, ) mock_socket = Mock() service = BenchmarkService(socket_service=mock_socket) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session: mock_session = Mock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_get_session.return_value = mock_session mock_run = Mock() mock_run.status = BenchmarkStatus.IN_PROGRESS mock_run.start_time = datetime.now(UTC) mock_run.end_time = None mock_run.error_message = None mock_session.query.return_value.filter.return_value.first.return_value = mock_run service.update_benchmark_status( 1, BenchmarkStatus.FAILED, error_message="Test error", ) assert mock_run.error_message == "Test error" class TestSessionRollbackExceptionHandling: """Tests for session rollback exception handling (PR #2029). PR #2029 changed bare `except:` to `except Exception:` in the session.rollback() error handler in sync_pending_results. """ def test_rollback_exception_is_caught(self): """Exception during rollback is caught and doesn't propagate.""" mock_session = Mock() mock_session.rollback.side_effect = Exception("Rollback failed") # Simulate the pattern from sync_pending_results caught = False try: mock_session.rollback() except Exception: caught = True assert caught def test_rollback_success_no_exception(self): """Successful rollback raises no exception.""" mock_session = Mock() mock_session.rollback.return_value = None error = None try: mock_session.rollback() except Exception as e: error = e assert error is None