learningcircuit--local-deep-research/tests/web/test_api_coverage.py

"""
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.<research_fn> 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 <think> 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