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

"""
Edge case tests for MCP server tools.

Tests for boundary conditions, unusual inputs, and error handling.
"""

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 TestQueryEdgeCases:
    """Tests for edge cases in query handling."""

    def test_empty_query(self):
        """Test behavior with empty string query returns validation error."""
        from local_deep_research.mcp.server import quick_research

        # Empty query should be caught by validation
        result = quick_research(query="")

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

    def test_whitespace_only_query(self):
        """Test behavior with whitespace-only query returns validation error."""
        from local_deep_research.mcp.server import quick_research

        # Whitespace-only query should be caught by validation
        result = quick_research(query="   \t\n   ")

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

    def test_very_long_query(self):
        """Test behavior with very long query (>10000 chars) returns validation error."""
        from local_deep_research.mcp.server import quick_research

        long_query = "What is quantum computing? " * 500  # ~13500 chars

        # Query exceeding max length should be caught by validation
        result = quick_research(query=long_query)

        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"
        assert "exceeds maximum length" in result["error"].lower()

    def test_query_with_special_characters(self):
        """Test query with unicode, emojis, special chars."""
        from local_deep_research.mcp.server import quick_research

        special_query = (
            'What is 量子计算? 🔬 Test with émojis & spëcial <chars> "quotes"'
        )

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": "Test",
                "findings": [],
                "sources": [],
                "iterations": 1,
            },
        ):
            result = quick_research(query=special_query)

        assert result["status"] == "success"

    def test_query_with_newlines(self):
        """Test query containing newline characters."""
        from local_deep_research.mcp.server import quick_research

        query_with_newlines = "What is quantum computing?\nHow does it work?\nWhat are its applications?"

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": "Test",
                "findings": [],
                "sources": [],
                "iterations": 1,
            },
        ):
            result = quick_research(query=query_with_newlines)

        assert result["status"] == "success"


class TestParameterEdgeCases:
    """Tests for edge cases in parameter handling.

    Note: The implementation validates parameters upfront and returns
    validation errors for invalid values like 0, negative, or exceeding max.
    """

    def test_iterations_zero(self):
        """Test behavior with iterations=0 returns validation error."""
        from local_deep_research.mcp.server import quick_research

        # iterations=0 is invalid (must be positive integer)
        result = quick_research(query="test", iterations=0)

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

    def test_iterations_negative(self):
        """Test behavior with negative iterations returns validation error."""
        from local_deep_research.mcp.server import quick_research

        # Negative iterations are invalid
        result = quick_research(query="test", iterations=-5)

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

    def test_iterations_very_large(self):
        """Test behavior with very large iterations value returns validation error."""
        from local_deep_research.mcp.server import quick_research

        # iterations=10000 exceeds max of 10 for quick_research
        result = quick_research(query="test", iterations=10000)

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

    def test_questions_per_iteration_zero(self):
        """Test behavior with questions_per_iteration=0 returns validation error."""
        from local_deep_research.mcp.server import quick_research

        # questions_per_iteration=0 is invalid (must be positive integer)
        result = quick_research(query="test", questions_per_iteration=0)

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

    def test_invalid_search_engine_name(self):
        """Test behavior with invalid search engine name."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            side_effect=Exception("Unknown search engine: nonexistent_engine"),
        ):
            result = quick_research(
                query="test", search_engine="nonexistent_engine"
            )

        assert result["status"] == "error"
        assert result[
            "error"
        ]  # Error message is present (sanitized, may not contain engine name)

    def test_invalid_strategy_name(self):
        """Test behavior with invalid strategy name."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            side_effect=Exception("Unknown strategy: fake_strategy"),
        ):
            result = quick_research(query="test", strategy="fake_strategy")

        assert result["status"] == "error"
        assert "fake_strategy" in result["error"]

    def test_max_results_zero(self):
        """Test analyze_documents with max_results=0 returns validation error."""
        from local_deep_research.mcp.server import analyze_documents

        # max_results=0 is invalid (must be positive integer)
        result = analyze_documents(
            query="test", collection_name="test", max_results=0
        )

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

    def test_max_results_negative(self):
        """Test analyze_documents with negative max_results returns validation error."""
        from local_deep_research.mcp.server import analyze_documents

        # Negative max_results is invalid
        result = analyze_documents(
            query="test", collection_name="test", max_results=-10
        )

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

    def test_empty_collection_name(self):
        """Test analyze_documents with empty collection name."""
        from local_deep_research.mcp.server import analyze_documents

        with patch(
            "local_deep_research.mcp.server.ldr_analyze_documents",
            side_effect=Exception("Collection name cannot be empty"),
        ):
            result = analyze_documents(
                query="test", collection_name="", max_results=10
            )

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


