""" Coverage tests for benchmarks/web_api/benchmark_routes.py Targets the uncovered code paths: route handler bodies, branching on provider types, evaluation config sources, dataset validation, error-handling except blocks, search-metric aggregation logic, and export/delete/cancel/validate/search-quality endpoints. """ import enum from contextlib import contextmanager from datetime import datetime, UTC from unittest.mock import MagicMock, patch from flask import Blueprint, Flask # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_app(): """Create a minimal Flask app with the benchmark blueprint registered.""" # Must import *after* patching decorators is no longer needed because # the blueprint is registered at import time. Instead we patch the # auth check at the decorator level so the routes execute normally. from local_deep_research.benchmarks.web_api.benchmark_routes import ( benchmark_bp, ) app = Flask(__name__) app.config["SECRET_KEY"] = "test-secret" app.config["TESTING"] = True app.config["WTF_CSRF_ENABLED"] = False # Stub auth blueprint so url_for("auth.login") resolves auth_bp = Blueprint("auth", __name__, url_prefix="/auth") @auth_bp.route("/login") def login(): return "login" app.register_blueprint(auth_bp) # Only register if not already registered (avoid duplicate blueprint error) if "benchmark" not in app.blueprints: app.register_blueprint(benchmark_bp) return app def _fake_settings(overrides=None): """Return a dict of settings and a mock SettingsManager.""" defaults = { "search.iterations": 8, "search.questions_per_iteration": 5, "search.tool": "searxng", "search.search_strategy": "focused_iteration", "llm.model": "gpt-4", "llm.provider": "openai_endpoint", "llm.temperature": 0.7, "llm.max_tokens": 30000, "llm.context_window_unrestricted": True, "llm.context_window_size": 128000, "llm.local_context_window_size": 4096, "llm.openai_endpoint.url": "http://localhost:8080", "llm.openai_endpoint.api_key": "sk-test", "llm.openai.api_key": "sk-openai", "llm.anthropic.api_key": "sk-anthropic", "benchmark.evaluation.provider": "openai_endpoint", "benchmark.evaluation.model": "anthropic/claude-3.7-sonnet", "benchmark.evaluation.temperature": 0, "benchmark.evaluation.endpoint_url": "https://openrouter.ai/api/v1", } if overrides: defaults.update(overrides) mgr = MagicMock() mgr.get_setting.side_effect = lambda key, default=None: defaults.get( key, default ) return defaults, mgr @contextmanager def _patch_auth_and_db(settings_overrides=None): """Context manager that patches login_required, db_manager, limiter, get_user_db_session, and SettingsManager so route bodies execute.""" _, mgr = _fake_settings(settings_overrides) mock_db_session = MagicMock() with ( patch( "local_deep_research.web.auth.decorators.db_manager" ) as mock_db_mgr, patch( "local_deep_research.benchmarks.web_api.benchmark_routes.get_user_db_session" ) as mock_get_session, patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_get_session2, patch( "local_deep_research.benchmarks.web_api.benchmark_routes.SettingsManager", return_value=mgr, ), patch( "local_deep_research.settings.SettingsManager", return_value=mgr, ), patch( "local_deep_research.benchmarks.web_api.benchmark_routes.benchmark_service" ) as mock_svc, ): mock_db_mgr.is_user_connected.return_value = True @contextmanager def _session_ctx(username, password=None): yield mock_db_session mock_get_session.side_effect = _session_ctx mock_get_session2.side_effect = _session_ctx yield mock_svc, mgr, mock_db_session def _make_routed_query( *, runs=None, avg_processing=None, results=None, search_calls=None, search_calls_exc=None, ): """Side-effect for mock_db.query that routes by model class. After the SearchTracker singleton removal, production code queries SearchCall directly on the same session. Tests must therefore route ``session.query(Model)`` to the right mock chain depending on the model class name. """ def _route(model, *args): chain = MagicMock() chain.filter.return_value = chain chain.order_by.return_value = chain chain.limit.return_value = chain if not isinstance(model, type): # func.avg(...) — not a class chain.scalar.return_value = avg_processing return chain name = getattr(model, "__name__", "") if "BenchmarkRun" in name: chain.all.return_value = runs or [] elif "SearchCall" in name: if search_calls_exc: raise search_calls_exc chain.all.return_value = search_calls or [] else: # BenchmarkResult or anything else chain.all.return_value = results or [] return chain return _route # Fake enum for BenchmarkStatus class _FakeStatus(enum.Enum): IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" class _FakeDatasetType(enum.Enum): SIMPLEQA = "simpleqa" BROWSECOMP = "browsecomp" # --------------------------------------------------------------------------- # index route # --------------------------------------------------------------------------- class TestIndex: def test_index_renders_template(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.benchmarks.web_api.benchmark_routes.render_template_with_defaults", return_value="ok", ) as mock_render: with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/") assert resp.status_code == 200 mock_render.assert_called_once() call_kwargs = mock_render.call_args assert ( "eval_settings" in call_kwargs.kwargs or "eval_settings" in (call_kwargs[1] if len(call_kwargs) > 1 else {}) ) class TestResults: def test_results_page(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.benchmarks.web_api.benchmark_routes.render_template_with_defaults", return_value="results", ): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/results") assert resp.status_code == 200 # --------------------------------------------------------------------------- # start_benchmark route # --------------------------------------------------------------------------- class TestStartBenchmark: def _post_start(self, client, json_data): return client.post( "/benchmark/api/start", json=json_data, content_type="application/json", ) def test_start_no_json_body(self): """No JSON body triggers require_json_body decorator -> 400.""" app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.post("/benchmark/api/start", data="not json") assert resp.status_code == 400 def test_start_empty_datasets(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start(client, {"datasets_config": {}}) assert resp.status_code == 400 def test_start_datasets_all_zero_count(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 0}}}, ) assert resp.status_code == 400 def test_start_success_openai_endpoint_provider(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = "pw123" mock_svc.create_benchmark_run.return_value = 42 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" sess["session_id"] = "sid1" resp = self._post_start( client, { "run_name": "test run", "datasets_config": {"simpleqa": {"count": 5}}, }, ) assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert data["benchmark_run_id"] == 42 def test_start_success_openai_provider(self): app = _make_app() with _patch_auth_and_db({"llm.provider": "openai"}) as ( mock_svc, mgr, _, ): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 10 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 2}}}, ) assert resp.status_code == 200 def test_start_success_anthropic_provider(self): app = _make_app() with _patch_auth_and_db({"llm.provider": "anthropic"}) as ( mock_svc, mgr, _, ): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 11 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_start_with_evaluation_config_in_data(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 99 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, { "datasets_config": {"simpleqa": {"count": 3}}, "evaluation_config": { "provider": "openai", "model_name": "gpt-4", }, }, ) assert resp.status_code == 200 # Verify evaluation_config was passed through call_kwargs = mock_svc.create_benchmark_run.call_args assert ( call_kwargs.kwargs["evaluation_config"]["provider"] == "openai" ) def test_start_eval_provider_openai(self): """Evaluation provider openai branch.""" app = _make_app() with _patch_auth_and_db( { "benchmark.evaluation.provider": "openai", } ) as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 50 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_start_eval_provider_anthropic(self): """Evaluation provider anthropic branch.""" app = _make_app() with _patch_auth_and_db( { "benchmark.evaluation.provider": "anthropic", } ) as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 51 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_start_benchmark_fails(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 77 mock_svc.start_benchmark.return_value = False with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 5}}}, ) assert resp.status_code == 500 assert resp.get_json()["success"] is False def test_start_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.side_effect = RuntimeError("boom") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_start( client, {"datasets_config": {"simpleqa": {"count": 5}}}, ) assert resp.status_code == 500 # --------------------------------------------------------------------------- # get_running_benchmark # --------------------------------------------------------------------------- class TestGetRunningBenchmark: def test_running_found(self): app = _make_app() mock_run = MagicMock() mock_run.id = 1 mock_run.run_name = "Run 1" mock_run.total_examples = 10 mock_run.completed_examples = 3 with _patch_auth_and_db() as (mock_svc, mgr, mock_db): with ( patch( "local_deep_research.benchmarks.web_api.benchmark_routes.BenchmarkRun", create=True, ), patch( "local_deep_research.benchmarks.web_api.benchmark_routes.BenchmarkStatus", create=True, ) as MockBS, ): MockBS.IN_PROGRESS = _FakeStatus.IN_PROGRESS mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value = mock_query mock_query.order_by.return_value = mock_query mock_query.first.return_value = mock_run with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/running") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert data["benchmark_run_id"] == 1 def test_no_running(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): with ( patch( "local_deep_research.benchmarks.web_api.benchmark_routes.BenchmarkRun", create=True, ), patch( "local_deep_research.benchmarks.web_api.benchmark_routes.BenchmarkStatus", create=True, ) as MockBS, ): MockBS.IN_PROGRESS = _FakeStatus.IN_PROGRESS mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value = mock_query mock_query.order_by.return_value = mock_query mock_query.first.return_value = None with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/running") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is False def test_running_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): # Force exception in the route mock_db.query.side_effect = RuntimeError("db error") with ( patch( "local_deep_research.benchmarks.web_api.benchmark_routes.BenchmarkRun", create=True, ), patch( "local_deep_research.benchmarks.web_api.benchmark_routes.BenchmarkStatus", create=True, ), ): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/running") assert resp.status_code == 500 # --------------------------------------------------------------------------- # get_benchmark_status # --------------------------------------------------------------------------- class TestGetBenchmarkStatus: def test_status_found(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): mock_svc.get_benchmark_status.return_value = { "completed_examples": 5, "overall_accuracy": 0.8, "avg_time_per_example": 12.5, "estimated_time_remaining": 60, } with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/status/1") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert data["status"]["completed_examples"] == 5 def test_status_not_found(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): mock_svc.get_benchmark_status.return_value = None with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/status/999") assert resp.status_code == 404 def test_status_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): mock_svc.get_benchmark_status.side_effect = RuntimeError("boom") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/status/1") assert resp.status_code == 500 # --------------------------------------------------------------------------- # cancel_benchmark # --------------------------------------------------------------------------- class TestCancelBenchmark: def test_cancel_success(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): mock_svc.cancel_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.post("/benchmark/api/cancel/1") assert resp.status_code == 200 assert resp.get_json()["success"] is True def test_cancel_failure(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): mock_svc.cancel_benchmark.return_value = False with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.post("/benchmark/api/cancel/1") assert resp.status_code == 500 def test_cancel_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): mock_svc.cancel_benchmark.side_effect = RuntimeError("oops") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.post("/benchmark/api/cancel/1") assert resp.status_code == 500 # --------------------------------------------------------------------------- # get_benchmark_history # --------------------------------------------------------------------------- class TestGetBenchmarkHistory: def _make_run(self, run_id, status_val="completed", run_name=None): run = MagicMock() run.id = run_id run.run_name = run_name run.created_at = datetime(2025, 1, 1, tzinfo=UTC) # Provenance fields added in migration 0014 — None mimics a pre-0014 # row, so the existing history tests cover the back-compat path. run.start_time = None run.ldr_version = None run.total_examples = 10 run.completed_examples = 8 run.overall_accuracy = 0.75 run.status = MagicMock() run.status.value = status_val run.search_config = {"tool": "searxng"} run.evaluation_config = {"provider": "openai"} run.datasets_config = {"simpleqa": {"count": 10}} return run def test_history_empty(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.order_by.return_value = mock_query mock_query.limit.return_value = mock_query mock_query.all.return_value = [] with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") assert resp.status_code == 200 assert resp.get_json()["runs"] == [] def test_history_with_runs_and_avg_processing_time(self): app = _make_app() run = self._make_run(1, run_name=None) mock_result = MagicMock() mock_result.research_id = "res-1" mock_search_call = MagicMock() mock_search_call.research_id = "res-1" mock_search_call.results_count = 20 with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( runs=[run], avg_processing=15.5, results=[mock_result], search_calls=[mock_search_call], ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert len(data["runs"]) == 1 # run_name falls back to "Benchmark #{id}" assert "Benchmark #1" in data["runs"][0]["run_name"] def test_history_avg_processing_time_none(self): """Branch: avg_result is None.""" app = _make_app() run = self._make_run(2, run_name="Named Run") with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( runs=[run], avg_processing=None, ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") assert resp.status_code == 200 data = resp.get_json() assert data["runs"][0]["avg_processing_time"] is None def test_history_search_metrics_exception(self): """Exception in search metrics calculation logged as warning.""" app = _make_app() run = self._make_run(3) with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( runs=[run], avg_processing=10.0, search_calls_exc=RuntimeError("no tracker"), ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") assert resp.status_code == 200 def test_history_avg_time_exception(self): """Exception in avg processing time calculation.""" app = _make_app() run = self._make_run(4) with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.order_by.return_value = mock_query mock_query.limit.return_value = mock_query mock_query.all.return_value = [run] mock_query.filter.side_effect = RuntimeError("avg fail") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") # Should still return 200 with avg_processing_time=None assert resp.status_code == 200 def test_history_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = RuntimeError("db fail") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") assert resp.status_code == 500 def test_history_no_research_ids(self): """Branch where research_ids list is empty.""" app = _make_app() run = self._make_run(5) with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.order_by.return_value = mock_query mock_query.limit.return_value = mock_query mock_query.all.side_effect = [[run], []] # runs, then results mock_query.filter.return_value = mock_query mock_query.scalar.return_value = 5.0 with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/history") assert resp.status_code == 200 # --------------------------------------------------------------------------- # get_benchmark_results # --------------------------------------------------------------------------- class TestGetBenchmarkResults: def _make_result(self, example_id="ex1", research_id="r1", completed=True): r = MagicMock() r.example_id = example_id r.dataset_type = _FakeDatasetType.SIMPLEQA r.question = "What is X?" r.correct_answer = "42" r.extracted_answer = "42" r.response = "The answer is 42" r.is_correct = True r.confidence = 0.95 r.grader_response = "Correct" r.processing_time = 10.5 r.sources = ["source1"] r.research_id = research_id r.completed_at = datetime(2025, 1, 1, tzinfo=UTC) if completed else None return r def test_results_success(self): app = _make_app() result = self._make_result() mock_search_call = MagicMock() mock_search_call.research_id = "r1" mock_search_call.results_count = 15 with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( results=[result], search_calls=[mock_search_call], ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert len(data["results"]) == 1 assert data["results"][0]["search_result_count"] == 15 def test_results_with_limit_param(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query() with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1?limit=5") assert resp.status_code == 200 def test_results_no_completed_at(self): """Result with completed_at = None.""" app = _make_app() result = self._make_result(completed=False) with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( results=[result], ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1") assert resp.status_code == 200 assert resp.get_json()["results"][0]["completed_at"] is None def test_results_no_research_id(self): """Result with research_id = None -> search_result_count = 0.""" app = _make_app() result = self._make_result(research_id=None) with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( results=[result], ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1") assert resp.status_code == 200 assert resp.get_json()["results"][0]["search_result_count"] == 0 def test_results_search_tracker_exception(self): """Exception fetching search metrics does not break the route.""" app = _make_app() result = self._make_result() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_routed_query( results=[result], search_calls_exc=RuntimeError("no tracker"), ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1") assert resp.status_code == 200 def test_results_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_svc.sync_pending_results.side_effect = RuntimeError("fail") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1") assert resp.status_code == 500 # --------------------------------------------------------------------------- # export_benchmark_results # --------------------------------------------------------------------------- def _make_export_query_router(*, run=None, results=None): """Side-effect for mock_db.query that routes by model class for the /export endpoint. The endpoint runs two queries: 1. session.query(BenchmarkRun).options(...).filter(...).one_or_none() 2. session.query(BenchmarkResult).options(...).filter(...).order_by(...).all() """ def _route(model, *args): chain = MagicMock() chain.options.return_value = chain chain.filter.return_value = chain chain.order_by.return_value = chain name = getattr(model, "__name__", "") if "BenchmarkRun" in name: chain.one_or_none.return_value = run else: chain.all.return_value = results or [] return chain return _route def _make_run_mock( *, ldr_version=None, start_time=None, settings_snapshot=None ): run = MagicMock() run.ldr_version = ldr_version run.start_time = start_time run.created_at = datetime(2025, 1, 1, tzinfo=UTC) run.settings_snapshot = settings_snapshot return run class TestExportBenchmarkResults: def test_export_success(self): app = _make_app() mock_result = MagicMock() mock_result.example_id = "ex1" mock_result.dataset_type = _FakeDatasetType.SIMPLEQA mock_result.question = "Q?" mock_result.correct_answer = "A" mock_result.extracted_answer = "A" mock_result.is_correct = True mock_result.confidence = 0.9 mock_result.processing_time = 5.0 mock_result.completed_at = datetime(2025, 1, 1, tzinfo=UTC) with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_export_query_router( run=_make_run_mock(ldr_version="1.6.10"), results=[mock_result], ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1/export") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert len(data["results"]) == 1 assert "full_response" not in data["results"][0] def test_export_no_completed_at(self): app = _make_app() mock_result = MagicMock() mock_result.example_id = "ex2" mock_result.dataset_type = _FakeDatasetType.BROWSECOMP mock_result.question = "Q2?" mock_result.correct_answer = "B" mock_result.extracted_answer = "B" mock_result.is_correct = False mock_result.confidence = 0.5 mock_result.processing_time = 3.0 mock_result.completed_at = None with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = _make_export_query_router( run=_make_run_mock(), results=[mock_result], ) with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1/export") assert resp.status_code == 200 assert resp.get_json()["results"][0]["completed_at"] is None def test_export_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = RuntimeError("fail") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/results/1/export") assert resp.status_code == 500 # --------------------------------------------------------------------------- # get_saved_configs # --------------------------------------------------------------------------- class TestGetSavedConfigs: def test_configs_success(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/configs") assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True assert len(data["configs"]) == 2 assert data["configs"][0]["name"] == "Quick Test" # --------------------------------------------------------------------------- # start_benchmark_simple # --------------------------------------------------------------------------- class TestStartBenchmarkSimple: def _post_simple(self, client, json_data): return client.post( "/benchmark/api/start-simple", json=json_data, content_type="application/json", ) def test_simple_no_json(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.post("/benchmark/api/start-simple", data="bad") assert resp.status_code == 400 def test_simple_empty_datasets(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple(client, {"datasets_config": {}}) assert resp.status_code == 400 def test_simple_success_openai_endpoint(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = "pw" mock_svc.create_benchmark_run.return_value = 100 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" sess["session_id"] = "sid" resp = self._post_simple( client, { "run_name": "simple test", "datasets_config": {"simpleqa": {"count": 3}}, }, ) assert resp.status_code == 200 assert resp.get_json()["success"] is True def test_simple_openai_provider(self): app = _make_app() with _patch_auth_and_db({"llm.provider": "openai"}) as ( mock_svc, mgr, _, ): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 101 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_simple_anthropic_provider(self): app = _make_app() with _patch_auth_and_db({"llm.provider": "anthropic"}) as ( mock_svc, mgr, _, ): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 102 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_simple_eval_openai_provider(self): app = _make_app() with _patch_auth_and_db( { "benchmark.evaluation.provider": "openai", } ) as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 103 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_simple_eval_anthropic_provider(self): app = _make_app() with _patch_auth_and_db( { "benchmark.evaluation.provider": "anthropic", } ) as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 104 mock_svc.start_benchmark.return_value = True with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 200 def test_simple_start_fails(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.return_value = 200 mock_svc.start_benchmark.return_value = False with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 500 def test_simple_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with patch( "local_deep_research.database.session_passwords.session_password_store" ) as mock_pw: mock_pw.get_session_password.return_value = None mock_svc.create_benchmark_run.side_effect = RuntimeError("boom") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_simple( client, {"datasets_config": {"simpleqa": {"count": 1}}}, ) assert resp.status_code == 500 # --------------------------------------------------------------------------- # validate_config # --------------------------------------------------------------------------- class TestValidateConfig: def _post_validate(self, client, json_data): return client.post( "/benchmark/api/validate-config", json=json_data, content_type="application/json", ) def test_validate_valid(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": { "search_tool": "searxng", "search_strategy": "focused_iteration", }, "datasets_config": {"simpleqa": {"count": 10}}, }, ) assert resp.status_code == 200 data = resp.get_json() assert data["valid"] is True assert data["total_examples"] == 10 def test_validate_no_data(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" # Send non-dict JSON (list) resp = client.post( "/benchmark/api/validate-config", data="[]", content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() assert data["valid"] is False def test_validate_missing_search_tool(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": { "search_strategy": "focused_iteration" }, "datasets_config": {"simpleqa": {"count": 10}}, }, ) data = resp.get_json() assert data["valid"] is False assert any("Search tool" in e for e in data["errors"]) def test_validate_missing_search_strategy(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": {"search_tool": "searxng"}, "datasets_config": {"simpleqa": {"count": 10}}, }, ) data = resp.get_json() assert data["valid"] is False def test_validate_no_datasets(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": { "search_tool": "searxng", "search_strategy": "focused_iteration", }, "datasets_config": {}, }, ) data = resp.get_json() assert data["valid"] is False assert any("dataset" in e.lower() for e in data["errors"]) def test_validate_zero_total_examples(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": { "search_tool": "searxng", "search_strategy": "focused_iteration", }, "datasets_config": {"simpleqa": {"count": 0}}, }, ) data = resp.get_json() assert data["valid"] is False def test_validate_large_count_accepted(self): """Large example counts should pass validation (no artificial cap).""" app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": { "search_tool": "searxng", "search_strategy": "focused_iteration", }, "datasets_config": {"simpleqa": {"count": 1001}}, }, ) data = resp.get_json() assert data["valid"] is True assert data["total_examples"] == 1001 def test_validate_datasets_config_with_non_dict_values(self): """datasets_config with valid structure but non-integer count.""" app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, _): with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = self._post_validate( client, { "search_config": { "search_tool": "searxng", "search_strategy": "focused_iteration", }, "datasets_config": {"simpleqa": {}}, }, ) data = resp.get_json() # count defaults to 0, so total_examples = 0 assert data["valid"] is False assert data["total_examples"] == 0 # --------------------------------------------------------------------------- # get_search_quality # --------------------------------------------------------------------------- class TestGetSearchQuality: """Tests for /benchmark/api/search-quality. The route reads RateLimitEstimate rows from the user DB and maps each to {engine_type, total_attempts, success_rate (0-100 scale), status (EXCELLENT/GOOD/CAUTION/WARNING/CRITICAL)}. """ @staticmethod def _make_estimate(engine_type, success_rate, total_attempts=100): est = MagicMock() est.engine_type = engine_type est.total_attempts = total_attempts est.success_rate = success_rate return est @staticmethod def _fetch(estimates): """Run the route with the given estimates and return (resp, json).""" app = _make_app() with _patch_auth_and_db() as (_svc, _mgr, mock_db): mock_db.query.return_value.all.return_value = estimates with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/search-quality") return resp, resp.get_json() def test_requires_authentication(self): app = _make_app() with _patch_auth_and_db(): with app.test_client() as client: resp = client.get("/benchmark/api/search-quality") assert resp.status_code in (401, 302) def test_status_tier_excellent(self): resp, data = self._fetch([self._make_estimate("pubmed", 0.95)]) assert resp.status_code == 200 assert data["success"] is True assert "timestamp" in data row = data["search_quality"][0] assert row["engine_type"] == "pubmed" assert row["status"] == "EXCELLENT" assert row["success_rate"] == 95.0 assert row["total_attempts"] == 100 def test_status_tier_good(self): _resp, data = self._fetch([self._make_estimate("google", 0.90)]) assert data["search_quality"][0]["status"] == "GOOD" assert data["search_quality"][0]["success_rate"] == 90.0 def test_status_tier_caution(self): _resp, data = self._fetch([self._make_estimate("google", 0.75)]) assert data["search_quality"][0]["status"] == "CAUTION" def test_status_tier_warning(self): _resp, data = self._fetch([self._make_estimate("google", 0.50)]) assert data["search_quality"][0]["status"] == "WARNING" def test_status_tier_critical(self): _resp, data = self._fetch([self._make_estimate("google", 0.30)]) assert data["search_quality"][0]["status"] == "CRITICAL" def test_success_rate_unit_conversion(self): # success_rate is stored as 0-1 in the DB; route multiplies by 100. _resp, data = self._fetch([self._make_estimate("google", 0.873)]) assert data["search_quality"][0]["success_rate"] == 87.3 def test_multiple_engines(self): resp, data = self._fetch( [ self._make_estimate("bing", 0.6), self._make_estimate("google", 0.95), ] ) assert resp.status_code == 200 assert len(data["search_quality"]) == 2 assert {r["engine_type"] for r in data["search_quality"]} == { "bing", "google", } def test_empty_engines(self): resp, data = self._fetch([]) assert resp.status_code == 200 assert data["success"] is True assert data["search_quality"] == [] def test_shape_excludes_legacy_fields(self): # Regression guard: the old get_search_quality_stats shape included # recent_avg_results / min_recent_results / max_recent_results / # sample_size. Those are gone (the underlying data lived only in # the per-request in-memory tracker). Asserting their absence # protects the benchmark.html JS from accidentally depending on # them again. _resp, data = self._fetch([self._make_estimate("google", 0.95)]) assert set(data["search_quality"][0].keys()) == { "engine_type", "total_attempts", "success_rate", "status", } def test_search_quality_exception(self): # The route now reads RateLimitEstimate rows from the user DB # (no get_tracker call), so a DB error is what surfaces a 500. app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = RuntimeError("db error") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.get("/benchmark/api/search-quality") assert resp.status_code == 500 # --------------------------------------------------------------------------- # delete_benchmark_run # --------------------------------------------------------------------------- class TestDeleteBenchmarkRun: def test_delete_success(self): app = _make_app() mock_run = MagicMock() mock_run.id = 1 mock_run.status = MagicMock() mock_run.status.value = "completed" with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value = mock_query mock_query.first.return_value = mock_run mock_query.delete.return_value = None with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.delete("/benchmark/api/delete/1") assert resp.status_code == 200 assert resp.get_json()["success"] is True def test_delete_not_found(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value = mock_query mock_query.first.return_value = None with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.delete("/benchmark/api/delete/999") assert resp.status_code == 404 def test_delete_in_progress(self): app = _make_app() mock_run = MagicMock() mock_run.status = MagicMock() mock_run.status.value = "in_progress" with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value = mock_query mock_query.first.return_value = mock_run with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.delete("/benchmark/api/delete/1") assert resp.status_code == 400 assert ( "running" in resp.get_json()["error"].lower() or "cancel" in resp.get_json()["error"].lower() ) def test_delete_exception(self): app = _make_app() with _patch_auth_and_db() as (mock_svc, mgr, mock_db): mock_db.query.side_effect = RuntimeError("db error") with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" resp = client.delete("/benchmark/api/delete/1") assert resp.status_code == 500