learningcircuit--local-deep-research/tests/advanced_search_system/tools/test_fetch_modes.py

"""Unit tests for ``advanced_search_system.tools.fetch.build_fetch_tool``.

Pins the mode dispatch and the prompt-content contract for the two
summary variants (focus-only vs focus + overall query). Avoids any real
HTTP — patches ``ContentFetcher`` and the model.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
    SearchResultsCollector,
)
from local_deep_research.advanced_search_system.tools.fetch import (
    FETCH_MODES,
    build_fetch_tool,
)


def _fetcher_cm(*, status="success", title="Page", content="Body text"):
    fetcher = MagicMock()
    fetcher.fetch.return_value = {
        "status": status,
        "title": title,
        "content": content,
    }
    cm = MagicMock()
    cm.__enter__.return_value = fetcher
    cm.__exit__.return_value = False
    return cm


def _model_returning(text: str):
    """A model whose ``invoke`` returns an object with ``.content == text``."""
    msg = MagicMock()
    msg.content = text
    model = MagicMock()
    model.invoke.return_value = msg
    return model


def test_fetch_modes_constant_lists_all_supported_values():
    assert FETCH_MODES == (
        "disabled",
        "full",
        "summary_focus",
        "summary_focus_query",
    )


def test_disabled_mode_returns_none():
    assert build_fetch_tool("disabled", SearchResultsCollector([])) is None


def test_full_mode_returns_tool_with_url_only_signature():
    tool = build_fetch_tool("full", SearchResultsCollector([]))
    assert tool is not None
    assert "url" in tool.args
    assert "focus" not in tool.args


def test_summary_focus_requires_model():
    with pytest.raises(ValueError, match="summary_focus"):
        build_fetch_tool("summary_focus", SearchResultsCollector([]))


def test_summary_focus_query_requires_model():
    with pytest.raises(ValueError, match="summary_focus_query"):
        build_fetch_tool("summary_focus_query", SearchResultsCollector([]))


def test_summary_focus_tool_calls_model_with_focus_only_prompt():
    collector = SearchResultsCollector([])
    model = _model_returning("relevant fact 1")
    tool = build_fetch_tool(
        "summary_focus",
        collector,
        model=model,
        overall_query="should not appear",
    )
    assert tool is not None
    assert "focus" in tool.args

    cm = _fetcher_cm(title="T", content="page body")
    with patch(
        "local_deep_research.content_fetcher.ContentFetcher", return_value=cm
    ):
        out = tool.invoke({"url": "http://example.com/", "focus": "year of X"})

    model.invoke.assert_called_once()
    prompt = model.invoke.call_args[0][0]
    # focus-only mode must NOT mention the overall query, even if one was passed
    assert "Why this page was fetched: year of X" in prompt
    assert "Overall research question" not in prompt
    # Output is prefixed with citation index so the agent can cite consistently
    assert out.startswith("[1] ")
    assert "relevant fact 1" in out


def test_summary_focus_query_includes_overall_query_in_prompt():
    collector = SearchResultsCollector([])
    model = _model_returning("relevant fact 2")
    tool = build_fetch_tool(
        "summary_focus_query",
        collector,
        model=model,
        overall_query="When did Liepmann receive the Prandtl-Ring Award?",
    )
    cm = _fetcher_cm(title="T", content="page body")
    with patch(
        "local_deep_research.content_fetcher.ContentFetcher", return_value=cm
    ):
        tool.invoke({"url": "http://example.com/", "focus": "year"})

    prompt = model.invoke.call_args[0][0]
    assert (
        "Overall research question: When did Liepmann receive the Prandtl-Ring Award?"
        in prompt
    )
    assert "Why this page was fetched: year" in prompt


def test_summary_focus_query_with_empty_overall_query_falls_back_to_focus_only():
    """An empty overall_query should not produce a stale 'Overall research question:' line."""
    collector = SearchResultsCollector([])
    model = _model_returning("ok")
    tool = build_fetch_tool(
        "summary_focus_query",
        collector,
        model=model,
        overall_query="",  # empty
    )
    cm = _fetcher_cm()
    with patch(
        "local_deep_research.content_fetcher.ContentFetcher", return_value=cm
    ):
        tool.invoke({"url": "http://example.com/", "focus": "x"})

    prompt = model.invoke.call_args[0][0]
    assert "Overall research question" not in prompt


def test_unknown_mode_raises_with_valid_modes_listed():
    with pytest.raises(ValueError, match="Unknown fetch mode"):
        build_fetch_tool("magic", SearchResultsCollector([]))


# ---- issue #3826: web.enable_javascript_rendering plumbing ----


def _captured_content_fetcher_kwargs(invoke_target):
    """Run *invoke_target* with ContentFetcher patched and return its
    constructor kwargs from the call.

    The patched factory returns a context manager whose ``fetch`` succeeds
    so the tool body runs end-to-end up to the call we care about.
    """
    cm = _fetcher_cm()
    with patch(
        "local_deep_research.content_fetcher.ContentFetcher", return_value=cm
    ) as factory:
        invoke_target()
    assert factory.call_args is not None
    return factory.call_args.kwargs


def test_full_mode_passes_js_off_when_snapshot_disables_it():
    """Full-mode fetch tool must pass enable_js_rendering=False to
    ContentFetcher when the snapshot disables JS rendering."""
    collector = SearchResultsCollector([])
    snapshot = {
        "web.enable_javascript_rendering": {
            "value": False,
            "ui_element": "checkbox",
        }
    }
    tool = build_fetch_tool("full", collector, settings_snapshot=snapshot)
    kwargs = _captured_content_fetcher_kwargs(
        lambda: tool.invoke({"url": "http://example.com/"})
    )
    assert kwargs.get("enable_js_rendering") is False


def test_full_mode_passes_js_on_when_snapshot_enables_it():
    """When the snapshot opts in, JS rendering is forwarded to ContentFetcher."""
    collector = SearchResultsCollector([])
    snapshot = {
        "web.enable_javascript_rendering": {
            "value": True,
            "ui_element": "checkbox",
        }
    }
    tool = build_fetch_tool("full", collector, settings_snapshot=snapshot)
    kwargs = _captured_content_fetcher_kwargs(
        lambda: tool.invoke({"url": "http://example.com/"})
    )
    assert kwargs.get("enable_js_rendering") is True


def test_full_mode_defaults_to_js_off_without_snapshot():
    """No snapshot, no thread-local context → JS disabled (safe default)."""
    collector = SearchResultsCollector([])
    tool = build_fetch_tool("full", collector)
    kwargs = _captured_content_fetcher_kwargs(
        lambda: tool.invoke({"url": "http://example.com/"})
    )
    assert kwargs.get("enable_js_rendering") is False


def test_fetch_error_scrubs_credentialed_url_from_exception():
    """A fetch exception that embeds a credentialed URL must be scrubbed
    before the tool return reaches the agent/LLM and the user-visible
    research output (credential-leak follow-up to #4625). Full detail stays
    in the server log."""
    collector = SearchResultsCollector([])
    tool = build_fetch_tool("full", collector)

    fetcher = MagicMock()
    fetcher.fetch.side_effect = Exception(
        "ConnectionError for https://user:SUPERSECRET123@proxy.example.com:8080"
    )
    cm = MagicMock()
    cm.__enter__.return_value = fetcher
    cm.__exit__.return_value = False

    with patch(
        "local_deep_research.content_fetcher.ContentFetcher", return_value=cm
    ):
        out = tool.invoke({"url": "http://example.com/"})

    assert "SUPERSECRET123" not in out  # credential scrubbed
    assert "Error fetching" in out  # still a useful error for the agent


def test_failed_fetch_scrubs_credentials_from_result_error():
    """A non-success fetch result carries ContentFetcher's raw error string,
    which can embed a credentialed URL. It must be scrubbed before the
    'Failed to fetch' tool return reaches the agent/LLM (#4633)."""
    collector = SearchResultsCollector([])
    tool = build_fetch_tool("full", collector)

    fetcher = MagicMock()
    fetcher.fetch.return_value = {
        "status": "error",
        "error": "blocked: https://user:SUPERSECRET999@cdn.example.com",
    }
    cm = MagicMock()
    cm.__enter__.return_value = fetcher
    cm.__exit__.return_value = False

    with patch(
        "local_deep_research.content_fetcher.ContentFetcher", return_value=cm
    ):
        out = tool.invoke({"url": "http://example.com/"})

    assert "SUPERSECRET999" not in out  # credential scrubbed
    assert "Failed to fetch" in out  # still a useful error for the agent


def test_summary_mode_forwards_js_setting():
    """Summary-mode tool also forwards the JS toggle from the snapshot."""
    collector = SearchResultsCollector([])
    model = _model_returning("ok")
    snapshot = {
        "web.enable_javascript_rendering": {
            "value": False,
            "ui_element": "checkbox",
        }
    }
    tool = build_fetch_tool(
        "summary_focus",
        collector,
        model=model,
        settings_snapshot=snapshot,
    )
    kwargs = _captured_content_fetcher_kwargs(
        lambda: tool.invoke({"url": "http://example.com/", "focus": "x"})
    )
    assert kwargs.get("enable_js_rendering") is False


# ---------------------------------------------------------------------------
# Egress policy: an out-of-scope URL is a RECOVERABLE tool message, not a
# raised exception. Re-raising used to abort a pooled subagent (and depended
# on each agent's tool-error layer); returning a message lets the lead agent
# and subagents handle it identically and stay in-scope. The URL is never
# fetched either way — only the reporting changes.
# ---------------------------------------------------------------------------


def test_full_mode_egress_denial_returns_message_not_raises():
    from local_deep_research.security.egress.policy import Decision

    collector = SearchResultsCollector([])
    tool = build_fetch_tool("full", collector, egress_context=object())

    with patch(
        "local_deep_research.security.egress.policy.evaluate_url",
        return_value=Decision(False, "scope_mismatch_private_only"),
    ):
        out = tool.invoke({"url": "https://public.example.com/x"})

    assert isinstance(out, str)
    assert "Cannot fetch" in out
    assert "scope_mismatch_private_only" in out  # the reason is surfaced
    assert "skip external URLs" in out  # the adapt-and-stay-local hint
    # The denied URL must never be registered as a citation.
    assert collector.results == []


def test_summary_mode_egress_denial_returns_message_not_raises():
    from local_deep_research.security.egress.policy import Decision

    collector = SearchResultsCollector([])
    model = _model_returning("should-never-be-called")
    tool = build_fetch_tool(
        "summary_focus", collector, model=model, egress_context=object()
    )

    with patch(
        "local_deep_research.security.egress.policy.evaluate_url",
        return_value=Decision(False, "scope_mismatch_private_only"),
    ):
        out = tool.invoke({"url": "https://public.example.com/x", "focus": "y"})

    assert isinstance(out, str)
    assert "Cannot fetch" in out
    assert "scope_mismatch_private_only" in out
    # The gate fires BEFORE any fetch/summarise work.
    model.invoke.assert_not_called()
    assert collector.results == []


def test_fetch_error_log_redacts_url_and_adds_mode():
    """The server-log line for a fetch error carries the MODE and a REDACTED
    scheme://host (no userinfo / path / query) — enough to locate the failure
    without leaking credentials, query tokens, page paths, or content. Asserted
    on the args passed to logger.exception (robust to loguru sink config)."""
    collector = SearchResultsCollector([])
    tool = build_fetch_tool("full", collector)

    fetcher = MagicMock()
    fetcher.fetch.side_effect = Exception("boom")
    cm = MagicMock()
    cm.__enter__.return_value = fetcher
    cm.__exit__.return_value = False

    url = (
        "https://user:SECRET123@proxy.example.com:8080/secretpath?token=ABCXYZ"
    )
    with patch(
        "local_deep_research.advanced_search_system.tools.fetch.logger"
    ) as mock_logger:
        with patch(
            "local_deep_research.content_fetcher.ContentFetcher",
            return_value=cm,
        ):
            tool.invoke({"url": url})

    mock_logger.exception.assert_called_once()
    args = mock_logger.exception.call_args.args
    # args = (template, mode_label, redacted_url)
    assert "mode={}" in args[0] and "url={}" in args[0]
    assert "full" in args  # the mode
    assert (
        "https://proxy.example.com:8080" in args
    )  # redacted scheme://host:port
    flat = " ".join(map(str, args))
    assert "SECRET123" not in flat  # userinfo dropped
    assert "ABCXYZ" not in flat  # query token dropped
    assert "secretpath" not in flat  # path dropped