class TestReturnValueEdgeCases:
    """Tests for edge cases in return value handling."""

    def test_handles_missing_summary_field(self):
        """Test graceful handling when API returns no summary."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "findings": [],
                "sources": [],
                "iterations": 1,
            },  # No summary
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        assert result["summary"] == ""  # Should default to empty string

    def test_handles_missing_findings_field(self):
        """Test graceful handling when API returns no findings."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": "Test",
                "sources": [],
                "iterations": 1,
            },  # No findings
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        assert result["findings"] == []  # Should default to empty list

    def test_handles_missing_sources_field(self):
        """Test graceful handling when API returns no sources."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": "Test",
                "findings": [],
                "iterations": 1,
            },  # No sources
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        assert result["sources"] == []  # Should default to empty list

    def test_handles_none_summary(self):
        """Test handling when summary is None."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": None,
                "findings": [],
                "sources": [],
                "iterations": 1,
            },
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        # None should be preserved or converted to empty string
        assert result["summary"] is None or result["summary"] == ""

    def test_handles_none_in_findings_list(self):
        """Test handling of None values in findings list."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": "Test",
                "findings": [None, {"content": "valid"}, None],
                "sources": [],
                "iterations": 1,
            },
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        assert len(result["findings"]) == 3  # Should preserve the list as-is

    def test_handles_completely_empty_response(self):
        """Test handling of completely empty API response."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={},  # Empty dict
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        assert result["summary"] == ""
        assert result["findings"] == []
        assert result["sources"] == []
        assert result["iterations"] == 0

    def test_handles_unexpected_extra_fields(self):
        """Test that unexpected fields in API response don't break the tool."""
        from local_deep_research.mcp.server import quick_research

        with patch(
            "local_deep_research.mcp.server.ldr_quick_summary",
            return_value={
                "summary": "Test",
                "findings": [],
                "sources": [],
                "iterations": 1,
                "unexpected_field": "some value",
                "another_unexpected": {"nested": "data"},
            },
        ):
            result = quick_research(query="test")

        assert result["status"] == "success"
        assert result["summary"] == "Test"


class TestDetailedResearchEdgeCases:
    """Edge case tests specific to detailed_research."""

    def test_handles_missing_research_id(self):
        """Test handling when research_id is missing."""
        from local_deep_research.mcp.server import detailed_research

        with patch(
            "local_deep_research.mcp.server.ldr_detailed_research",
            return_value={
                "query": "test",
                "summary": "Test",
                "findings": [],
                "sources": [],
                "iterations": 1,
            },
        ):
            result = detailed_research(query="test")

        assert result["status"] == "success"
        assert result["research_id"] == ""  # Should default to empty string

    def test_handles_missing_metadata(self):
        """Test handling when metadata is missing."""
        from local_deep_research.mcp.server import detailed_research

        with patch(
            "local_deep_research.mcp.server.ldr_detailed_research",
            return_value={
                "query": "test",
                "research_id": "123",
                "summary": "Test",
                "findings": [],
                "sources": [],
                "iterations": 1,
            },
        ):
            result = detailed_research(query="test")

        assert result["status"] == "success"
        assert result["metadata"] == {}  # Should default to empty dict


