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

"""
MCP protocol tests for the MCP server.

Tests for server setup, tool registration, and logging configuration.
"""

import sys
import io
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 TestMCPServerSetup:
    """Tests for MCP server initialization and configuration."""

    def test_mcp_server_instance_created(self):
        """Verify FastMCP server instance is created correctly."""
        from local_deep_research.mcp.server import mcp

        assert mcp is not None
        assert mcp.name == "local-deep-research"

    def test_mcp_server_instructions_set(self):
        """Verify server instructions/description is set."""
        from local_deep_research.mcp.server import mcp

        # FastMCP stores instructions
        assert mcp.instructions is not None
        assert "research" in mcp.instructions.lower()

    def test_run_server_function_exists(self):
        """Verify run_server function is exported."""
        from local_deep_research.mcp import run_server

        assert callable(run_server)


class TestMCPToolRegistration:
    """Tests for MCP tool decorators and registration."""

    def test_all_expected_tools_are_defined(self):
        """Verify all 7 expected tools are defined as functions."""
        from local_deep_research.mcp.server import (
            quick_research,
            detailed_research,
            generate_report,
            analyze_documents,
            list_search_engines,
            list_strategies,
            get_configuration,
        )

        # All tools should be callable
        assert callable(quick_research)
        assert callable(detailed_research)
        assert callable(generate_report)
        assert callable(analyze_documents)
        assert callable(list_search_engines)
        assert callable(list_strategies)
        assert callable(get_configuration)

    def test_quick_research_has_docstring(self):
        """Verify quick_research has documentation."""
        from local_deep_research.mcp.server import quick_research

        assert quick_research.__doc__ is not None
        assert len(quick_research.__doc__) > 50  # Substantial docstring
        assert "query" in quick_research.__doc__.lower()

    def test_detailed_research_has_docstring(self):
        """Verify detailed_research has documentation."""
        from local_deep_research.mcp.server import detailed_research

        assert detailed_research.__doc__ is not None
        assert "detailed" in detailed_research.__doc__.lower()

    def test_generate_report_has_docstring(self):
        """Verify generate_report has documentation."""
        from local_deep_research.mcp.server import generate_report

        assert generate_report.__doc__ is not None
        assert "report" in generate_report.__doc__.lower()

    def test_analyze_documents_has_docstring(self):
        """Verify analyze_documents has documentation."""
        from local_deep_research.mcp.server import analyze_documents

        assert analyze_documents.__doc__ is not None
        assert "document" in analyze_documents.__doc__.lower()

    def test_list_search_engines_has_docstring(self):
        """Verify list_search_engines has documentation."""
        from local_deep_research.mcp.server import list_search_engines

        assert list_search_engines.__doc__ is not None
        assert "search" in list_search_engines.__doc__.lower()

    def test_list_strategies_has_docstring(self):
        """Verify list_strategies has documentation."""
        from local_deep_research.mcp.server import list_strategies

        assert list_strategies.__doc__ is not None
        assert "strateg" in list_strategies.__doc__.lower()

    def test_get_configuration_has_docstring(self):
        """Verify get_configuration has documentation."""
        from local_deep_research.mcp.server import get_configuration

        assert get_configuration.__doc__ is not None
        assert "config" in get_configuration.__doc__.lower()


class TestMCPLogging:
    """Tests for MCP logging configuration (critical for STDIO)."""

    def test_no_stdout_pollution_from_list_strategies(self):
        """Verify list_strategies doesn't write to stdout."""
        from local_deep_research.mcp.server import list_strategies

        # Capture stdout
        old_stdout = sys.stdout
        sys.stdout = captured_stdout = io.StringIO()

        try:
            result = list_strategies()
            stdout_output = captured_stdout.getvalue()
        finally:
            sys.stdout = old_stdout

        # Should have no stdout output (MCP uses stdout for JSON-RPC)
        assert stdout_output == "", f"Unexpected stdout output: {stdout_output}"
        assert result["status"] == "success"

    def test_no_stdout_pollution_from_get_configuration(self):
        """Verify get_configuration doesn't write to stdout."""
        from local_deep_research.mcp.server import get_configuration

        # Capture stdout
        old_stdout = sys.stdout
        sys.stdout = captured_stdout = io.StringIO()

        try:
            result = get_configuration()
            stdout_output = captured_stdout.getvalue()
        finally:
            sys.stdout = old_stdout

        # Should have no stdout output
        assert stdout_output == "", f"Unexpected stdout output: {stdout_output}"
        assert result["status"] == "success"

    def test_no_stdout_pollution_during_error(self):
        """Verify error handling doesn't write to stdout."""
        from local_deep_research.mcp.server import quick_research

        # Capture stdout
        old_stdout = sys.stdout
        sys.stdout = captured_stdout = io.StringIO()

        try:
            with patch(
                "local_deep_research.mcp.server.ldr_quick_summary",
                side_effect=Exception("Test error"),
            ):
                result = quick_research(query="test")
            stdout_output = captured_stdout.getvalue()
        finally:
            sys.stdout = old_stdout

        # Should have no stdout output even during errors
        assert stdout_output == "", f"Unexpected stdout output: {stdout_output}"
        assert result["status"] == "error"


class TestAvailableStrategies:
    """Tests for the get_available_strategies function.

    MCP server exposes all strategies (show_all=True), so these tests
    validate the full list.
    """

    def test_available_strategies_is_list(self):
        """Verify get_available_strategies returns a list."""
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        strategies = get_available_strategies()
        assert isinstance(strategies, list)

    def test_available_strategies_has_entries(self):
        """Verify get_available_strategies has multiple entries."""
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        strategies = get_available_strategies()
        assert len(strategies) >= 5

    def test_available_strategies_have_required_fields(self):
        """Verify each strategy has name and description."""
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        strategies = get_available_strategies()
        for strategy in strategies:
            assert "name" in strategy, f"Strategy missing 'name': {strategy}"
            assert "description" in strategy, (
                f"Strategy missing 'description': {strategy}"
            )
            assert isinstance(strategy["name"], str)
            assert isinstance(strategy["description"], str)
            assert len(strategy["name"]) > 0
            assert len(strategy["description"]) > 0

    def test_source_based_strategy_exists(self):
        """Verify source-based strategy is in the list."""
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        strategies = get_available_strategies()
        strategy_names = [s["name"] for s in strategies]
        assert "source-based" in strategy_names

    def test_focused_iteration_strategy_exists(self):
        """Verify focused-iteration strategy is in the list."""
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        strategies = get_available_strategies()
        strategy_names = [s["name"] for s in strategies]
        assert "focused-iteration" in strategy_names


class TestHelperFunctions:
    """Tests for helper functions in server.py."""

    def test_classify_error_is_callable(self):
        """Verify _classify_error function exists and is callable."""
        from local_deep_research.mcp.server import _classify_error

        assert callable(_classify_error)

    def test_build_settings_overrides_is_callable(self):
        """Verify _build_settings_overrides function exists and is callable."""
        from local_deep_research.mcp.server import _build_settings_overrides

        assert callable(_build_settings_overrides)

    def test_classify_error_returns_string(self):
        """Verify _classify_error returns a string."""
        from local_deep_research.mcp.server import _classify_error

        result = _classify_error("Some error")
        assert isinstance(result, str)

    def test_build_settings_overrides_returns_dict(self):
        """Verify _build_settings_overrides returns a dict."""
        from local_deep_research.mcp.server import _build_settings_overrides

        result = _build_settings_overrides()
        assert isinstance(result, dict)