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

"""
Tests for MCP server validation functions.

These tests verify input validation for all MCP tool parameters.
"""

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 TestValidationFunctions:
    """Tests for parameter validation helper functions."""

    def test_validate_query_valid(self):
        """Test valid query passes validation."""
        from local_deep_research.mcp.server import _validate_query

        result = _validate_query("What is quantum computing?")
        assert result == "What is quantum computing?"

    def test_validate_query_strips_whitespace(self):
        """Test query is stripped of whitespace."""
        from local_deep_research.mcp.server import _validate_query

        result = _validate_query("  test query  ")
        assert result == "test query"

    def test_validate_query_empty_raises(self):
        """Test empty query raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_query,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="cannot be empty"):
            _validate_query("")

    def test_validate_query_whitespace_only_raises(self):
        """Test whitespace-only query raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_query,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="cannot be empty"):
            _validate_query("   ")

    def test_validate_query_too_long_raises(self):
        """Test query exceeding max length raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_query,
            ValidationError,
        )

        long_query = "x" * 10001
        with pytest.raises(ValidationError, match="exceeds maximum length"):
            _validate_query(long_query)

    def test_validate_query_at_max_length(self):
        """Test query at exactly max length is valid."""
        from local_deep_research.mcp.server import _validate_query

        max_query = "x" * 10000
        result = _validate_query(max_query)
        assert len(result) == 10000

    def test_validate_iterations_valid(self):
        """Test valid iterations pass validation."""
        from local_deep_research.mcp.server import _validate_iterations

        assert _validate_iterations(5) == 5
        assert _validate_iterations(1) == 1
        assert _validate_iterations(20) == 20

    def test_validate_iterations_none_allowed(self):
        """Test None iterations returns None."""
        from local_deep_research.mcp.server import _validate_iterations

        assert _validate_iterations(None) is None

    def test_validate_iterations_zero_raises(self):
        """Test zero iterations raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_iterations,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="positive integer"):
            _validate_iterations(0)

    def test_validate_iterations_negative_raises(self):
        """Test negative iterations raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_iterations,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="positive integer"):
            _validate_iterations(-5)

    def test_validate_iterations_exceeds_max_raises(self):
        """Test iterations exceeding max raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_iterations,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="cannot exceed"):
            _validate_iterations(21, max_val=20)

    def test_validate_iterations_custom_max(self):
        """Test iterations with custom max value."""
        from local_deep_research.mcp.server import (
            _validate_iterations,
            ValidationError,
        )

        assert _validate_iterations(5, max_val=10) == 5
        with pytest.raises(ValidationError, match="cannot exceed 10"):
            _validate_iterations(11, max_val=10)

    def test_validate_questions_per_iteration_valid(self):
        """Test valid questions_per_iteration passes validation."""
        from local_deep_research.mcp.server import (
            _validate_questions_per_iteration,
        )

        assert _validate_questions_per_iteration(3) == 3
        assert _validate_questions_per_iteration(1) == 1
        assert _validate_questions_per_iteration(10) == 10

    def test_validate_questions_per_iteration_none_allowed(self):
        """Test None questions_per_iteration returns None."""
        from local_deep_research.mcp.server import (
            _validate_questions_per_iteration,
        )

        assert _validate_questions_per_iteration(None) is None

    def test_validate_questions_per_iteration_zero_raises(self):
        """Test zero questions_per_iteration raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_questions_per_iteration,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="positive integer"):
            _validate_questions_per_iteration(0)

    def test_validate_questions_per_iteration_exceeds_max_raises(self):
        """Test questions_per_iteration exceeding max raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_questions_per_iteration,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="cannot exceed 10"):
            _validate_questions_per_iteration(11)

    def test_validate_max_results_valid(self):
        """Test valid max_results passes validation."""
        from local_deep_research.mcp.server import _validate_max_results

        assert _validate_max_results(10) == 10
        assert _validate_max_results(1) == 1
        assert _validate_max_results(100) == 100

    def test_validate_max_results_zero_raises(self):
        """Test zero max_results raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_max_results,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="positive integer"):
            _validate_max_results(0)

    def test_validate_max_results_exceeds_max_raises(self):
        """Test max_results exceeding max raises ValidationError."""
        from local_deep_research.mcp.server import (
            _validate_max_results,
            ValidationError,
        )

        with pytest.raises(ValidationError, match="cannot exceed 100"):
            _validate_max_results(101)


class TestValidationInTools:
    """Tests for validation being applied in tool functions."""

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

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

    def test_quick_research_invalid_iterations_returns_error(self):
        """Test quick_research with invalid iterations returns validation error."""
        from local_deep_research.mcp.server import quick_research

        result = quick_research(query="Test query", iterations=0)
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_quick_research_iterations_exceeds_max_returns_error(self):
        """Test quick_research with iterations exceeding max returns validation error."""
        from local_deep_research.mcp.server import quick_research

        result = quick_research(query="Test query", iterations=100)
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_detailed_research_empty_query_returns_error(self):
        """Test detailed_research with empty query returns validation error."""
        from local_deep_research.mcp.server import detailed_research

        result = detailed_research(query="   ")
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_generate_report_empty_query_returns_error(self):
        """Test generate_report with empty query returns validation error."""
        from local_deep_research.mcp.server import generate_report

        result = generate_report(query="")
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_generate_report_invalid_searches_per_section(self):
        """Test generate_report with invalid searches_per_section returns error."""
        from local_deep_research.mcp.server import generate_report

        result = generate_report(query="Test query", searches_per_section=0)
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_generate_report_searches_per_section_exceeds_max(self):
        """Test generate_report with searches_per_section exceeding max returns error."""
        from local_deep_research.mcp.server import generate_report

        result = generate_report(query="Test query", searches_per_section=11)
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_analyze_documents_empty_query_returns_error(self):
        """Test analyze_documents with empty query returns validation error."""
        from local_deep_research.mcp.server import analyze_documents

        result = analyze_documents(query="", collection_name="test")
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_analyze_documents_empty_collection_returns_error(self):
        """Test analyze_documents with empty collection returns validation error."""
        from local_deep_research.mcp.server import analyze_documents

        result = analyze_documents(query="Test", collection_name="")
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"

    def test_analyze_documents_invalid_max_results_returns_error(self):
        """Test analyze_documents with invalid max_results returns error."""
        from local_deep_research.mcp.server import analyze_documents

        result = analyze_documents(
            query="Test", collection_name="test", max_results=0
        )
        assert result["status"] == "error"
        assert result["error_type"] == "validation_error"


class TestErrorClassification:
    """Tests for the _classify_error function."""

    def test_classify_service_unavailable(self):
        """Test 503 error is classified as service_unavailable."""
        from local_deep_research.mcp.server import _classify_error

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

    def test_classify_model_not_found(self):
        """Test 404 error is classified as model_not_found."""
        from local_deep_research.mcp.server import _classify_error

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

    def test_classify_auth_error(self):
        """Test auth errors are classified correctly."""
        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"
        assert _classify_error("Error 401: Unauthorized") == "auth_error"

    def test_classify_timeout(self):
        """Test timeout errors are classified correctly."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Request timeout") == "timeout"
        # "Connection timed out" matches "timed out" check which comes before
        # the "connection" check, so it returns "timeout"
        assert _classify_error("Connection timed out") == "timeout"
        # "Operation timed out" also matches "timed out"
        assert _classify_error("Operation timed out") == "timeout"
        assert _classify_error("A timeout occurred") == "timeout"

    def test_classify_rate_limit(self):
        """Test rate limit errors are classified correctly."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Rate limit exceeded") == "rate_limit"
        assert _classify_error("Error 429: Too many requests") == "rate_limit"

    def test_classify_connection_error(self):
        """Test connection errors are classified correctly."""
        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"
        )

    def test_classify_validation_error(self):
        """Test validation errors are classified correctly."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Validation failed") == "validation_error"
        assert _classify_error("Invalid parameter") == "validation_error"

    def test_classify_unknown_error(self):
        """Test unknown errors are classified as unknown."""
        from local_deep_research.mcp.server import _classify_error

        assert _classify_error("Some random error") == "unknown"
        assert _classify_error("Unexpected exception") == "unknown"


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

    Note: These tests don't require the MCP package as they test the factory module.
    """

    @pytest.mark.skipif(
        True, reason="Tested in test_search_system_factory.py instead"
    )
    def test_get_available_strategies_returns_list(self):
        """Test get_available_strategies returns a list."""
        pass

    def test_list_strategies_uses_get_available_strategies(self):
        """Test MCP list_strategies tool uses the factory function."""
        from local_deep_research.mcp.server import list_strategies
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        result = list_strategies()
        expected = get_available_strategies()

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