learningcircuit--local-deep-research/tests/core/test_search_system.py

"""
Tests for search_system.py - AdvancedSearchSystem class.

Tests cover:
- __init__ settings extraction, defaults, factory wiring, follow-up delegation, logging
- set_progress_callback forwarding to strategy
- analyze_topic search_id generation and delegation
- _perform_search settings extraction, progress callbacks, result assembly
- Duplicate link avoidance (id check)
- NewsSearchCallback error resilience
"""

from unittest.mock import MagicMock, patch

import pytest


# ---------------------------------------------------------------------------
# Helpers / fixtures
# ---------------------------------------------------------------------------


def _make_strategy_mock():
    """Return a MagicMock that behaves like a strategy returned by create_strategy."""
    strategy = MagicMock()
    strategy.analyze_topic.return_value = {"current_knowledge": "test"}
    strategy.questions_by_iteration = []
    strategy.all_links_of_system = []
    strategy.set_progress_callback = MagicMock()
    return strategy


# Patch paths: top-level imports can be patched on search_system directly,
# but create_strategy and NewsSearchCallback are lazily imported inside method
# bodies, so they must be patched at their source modules.
PATCH_CITATION = "local_deep_research.search_system.CitationHandler"
PATCH_QUESTION_GEN = (
    "local_deep_research.search_system.StandardQuestionGenerator"
)
PATCH_FINDINGS = "local_deep_research.search_system.FindingsRepository"
PATCH_CREATE_STRATEGY = (
    "local_deep_research.search_system_factory.create_strategy"
)
PATCH_ENHANCED_FOLLOWUP = (
    "local_deep_research.search_system.EnhancedContextualFollowUpStrategy"
)
PATCH_NEWS_CALLBACK = (
    "local_deep_research.news.core.search_integration.NewsSearchCallback"
)


@pytest.fixture()
def mock_deps():
    """Patch CitationHandler, StandardQuestionGenerator, FindingsRepository,
    and create_strategy so that AdvancedSearchSystem can be instantiated
    without real dependencies.  Yields a dict of mock objects."""
    with (
        patch(PATCH_CITATION) as m_citation,
        patch(PATCH_QUESTION_GEN) as m_qgen,
        patch(PATCH_FINDINGS) as m_findings,
        patch(PATCH_CREATE_STRATEGY) as m_create,
    ):
        strategy = _make_strategy_mock()
        m_create.return_value = strategy

        yield {
            "citation": m_citation,
            "question_gen": m_qgen,
            "findings": m_findings,
            "create_strategy": m_create,
            "strategy": strategy,
        }


@pytest.fixture()
def mock_model():
    return MagicMock(name="BaseChatModel")


@pytest.fixture()
def mock_search():
    return MagicMock(name="BaseSearchEngine")


def _build_system(mock_model, mock_search, mock_deps, **kwargs):
    """Convenience: build an AdvancedSearchSystem with the standard mocks."""
    from local_deep_research.search_system import AdvancedSearchSystem

    return AdvancedSearchSystem(llm=mock_model, search=mock_search, **kwargs)


# ===================================================================
# __init__ tests
# ===================================================================


