learningcircuit--local-deep-research/tests/web/services/test_research_service_execution.py

"""
Tests for run_research_process() core execution logic.

Covers:
- Quick vs detailed mode branching
- Settings context creation
- Progress callback invocation
- Termination handling
- Error handling
- Research status updates
"""

import pytest

from local_deep_research.web.services.research_service import (
    _DETAILED_REPORT_PROGRESS_END,
    _DETAILED_SEARCH_PROGRESS_CAP,
    _REPORT_PHASES,
)


class TestSettingsContext:
    """Tests for SettingsContext created inside run_research_process."""

    def test_settings_context_extracts_values_from_setting_objects(self):
        """SettingsContext extracts 'value' from setting dicts."""
        # Replicate the SettingsContext class logic from research_service.py
        snapshot = {
            "llm.provider": {"value": "openai", "type": "string"},
            "search.tool": "google",  # plain value
        }

        # Simulate the extraction logic
        values = {}
        for key, setting in snapshot.items():
            if isinstance(setting, dict) and "value" in setting:
                values[key] = setting["value"]
            else:
                values[key] = setting

        assert values["llm.provider"] == "openai"
        assert values["search.tool"] == "google"

    def test_settings_context_get_setting_from_snapshot(self):
        """get_setting returns value from snapshot, default for missing."""
        values = {"llm.provider": "openai"}

        def get_setting(key, default=None):
            return values.get(key, default)

        assert get_setting("llm.provider") == "openai"
        assert get_setting("missing.key", "fallback") == "fallback"

    def test_settings_context_empty_snapshot(self):
        """Empty snapshot → all get_setting calls return default."""
        values = {}

        def get_setting(key, default=None):
            return values.get(key, default)

        assert get_setting("any.key", 42) == 42


class TestProgressCallback:
    """Tests for progress callback logic."""

    def test_progress_adjusted_for_detailed_output_generation(self):
        """Detailed mode output_generation → capped at search cap."""
        mode = "detailed"
        metadata = {"phase": "output_generation"}
        progress_percent = 90

        adjusted_progress = progress_percent
        phase = metadata.get("phase", "")
        if mode == "detailed":
            if phase not in _REPORT_PHASES and progress_percent is not None:
                adjusted_progress = min(
                    _DETAILED_SEARCH_PROGRESS_CAP, progress_percent
                )

        assert adjusted_progress == _DETAILED_SEARCH_PROGRESS_CAP

    def test_progress_adjusted_for_detailed_report_generation(self):
        """Detailed mode report_generation → passes through (wrapper maps range)."""
        mode = "detailed"
        metadata = {"phase": "report_generation"}
        progress_percent = 50

        adjusted_progress = progress_percent
        phase = metadata.get("phase", "")
        if mode == "detailed" and phase not in _REPORT_PHASES:
            adjusted_progress = min(
                _DETAILED_SEARCH_PROGRESS_CAP, progress_percent
            )

        assert adjusted_progress == 50

    def test_progress_adjusted_for_quick_output_generation(self):
        """Quick mode output_generation → scaled 85-95%."""
        mode = "quick"
        metadata = {"phase": "output_generation"}
        progress_percent = 50

        adjusted_progress = progress_percent
        if (
            mode == "quick"
            and metadata.get("phase") == "output_generation"
            and progress_percent is not None
        ):
            if progress_percent > 0:
                adjusted_progress = 85 + (progress_percent / 100) * 10
            else:
                adjusted_progress = 85

        assert adjusted_progress == 90.0

    def test_quick_output_generation_with_none_progress_passes_through(self):
        """Quick mode output_generation with None progress must not crash."""
        mode = "quick"
        metadata = {"phase": "output_generation"}
        progress_percent = None

        adjusted_progress = progress_percent
        if (
            mode == "quick"
            and metadata.get("phase") == "output_generation"
            and progress_percent is not None
        ):
            if progress_percent > 0:
                adjusted_progress = 85 + (progress_percent / 100) * 10
            else:
                adjusted_progress = 85

        assert adjusted_progress is None

    def test_search_plan_extracted_from_message(self):
        """SEARCH_PLAN: in message → engines extracted."""
        message = "Planning SEARCH_PLAN: google, bing, wikipedia"
        metadata = {}

        if "SEARCH_PLAN:" in message:
            engines = message.split("SEARCH_PLAN:")[1].strip()
            metadata["planned_engines"] = engines
            metadata["phase"] = "search_planning"

        assert metadata["planned_engines"] == "google, bing, wikipedia"
        assert metadata["phase"] == "search_planning"

    def test_engine_selected_extracted_from_message(self):
        """ENGINE_SELECTED: in message → engine extracted."""
        message = "Selected ENGINE_SELECTED: google"
        metadata = {}

        if "ENGINE_SELECTED:" in message:
            engine = message.split("ENGINE_SELECTED:")[1].strip()
            metadata["selected_engine"] = engine
            metadata["phase"] = "search"

        assert metadata["selected_engine"] == "google"


