learningcircuit--local-deep-research/tests/mcp/test_server.py

"""
Tests for the MCP server tools.

These tests use mocked research functions to verify the MCP tool implementations
without actually running research (which would take minutes).
"""

from unittest.mock import patch

import pytest

# Skip all tests if MCP is not available
try:
    import mcp  # noqa: F401

    MCP_AVAILABLE = True
except ImportError:
    MCP_AVAILABLE = False

pytestmark = pytest.mark.skipif(
    not MCP_AVAILABLE, reason="MCP package not installed"
)


class TestQuickResearch:
    """Tests for the quick_research tool."""

    def test_quick_research_success(self, mock_quick_summary):
        """Test successful quick research."""
        from local_deep_research.mcp.server import quick_research

        result = quick_research(query="What is quantum computing?")

        assert result["status"] == "success"
        assert "summary" in result
        assert (
            result["summary"]
            == "This is a test summary about quantum computing."
        )
        assert "findings" in result
        assert "sources" in result
        assert result["iterations"] == 1
        mock_quick_summary.assert_called_once()

    def test_quick_research_with_overrides(self, mock_quick_summary):
        """Test quick research with parameter overrides."""
        from local_deep_research.mcp.server import quick_research

        result = quick_research(
            query="Test query",
            search_engine="wikipedia",
            strategy="source-based",
            iterations=2,
        )

        assert result["status"] == "success"
        mock_quick_summary.assert_called_once()
        # Verify settings_snapshot was passed
        call_kwargs = mock_quick_summary.call_args[1]
        assert "settings_snapshot" in call_kwargs

    def test_quick_research_error_handling(self):
        """Test error handling in quick research."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            side_effect=Exception("API key invalid"),
        ):
            result = quick_research(query="Test query")

        assert result["status"] == "error"
        assert "error" in result
        assert result["error_type"] == "auth_error"


class TestDetailedResearch:
    """Tests for the detailed_research tool."""

    def test_detailed_research_success(self, mock_detailed_research):
        """Test successful detailed research."""
        from local_deep_research.mcp.server import detailed_research

        result = detailed_research(query="quantum computing applications")

        assert result["status"] == "success"
        assert result["query"] == "quantum computing applications"
        assert result["research_id"] == "test-research-123"
        assert "summary" in result
        assert "findings" in result
        assert len(result["findings"]) == 2
        assert "metadata" in result

    def test_detailed_research_error_handling(self):
        """Test error handling in detailed research."""
        from local_deep_research.mcp.server import detailed_research

        with patch(
            "local_deep_research.mcp.server.ldr_detailed_research",
            side_effect=Exception("Service unavailable: 503"),
        ):
            result = detailed_research(query="Test query")

        assert result["status"] == "error"
        assert result["error_type"] == "service_unavailable"


class TestGenerateReport:
    """Tests for the generate_report tool."""

    def test_generate_report_success(self, mock_generate_report):
        """Test successful report generation."""
        from local_deep_research.mcp.server import generate_report

        result = generate_report(query="quantum computing")

        assert result["status"] == "success"
        assert "content" in result
        assert result["content"].startswith("# Research Report")
        assert "metadata" in result

    def test_generate_report_with_options(self, mock_generate_report):
        """Test report generation with options."""
        from local_deep_research.mcp.server import generate_report

        result = generate_report(
            query="Test query",
            search_engine="arxiv",
            searches_per_section=3,
        )

        assert result["status"] == "success"
        call_kwargs = mock_generate_report.call_args[1]
        assert call_kwargs["searches_per_section"] == 3

    def test_generate_report_preserves_assembled_sources_block(self):
        """MCP surfaces the generator's in-memory assembled content verbatim,
        including the ``## Sources`` block — it never DB-reads or strips it.

        Pins the MCP-keeps-assembled-shape invariant the #3665 audit
        confirmed (in-memory generator output, never a DB read).
        """
        from local_deep_research.mcp.server import generate_report

        assembled = (
            "# Research Report\n\n## Introduction\n\nBody.\n\n"
            "## Sources\n\n[1] https://src.example\n\n"
            "## Research Metrics\n\nSearch Iterations: 2\n"
        )
        with patch(
            "local_deep_research.mcp.server.ldr_generate_report",
            return_value={"content": assembled, "metadata": {"query": "q"}},
        ):
            result = generate_report(query="q")

        assert result["status"] == "success"
        # Verbatim passthrough: the assembled content (incl. ## Sources and the
        # trailing ## Research Metrics) survives unmodified. Full equality —
        # substring presence alone wouldn't catch truncation after ## Sources.
        assert result["content"] == assembled
        assert result["metadata"] == {"query": "q"}


class TestAnalyzeDocuments:
    """Tests for the analyze_documents tool."""

    def test_analyze_documents_success(self, mock_analyze_documents):
        """Test successful document analysis."""
        from local_deep_research.mcp.server import analyze_documents

        result = analyze_documents(
            query="test query",
            collection_name="test_collection",
        )

        assert result["status"] == "success"
        assert result["collection"] == "test_collection"
        assert result["document_count"] == 1
        assert "summary" in result
        assert "documents" in result

    def test_analyze_documents_error_handling(self):
        """Test error handling in document analysis."""
        from local_deep_research.mcp.server import analyze_documents

        with patch(
            "local_deep_research.mcp.server.ldr_analyze_documents",
            side_effect=Exception("Collection not found"),
        ):
            result = analyze_documents(
                query="test query",
                collection_name="nonexistent",
            )

        assert result["status"] == "error"
        assert result["error_type"] == "model_not_found"


class TestDiscoveryTools:
    """Tests for discovery tools (list_search_engines, list_strategies, get_configuration)."""

    def test_list_strategies(self):
        """Test listing available strategies."""
        from local_deep_research.mcp.server import list_strategies

        result = list_strategies()

        assert result["status"] == "success"
        assert "strategies" in result
        assert len(result["strategies"]) > 0
        # Check that each strategy has required fields
        for strategy in result["strategies"]:
            assert "name" in strategy
            assert "description" in strategy

    def test_list_search_engines(self):
        """Test listing available search engines."""
        from local_deep_research.mcp.server import list_search_engines

        # Mock the search_config function at the right location
        mock_engines = {
            "wikipedia": {
                "description": "Wikipedia search",
                "strengths": ["Free", "Reliable"],
                "weaknesses": ["Limited depth"],
                "requires_api_key": False,
            },
            "arxiv": {
                "description": "arXiv academic papers",
                "strengths": ["Academic sources"],
                "weaknesses": ["Science only"],
                "requires_api_key": False,
            },
        }

        with patch(
            "local_deep_research.web_search_engines.search_engines_config.search_config",
            return_value=mock_engines,
        ):
            result = list_search_engines()

        assert result["status"] == "success"
        assert "engines" in result
        assert len(result["engines"]) == 2

    def test_get_configuration(self, mock_settings_snapshot):
        """Test getting server configuration."""
        from local_deep_research.mcp.server import get_configuration

        result = get_configuration()

        assert result["status"] == "success"
        assert "config" in result
        config = result["config"]
        assert "llm" in config
        assert "search" in config


class TestSearch:
    """Tests for the search tool."""

    def test_search_success(self):
        """Test successful raw search."""
        from local_deep_research.mcp.server import search

        mock_engine = type(
            "MockEngine",
            (),
            {
                "run": lambda self, q: [
                    {
                        "title": "Result 1",
                        "link": "https://example.com/1",
                        "snippet": "Snippet 1",
                    },
                    {
                        "title": "Result 2",
                        "link": "https://example.com/2",
                        "snippet": "Snippet 2",
                    },
                ]
            },
        )()
        mock_engines_config = {
            "wikipedia": {
                "description": "Wikipedia",
                "requires_api_key": False,
            },
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
            patch(
                "local_deep_research.web_search_engines.search_engine_factory.create_search_engine",
                return_value=mock_engine,
            ),
        ):
            result = search(query="quantum computing", engine="wikipedia")

        assert result["status"] == "success"
        assert result["query"] == "quantum computing"
        assert result["engine"] == "wikipedia"
        assert result["result_count"] == 2
        assert len(result["results"]) == 2
        assert result["results"][0]["title"] == "Result 1"

    def test_search_with_max_results(self):
        """Test that max_results is passed to factory."""
        from local_deep_research.mcp.server import search

        mock_engine = type("MockEngine", (), {"run": lambda self, q: []})()
        mock_engines_config = {
            "arxiv": {"description": "arXiv", "requires_api_key": False},
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
            patch(
                "local_deep_research.web_search_engines.search_engine_factory.create_search_engine",
                return_value=mock_engine,
            ) as mock_factory,
        ):
            result = search(query="test", engine="arxiv", max_results=25)

        assert result["status"] == "success"
        mock_factory.assert_called_once()
        call_kwargs = mock_factory.call_args[1]
        assert call_kwargs["max_results"] == 25

    def test_search_empty_query_error(self):
        """Test that empty query returns validation error."""
        from local_deep_research.mcp.server import search

        result = search(query="", engine="wikipedia")

        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"
        assert "empty" in result["error"].lower()

    def test_search_invalid_engine_error(self):
        """Test that unknown engine name returns validation error with available engines."""
        from local_deep_research.mcp.server import search

        mock_engines_config = {
            "wikipedia": {"description": "Wikipedia"},
            "arxiv": {"description": "arXiv"},
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
        ):
            result = search(query="test", engine="nonexistent")

        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"
        assert "nonexistent" in result["error"]
        assert "arxiv" in result["error"]
        assert "wikipedia" in result["error"]

    def test_search_engine_creation_failure(self):
        """Test error when factory returns None."""
        from local_deep_research.mcp.server import search

        mock_engines_config = {
            "searxng": {"description": "SearXNG", "requires_api_key": False},
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
            patch(
                "local_deep_research.web_search_engines.search_engine_factory.create_search_engine",
                return_value=None,
            ),
        ):
            result = search(query="test", engine="searxng")

        assert result["status"] == "error"
        assert "configuration_error" in result["error_type"]

    def test_search_empty_results(self):
        """Test success with empty results list."""
        from local_deep_research.mcp.server import search

        mock_engine = type("MockEngine", (), {"run": lambda self, q: []})()
        mock_engines_config = {
            "arxiv": {"description": "arXiv", "requires_api_key": False},
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
            patch(
                "local_deep_research.web_search_engines.search_engine_factory.create_search_engine",
                return_value=mock_engine,
            ),
        ):
            result = search(query="obscure topic", engine="arxiv")

        assert result["status"] == "success"
        assert result["result_count"] == 0
        assert result["results"] == []

    def test_search_missing_api_key_error(self):
        """Test error when engine requires API key but none is configured."""
        from local_deep_research.mcp.server import search

        mock_engines_config = {
            "brave": {"description": "Brave Search", "requires_api_key": True},
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
        ):
            result = search(query="test", engine="brave")

        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"
        assert "api key" in result["error"].lower()
        assert "brave" in result["error"].lower()

    def test_search_requires_llm_engine_without_llm(self):
        """Test that engines with requires_llm=True work without LLM (degraded mode)."""
        from local_deep_research.mcp.server import search

        mock_engine = type(
            "MockEngine",
            (),
            {
                "run": lambda self, q: [
                    {
                        "title": "PubMed Result",
                        "link": "https://pubmed.ncbi.nlm.nih.gov/123",
                        "snippet": "CRISPR gene therapy results",
                    },
                ]
            },
        )()
        mock_engines_config = {
            "pubmed": {
                "description": "PubMed biomedical literature",
                "requires_api_key": False,
                "requires_llm": True,
            },
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
            patch(
                "local_deep_research.web_search_engines.search_engine_factory.create_search_engine",
                return_value=mock_engine,
            ) as mock_factory,
        ):
            result = search(query="CRISPR gene therapy", engine="pubmed")

        assert result["status"] == "success"
        assert result["engine"] == "pubmed"
        assert result["result_count"] == 1
        assert result["results"][0]["title"] == "PubMed Result"
        # Verify factory was called with llm=None (search tool doesn't use LLM)
        call_kwargs = mock_factory.call_args[1]
        assert call_kwargs["llm"] is None

    def test_search_normalizes_body_to_snippet(self):
        """Test that results with 'body' key get 'snippet' alias."""
        from local_deep_research.mcp.server import search

        mock_engine = type(
            "MockEngine",
            (),
            {
                "run": lambda self, q: [
                    {
                        "title": "Result",
                        "link": "https://example.com",
                        "body": "Body text",
                    },
                ]
            },
        )()
        mock_engines_config = {
            "wikipedia": {
                "description": "Wikipedia",
                "requires_api_key": False,
            },
        }

        with (
            patch(
                "local_deep_research.mcp.server.create_settings_snapshot",
                return_value={},
            ),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.search_config",
                return_value=mock_engines_config,
            ),
            patch(
                "local_deep_research.web_search_engines.search_engine_factory.create_search_engine",
                return_value=mock_engine,
            ),
        ):
            result = search(query="test", engine="wikipedia")

        assert result["status"] == "success"
        assert result["results"][0]["snippet"] == "Body text"


class TestErrorClassification:
    """Tests for error classification logic."""

    def test_classify_service_unavailable(self):
        """Test classification of service unavailable errors."""
        from local_deep_research.mcp.server import _classify_error

        assert (
            _classify_error("Error 503: Service unavailable")
            == "service_unavailable"
        )
        assert (
            _classify_error("The service is unavailable")
            == "service_unavailable"
        )

    def test_classify_model_not_found(self):
        """Test classification of model not found errors."""
        from local_deep_research.mcp.server import _classify_error

        assert (
            _classify_error("Error 404: Model not found") == "model_not_found"
        )
        assert _classify_error("Model llama3 not found") == "model_not_found"

    def test_classify_auth_error(self):
        """Test classification of authentication errors."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Invalid API key") == "auth_error"
        assert _classify_error("Authentication failed") == "auth_error"
        assert _classify_error("Unauthorized access") == "auth_error"

    def test_classify_timeout(self):
        """Test classification of timeout errors."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Request timeout") == "timeout"
        assert _classify_error("Connection timeout after 30s") == "timeout"

    def test_classify_rate_limit(self):
        """Test classification of rate limit errors."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Rate limit exceeded") == "rate_limit"

    def test_classify_unknown(self):
        """Test classification of unknown errors."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Something weird happened") == "unknown"


class TestSettingsOverrides:
    """Tests for settings override building."""

    def test_build_settings_overrides_empty(self):
        """Test building overrides with no parameters."""
        from local_deep_research.mcp.server import _build_settings_overrides

        result = _build_settings_overrides()
        assert result == {}

    def test_build_settings_overrides_all_params(self):
        """Test building overrides with all parameters."""
        from local_deep_research.mcp.server import _build_settings_overrides

        result = _build_settings_overrides(
            search_engine="wikipedia",
            strategy="source-based",
            iterations=3,
            questions_per_iteration=5,
            temperature=0.5,
        )

        assert result["search.tool"] == "wikipedia"
        assert result["search.search_strategy"] == "source-based"
        assert result["search.iterations"] == 3
        assert result["search.questions_per_iteration"] == 5
        assert result["llm.temperature"] == 0.5

    def test_build_settings_overrides_partial(self):
        """Test building overrides with some parameters."""
        from local_deep_research.mcp.server import _build_settings_overrides

        result = _build_settings_overrides(search_engine="arxiv", iterations=2)

        assert result["search.tool"] == "arxiv"
        assert result["search.iterations"] == 2
        assert "search.search_strategy" not in result
        assert "llm.temperature" not in result


class TestMcpToolToResearchFunctionContext:
    """Contract tests: every MCP tool that calls a research function must
    build a settings_snapshot via create_settings_snapshot() and thread it
    to the underlying call. Without this, the user's stored settings (LLM
    provider/model, API keys, embedding model, etc.) are silently ignored
    and the tool falls back to JSON defaults + LDR_* env vars.

    Pattern: capture-list mock (assertions run on a captured kwargs dict
    after the response, not inside the mock itself — the MCP tool's broad
    ``except Exception`` would otherwise swallow AssertionError).
    """

    def test_quick_research_threads_settings_snapshot(self):
        """quick_research → ldr_quick_summary must include settings_snapshot."""
        from local_deep_research.mcp.server import quick_research

        captured = {}

        def _capture(*args, **kwargs):
            captured["args"] = args
            captured["kwargs"] = kwargs
            return {
                "summary": "ok",
                "findings": [],
                "sources": [],
                "iterations": 0,
                "formatted_findings": "",
            }

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            side_effect=_capture,
        ):
            result = quick_research(query="test query")

        assert result["status"] == "success"
        assert "settings_snapshot" in captured["kwargs"], (
            f"ldr_quick_summary called without settings_snapshot. "
            f"kwargs: {list(captured['kwargs'].keys())}"
        )
        assert captured["kwargs"]["settings_snapshot"] is not None

    def test_detailed_research_threads_settings_snapshot(self):
        """detailed_research → ldr_detailed_research must include settings_snapshot."""
        from local_deep_research.mcp.server import detailed_research

        captured = {}

        def _capture(*args, **kwargs):
            captured["args"] = args
            captured["kwargs"] = kwargs
            return {
                "query": "test",
                "research_id": "test-123",
                "summary": "ok",
                "findings": [],
                "iterations": 0,
                "questions": {},
                "formatted_findings": "",
                "sources": [],
                "metadata": {},
            }

        with patch(
            "local_deep_research.mcp.server.ldr_detailed_research",
            side_effect=_capture,
        ):
            result = detailed_research(query="test query")

        assert result["status"] == "success"
        assert "settings_snapshot" in captured["kwargs"], (
            f"ldr_detailed_research called without settings_snapshot. "
            f"kwargs: {list(captured['kwargs'].keys())}"
        )
        assert captured["kwargs"]["settings_snapshot"] is not None

    def test_generate_report_threads_settings_snapshot(self):
        """generate_report → ldr_generate_report must include settings_snapshot."""
        from local_deep_research.mcp.server import generate_report

        captured = {}

        def _capture(*args, **kwargs):
            captured["args"] = args
            captured["kwargs"] = kwargs
            return {"content": "# Report", "metadata": {}}

        with patch(
            "local_deep_research.mcp.server.ldr_generate_report",
            side_effect=_capture,
        ):
            result = generate_report(query="test query")

        assert result["status"] == "success"
        assert "settings_snapshot" in captured["kwargs"], (
            f"ldr_generate_report called without settings_snapshot. "
            f"kwargs: {list(captured['kwargs'].keys())}"
        )
        assert captured["kwargs"]["settings_snapshot"] is not None

    def test_analyze_documents_threads_settings_snapshot(self):
        """analyze_documents → ldr_analyze_documents must include
        settings_snapshot. This is the regression fence for the bug that
        previously omitted user context from the call."""
        from local_deep_research.mcp.server import analyze_documents

        captured = {}

        def _capture(*args, **kwargs):
            captured["args"] = args
            captured["kwargs"] = kwargs
            return {
                "summary": "ok",
                "documents": [],
                "collection": "test",
                "document_count": 0,
            }

        with patch(
            "local_deep_research.mcp.server.ldr_analyze_documents",
            side_effect=_capture,
        ):
            result = analyze_documents(
                query="test query",
                collection_name="test_collection",
            )

        assert result["status"] == "success"
        assert "settings_snapshot" in captured["kwargs"], (
            f"ldr_analyze_documents called without settings_snapshot — "
            f"user-configured embedding model / LLM provider are ignored. "
            f"kwargs: {list(captured['kwargs'].keys())}"
        )
        assert captured["kwargs"]["settings_snapshot"] is not None