""" Coverage tests for benchmarks/runners.py. Targets uncovered paths: dataset fallback loading, example processing with dataset class methods, legacy browsecomp field handling, quick_summary calls, error handling, evaluation paths (human, automated, fallback), metrics/report generation, progress callbacks at every stage, and convenience wrappers. """ import json import tempfile from pathlib import Path from unittest.mock import Mock, patch MODULE = "local_deep_research.benchmarks.runners" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_example(problem="Q1", answer="A1", example_id="ex_1", **extra): d = {"id": example_id, "problem": problem, "answer": answer} d.update(extra) return d def _quick_summary_response(summary="Response text", sources=None): return {"summary": summary, "sources": sources or []} def _setup_registry_with_fake_class(mock_registry, examples): """Set up mock registry so isinstance check fails (legacy path).""" FakeClass = type("OtherClass", (), {}) mock_dataset = Mock() mock_dataset.load.return_value = examples mock_registry.create_dataset.return_value = mock_dataset mock_registry.get_dataset_class.return_value = FakeClass # --------------------------------------------------------------------------- # format_query (small gap: default parameter) # --------------------------------------------------------------------------- class TestFormatQueryDefaults: def test_default_dataset_type_is_simpleqa(self): from local_deep_research.benchmarks.runners import format_query assert format_query("hello") == "hello" # --------------------------------------------------------------------------- # run_benchmark - dataset loading # --------------------------------------------------------------------------- class TestDatasetLoadingFallback: """Cover lines 97-107: fallback to legacy load_dataset on registry error.""" @patch(f"{MODULE}.load_dataset") @patch(f"{MODULE}.DatasetRegistry") def test_fallback_to_legacy_load_dataset(self, mock_registry, mock_load): from local_deep_research.benchmarks.runners import run_benchmark mock_registry.create_dataset.side_effect = ValueError("no such dataset") mock_load.return_value = [] with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) mock_load.assert_called_once() assert result["status"] == "complete_no_eval" # --------------------------------------------------------------------------- # run_benchmark - example processing with dataset class methods # --------------------------------------------------------------------------- class TestExampleProcessingDatasetClass: """Cover lines 146-155: using dataset_instance.get_question / get_answer.""" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_uses_dataset_class_methods( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark FakeDatasetClass = type( "FakeDataset", (), { "load": lambda self: [ {"id": "1", "problem": "Q", "answer": "A"} ], "get_question": lambda self, ex: "class_question", "get_answer": lambda self, ex: "class_answer", }, ) instance = FakeDatasetClass() mock_registry.create_dataset.return_value = instance mock_registry.get_dataset_class.return_value = FakeDatasetClass mock_qs.return_value = _quick_summary_response() mock_extract.return_value = { "extracted_answer": "ans", "confidence": 0.9, } with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) assert result["total_examples"] == 1 with open(result["results_path"]) as f: line = json.loads(f.readline()) assert line["correct_answer"] == "class_answer" assert line["problem"] == "class_question" # --------------------------------------------------------------------------- # run_benchmark - legacy fallback for simpleqa and browsecomp # --------------------------------------------------------------------------- class TestLegacyFieldExtraction: """Cover lines 156-167: legacy approach for extracting question/answer.""" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_legacy_simpleqa_fields(self, mock_registry, mock_qs, mock_extract): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class( mock_registry, [{"id": "1", "problem": "legacy_q", "answer": "legacy_a"}], ) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "x", "confidence": 0.5} with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) with open(result["results_path"]) as f: line = json.loads(f.readline()) assert line["problem"] == "legacy_q" assert line["correct_answer"] == "legacy_a" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_legacy_browsecomp_correct_answer_field( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class( mock_registry, [{"id": "1", "problem": "bc_q", "correct_answer": "bc_correct"}], ) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "x", "confidence": 0.5} with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="browsecomp", output_dir=tmpdir, run_evaluation=False, ) with open(result["results_path"]) as f: line = json.loads(f.readline()) assert line["correct_answer"] == "bc_correct" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_legacy_browsecomp_fallback_to_answer_field( self, mock_registry, mock_qs, mock_extract ): """Cover line 166-167: fallback to 'answer' when 'correct_answer' missing.""" from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class( mock_registry, [{"id": "1", "problem": "bc_q", "answer": "bc_fallback_answer"}], ) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "x", "confidence": 0.5} with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="browsecomp", output_dir=tmpdir, run_evaluation=False, ) with open(result["results_path"]) as f: line = json.loads(f.readline()) assert line["correct_answer"] == "bc_fallback_answer" # --------------------------------------------------------------------------- # run_benchmark - successful example processing # --------------------------------------------------------------------------- class TestSuccessfulExampleProcessing: """Cover lines 188-245: quick_summary call, extract, write result.""" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_result_contains_expected_fields( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.return_value = _quick_summary_response("my summary", ["src1"]) mock_extract.return_value = { "extracted_answer": "extracted", "confidence": 0.8, } with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, search_config={ "iterations": 2, "questions_per_iteration": 1, "search_tool": "wiki", }, ) with open(result["results_path"]) as f: line = json.loads(f.readline()) assert line["response"] == "my summary" assert line["extracted_answer"] == "extracted" assert line["confidence"] == 0.8 assert line["sources"] == ["src1"] assert "processing_time" in line assert line["search_config"]["search_tool"] == "wiki" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_quick_summary_called_with_search_config( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} cfg = { "iterations": 7, "questions_per_iteration": 5, "search_tool": "google", } with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, search_config=cfg, ) mock_qs.assert_called_once() _, kwargs = mock_qs.call_args assert kwargs["iterations"] == 7 assert kwargs["questions_per_iteration"] == 5 assert kwargs["search_tool"] == "google" # --------------------------------------------------------------------------- # run_benchmark - error handling during processing # --------------------------------------------------------------------------- class TestExampleProcessingError: """Cover lines 247-280: error during quick_summary.""" @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_error_result_written_on_exception(self, mock_registry, mock_qs): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.side_effect = RuntimeError("search failed") with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) with open(result["results_path"]) as f: line = json.loads(f.readline()) assert "error" in line assert "search failed" in line["error"] assert line["id"] == "ex_1" @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_error_with_progress_callback(self, mock_registry, mock_qs): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.side_effect = RuntimeError("boom") callback = Mock() with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, progress_callback=callback, ) statuses = [ c[0][2]["status"] for c in callback.call_args_list if len(c[0]) >= 3 ] assert "error" in statuses assert "started" in statuses @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_error_result_uses_fallback_id(self, mock_registry, mock_qs): """When example has no 'id', fallback to example_{i}.""" from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class( mock_registry, [{"problem": "q", "answer": "a"}] ) mock_qs.side_effect = RuntimeError("fail") with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) with open(result["results_path"]) as f: line = json.loads(f.readline()) assert line["id"] == "example_0" # --------------------------------------------------------------------------- # run_benchmark - progress callbacks throughout # --------------------------------------------------------------------------- class TestProgressCallbacks: """Cover all progress_callback invocations.""" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_all_progress_stages_with_evaluation( self, mock_registry, mock_qs, mock_extract, mock_grade, mock_calc, mock_report, ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} mock_grade.return_value = [] mock_calc.return_value = {"accuracy": 0.5} mock_report.return_value = "/report.md" callback = Mock() with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, progress_callback=callback, ) statuses = [c[0][2]["status"] for c in callback.call_args_list] assert "started" in statuses assert "processing" in statuses assert "completed_example" in statuses assert "evaluating" in statuses assert "calculating_metrics" in statuses assert "generating_report" in statuses assert "complete" in statuses @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_progress_truncates_long_question( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class( mock_registry, [{"id": "1", "problem": "x" * 100, "answer": "a"}] ) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} callback = Mock() with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, progress_callback=callback, ) processing_calls = [ c for c in callback.call_args_list if len(c[0]) >= 3 and c[0][2].get("status") == "processing" ] assert len(processing_calls) == 1 assert processing_calls[0][0][2]["question"].endswith("...") @patch(f"{MODULE}.DatasetRegistry") def test_no_eval_progress_callback(self, mock_registry): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset callback = Mock() with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, progress_callback=callback, ) statuses = [c[0][2]["status"] for c in callback.call_args_list] assert "complete_no_eval" in statuses @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_progress_short_question_no_truncation( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark short_q = "Short question" _setup_registry_with_fake_class( mock_registry, [{"id": "1", "problem": short_q, "answer": "a"}] ) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} callback = Mock() with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, progress_callback=callback, ) processing_calls = [ c for c in callback.call_args_list if len(c[0]) >= 3 and c[0][2].get("status") == "processing" ] assert processing_calls[0][0][2]["question"] == short_q # --------------------------------------------------------------------------- # run_benchmark - evaluation paths # --------------------------------------------------------------------------- class TestEvaluationPaths: """Cover lines 285-425: evaluation with grading, metrics, report.""" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_automated_evaluation_success( self, mock_registry, mock_qs, mock_extract, mock_grade, mock_calc, mock_report, ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.9} mock_grade.return_value = [{"grade": "correct"}] mock_calc.return_value = {"accuracy": 1.0} mock_report.return_value = "/tmp/report.md" with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, ) assert result["status"] == "complete" assert result["accuracy"] == 1.0 assert "report_path" in result assert "evaluation_path" in result mock_grade.assert_called_once() mock_calc.assert_called_once() @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_automated_evaluation_with_eval_config( self, mock_registry, mock_qs, mock_extract, mock_grade, mock_calc, mock_report, ): from local_deep_research.benchmarks.runners import run_benchmark _setup_registry_with_fake_class(mock_registry, [_make_example()]) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} mock_grade.return_value = [] mock_calc.return_value = {"accuracy": 0.0} mock_report.return_value = "/report.md" eval_config = {"model": "test-model"} with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, evaluation_config=eval_config, ) _, kwargs = mock_grade.call_args assert kwargs["evaluation_config"] == eval_config @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.DatasetRegistry") def test_human_evaluation_path(self, mock_registry, mock_calc, mock_report): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_calc.return_value = {"accuracy": 0.5} mock_report.return_value = "/report.md" with ( patch(f"{MODULE}.grade_results") as mock_grade, patch( "local_deep_research.benchmarks.graders.human_evaluation" ) as mock_human, ): mock_human.return_value = [{"grade": "correct"}] with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, human_evaluation=True, ) mock_human.assert_called_once() mock_grade.assert_not_called() assert result["status"] == "complete" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_evaluation_report_config_info( self, mock_registry, mock_grade, mock_calc, mock_report ): """Cover the config_info dict passed to generate_report.""" from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_grade.return_value = [] mock_calc.return_value = {"accuracy": 0.0} mock_report.return_value = "/report.md" with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, human_evaluation=False, dataset_path="/custom/path.json", search_config={ "iterations": 10, "questions_per_iteration": 2, "search_tool": "bing", }, ) _, kwargs = mock_report.call_args assert kwargs["dataset_name"] == "Simpleqa" config = kwargs["config_info"] assert config["Dataset"] == "/custom/path.json" assert config["Iterations"] == 10 assert config["Search tool"] == "bing" assert config["Evaluation method"] == "Automated" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.DatasetRegistry") def test_evaluation_report_human_method_label( self, mock_registry, mock_calc, mock_report ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_calc.return_value = {"accuracy": 0.0} mock_report.return_value = "/report.md" with patch( "local_deep_research.benchmarks.graders.human_evaluation" ) as mock_human: mock_human.return_value = [] with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, human_evaluation=True, ) _, kwargs = mock_report.call_args assert kwargs["config_info"]["Evaluation method"] == "Human" # --------------------------------------------------------------------------- # run_benchmark - automated evaluation failure # --------------------------------------------------------------------------- class TestEvaluationFailure: """Cover lines 320-367: grade_results raises, fallback decision.""" @patch("builtins.input", return_value="n") @patch("builtins.print") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_eval_failure_skip_human_fallback( self, mock_registry, mock_grade, mock_print, mock_input ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_grade.side_effect = RuntimeError("eval error") with patch( "local_deep_research.security.file_write_verifier.write_file_verified" ) as mock_write: with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, ) assert result["status"] == "evaluation_error" assert "eval error" in result["evaluation_error"] mock_write.assert_called_once() @patch("builtins.input", return_value="y") @patch("builtins.print") @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_eval_failure_accept_human_fallback( self, mock_registry, mock_grade, mock_calc, mock_report, mock_print, mock_input, ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_grade.side_effect = RuntimeError("eval error") mock_calc.return_value = {"accuracy": 0.5} mock_report.return_value = "/report.md" with patch( "local_deep_research.benchmarks.graders.human_evaluation" ) as mock_human: mock_human.return_value = [] with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, ) mock_human.assert_called_once() assert result["status"] == "complete" @patch("builtins.input", return_value="n") @patch("builtins.print") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_eval_failure_with_progress_callback( self, mock_registry, mock_grade, mock_print, mock_input ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_grade.side_effect = RuntimeError("eval error") callback = Mock() with patch( "local_deep_research.security.file_write_verifier.write_file_verified" ): with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, progress_callback=callback, ) statuses = [ c[0][2]["status"] for c in callback.call_args_list if len(c[0]) >= 3 ] assert "evaluation_fallback" in statuses # --------------------------------------------------------------------------- # run_benchmark - grade_results progress_callback lambda # --------------------------------------------------------------------------- class TestGradeProgressCallback: """Cover lines 310-318: the lambda passed as progress_callback to grade_results.""" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_grade_results_progress_lambda_invoked( self, mock_registry, mock_grade, mock_calc, mock_report ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset def fake_grade(**kwargs): cb = kwargs.get("progress_callback") if cb: cb(0, 10, {"detail": "grading"}) cb(5, 10, {"detail": "grading"}) return [] mock_grade.side_effect = fake_grade mock_calc.return_value = {"accuracy": 0.0} mock_report.return_value = "/report.md" outer_callback = Mock() with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, progress_callback=outer_callback, ) evaluating_calls = [ c for c in outer_callback.call_args_list if len(c[0]) >= 3 and c[0][2].get("status") == "evaluating" ] assert len(evaluating_calls) >= 2 @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_grade_results_progress_lambda_without_outer_callback( self, mock_registry, mock_grade, mock_calc, mock_report ): """When no outer callback, the lambda should still work (returns None).""" from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset captured_cb = {} def fake_grade(**kwargs): captured_cb["cb"] = kwargs.get("progress_callback") return [] mock_grade.side_effect = fake_grade mock_calc.return_value = {"accuracy": 0.0} mock_report.return_value = "/report.md" with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, progress_callback=None, ) cb = captured_cb["cb"] assert cb is not None result = cb(0, 10, {"x": 1}) assert result is None # --------------------------------------------------------------------------- # run_benchmark - no evaluation path # --------------------------------------------------------------------------- class TestNoEvaluation: """Cover lines 427-441: run_evaluation=False.""" @patch(f"{MODULE}.DatasetRegistry") def test_no_eval_returns_correct_status(self, mock_registry): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) assert result["status"] == "complete_no_eval" assert "evaluation_path" not in result assert "metrics" not in result # --------------------------------------------------------------------------- # run_benchmark - file cleanup of existing output files # --------------------------------------------------------------------------- class TestOutputFileCleanup: """Cover lines 122-125: unlinking existing output files.""" @patch(f"{MODULE}.DatasetRegistry") @patch(f"{MODULE}.time") def test_existing_files_are_removed(self, mock_time, mock_registry): from local_deep_research.benchmarks.runners import run_benchmark mock_time.strftime.return_value = "20260101_000000" mock_time.time.return_value = 0 mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset with tempfile.TemporaryDirectory() as tmpdir: for suffix in ["_results.jsonl", "_evaluation.jsonl", "_report.md"]: p = Path(tmpdir) / f"simpleqa_20260101_000000{suffix}" p.write_text("old content") run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) results_file = ( Path(tmpdir) / "simpleqa_20260101_000000_results.jsonl" ) assert not results_file.exists() or results_file.read_text() == "" # --------------------------------------------------------------------------- # Convenience wrapper functions # --------------------------------------------------------------------------- class TestConvenienceWrappers: """Cover run_simpleqa_benchmark, run_browsecomp_benchmark, run_xbench_deepsearch_benchmark.""" @patch(f"{MODULE}.run_benchmark") def test_run_simpleqa_benchmark(self, mock_rb): from local_deep_research.benchmarks.runners import ( run_simpleqa_benchmark, ) mock_rb.return_value = {"status": "ok"} result = run_simpleqa_benchmark(num_examples=50, output_dir="/tmp/test") mock_rb.assert_called_once_with( dataset_type="simpleqa", num_examples=50, output_dir="/tmp/test" ) assert result == {"status": "ok"} @patch(f"{MODULE}.run_benchmark") def test_run_browsecomp_benchmark(self, mock_rb): from local_deep_research.benchmarks.runners import ( run_browsecomp_benchmark, ) mock_rb.return_value = {"status": "ok"} result = run_browsecomp_benchmark(num_examples=25) mock_rb.assert_called_once_with( dataset_type="browsecomp", num_examples=25 ) assert result == {"status": "ok"} @patch(f"{MODULE}.run_benchmark") def test_run_xbench_deepsearch_benchmark(self, mock_rb): from local_deep_research.benchmarks.runners import ( run_xbench_deepsearch_benchmark, ) mock_rb.return_value = {"status": "ok"} result = run_xbench_deepsearch_benchmark(num_examples=10, seed=99) mock_rb.assert_called_once_with( dataset_type="xbench_deepsearch", num_examples=10, seed=99 ) assert result == {"status": "ok"} # --------------------------------------------------------------------------- # run_benchmark - multiple examples # --------------------------------------------------------------------------- class TestMultipleExamples: """Cover loop iteration with multiple examples.""" @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_multiple_examples_all_written( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark examples = [ _make_example(f"Q{i}", f"A{i}", f"id_{i}") for i in range(3) ] _setup_registry_with_fake_class(mock_registry, examples) mock_qs.return_value = _quick_summary_response() mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) assert result["total_examples"] == 3 with open(result["results_path"]) as f: lines = f.readlines() assert len(lines) == 3 @patch(f"{MODULE}.extract_answer_from_response") @patch(f"{MODULE}.quick_summary") @patch(f"{MODULE}.DatasetRegistry") def test_mix_of_success_and_error( self, mock_registry, mock_qs, mock_extract ): from local_deep_research.benchmarks.runners import run_benchmark examples = [ _make_example(f"Q{i}", f"A{i}", f"id_{i}") for i in range(3) ] _setup_registry_with_fake_class(mock_registry, examples) mock_qs.side_effect = [ _quick_summary_response(), RuntimeError("fail"), _quick_summary_response(), ] mock_extract.return_value = {"extracted_answer": "a", "confidence": 0.5} with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=False, ) with open(result["results_path"]) as f: lines = [json.loads(line) for line in f.readlines()] assert len(lines) == 3 assert "error" not in lines[0] assert "error" in lines[1] assert "error" not in lines[2] # --------------------------------------------------------------------------- # run_benchmark - DEFAULT_DATASET_URLS fallback in report config # --------------------------------------------------------------------------- class TestDefaultDatasetUrlInReport: """Cover line 391: when dataset_path is None, use DEFAULT_DATASET_URLS.""" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_report_uses_default_url_when_no_path( self, mock_registry, mock_grade, mock_calc, mock_report ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_grade.return_value = [] mock_calc.return_value = {"accuracy": 0.0} mock_report.return_value = "/report.md" with tempfile.TemporaryDirectory() as tmpdir: run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, dataset_path=None, ) _, kwargs = mock_report.call_args assert kwargs["config_info"]["Dataset"] is not None # --------------------------------------------------------------------------- # run_benchmark - metrics accuracy key missing # --------------------------------------------------------------------------- class TestMetricsAccuracyFallback: """Cover line 424: metrics.get('accuracy', 0) when key missing.""" @patch(f"{MODULE}.generate_report") @patch(f"{MODULE}.calculate_metrics") @patch(f"{MODULE}.grade_results") @patch(f"{MODULE}.DatasetRegistry") def test_accuracy_defaults_to_zero( self, mock_registry, mock_grade, mock_calc, mock_report ): from local_deep_research.benchmarks.runners import run_benchmark mock_dataset = Mock() mock_dataset.load.return_value = [] mock_registry.create_dataset.return_value = mock_dataset mock_grade.return_value = [] mock_calc.return_value = {} # No accuracy key mock_report.return_value = "/report.md" with tempfile.TemporaryDirectory() as tmpdir: result = run_benchmark( dataset_type="simpleqa", output_dir=tmpdir, run_evaluation=True, ) assert result["accuracy"] == 0