""" Comprehensive tests for local_deep_research.web.api – REST API endpoints. Covers all blueprint routes, the api_access_control decorator (auth, API-disabled, rate-limiting), and the _serialize_results helper. """ import sys import time from contextlib import contextmanager from unittest.mock import MagicMock, patch import pytest from flask import Flask from local_deep_research.web.api import api_blueprint from local_deep_research.security.rate_limiter import limiter # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _auth_session(client, username="testuser"): """Inject an authenticated session.""" with client.session_transaction() as sess: sess["username"] = username @contextmanager def _mock_access_control(*, api_enabled=True, rate_limit=60): """Context manager that patches get_user_db_session + get_settings_manager so that the api_access_control decorator lets requests through (or blocks them if api_enabled=False). The rate_limit parameter controls the per-user API rate limit cached on ``g._api_rate_limit`` by the ``api_access_control`` decorator. Setting it here ensures that the real ``_get_user_api_rate_limit`` function reads the correct value from the cache, which keeps Flask-Limiter's enforcement consistent even under xdist test-parallelism where the global ``limiter`` singleton may carry stale state from other test modules. """ with patch("local_deep_research.web.api.get_user_db_session") as mock_ctx: mock_session = MagicMock() mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session) mock_ctx.return_value.__exit__ = MagicMock(return_value=None) with patch( "local_deep_research.web.api.get_settings_manager" ) as mock_sm: mock_manager = MagicMock() mock_manager.get_setting.side_effect = lambda key, default: { "app.enable_api": api_enabled, "app.api_rate_limit": rate_limit, }.get(key, default) # Tracer setting so contract tests can verify the user's # snapshot reaches the underlying research function. # `_ldr_test_tracer` is reserved for tests; no production # code reads this key. mock_manager.get_settings_snapshot.return_value = { "_ldr_test_tracer": "tracer-value" } mock_sm.return_value = mock_manager yield mock_ctx, mock_sm # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _reset_limiter(app): """Reset Flask-Limiter storage before and after each test. Resetting *before* the test prevents stale counters left by other test modules that share the global ``limiter`` singleton (common under xdist). """ with app.app_context(): try: limiter.reset() except Exception: pass yield with app.app_context(): try: limiter.reset() except Exception: pass @pytest.fixture def app(): flask_app = Flask(__name__) flask_app.config["SECRET_KEY"] = "test-secret" flask_app.config["TESTING"] = True flask_app.config["WTF_CSRF_ENABLED"] = False flask_app.config["RATELIMIT_ENABLED"] = True flask_app.config["RATELIMIT_STRATEGY"] = "moving-window" flask_app.register_blueprint(api_blueprint) limiter.init_app(flask_app) return flask_app @pytest.fixture def client(app): return app.test_client() @pytest.fixture def authed_client(client): _auth_session(client) return client # =================================================================== # /api/v1/health # =================================================================== class TestHealthCheck: """GET /api/v1/health – no auth required.""" def test_returns_ok(self, client): resp = client.get("/api/v1/health") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "ok" assert data["message"] == "API is running" def test_timestamp_is_recent(self, client): resp = client.get("/api/v1/health") ts = resp.get_json()["timestamp"] assert abs(ts - time.time()) < 5 def test_unauthenticated_access_allowed(self, client): """Health check must be accessible without a session.""" resp = client.get("/api/v1/health") assert resp.status_code == 200 def test_no_resources_when_unauthenticated(self, client): """Unauthenticated requests do not get resource diagnostics.""" resp = client.get("/api/v1/health") data = resp.get_json() assert data["status"] == "ok" assert "resources" not in data @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_resources_structure(self, _mock_user, client): """Authenticated response contains resources dict with expected keys.""" resp = client.get("/api/v1/health") data = resp.get_json() assert "resources" in data res = data["resources"] expected_keys = { "fd_count", "fd_soft_limit", "fd_hard_limit", "fd_usage_percent", "thread_count", } assert set(res.keys()) == expected_keys @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_thread_count_positive(self, _mock_user, client): """thread_count is always >= 1 (main thread).""" resp = client.get("/api/v1/health") tc = resp.get_json()["resources"]["thread_count"] assert isinstance(tc, int) assert tc >= 1 @pytest.mark.skipif( sys.platform != "linux", reason="/proc/self/fd only on Linux" ) @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_fd_count_on_linux(self, _mock_user, client): """On Linux, fd_count is a non-negative integer.""" resp = client.get("/api/v1/health") fd = resp.get_json()["resources"]["fd_count"] assert isinstance(fd, int) assert fd >= 0 @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_no_resource_module(self, _mock_user, client): """When resource module is unavailable, FD limits are None.""" with patch("local_deep_research.web.api._resource_mod", None): resp = client.get("/api/v1/health") data = resp.get_json() assert data["status"] == "ok" assert data["resources"]["fd_soft_limit"] is None assert data["resources"]["fd_hard_limit"] is None @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_no_proc_fs(self, _mock_user, client): """When /proc/self/fd is unavailable, fd_count and percent are None.""" with patch("os.listdir", side_effect=OSError("no /proc")): resp = client.get("/api/v1/health") res = resp.get_json()["resources"] assert res["fd_count"] is None assert res["fd_usage_percent"] is None @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_warning_status_high_fd(self, _mock_user, client): """Status becomes 'warning' when FD usage exceeds 70%.""" fake_fds = [str(i) for i in range(80)] with ( patch("os.listdir", return_value=fake_fds), patch("local_deep_research.web.api._resource_mod") as mock_res, ): mock_res.RLIM_INFINITY = -1 mock_res.RLIMIT_NOFILE = 7 mock_res.getrlimit.return_value = (100, 100) resp = client.get("/api/v1/health") data = resp.get_json() assert data["status"] == "warning" assert "High FD usage" in data["message"] assert data["resources"]["fd_usage_percent"] == 80.0 @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_ok_status_normal_fd(self, _mock_user, client): """Status stays 'ok' when FD usage is low.""" fake_fds = [str(i) for i in range(10)] with ( patch("os.listdir", return_value=fake_fds), patch("local_deep_research.web.api._resource_mod") as mock_res, ): mock_res.RLIM_INFINITY = -1 mock_res.RLIMIT_NOFILE = 7 mock_res.getrlimit.return_value = (1000, 1000) resp = client.get("/api/v1/health") data = resp.get_json() assert data["status"] == "ok" assert data["resources"]["fd_usage_percent"] == 1.0 @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_getrlimit_oserror_does_not_500(self, _mock_user, client): """A getrlimit OSError must not break the health endpoint.""" with patch("local_deep_research.web.api._resource_mod") as mock_res: mock_res.RLIM_INFINITY = -1 mock_res.RLIMIT_NOFILE = 7 mock_res.getrlimit.side_effect = OSError("getrlimit failed") resp = client.get("/api/v1/health") assert resp.status_code == 200 res = resp.get_json()["resources"] assert res["fd_soft_limit"] is None assert res["fd_hard_limit"] is None assert res["fd_usage_percent"] is None @patch( "local_deep_research.web.api.get_current_username", return_value="testuser", ) def test_rlim_infinity_becomes_none(self, _mock_user, client): """RLIM_INFINITY limits are reported as None.""" with patch("local_deep_research.web.api._resource_mod") as mock_res: mock_res.RLIM_INFINITY = -1 mock_res.RLIMIT_NOFILE = 7 mock_res.getrlimit.return_value = (-1, -1) resp = client.get("/api/v1/health") res = resp.get_json()["resources"] assert res["fd_soft_limit"] is None assert res["fd_hard_limit"] is None assert res["fd_usage_percent"] is None # =================================================================== # /api/v1/ (api_documentation) # =================================================================== class TestApiDocumentation: """GET /api/v1/ – requires auth + api_access_control.""" def test_unauthenticated_returns_401(self, client): resp = client.get("/api/v1/") assert resp.status_code == 401 assert "authentication" in resp.get_json()["error"].lower() def test_returns_api_docs(self, authed_client): with _mock_access_control(): resp = authed_client.get("/api/v1/") assert resp.status_code == 200 data = resp.get_json() assert data["api_version"] == "v1" assert data["description"] == "REST API for Local Deep Research" assert len(data["endpoints"]) == 3 def test_lists_all_endpoint_paths(self, authed_client): with _mock_access_control(): resp = authed_client.get("/api/v1/") paths = [ep["path"] for ep in resp.get_json()["endpoints"]] assert "/api/v1/quick_summary" in paths assert "/api/v1/generate_report" in paths assert "/api/v1/analyze_documents" in paths def test_endpoint_entries_have_method_and_description(self, authed_client): with _mock_access_control(): resp = authed_client.get("/api/v1/") for ep in resp.get_json()["endpoints"]: assert "method" in ep assert "description" in ep assert "parameters" in ep def test_all_endpoints_document_allow_default_settings(self, authed_client): """allow_default_settings is accepted by every research endpoint (fail-closed opt-out) — the docs endpoint must advertise it.""" with _mock_access_control(): resp = authed_client.get("/api/v1/") for ep in resp.get_json()["endpoints"]: assert "allow_default_settings" in ep["parameters"], ( f"{ep['path']} does not document allow_default_settings" ) # =================================================================== # api_access_control decorator # =================================================================== class TestApiAccessControl: """Tests exercising the api_access_control decorator paths.""" def test_no_session_returns_401(self, client): resp = client.get("/api/v1/") assert resp.status_code == 401 def test_api_disabled_returns_403(self, authed_client): with _mock_access_control(api_enabled=False): resp = authed_client.get("/api/v1/") assert resp.status_code == 403 assert "disabled" in resp.get_json()["error"].lower() def test_rate_limit_exceeded_returns_429(self, authed_client): with _mock_access_control(rate_limit=2): # First two should pass resp1 = authed_client.get("/api/v1/") assert resp1.status_code == 200 resp2 = authed_client.get("/api/v1/") assert resp2.status_code == 200 # Third should be rate-limited resp3 = authed_client.get("/api/v1/") assert resp3.status_code == 429 def test_rate_limit_429_returns_json_with_custom_handler(self, app): """429 response has JSON body with 'error' and 'message' keys when the custom handler from app_factory is registered.""" from flask import jsonify # Register the same custom 429 handler as app_factory.py @app.errorhandler(429) def ratelimit_handler(e): return ( jsonify( error="Too many requests", message="Too many attempts. Please try again later.", ), 429, ) with app.test_client() as client: _auth_session(client) with _mock_access_control(rate_limit=1): # First request passes resp1 = client.get("/api/v1/") assert resp1.status_code == 200 # Second hits the limit resp2 = client.get("/api/v1/") assert resp2.status_code == 429 body = resp2.get_json() assert body is not None, "429 response should be JSON" assert body["error"] == "Too many requests" assert "message" in body def test_rate_limit_headers_present_on_success(self, authed_client): """Successful responses include X-RateLimit headers.""" with _mock_access_control(rate_limit=10): resp = authed_client.get("/api/v1/") assert resp.status_code == 200 # Flask-Limiter adds these headers when headers_enabled=True assert "X-RateLimit-Limit" in resp.headers assert "X-RateLimit-Remaining" in resp.headers assert "X-RateLimit-Reset" in resp.headers def test_rate_limit_remaining_header_decrements(self, authed_client): """X-RateLimit-Remaining decrements with each request.""" with _mock_access_control(rate_limit=5): resp1 = authed_client.get("/api/v1/") assert resp1.status_code == 200 remaining1 = int(resp1.headers["X-RateLimit-Remaining"]) resp2 = authed_client.get("/api/v1/") assert resp2.status_code == 200 remaining2 = int(resp2.headers["X-RateLimit-Remaining"]) assert remaining2 == remaining1 - 1 def test_different_users_have_independent_buckets(self, app): """User A hitting their limit does not affect User B.""" with _mock_access_control(rate_limit=2): # User A exhausts their limit client_a = app.test_client() _auth_session(client_a, username="alice") assert client_a.get("/api/v1/").status_code == 200 assert client_a.get("/api/v1/").status_code == 200 assert client_a.get("/api/v1/").status_code == 429 # User B is unaffected client_b = app.test_client() _auth_session(client_b, username="bob") assert client_b.get("/api/v1/").status_code == 200 assert client_b.get("/api/v1/").status_code == 200 def test_rate_limit_zero_means_no_limiting(self, authed_client): """rate_limit=0 (falsy) should exempt from rate limiting.""" with _mock_access_control(rate_limit=0): for _ in range(5): resp = authed_client.get("/api/v1/") assert resp.status_code == 200 def test_g_current_user_fallback(self, app): """When g.current_user is set, it should be used instead of session.""" with app.test_request_context(): from flask import g g.current_user = "guser" with _mock_access_control(): with app.test_client() as c: with c.session_transaction() as sess: sess["username"] = "guser" resp = c.get("/api/v1/") assert resp.status_code == 200 def test_db_session_none_still_allows(self, authed_client): """When get_user_db_session returns None, api_enabled stays True.""" with patch( "local_deep_research.web.api.get_user_db_session" ) as mock_ctx: mock_ctx.return_value.__enter__ = MagicMock(return_value=None) mock_ctx.return_value.__exit__ = MagicMock(return_value=None) resp = authed_client.get("/api/v1/") assert resp.status_code == 200 # =================================================================== # /api/v1/quick_summary # =================================================================== class TestQuickSummary: """POST /api/v1/quick_summary""" def test_unauthenticated_returns_401(self, client): resp = client.post("/api/v1/quick_summary", json={"query": "hi"}) assert resp.status_code == 401 def test_no_json_body_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", content_type="application/json" ) assert resp.status_code == 400 def test_missing_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", json={"search_tool": "searxng"} ) assert resp.status_code == 400 def test_non_string_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", json={"query": 42} ) assert resp.status_code == 400 assert "string" in resp.get_json()["error"].lower() def test_null_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", json={"query": None} ) assert resp.status_code == 400 def test_list_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", json={"query": ["a", "b"]} ) assert resp.status_code == 400 def test_dict_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", json={"query": {"nested": True}} ) assert resp.status_code == 400 def test_successful_quick_summary(self, authed_client): mock_result = {"findings": [], "summary": "done"} with _mock_access_control(): with patch( "local_deep_research.api.research_functions.quick_summary", return_value=mock_result, ): with patch( "local_deep_research.web.api.get_user_db_session" ) as inner_ctx: inner_ctx.return_value.__enter__ = MagicMock( return_value=None ) inner_ctx.return_value.__exit__ = MagicMock( return_value=None ) resp = authed_client.post( "/api/v1/quick_summary", json={"query": "test"} ) assert resp.status_code == 200 def test_error_summary_scrubbed_end_to_end(self, authed_client): # CWE-209 / CodeQL #8019: drive the real route with the real # quick_summary payload shape (error text arrives under `summary`, # NOT `current_knowledge`) and assert the boundary scrub fires on # what the client actually receives. leaked = ( "Error: LLM call failed: " "https://api.example.com/v1?api_key=sk-ENDTOEND1234567890AB" ) mock_result = { "research_id": "abc-123", "summary": leaked, "findings": [{"phase": "Error", "content": leaked}], "iterations": 0, "questions": {}, "formatted_findings": leaked, "sources": [], } with _mock_access_control(): with patch( "local_deep_research.api.research_functions.quick_summary", return_value=mock_result, ): with patch( "local_deep_research.web.api.get_user_db_session" ) as inner_ctx: inner_ctx.return_value.__enter__ = MagicMock( return_value=None ) inner_ctx.return_value.__exit__ = MagicMock( return_value=None ) resp = authed_client.post( "/api/v1/quick_summary", json={"query": "test"} ) assert resp.status_code == 200 data = resp.get_json() for field in ("summary", "formatted_findings"): assert "sk-ENDTOEND1234567890AB" not in data[field] assert data[field].startswith("Error:") content = data["findings"][0]["content"] assert "sk-ENDTOEND1234567890AB" not in content assert content.startswith("Error:") def test_timeout_returns_504(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.api.research_functions.quick_summary", side_effect=TimeoutError("slow"), ): with patch( "local_deep_research.web.api.get_user_db_session" ) as inner_ctx: inner_ctx.return_value.__enter__ = MagicMock( return_value=None ) inner_ctx.return_value.__exit__ = MagicMock( return_value=None ) resp = authed_client.post( "/api/v1/quick_summary", json={"query": "slow query"} ) assert resp.status_code == 504 assert "timed out" in resp.get_json()["error"].lower() def test_generic_error_returns_500(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.api.research_functions.quick_summary", side_effect=ValueError("bad"), ): with patch( "local_deep_research.web.api.get_user_db_session" ) as inner_ctx: inner_ctx.return_value.__enter__ = MagicMock( return_value=None ) inner_ctx.return_value.__exit__ = MagicMock( return_value=None ) resp = authed_client.post( "/api/v1/quick_summary", json={"query": "fail"} ) assert resp.status_code == 500 assert "internal error" in resp.get_json()["error"].lower() def test_optional_params_forwarded(self, authed_client): """Extra params like search_tool, iterations, temperature should be forwarded to the research function.""" with _mock_access_control(): with patch( "local_deep_research.api.research_functions.quick_summary", return_value={"findings": []}, ) as mock_qs: with patch( "local_deep_research.web.api.get_user_db_session" ) as inner_ctx: inner_ctx.return_value.__enter__ = MagicMock( return_value=None ) inner_ctx.return_value.__exit__ = MagicMock( return_value=None ) resp = authed_client.post( "/api/v1/quick_summary", json={ "query": "test", "search_tool": "wikipedia", "iterations": 3, "temperature": 0.5, }, ) assert resp.status_code == 200 call_kwargs = mock_qs.call_args assert call_kwargs[0][0] == "test" # The explicit values should override defaults assert call_kwargs[1]["search_tool"] == "wikipedia" assert call_kwargs[1]["iterations"] == 3 assert call_kwargs[1]["temperature"] == 0.5 def test_settings_snapshot_loaded(self, authed_client): """When the user has a valid db session, settings_snapshot is populated.""" mock_snapshot = { "some.key": 42, "another.key": "raw_value", } mock_sm_instance = MagicMock() mock_sm_instance.get_settings_snapshot.return_value = mock_snapshot # Also handle the api_access_control decorator calls mock_sm_instance.get_setting.side_effect = lambda key, default: { "app.enable_api": True, }.get(key, default) # The decorator uses module-level imports while the function body # re-imports from the original modules. Patch both paths. with ( patch( "local_deep_research.web.api.get_user_db_session" ) as mock_ctx, patch( "local_deep_research.database.session_context.get_user_db_session", mock_ctx, ), patch( "local_deep_research.web.api.get_settings_manager", return_value=mock_sm_instance, ), patch( "local_deep_research.utilities.db_utils.get_settings_manager", return_value=mock_sm_instance, ), patch( "local_deep_research.api.research_functions.quick_summary", # autospec=True catches param renames on existing named # args (e.g. username → user). quick_summary has **kwargs # so unknown kwarg names are NOT rejected — for that # bug class see TestResearchFunctionSignatures. autospec=True, ) as mock_qs, ): mock_qs.return_value = {"findings": []} mock_db = MagicMock() mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_db) mock_ctx.return_value.__exit__ = MagicMock(return_value=None) resp = authed_client.post( "/api/v1/quick_summary", json={"query": "test"}, ) assert resp.status_code == 200 snapshot = mock_qs.call_args.kwargs.get("settings_snapshot") assert snapshot is not None assert snapshot["some.key"] == 42 assert snapshot["another.key"] == "raw_value" # Contract: username + programmatic_mode=False also reach # the research function. See _load_user_context_into_params # in src/local_deep_research/web/api.py. assert mock_qs.call_args.kwargs.get("username") == "testuser" assert mock_qs.call_args.kwargs.get("programmatic_mode") is False def test_settings_load_failure_fails_closed(self, authed_client): """If loading the settings snapshot fails, the request is REFUSED (HTTP 503) rather than silently continuing with an empty snapshot. Continuing with ``{}`` would resolve to the permissive BOTH scope, downgrading a configured PRIVATE_ONLY / require-local user — so the endpoint fails closed and the research function is never called. We fail the endpoint's full-snapshot load (``get_settings_snapshot``) on the shared settings-manager mock, so the auth decorator's ``get_setting`` calls keep working — only the snapshot build inside ``_load_user_context_into_params`` is broken. (Both get_user_db_session and get_settings_manager are bound at module level, so they share one patch surface.) """ with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.api.research_functions.quick_summary", return_value={"findings": []}, ) as mock_qs: resp = authed_client.post( "/api/v1/quick_summary", json={"query": "test"}, ) assert resp.status_code == 503 mock_qs.assert_not_called() # The 503 carries actionable guidance, not just a code. body = resp.get_json() assert "how_to_fix" in body assert "allow_default_settings" in body["how_to_fix"] def test_settings_load_failure_opt_in_continues_with_defaults( self, authed_client ): """With ``allow_default_settings=true`` the caller consciously opts in to run with defaults (empty snapshot) when settings can't load — the request proceeds (200) instead of failing closed, and quick_summary is called with an empty settings_snapshot.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.api.research_functions.quick_summary", return_value={"findings": []}, ) as mock_qs: resp = authed_client.post( "/api/v1/quick_summary", json={"query": "test", "allow_default_settings": True}, ) assert resp.status_code == 200 mock_qs.assert_called_once() assert mock_qs.call_args.kwargs["settings_snapshot"] == {} # The opt-in flag must NOT be forwarded to quick_summary. assert "allow_default_settings" not in mock_qs.call_args.kwargs def test_opt_in_requires_real_true_not_truthy_string(self, authed_client): """Security-boundary flag: a truthy STRING like "false" must NOT opt in — only a real JSON ``true`` does. Otherwise it still fails closed.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.api.research_functions.quick_summary", return_value={"findings": []}, ) as mock_qs: resp = authed_client.post( "/api/v1/quick_summary", # JSON string "false" is truthy but is not boolean true. json={"query": "t", "allow_default_settings": "false"}, ) assert resp.status_code == 503 mock_qs.assert_not_called() def test_opt_in_path_emits_policy_audit_warning(self, authed_client): """The opt-in (run-without-settings) path must log a loud policy_audit warning — the security claim is that it is never silent.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with ( patch( "local_deep_research.api.research_functions.quick_summary", return_value={"findings": []}, ), patch("local_deep_research.web.api.logger") as mock_logger, ): resp = authed_client.post( "/api/v1/quick_summary", json={"query": "t", "allow_default_settings": True}, ) assert resp.status_code == 200 # logger.bind(policy_audit=True) was used … bind_kwargs = [ c.kwargs for c in mock_logger.bind.call_args_list ] assert {"policy_audit": True} in bind_kwargs # … and the bound logger emitted a warning naming the opt-in. bound = mock_logger.bind.return_value assert any( "DEFAULT settings" in str(c) for c in bound.warning.call_args_list ) def test_non_dict_body_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/quick_summary", data="[1]", content_type="application/json", ) assert resp.status_code == 400 # =================================================================== # /api/v1/generate_report # =================================================================== class TestGenerateReport: """POST /api/v1/generate_report""" def test_unauthenticated_returns_401(self, client): resp = client.post("/api/v1/generate_report", json={"query": "hi"}) assert resp.status_code == 401 def test_no_json_body_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/generate_report", content_type="application/json" ) assert resp.status_code == 400 def test_missing_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/generate_report", json={"temperature": 0.5} ) assert resp.status_code == 400 def test_successful_report(self, authed_client): mock_result = {"content": "short report", "title": "Report"} with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", return_value=mock_result, ): resp = authed_client.post( "/api/v1/generate_report", json={"query": "test"} ) assert resp.status_code == 200 data = resp.get_json() assert data["content"] == "short report" assert data["title"] == "Report" def test_large_report_is_truncated(self, authed_client): long_content = "x" * 15000 mock_result = {"content": long_content, "title": "Big Report"} with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", return_value=mock_result, ): resp = authed_client.post( "/api/v1/generate_report", json={"query": "big"} ) assert resp.status_code == 200 data = resp.get_json() assert data["content_truncated"] is True assert len(data["content"]) < 15000 assert data["content"].endswith("... [Content truncated]") def test_content_exactly_10000_not_truncated(self, authed_client): content = "a" * 10000 mock_result = {"content": content} with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", return_value=mock_result, ): resp = authed_client.post( "/api/v1/generate_report", json={"query": "q"} ) data = resp.get_json() assert "content_truncated" not in data assert len(data["content"]) == 10000 def test_timeout_returns_504(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", side_effect=TimeoutError("slow"), ): resp = authed_client.post( "/api/v1/generate_report", json={"query": "slow"} ) assert resp.status_code == 504 def test_generic_error_returns_500(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", side_effect=RuntimeError("boom"), ): resp = authed_client.post( "/api/v1/generate_report", json={"query": "fail"} ) assert resp.status_code == 500 def test_optional_params_forwarded(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", return_value={"content": "ok"}, ) as mock_gr: resp = authed_client.post( "/api/v1/generate_report", json={ "query": "test", "output_file": "/tmp/out.md", "searches_per_section": 5, "model_name": "gpt-4", "temperature": 0.3, }, ) assert resp.status_code == 200 kw = mock_gr.call_args[1] assert kw["output_file"] == "/tmp/out.md" assert kw["model_name"] == "gpt-4" assert kw["temperature"] == 0.3 assert kw["searches_per_section"] == 5 def test_default_params_applied(self, authed_client): """searches_per_section and temperature get defaults if omitted.""" with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", return_value={"content": "ok"}, ) as mock_gr: resp = authed_client.post( "/api/v1/generate_report", json={"query": "test"} ) assert resp.status_code == 200 kw = mock_gr.call_args[1] assert kw["searches_per_section"] == 1 assert kw["temperature"] == 0.7 def test_non_dict_body_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/generate_report", data='"just a string"', content_type="application/json", ) assert resp.status_code == 400 def test_result_none_content_not_truncated(self, authed_client): """If content is not a string (e.g. None), truncation is skipped.""" mock_result = {"content": None, "title": "T"} with _mock_access_control(): with patch( "local_deep_research.api.research_functions.generate_report", return_value=mock_result, ): resp = authed_client.post( "/api/v1/generate_report", json={"query": "q"} ) assert resp.status_code == 200 assert resp.get_json()["content"] is None def test_user_context_loaded(self, authed_client): """Authenticated requests must thread username + settings_snapshot + programmatic_mode=False down to generate_report. Pre-fix this endpoint silently dropped user context, so users' encrypted-DB API keys / model preferences / search tool config never reached the research function.""" mock_snapshot = { "llm.provider": "openai", "search.tool": "tavily", } mock_sm_instance = MagicMock() mock_sm_instance.get_settings_snapshot.return_value = mock_snapshot mock_sm_instance.get_setting.side_effect = lambda key, default: { "app.enable_api": True, }.get(key, default) with ( patch( "local_deep_research.web.api.get_user_db_session" ) as mock_ctx, patch( "local_deep_research.web.api.get_settings_manager", return_value=mock_sm_instance, ), patch( "local_deep_research.api.research_functions.generate_report", # autospec=True catches param renames on existing named # args. generate_report has **kwargs so unknown kwarg # names are NOT rejected — TestResearchFunctionSignatures # covers that bug class via sig.bind_partial. autospec=True, ) as mock_gr, ): mock_gr.return_value = {"content": "ok"} mock_ctx.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_ctx.return_value.__exit__ = MagicMock(return_value=None) resp = authed_client.post( "/api/v1/generate_report", json={"query": "test"} ) assert resp.status_code == 200 kw = mock_gr.call_args.kwargs assert kw.get("username") == "testuser" assert kw.get("programmatic_mode") is False snapshot = kw.get("settings_snapshot") assert snapshot is not None assert snapshot["llm.provider"] == "openai" assert snapshot["search.tool"] == "tavily" def test_settings_load_failure_fails_closed(self, authed_client): """Like /quick_summary, /generate_report refuses (503) when the settings snapshot can't be loaded, instead of silently running with defaults — same egress-policy boundary, same helper.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.api.research_functions.generate_report", return_value={"content": "ok"}, ) as mock_gr: resp = authed_client.post( "/api/v1/generate_report", json={"query": "test"}, ) assert resp.status_code == 503 mock_gr.assert_not_called() assert "how_to_fix" in resp.get_json() def test_settings_load_failure_opt_in_continues_with_defaults( self, authed_client ): """allow_default_settings=true opts in to run with an empty snapshot when settings can't load; the flag itself must not be forwarded to generate_report.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.api.research_functions.generate_report", return_value={"content": "ok"}, ) as mock_gr: resp = authed_client.post( "/api/v1/generate_report", json={"query": "test", "allow_default_settings": True}, ) assert resp.status_code == 200 mock_gr.assert_called_once() assert mock_gr.call_args.kwargs["settings_snapshot"] == {} assert "allow_default_settings" not in mock_gr.call_args.kwargs # =================================================================== # /api/v1/analyze_documents # =================================================================== class TestAnalyzeDocuments: """POST /api/v1/analyze_documents""" def test_unauthenticated_returns_401(self, client): resp = client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "c"}, ) assert resp.status_code == 401 def test_no_json_body_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/analyze_documents", content_type="application/json" ) assert resp.status_code == 400 def test_missing_query_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/analyze_documents", json={"collection_name": "c"} ) assert resp.status_code == 400 def test_missing_collection_name_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q"} ) assert resp.status_code == 400 def test_missing_both_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/analyze_documents", json={"temperature": 0.5} ) assert resp.status_code == 400 assert "both" in resp.get_json()["error"].lower() def test_successful_analyze(self, authed_client): mock_result = {"analysis": "done", "documents": []} with _mock_access_control(): with patch( "local_deep_research.web.api.analyze_documents", return_value=mock_result, ): resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "neural nets", "collection_name": "papers"}, ) assert resp.status_code == 200 assert resp.get_json() == mock_result def test_error_summary_scrubbed_end_to_end(self, authed_client): # CWE-209 / CodeQL #8019: analyze_documents puts "Error: ..." text # in `summary` (its real payload field) — assert the boundary # scrub fires on the actual route response. leaked = ( "Error: collection lookup failed: " "https://db.example.com:5432/db?password=supersecret123" ) with _mock_access_control(): with patch( "local_deep_research.web.api.analyze_documents", return_value={"summary": leaked, "documents": []}, ): resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "papers"}, ) assert resp.status_code == 200 data = resp.get_json() assert "supersecret123" not in data["summary"] assert data["summary"].startswith("Error:") def test_extra_params_forwarded(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.web.api.analyze_documents", return_value={}, ) as mock_ad: resp = authed_client.post( "/api/v1/analyze_documents", json={ "query": "q", "collection_name": "c", "max_results": 10, "temperature": 0.3, "force_reindex": True, }, ) assert resp.status_code == 200 kw = mock_ad.call_args[1] assert kw["max_results"] == 10 assert kw["temperature"] == 0.3 assert kw["force_reindex"] is True def test_unknown_param_returns_400(self, authed_client): """analyze_documents has no **kwargs, so an unknown body key would TypeError at call time and surface as an opaque 500. The endpoint must reject it up front with a 400 naming the parameter.""" with _mock_access_control(): with patch( "local_deep_research.web.api.analyze_documents", return_value={}, ) as mock_ad: resp = authed_client.post( "/api/v1/analyze_documents", json={ "query": "q", "collection_name": "c", "max_result": 5, # typo: should be max_results }, ) assert resp.status_code == 400 mock_ad.assert_not_called() body = resp.get_json() assert "max_result" in body["error"] assert "max_results" in body["allowed_parameters"] @pytest.mark.parametrize("key", ["username", "settings_snapshot"]) def test_server_set_params_rejected_in_body(self, authed_client, key): """username/settings_snapshot are set server-side by _load_user_context_into_params; a body that supplies them is rejected rather than silently overwritten.""" with _mock_access_control(): with patch( "local_deep_research.web.api.analyze_documents", return_value={}, ) as mock_ad: resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "c", key: "x"}, ) assert resp.status_code == 400 mock_ad.assert_not_called() def test_error_returns_500(self, authed_client): with _mock_access_control(): with patch( "local_deep_research.web.api.analyze_documents", side_effect=RuntimeError("boom"), ): resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "c"}, ) assert resp.status_code == 500 assert "internal error" in resp.get_json()["error"].lower() def test_non_dict_body_returns_400(self, authed_client): with _mock_access_control(): resp = authed_client.post( "/api/v1/analyze_documents", data="42", content_type="application/json", ) assert resp.status_code == 400 def test_user_context_loaded(self, authed_client): """Authenticated requests must thread username + settings_snapshot + programmatic_mode=False down to analyze_documents. Pre-fix this endpoint silently dropped user context, so users' encrypted-DB embedding model / collection settings never reached the research function.""" mock_snapshot = { "rag.embedding_model": "BAAI/bge-base", } mock_sm_instance = MagicMock() mock_sm_instance.get_settings_snapshot.return_value = mock_snapshot mock_sm_instance.get_setting.side_effect = lambda key, default: { "app.enable_api": True, }.get(key, default) with ( patch( "local_deep_research.web.api.get_user_db_session" ) as mock_ctx, patch( "local_deep_research.web.api.get_settings_manager", return_value=mock_sm_instance, ), patch( "local_deep_research.web.api.analyze_documents", # autospec=True so the mock mirrors the real function's # signature; if the endpoint passes kwargs the function # doesn't accept, the mock raises TypeError just like # the real call would. Caught the original /analyze_documents # bug where MagicMock(return_value={}) silently swallowed # incompatible kwargs. autospec=True, ) as mock_ad, ): mock_ad.return_value = {} mock_ctx.return_value.__enter__ = MagicMock( return_value=MagicMock() ) mock_ctx.return_value.__exit__ = MagicMock(return_value=None) resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "c"}, ) assert resp.status_code == 200 kw = mock_ad.call_args.kwargs assert kw.get("username") == "testuser" assert kw.get("programmatic_mode") is False snapshot = kw.get("settings_snapshot") assert snapshot is not None assert snapshot["rag.embedding_model"] == "BAAI/bge-base" def test_settings_load_failure_fails_closed(self, authed_client): """Like /quick_summary, /analyze_documents refuses (503) when the settings snapshot can't be loaded, instead of silently running with defaults — same egress-policy boundary, same helper.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.web.api.analyze_documents", return_value={}, ) as mock_ad: resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "c"}, ) assert resp.status_code == 503 mock_ad.assert_not_called() assert "how_to_fix" in resp.get_json() def test_settings_load_failure_opt_in_continues_with_defaults( self, authed_client ): """allow_default_settings=true opts in to run with an empty snapshot when settings can't load; the flag itself must not be forwarded to analyze_documents.""" with _mock_access_control() as (_ctx, mock_sm): mock_sm.return_value.get_settings_snapshot.side_effect = ( RuntimeError("settings fail") ) with patch( "local_deep_research.web.api.analyze_documents", return_value={}, ) as mock_ad: resp = authed_client.post( "/api/v1/analyze_documents", json={ "query": "q", "collection_name": "c", "allow_default_settings": True, }, ) assert resp.status_code == 200 mock_ad.assert_called_once() assert mock_ad.call_args.kwargs["settings_snapshot"] == {} assert "allow_default_settings" not in mock_ad.call_args.kwargs # =================================================================== # End-to-end REST → research-function call path # =================================================================== class TestRestToResearchFunctionCallPath: """Real-path tests that DO NOT mock the research functions themselves. Instead they mock the LLM and search-engine factories one layer down, so the actual REST handler → research function → signature-unpack → function-body chain runs. Catches the bug class where the REST handler passes kwargs the research function does not accept. Mock-based tests at the web.api. boundary use MagicMock which silently swallows any kwargs and never raises TypeError. The end-to-end path test in tests/api_tests/test_rest_api.py is marked @requires_llm and therefore auto-skipped in CI (which runs with LDR_TESTING_WITH_MOCKS=true), so until now no test verified the REST endpoint could actually invoke the research function with the kwargs it passes. """ def test_analyze_documents_full_call_path(self, authed_client): """Fires a real /api/v1/analyze_documents request. Does NOT mock analyze_documents itself — only get_llm and get_search. The real analyze_documents() body executes; if the REST endpoint passes kwargs it does not accept, this surfaces as a 500 (caught by the handler's broad except), not a silent green test.""" mock_llm = MagicMock() mock_response = MagicMock() mock_response.content = "stub summary" mock_llm.invoke.return_value = mock_response mock_search = MagicMock() mock_search.run.return_value = [ {"content": "stub document content", "title": "Stub Doc"} ] with _mock_access_control(): with ( # autospec=True so analyze_documents' calls into # get_llm/get_search are validated against the real # signatures — same kwargs-swallowing blind spot the # endpoint-level tests close, one layer down. patch( "local_deep_research.api.research_functions.get_llm", autospec=True, return_value=mock_llm, ), patch( "local_deep_research.api.research_functions.get_search", autospec=True, return_value=mock_search, ), ): resp = authed_client.post( "/api/v1/analyze_documents", json={"query": "q", "collection_name": "c"}, ) assert resp.status_code == 200, ( f"REST endpoint failed to invoke analyze_documents() " f"with the kwargs it passes. Response: {resp.get_data(as_text=True)}" ) data = resp.get_json() assert "summary" in data # --------------------------------------------------------------- # Stub data shared across the per-output assertions below. The # LLM content includes [1]/[2] citations so the test also covers # the "summary preserves citation markers" passthrough — if a # future refactor adds citation normalization to analyze_documents, # the citation-passthrough test below will flag it for review. # --------------------------------------------------------------- _QUERY = "What is quantum computing?" _COLLECTION = "physics_papers" _STUB_LLM_SUMMARY = "Qubits [1] enable superposition and entanglement [2]." _STUB_DOCUMENTS = [ { "content": "Qubits are quantum bits.", "title": "Qubit Basics", "link": "https://example.com/qubit", }, { "content": "Superposition allows simultaneous states.", "title": "Superposition", "link": "https://example.com/super", }, ] def _post_analyze_documents(self, authed_client): """Fire a real /api/v1/analyze_documents request with the stub LLM + stub search above. Returns (response, mock_llm, mock_search) so individual tests can assert on whichever slice they care about.""" mock_llm = MagicMock() mock_llm_response = MagicMock() mock_llm_response.content = self._STUB_LLM_SUMMARY mock_llm.invoke.return_value = mock_llm_response mock_search = MagicMock() mock_search.run.return_value = self._STUB_DOCUMENTS with _mock_access_control(): with ( # autospec=True validates analyze_documents' calls into # get_llm/get_search against the real signatures. patch( "local_deep_research.api.research_functions.get_llm", autospec=True, return_value=mock_llm, ), patch( "local_deep_research.api.research_functions.get_search", autospec=True, return_value=mock_search, ), ): resp = authed_client.post( "/api/v1/analyze_documents", json={ "query": self._QUERY, "collection_name": self._COLLECTION, }, ) return resp, mock_llm, mock_search # --- Per-output assertions (one test per response field) ------- def test_analyze_documents_response_shape_exact(self, authed_client): """The full response equals an exact predetermined dict — no unexpected keys, no missing keys. Catches new fields silently added or existing fields silently removed.""" resp, _, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 assert resp.get_json() == { "summary": self._STUB_LLM_SUMMARY, "documents": self._STUB_DOCUMENTS, "collection": self._COLLECTION, "document_count": len(self._STUB_DOCUMENTS), } def test_analyze_documents_summary_equals_llm_content(self, authed_client): """The ``summary`` field must equal the LLM's ``.content`` verbatim (modulo the no-op ``remove_think_tags`` transform — our stub content has no tags).""" resp, _, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 assert resp.get_json()["summary"] == self._STUB_LLM_SUMMARY def test_analyze_documents_summary_preserves_citations(self, authed_client): """``[1]``/``[2]`` citation markers in the LLM response must reach the client unchanged. analyze_documents() must not strip or rewrite them. If a future refactor adds citation processing here, this test will flag the behavior change.""" resp, _, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 summary = resp.get_json()["summary"] assert "[1]" in summary, ( f"Citation [1] was stripped from summary: {summary!r}" ) assert "[2]" in summary, ( f"Citation [2] was stripped from summary: {summary!r}" ) def test_analyze_documents_documents_field_passes_through( self, authed_client ): """The ``documents`` field equals the search engine's output verbatim — analyze_documents() does not filter, reorder, or mutate it.""" resp, _, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 assert resp.get_json()["documents"] == self._STUB_DOCUMENTS def test_analyze_documents_collection_echoed(self, authed_client): """The ``collection`` field equals the collection name the client requested.""" resp, _, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 assert resp.get_json()["collection"] == self._COLLECTION def test_analyze_documents_document_count_matches(self, authed_client): """The ``document_count`` field equals ``len(documents)``.""" resp, _, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 assert resp.get_json()["document_count"] == len(self._STUB_DOCUMENTS) # --- Inputs flowed to LLM/search assertions -------------------- def test_analyze_documents_search_called_with_query(self, authed_client): """The search engine receives the user's query verbatim.""" resp, _, mock_search = self._post_analyze_documents(authed_client) assert resp.status_code == 200 mock_search.run.assert_called_once_with(self._QUERY) def test_analyze_documents_llm_prompt_includes_query(self, authed_client): """The LLM's summarisation prompt embeds the user's query.""" resp, mock_llm, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 prompt = mock_llm.invoke.call_args[0][0] assert self._QUERY in prompt def test_analyze_documents_llm_prompt_includes_documents( self, authed_client ): """The LLM's prompt embeds each stub document's content.""" resp, mock_llm, _ = self._post_analyze_documents(authed_client) assert resp.status_code == 200 prompt = mock_llm.invoke.call_args[0][0] for doc in self._STUB_DOCUMENTS: assert doc["content"] in prompt, ( f"Document content missing from LLM prompt: {doc['content']!r}" ) def test_analyze_documents_output_file_branch_passes_snapshot( self, authed_client ): """When ``output_file`` is in the request body, ``analyze_documents`` calls ``write_file_verified`` to enforce the user's ``api.allow_file_output`` setting. Verify the user's ``settings_snapshot`` reaches ``write_file_verified`` so the setting check uses user config, not JSON defaults / env vars. This is the file-write branch of ``analyze_documents`` that the other end-to-end tests skip because they don't pass ``output_file``. Locks in the third of the four coordinated threadings (signature → get_llm → get_search → write_file_verified) — without this, a future revert of the ``settings_snapshot=None`` line at ``research_functions.py`` would silently regress the file-output gate to ignore the user's setting. """ write_call = {} def _capture_write(*args, **kwargs): write_call["args"] = args write_call["kwargs"] = kwargs # Return None — the real write_file_verified returns # nothing on the no-op path, and analyze_documents only # uses the side effect (file write). return mock_llm = MagicMock() mock_llm_response = MagicMock() mock_llm_response.content = self._STUB_LLM_SUMMARY mock_llm.invoke.return_value = mock_llm_response mock_search = MagicMock() mock_search.run.return_value = self._STUB_DOCUMENTS with _mock_access_control(): with ( # autospec=True validates analyze_documents' calls into # get_llm/get_search against the real signatures. patch( "local_deep_research.api.research_functions.get_llm", autospec=True, return_value=mock_llm, ), patch( "local_deep_research.api.research_functions.get_search", autospec=True, return_value=mock_search, ), patch( "local_deep_research.security.file_write_verifier.write_file_verified", side_effect=_capture_write, ), ): resp = authed_client.post( "/api/v1/analyze_documents", json={ "query": self._QUERY, "collection_name": self._COLLECTION, "output_file": "/tmp/ldr_test_output.md", }, ) assert resp.status_code == 200 assert write_call, ( "write_file_verified was not called — analyze_documents may " "have skipped the output_file branch entirely." ) snapshot = write_call["kwargs"].get("settings_snapshot") assert snapshot is not None, ( "write_file_verified got settings_snapshot=None — user's " "api.allow_file_output setting is ignored on the file-write " "branch." ) assert snapshot.get("_ldr_test_tracer") == "tracer-value", ( f"settings_snapshot reaching write_file_verified does not " f"match the user's snapshot. Got: {snapshot!r}" ) # The setting key write_file_verified is asked to enforce. # Pinning it ensures the file-output gate isn't accidentally # rerouted to a different (or no) setting. assert "api.allow_file_output" in write_call["args"], ( f"write_file_verified called without 'api.allow_file_output' " f"setting key. args: {write_call['args']!r}" ) # --------------------------------------------------------------- # /generate_report real-path tests (issue #4396) # --------------------------------------------------------------- # Unlike analyze_documents (a flat get_search -> get_llm -> summarize # function), generate_report runs the whole research engine # (_init_search_system -> AdvancedSearchSystem.analyze_topic -> # IntegratedReportGenerator), so get_llm/get_search alone is the wrong # seam — it would run the full engine. The deepest function specific to # the generate_report wrapper is _init_search_system, and its # get_llm(settings_snapshot=...) call (research_functions.py) is the # EXACT line that 500'd in #4396 when the REST endpoint failed to inject # the user's snapshot (no provider/api_key -> LLM init failure). So mock # _init_search_system + IntegratedReportGenerator and let the real # generate_report() body run: snapshot threading, search-context setup, # report assembly, and the REST handler's truncation/response logic. _GR_REPORT = {"content": "Final report body.", "metadata": {"query": "q"}} def _post_generate_report(self, authed_client): """Fire a real /api/v1/generate_report request WITHOUT mocking generate_report itself — only _init_search_system (one layer down, whose get_llm call is the #4396 failure point) and the report generator. Returns ``(response, init_kwargs)`` where ``init_kwargs`` is the kwargs _init_search_system actually received (or None if it was never called), so individual tests assert on the response or on the user context that reached it.""" captured = {} def _capture_init(*args, **kwargs): captured["kwargs"] = kwargs stub_system = MagicMock() stub_system.analyze_topic.return_value = { "findings": [], "current_knowledge": "", } return stub_system with _mock_access_control(): with ( patch( "local_deep_research.api.research_functions._init_search_system", side_effect=_capture_init, ), patch( "local_deep_research.api.research_functions.IntegratedReportGenerator" ) as mock_rg_cls, # _close_system runs in generate_report's finally; with a # MagicMock system it would call safe_close on auto-created # attributes. No-op it to keep the test focused. patch( "local_deep_research.api.research_functions._close_system" ), ): mock_rg_cls.return_value.generate_report.return_value = ( self._GR_REPORT ) resp = authed_client.post( "/api/v1/generate_report", json={"query": "q"} ) return resp, captured.get("kwargs") def test_generate_report_full_call_path(self, authed_client): """The real generate_report() body runs end-to-end via REST (only _init_search_system + the report generator are mocked) and returns 200 — not the original #4396 500 — with the report content passed back to the client. The per-input user-context assertions live in the next test.""" resp, _ = self._post_generate_report(authed_client) assert resp.status_code == 200, ( f"REST endpoint failed to invoke generate_report() end-to-end. " f"Issue #4396 was a 500 here (no provider/api_key reached LLM " f"init). Response: {resp.get_data(as_text=True)}" ) assert resp.get_json()["content"] == self._GR_REPORT["content"] def test_generate_report_threads_user_context_to_search_system( self, authed_client ): """The #4396 fix, verified at the failure site: the authenticated user's settings_snapshot (and username) must reach _init_search_system, whose get_llm(settings_snapshot=...) call is what raised when the snapshot was missing. A future revert of the user-context injection in api_generate_report fails loudly here — not as a silent green mock test.""" resp, init_kwargs = self._post_generate_report(authed_client) assert resp.status_code == 200, resp.get_data(as_text=True) assert init_kwargs is not None, ( "_init_search_system was never called — generate_report " "returned 200 without initializing the search system." ) snapshot = init_kwargs.get("settings_snapshot") assert snapshot is not None, ( "_init_search_system received no settings_snapshot — this is " "exactly the #4396 regression (no provider/api_key -> LLM init " "500)." ) assert snapshot.get("_ldr_test_tracer") == "tracer-value", ( f"settings_snapshot reaching _init_search_system is not the " f"user's snapshot. Got: {snapshot!r}" ) assert init_kwargs.get("username") == "testuser", ( "username did not reach _init_search_system" ) # =================================================================== # Research-function signature compatibility # =================================================================== class TestResearchFunctionSignatures: """Static checks that the programmatic-API research functions accept the kwargs that authed REST endpoints pass through ``_load_user_context_into_params``. No mocking — operates purely on ``inspect.signature``. Catches the bug class where an endpoint passes ``**params`` to a research function whose signature doesn't accept the keys, which would TypeError at runtime but pass mock-based tests (``MagicMock(return_value=...)`` swallows arbitrary kwargs). """ @pytest.mark.parametrize( "fn_name,extra_required", [ ("quick_summary", {}), ("generate_report", {}), ("analyze_documents", {"collection_name": "c"}), ], ) def test_endpoint_call_binds_to_function_signature( self, fn_name, extra_required ): """Reproduce the exact call shape REST endpoints use: ``fn(query=..., **params)`` where params contains the keys ``_load_user_context_into_params`` writes. ``sig.bind_partial`` raises TypeError if any kwarg is rejected by the signature — catches the bug class for all three functions, including ones with ``**kwargs`` where a plain "is the kwarg accepted" check would short-circuit on ``has_var_keyword=True`` and never actually validate. """ import inspect from local_deep_research.api import research_functions fn = getattr(research_functions, fn_name) sig = inspect.signature(fn) endpoint_call_kwargs = { "query": "q", "username": "u", "settings_snapshot": {}, "programmatic_mode": False, **extra_required, } try: sig.bind_partial(**endpoint_call_kwargs) except TypeError as exc: pytest.fail( f"REST endpoint cannot call {fn_name}{sig}: {exc}. " f"This bug class — endpoint passes kwargs the function " f"rejects — caused the original /analyze_documents " f"runtime TypeError that mock-based tests missed." ) # =================================================================== # Endpoint completeness: every research endpoint threads user context # =================================================================== # Contract registry for TestEndpointUserContextCompleteness. Maps each # POST view function on the api_v1 blueprint to (patch target for its # research function, minimal valid request body, stub return value). # A NEW POST endpoint added to the blueprint fails # test_every_post_endpoint_has_contract until it gets an entry here — # which routes it through test_endpoint_threads_user_context and # therefore through _load_user_context_into_params. See the contract # comment above that helper in src/local_deep_research/web/api.py. # # Escape hatch: an endpoint that genuinely calls NO research function # maps to None, with a comment on its entry explaining why — the # exemption is then visible and reviewable here instead of the endpoint # being silently absent. None entries are skipped by # test_endpoint_threads_user_context but still checked for staleness. _ENDPOINT_CONTRACTS = { "api_quick_summary": ( "local_deep_research.api.research_functions.quick_summary", {"query": "q"}, {"findings": []}, ), "api_generate_report": ( "local_deep_research.api.research_functions.generate_report", {"query": "q"}, {"content": "ok"}, ), "api_analyze_documents": ( # analyze_documents is imported at module level in web.api, so # the patch target is the web.api binding, not research_functions. "local_deep_research.web.api.analyze_documents", {"query": "q", "collection_name": "c"}, {}, ), } class TestEndpointUserContextCompleteness: """Guard against the next variant of the original #3661 bug: a NEW endpoint that calls a research function but never loads the user's encrypted-DB context. The per-endpoint contract tests above can only cover endpoints that exist today. This class iterates the api_v1 blueprint's actual URL map (not the route registry, which could drift from the blueprint), so a POST endpoint added to the blueprint without a contract entry fails loudly here instead of shipping silently uncovered. """ @staticmethod def _post_view_names(app): return sorted( rule.endpoint.removeprefix("api_v1.") for rule in app.url_map.iter_rules() if rule.endpoint.startswith("api_v1.") and "POST" in rule.methods ) def test_every_post_endpoint_has_contract(self, app): missing = [ name for name in self._post_view_names(app) if name not in _ENDPOINT_CONTRACTS ] assert not missing, ( f"POST endpoint(s) on the api_v1 blueprint have no entry in " f"_ENDPOINT_CONTRACTS: {missing}. Every research-calling REST " f"endpoint must invoke _load_user_context_into_params (see the " f"contract comment in web/api.py) and be wired into this " f"contract registry so test_endpoint_threads_user_context " f"verifies the user's context actually reaches the research " f"function. If the new endpoint genuinely calls no research " f"function, map it to None with a comment explaining why." ) def test_no_stale_contract_entries(self, app): registered = set(self._post_view_names(app)) stale = [name for name in _ENDPOINT_CONTRACTS if name not in registered] assert not stale, ( f"_ENDPOINT_CONTRACTS lists endpoint(s) that no longer exist " f"on the api_v1 blueprint: {stale}. Remove their entries." ) @pytest.mark.parametrize( "view_name", sorted( name for name, contract in _ENDPOINT_CONTRACTS.items() if contract is not None ), ) def test_endpoint_threads_user_context(self, authed_client, app, view_name): """Generic contract: an authenticated POST to the endpoint must deliver username, the user's settings snapshot (tracer key from _mock_access_control), and programmatic_mode=False to its research function.""" patch_target, body, stub_return = _ENDPOINT_CONTRACTS[view_name] rule = next( r for r in app.url_map.iter_rules() if r.endpoint == f"api_v1.{view_name}" ) with _mock_access_control(): with patch(patch_target, return_value=stub_return) as mock_fn: resp = authed_client.post(rule.rule, json=body) assert resp.status_code == 200, ( f"{view_name} returned {resp.status_code}: " f"{resp.get_data(as_text=True)}" ) kwargs = mock_fn.call_args.kwargs assert kwargs.get("username") == "testuser", ( f"{view_name} did not pass the authenticated username " f"to its research function" ) assert kwargs.get("programmatic_mode") is False, ( f"{view_name} did not pass programmatic_mode=False" ) snapshot = kwargs.get("settings_snapshot") assert snapshot is not None, ( f"{view_name} did not pass settings_snapshot" ) assert snapshot.get("_ldr_test_tracer") == "tracer-value", ( f"{view_name} passed a settings_snapshot that is not " f"the user's snapshot. Got: {snapshot!r}" ) # =================================================================== # _serialize_results helper # =================================================================== class TestSerializeResults: """Unit tests for the _serialize_results helper.""" def test_converts_documents(self, app): from local_deep_research.web.api import _serialize_results mock_doc = MagicMock() mock_doc.metadata = {"source": "wiki"} mock_doc.page_content = "some text" results = { "findings": [{"documents": [mock_doc], "query": "test"}], "summary": "ok", } with app.app_context(): resp = _serialize_results(results) data = resp.get_json() assert data["summary"] == "ok" doc = data["findings"][0]["documents"][0] assert doc["metadata"] == {"source": "wiki"} assert doc["content"] == "some text" def test_no_findings_key(self, app): from local_deep_research.web.api import _serialize_results results = {"summary": "nothing"} with app.app_context(): resp = _serialize_results(results) assert resp.get_json() == {"summary": "nothing"} def test_empty_findings(self, app): from local_deep_research.web.api import _serialize_results results = {"findings": []} with app.app_context(): resp = _serialize_results(results) assert resp.get_json()["findings"] == [] def test_finding_without_documents(self, app): from local_deep_research.web.api import _serialize_results results = {"findings": [{"query": "test"}]} with app.app_context(): resp = _serialize_results(results) assert resp.get_json()["findings"][0]["query"] == "test" def test_does_not_mutate_original(self, app): from local_deep_research.web.api import _serialize_results mock_doc = MagicMock() mock_doc.metadata = {"a": 1} mock_doc.page_content = "text" original_findings = [{"documents": [mock_doc]}] results = {"findings": original_findings} with app.app_context(): _serialize_results(results) # Original list should still contain the mock object assert original_findings[0]["documents"][0] is not None # -- CWE-209 / CodeQL #8019: scrub exception-derived fields at the # HTTP boundary so credentials and stack-trace text cannot leak via # the API path (which bypasses the web-UI ErrorReportGenerator). def test_error_prefixed_current_knowledge_is_scrubbed(self, app): from local_deep_research.web.api import _serialize_results # API key glued into an exception message (realistic: many HTTP # library errors echo the request URL). leaked = ( "Error: LLM call failed: " "https://api.example.com/v1?api_key=sk-SECRETKEY1234567890AB" ) results = {"current_knowledge": leaked} with app.app_context(): data = _serialize_results(results).get_json() assert "sk-SECRETKEY1234567890AB" not in data["current_knowledge"] assert data["current_knowledge"].startswith("Error:") def test_error_prefixed_summary_is_scrubbed_real_payload_shape(self, app): from local_deep_research.web.api import _serialize_results # quick_summary() returns the strategy's current_knowledge under # the key `summary` (research_functions.py) — the field an API # client actually receives. Regression guard for the field-name # mismatch where only `current_knowledge` was scrubbed and the # real payload leaked through untouched. leaked = ( "Error: LLM call failed: " "https://api.example.com/v1?api_key=sk-REALSHAPE12345678901" ) results = { "research_id": "abc-123", "summary": leaked, "findings": [], "iterations": 1, "questions": {}, "formatted_findings": "", "sources": [], } with app.app_context(): data = _serialize_results(results).get_json() assert "sk-REALSHAPE12345678901" not in data["summary"] assert data["summary"].startswith("Error:") def test_bare_api_key_in_prose_is_scrubbed(self, app): from local_deep_research.web.api import _serialize_results # Not URL-embedded: exercises the sk-* prefix regex directly # (>= 20 chars after the prefix), not the ?api_key= query-param # pattern the other fixtures rely on. leaked = ( "Error: OpenAI authentication failed for key " "sk-proj-AbCdEfGhIjKlMnOpQrStUvWx" ) results = {"summary": leaked} with app.app_context(): data = _serialize_results(results).get_json() assert "sk-proj-AbCdEfGhIjKlMnOpQrStUvWx" not in data["summary"] assert data["summary"].startswith("Error:") def test_error_prefixed_formatted_findings_is_scrubbed(self, app): from local_deep_research.web.api import _serialize_results leaked = ( "Error: request to " "https://api.example.com/v1?api_key=sk-LEAK9876543210abcdefgh" ) results = {"formatted_findings": leaked} with app.app_context(): data = _serialize_results(results).get_json() assert "sk-LEAK9876543210abcdefgh" not in data["formatted_findings"] assert data["formatted_findings"].startswith("Error:") def test_error_prefixed_finding_content_is_scrubbed(self, app): from local_deep_research.web.api import _serialize_results leaked = ( "Error: connection to " "https://db.example.com:5432/db?password=supersecret123" ) results = {"findings": [{"phase": "Error", "content": leaked}]} with app.app_context(): data = _serialize_results(results).get_json() assert "supersecret123" not in data["findings"][0]["content"] assert data["findings"][0]["content"].startswith("Error:") def test_long_non_error_current_knowledge_passes_through_unchanged( self, app ): from local_deep_research.web.api import _serialize_results # The boundary sanitizer only fires on fields starting with # "Error:". A long, legitimate research summary that is NOT # exception-derived must pass through untouched -- no truncation # (the 500-char cap only applies via sanitize_error_for_client, # which never runs for non-error content) and no credential-regex # false positives on prose mentions of "Authorization"/"Bearer". long_summary = ( "Research summary: Authentication systems often use Bearer " "tokens in the Authorization header, but a well-designed API " "also supports API keys, OAuth, and session cookies. " ) * 20 # ~1800 chars, well over the boundary cap assert not long_summary.startswith("Error:") results = {"current_knowledge": long_summary} with app.app_context(): data = _serialize_results(results).get_json() assert data["current_knowledge"] == long_summary def test_error_field_categorizable_token_survives_boundary_cap(self, app): from local_deep_research.web.api import _serialize_results # Regression guard for the strategy cap alignment: an error # message whose categorizable signal ("Connection refused") sits # past the 200-char HTTP-client default must survive the boundary # scrub. If _ERROR_BOUNDARY_MAX_LEN regresses to 200, the token # is truncated and ErrorReportGenerator classification breaks. leading = "x" * 230 leaked = f"Error: {leading} Connection refused [Errno 111]" results = {"current_knowledge": leaked} with app.app_context(): data = _serialize_results(results).get_json() assert "Connection refused" in data["current_knowledge"] assert data["current_knowledge"].startswith("Error:") def test_scrub_error_fields_helper_covers_all_field_spellings(self, app): # _scrub_error_fields is the helper called by all three jsonify # sinks (_serialize_results, generate_report, analyze_documents). # Verify every scrubbed spelling as a unit: `current_knowledge` # (raw strategy dicts), `summary` (quick_summary / # analyze_documents payloads), `formatted_findings`, and # per-finding `content`. Endpoint-level coverage lives in # TestQuickSummary / TestAnalyzeDocuments. from local_deep_research.web.api import _scrub_error_fields leaked = ( "Error: connect to " "https://api.example.com/v1?api_key=sk-LEAK1234567890abcd" ) results = { "current_knowledge": leaked, "summary": leaked, "formatted_findings": leaked, "findings": [{"phase": "Error", "content": leaked}], } _scrub_error_fields(results) assert "sk-LEAK1234567890abcd" not in results["current_knowledge"] assert "sk-LEAK1234567890abcd" not in results["summary"] assert "sk-LEAK1234567890abcd" not in results["formatted_findings"] assert "sk-LEAK1234567890abcd" not in results["findings"][0]["content"] assert results["current_knowledge"].startswith("Error:") assert results["summary"].startswith("Error:") # =================================================================== # HTTP method enforcement # =================================================================== class TestHttpMethods: """Verify endpoints reject wrong HTTP methods.""" def test_health_rejects_post(self, client): resp = client.post("/api/v1/health") assert resp.status_code == 405 def test_docs_rejects_post(self, authed_client): resp = authed_client.post("/api/v1/") assert resp.status_code == 405 def test_quick_summary_rejects_get(self, authed_client): resp = authed_client.get("/api/v1/quick_summary") assert resp.status_code == 405 def test_generate_report_rejects_get(self, authed_client): resp = authed_client.get("/api/v1/generate_report") assert resp.status_code == 405 def test_analyze_documents_rejects_get(self, authed_client): resp = authed_client.get("/api/v1/analyze_documents") assert resp.status_code == 405