def _apply_detailed_progress(progress_percent, phase):
    """Replicate detailed-mode progress adjustment from research_service.py.

    Mirrors the production logic so tests stay self-contained, matching the
    existing pattern in this file.
    """
    adjusted_progress = progress_percent
    if phase not in _REPORT_PHASES and progress_percent is not None:
        adjusted_progress = min(_DETAILED_SEARCH_PROGRESS_CAP, progress_percent)
    return adjusted_progress


class TestDetailedModeSearchCap:
    """Tests for the search-phase cap in detailed mode.

    Global monotonicity is enforced by update_progress_and_check_active in
    web/routes/globals.py; these tests cover only the closure's per-call
    capping. The integration tests in
    test_research_service_progress_integration.py exercise the full
    closure + global guard pipeline.
    """

    def test_high_search_values_clamp_to_cap(self):
        """Search progress above the cap clamps to the cap."""
        emitted = [
            _apply_detailed_progress(value, "search")
            for value in (90, 33, 50, 5, 100)
        ]
        assert emitted == [8, 8, 8, 5, 8]

    def test_low_search_values_pass_through_below_cap(self):
        """Search progress below the cap passes through unchanged."""
        assert _apply_detailed_progress(3, "search") == 3
        assert _apply_detailed_progress(7, "search") == 7
        assert _apply_detailed_progress(8, "search") == 8

    def test_none_progress_passes_through_for_error_phase(self):
        """None progress with phase='error' must not crash (regression #3806)."""
        assert _apply_detailed_progress(None, "error") is None

    def test_none_progress_passes_through_for_sub_search(self):
        """None progress from constrained-search sub-callback must not crash.

        The sub-callback in constrained_search_strategy.py emits None with
        phase in {'search_complete', 'final_results'}.
        """
        for phase in ("search_complete", "final_results"):
            assert _apply_detailed_progress(None, phase) is None, (
                f"None should pass through for phase={phase}"
            )

    def test_report_phase_unaffected_by_search_cap(self):
        """Report phases pass through; the wrapper already maps the range."""
        for phase in (
            "report_generation",
            "report_section_research",
            "report_formatting",
            "report_structure",
            "report_complete",
        ):
            assert _apply_detailed_progress(55, phase) == 55

    def test_strategy_complete_phase_does_not_jump_bar_to_100_mid_report(self):
        """phase='complete' from a strategy mid-report must NOT pin bar to 100.

        Regression: every search strategy emits {'phase': 'complete'} when its
        analyze_topic finishes (e.g. standard_strategy.py:334 at value 95).
        report_generator runs analyze_topic per subsection via
        self.search_system.analyze_topic, and the SearchSystem's
        progress_callback is the outer callback (set at research_service.py:711,
        NOT the wrapper). So a strategy 'complete' fires AFTER each subsection
        and would jump the bar to 100 mid-report if treated as the final marker.

        Expected behavior: 'complete' is treated as a search-phase emission —
        capped at the search cap. The legitimate 100% emit uses phase=
        'report_complete' (in _REPORT_PHASES) and passes through.
        """
        # First subsection's strategy finishes: emits phase='complete' at 95
        adjusted = _apply_detailed_progress(95, "complete")
        assert adjusted != 100, (
            "regression: strategy 'complete' jumped bar to 100 mid-report"
        )
        assert adjusted == _DETAILED_SEARCH_PROGRESS_CAP

    def test_legitimate_final_complete_uses_report_complete_phase(self):
        """The legitimate end-of-research 100 emit uses phase='report_complete'.

        Confirms that the production code's final detailed-mode emission
        uses a phase that's in _REPORT_PHASES, so it passes through
        unchanged to 100 — independent of the strategy-level 'complete'
        phase that is intentionally capped.
        """
        assert "report_complete" in _REPORT_PHASES
        assert "complete" not in _REPORT_PHASES
        adjusted = _apply_detailed_progress(
            _DETAILED_REPORT_PROGRESS_END, "report_complete"
        )
        assert adjusted == _DETAILED_REPORT_PROGRESS_END