class TestGenerateReportEdgeCases:
    """Edge case tests specific to generate_report."""

    def test_handles_missing_content(self):
        """Test handling when report content is missing."""
        from local_deep_research.mcp.server import generate_report

        with patch(
            "local_deep_research.mcp.server.ldr_generate_report",
            return_value={"metadata": {"query": "test"}},  # No content
        ):
            result = generate_report(query="test")

        assert result["status"] == "success"
        assert result["content"] == ""  # Should default to empty string

    def test_searches_per_section_zero(self):
        """Test generate_report with searches_per_section=0 returns validation error."""
        from local_deep_research.mcp.server import generate_report

        # searches_per_section=0 is invalid (must be positive integer)
        result = generate_report(query="test", searches_per_section=0)

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

    def test_searches_per_section_very_large(self):
        """Test generate_report with very large searches_per_section returns validation error."""
        from local_deep_research.mcp.server import generate_report

        # searches_per_section=1000 exceeds max of 10
        result = generate_report(query="test", searches_per_section=1000)

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


class TestConnectionErrorClassification:
    """Tests for connection error classification."""

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

        assert _classify_error("Connection refused") == "connection_error"
        assert (
            _classify_error("Failed to establish connection")
            == "connection_error"
        )
        assert _classify_error("Connection reset by peer") == "connection_error"

    def test_classify_combined_errors(self):
        """Test error classification with multiple error patterns."""
        from local_deep_research.mcp.server import _classify_error

        # Service unavailable should take precedence
        assert (
            _classify_error("Connection failed: 503 Service unavailable")
            == "service_unavailable"
        )
        # Timeout should be detected
        assert (
            _classify_error("Connection timeout after 30 seconds") == "timeout"
        )


class TestMCPClientEdgeCases:
    """Edge case tests for MCP client result handling."""

    def test_result_content_none_handling(self):
        """Test client handles None result.content gracefully."""
        from unittest.mock import MagicMock

        # Create a mock result with None content
        mock_result = MagicMock()
        mock_result.isError = False
        mock_result.content = None

        # Verify the handling logic works with None
        content = []
        if mock_result.content is not None and hasattr(
            mock_result.content, "__iter__"
        ):
            for item in mock_result.content:
                content.append(str(item))
        elif mock_result.content is not None:
            content.append(str(mock_result.content))

        # Should produce empty content list
        assert content == []

    def test_result_content_non_iterable_handling(self):
        """Test client handles non-iterable result.content gracefully."""
        from unittest.mock import MagicMock

        # Create a mock result with non-iterable content (e.g., an integer)
        mock_result = MagicMock()
        mock_result.isError = False
        mock_result.content = 42  # Non-iterable

        # Verify the handling logic works with non-iterable
        content = []
        if mock_result.content is not None and hasattr(
            mock_result.content, "__iter__"
        ):
            for item in mock_result.content:
                content.append(str(item))
        elif mock_result.content is not None:
            content.append(str(mock_result.content))

        # Should convert to string
        assert content == ["42"]

    def test_result_content_string_handling(self):
        """Test client handles string result.content (which is iterable) correctly."""
        from unittest.mock import MagicMock

        # Create a mock result with string content
        # Note: strings are iterable but we want to treat them as a single value
        mock_result = MagicMock()
        mock_result.isError = False
        mock_result.content = "test string"

        content = []
        if mock_result.content is not None and hasattr(
            mock_result.content, "__iter__"
        ):
            # String has __iter__ but we iterate over characters
            # This tests the current behavior - strings will be iterated
            for item in mock_result.content:
                if hasattr(item, "text"):
                    content.append(item.text)
                elif hasattr(item, "data"):
                    content.append(str(item.data))
                else:
                    content.append(str(item))

        # Each character becomes a string
        assert len(content) == len("test string")

    def test_result_content_list_with_text_items(self):
        """Test client handles list of items with text attribute."""
        from unittest.mock import MagicMock

        # Create mock items with text attribute
        item1 = MagicMock()
        item1.text = "First text"

        item2 = MagicMock()
        item2.text = "Second text"

        mock_result = MagicMock()
        mock_result.isError = False
        mock_result.content = [item1, item2]

        content = []
        if mock_result.content is not None and hasattr(
            mock_result.content, "__iter__"
        ):
            for item in mock_result.content:
                if hasattr(item, "text"):
                    content.append(item.text)
                elif hasattr(item, "data"):
                    content.append(str(item.data))
                else:
                    content.append(str(item))

        assert content == ["First text", "Second text"]