class TestInit:
    """Tests for AdvancedSearchSystem.__init__."""

    def test_settings_snapshot_defaults_to_empty_dict(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(
            mock_model, mock_search, mock_deps, settings_snapshot=None
        )
        assert system.settings_snapshot == {}

    # -- max_iterations -------------------------------------------------

    def test_max_iterations_default_is_one(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        assert system.max_iterations == 1

    def test_max_iterations_from_settings_plain_value(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(
            mock_model,
            mock_search,
            mock_deps,
            settings_snapshot={"search.iterations": 5},
        )
        assert system.max_iterations == 5

    def test_max_iterations_extracts_value_from_dict(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(
            mock_model,
            mock_search,
            mock_deps,
            settings_snapshot={"search.iterations": {"value": 8}},
        )
        assert system.max_iterations == 8

    def test_max_iterations_explicit_overrides_snapshot(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(
            mock_model,
            mock_search,
            mock_deps,
            max_iterations=12,
            settings_snapshot={"search.iterations": {"value": 8}},
        )
        assert system.max_iterations == 12

    # -- questions_per_iteration ----------------------------------------

    def test_questions_per_iteration_default_is_three(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        assert system.questions_per_iteration == 3

    def test_questions_per_iteration_from_settings_dict(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(
            mock_model,
            mock_search,
            mock_deps,
            settings_snapshot={"search.questions_per_iteration": {"value": 7}},
        )
        assert system.questions_per_iteration == 7

    # -- strategy creation via factory ----------------------------------

    def test_calls_create_strategy_factory(
        self, mock_model, mock_search, mock_deps
    ):
        _build_system(mock_model, mock_search, mock_deps)
        mock_deps["create_strategy"].assert_called_once()

    # -- enhanced-contextual-followup strategy --------------------------

    def test_enhanced_contextual_followup_creates_delegate(
        self, mock_model, mock_search, mock_deps
    ):
        with patch(PATCH_ENHANCED_FOLLOWUP) as m_followup:
            m_followup.return_value = _make_strategy_mock()
            system = _build_system(
                mock_model,
                mock_search,
                mock_deps,
                strategy_name="enhanced-contextual-followup",
            )
            # create_strategy should have been called for the delegate
            mock_deps["create_strategy"].assert_called_once()
            delegate_kwargs = mock_deps["create_strategy"].call_args[1]
            assert delegate_kwargs["strategy_name"] == "source-based"
            # EnhancedContextualFollowUpStrategy should have been instantiated
            m_followup.assert_called_once()
            assert system.strategy is m_followup.return_value

    # -- programmatic_mode warning --------------------------------------

    def test_programmatic_mode_logs_warning(
        self, mock_model, mock_search, mock_deps
    ):
        with patch("local_deep_research.search_system.logger") as m_logger:
            _build_system(
                mock_model, mock_search, mock_deps, programmatic_mode=True
            )
            m_logger.warning.assert_called_once()
            assert (
                "programmatic mode" in m_logger.warning.call_args[0][0].lower()
            )


# ===================================================================
# set_progress_callback tests
# ===================================================================


class TestSetProgressCallback:
    """Tests for set_progress_callback."""

    def test_forwards_callback_to_strategy(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        cb = MagicMock(name="my_callback")
        system.set_progress_callback(cb)
        mock_deps["strategy"].set_progress_callback.assert_called_with(cb)


# ===================================================================
# analyze_topic tests
# ===================================================================


class TestAnalyzeTopic:
    """Tests for analyze_topic."""

    def test_generates_search_id_when_not_provided(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        with patch.object(system, "_perform_search", return_value={}) as m_ps:
            system.analyze_topic("quantum computing")
            # search_id (2nd positional arg) should be a non-empty string
            search_id = m_ps.call_args[0][1]
            assert isinstance(search_id, str) and len(search_id) > 0

    def test_passes_explicit_search_id(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        with patch.object(system, "_perform_search", return_value={}) as m_ps:
            system.analyze_topic("q", search_id="my-id-99")
            assert m_ps.call_args[0][1] == "my-id-99"

    def test_returns_perform_search_result(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        expected = {"current_knowledge": "deep"}
        with patch.object(system, "_perform_search", return_value=expected):
            result = system.analyze_topic("topic")
            assert result is expected


# ===================================================================
# _perform_search tests
# ===================================================================


class TestPerformSearch:
    """Tests for _perform_search."""

    def _run_perform_search(
        self,
        mock_model,
        mock_search,
        mock_deps,
        settings=None,
        query="test q",
    ):
        system = _build_system(
            mock_model,
            mock_search,
            mock_deps,
            settings_snapshot=settings or {},
        )
        progress_calls = []
        system.progress_callback = lambda m, p, md: progress_calls.append(
            (m, p, md)
        )
        with patch(PATCH_NEWS_CALLBACK):
            result = system._perform_search(
                query, "sid-1", True, False, "user1"
            )
        return system, result, progress_calls

    def test_extracts_llm_provider_from_settings(
        self, mock_model, mock_search, mock_deps
    ):
        settings = {"llm.provider": {"value": "anthropic"}}
        _, _, calls = self._run_perform_search(
            mock_model, mock_search, mock_deps, settings=settings
        )
        assert any("anthropic" in str(c) for c in calls)

    def test_extracts_llm_model_from_settings(
        self, mock_model, mock_search, mock_deps
    ):
        settings = {"llm.model": "claude-3-opus"}
        _, _, calls = self._run_perform_search(
            mock_model, mock_search, mock_deps, settings=settings
        )
        assert any("claude-3-opus" in str(c) for c in calls)

    def test_extracts_search_tool_from_settings(
        self, mock_model, mock_search, mock_deps
    ):
        settings = {"search.tool": {"value": "google"}}
        _, _, calls = self._run_perform_search(
            mock_model, mock_search, mock_deps, settings=settings
        )
        assert any("google" in str(c) for c in calls)

    def test_calls_progress_callback_with_setup_info(
        self, mock_model, mock_search, mock_deps
    ):
        _, _, calls = self._run_perform_search(
            mock_model, mock_search, mock_deps
        )
        # At least two calls: one for LLM info, one for search tool info
        assert len(calls) >= 2
        phases = [c[2].get("phase") for c in calls]
        assert "setup" in phases

    def test_calls_strategy_analyze_topic(
        self, mock_model, mock_search, mock_deps
    ):
        self._run_perform_search(
            mock_model, mock_search, mock_deps, query="my query"
        )
        mock_deps["strategy"].analyze_topic.assert_called_once_with("my query")

    def test_updates_questions_by_iteration(
        self, mock_model, mock_search, mock_deps
    ):
        mock_deps["strategy"].questions_by_iteration = ["q1", "q2"]
        system, result, _ = self._run_perform_search(
            mock_model, mock_search, mock_deps
        )
        assert system.questions_by_iteration == ["q1", "q2"]
        assert result["questions_by_iteration"] == ["q1", "q2"]

    def test_result_contains_search_system(
        self, mock_model, mock_search, mock_deps
    ):
        system, result, _ = self._run_perform_search(
            mock_model, mock_search, mock_deps
        )
        assert result["search_system"] is system

    def test_result_contains_all_links_of_system(
        self, mock_model, mock_search, mock_deps
    ):
        mock_deps["strategy"].all_links_of_system = [{"url": "http://a.com"}]
        _, result, _ = self._run_perform_search(
            mock_model, mock_search, mock_deps
        )
        assert "all_links_of_system" in result

    def test_result_contains_query(self, mock_model, mock_search, mock_deps):
        _, result, _ = self._run_perform_search(
            mock_model, mock_search, mock_deps, query="deep research"
        )
        assert result["query"] == "deep research"

    def test_avoids_duplicate_links_when_same_object(
        self, mock_model, mock_search, mock_deps
    ):
        """When strategy.all_links_of_system is the SAME object as
        system.all_links_of_system, no extend should happen (id check)."""
        system = _build_system(mock_model, mock_search, mock_deps)
        # Make the strategy share the exact same list object
        mock_deps["strategy"].all_links_of_system = system.all_links_of_system
        system.all_links_of_system.extend(
            [{"url": "http://a.com"}, {"url": "http://b.com"}]
        )

        system.progress_callback = lambda m, p, md: None
        with patch(PATCH_NEWS_CALLBACK):
            system._perform_search("q", "sid", True, False, "u")

        # If id-check were removed, self-extend would double to 4
        assert len(system.all_links_of_system) == 2

    def test_extends_links_when_different_objects(
        self, mock_model, mock_search, mock_deps
    ):
        """When strategy.all_links_of_system is a different object,
        the links should be extended."""
        system = _build_system(mock_model, mock_search, mock_deps)
        strategy_links = [{"url": "http://new.com"}]
        mock_deps["strategy"].all_links_of_system = strategy_links

        system.progress_callback = lambda m, p, md: None
        with patch(PATCH_NEWS_CALLBACK):
            system._perform_search("q", "sid", True, False, "u")

        assert {"url": "http://new.com"} in system.all_links_of_system

    def test_news_callback_error_does_not_break_search(
        self, mock_model, mock_search, mock_deps
    ):
        system = _build_system(mock_model, mock_search, mock_deps)
        system.progress_callback = lambda m, p, md: None

        with patch(PATCH_NEWS_CALLBACK, side_effect=Exception("boom")):
            # Should not raise
            result = system._perform_search("q", "sid", True, False, "u")
            assert isinstance(result, dict)