class TestTerminationHandling:
    """Tests for termination checks in research process."""

    def test_termination_during_progress_raises(self):
        """Termination requested during progress → raises exception."""
        from local_deep_research.web.services.research_service import (
            ResearchTerminatedException,
        )

        with pytest.raises(ResearchTerminatedException):
            raise ResearchTerminatedException("Research was terminated by user")


def _classify_error(error_message):
    """Replicate error classification logic from run_research_process (lines 714-733)."""
    if "status code: 503" in error_message:
        return "ollama_unavailable"
    if "status code: 404" in error_message:
        return "model_not_found"
    if "status code:" in error_message:
        return "api_error"
    if "connection" in error_message.lower():
        return "connection_error"
    return "unknown"


class TestErrorClassification:
    """Tests for search error classification in run_research_process."""

    @pytest.mark.parametrize(
        "error_message, expected_type",
        [
            ("Request failed with status code: 503", "ollama_unavailable"),
            ("status code: 404 not found", "model_not_found"),
            ("status code: 429 rate limited", "api_error"),
            ("status code: 500 internal error", "api_error"),
            ("Connection refused to localhost:11434", "connection_error"),
            ("TCP connection reset by peer", "connection_error"),
            ("Something unexpected happened", "unknown"),
            ("", "unknown"),
        ],
    )
    def test_error_classification(self, error_message, expected_type):
        assert _classify_error(error_message) == expected_type

    def test_503_takes_priority_over_generic_status_code(self):
        """503 is matched specifically before generic 'status code:' pattern."""
        assert _classify_error("status code: 503") == "ollama_unavailable"

    def test_404_takes_priority_over_generic_status_code(self):
        """404 is matched specifically before generic 'status code:' pattern."""
        assert _classify_error("status code: 404") == "model_not_found"


class TestResearchModes:
    """Tests for mode-specific behavior."""

    def test_quick_mode_checks_findings(self):
        """Quick mode checks for findings or formatted_findings."""
        results = {"findings": ["f1"], "formatted_findings": "Summary text"}
        mode = "quick"

        if mode == "quick":
            has_findings = bool(
                results.get("findings") or results.get("formatted_findings")
            )
        else:
            has_findings = False

        assert has_findings is True

    def test_quick_mode_detects_error_in_findings(self):
        """Quick mode detects 'Error:' prefix in formatted_findings."""
        results = {"formatted_findings": "Error: token limit exceeded"}

        raw = results["formatted_findings"]
        is_error = isinstance(raw, str) and raw.startswith("Error:")

        assert is_error is True

    def test_quick_mode_no_error_prefix(self):
        """Normal formatted_findings → not detected as error."""
        results = {"formatted_findings": "This is a normal summary."}

        raw = results["formatted_findings"]
        is_error = isinstance(raw, str) and raw.startswith("Error:")

        assert is_error is False