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

"""Tests for pure validation functions in local_deep_research.mcp.server."""

from unittest.mock import patch

import pytest

# Skip all tests if MCP package is not installed
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"
)


def _import_server():
    """Lazy import to avoid collection-time failures when mcp is missing."""
    from local_deep_research.mcp.server import (
        ValidationError,
        _COLLECTION_NAME_RE,
        _build_settings_overrides,
        _classify_error,
        _validate_iterations,
        _validate_max_results,
        _validate_query,
        _validate_questions_per_iteration,
        _validate_search_engine,
    )

    return {
        "ValidationError": ValidationError,
        "_COLLECTION_NAME_RE": _COLLECTION_NAME_RE,
        "_build_settings_overrides": _build_settings_overrides,
        "_classify_error": _classify_error,
        "_validate_iterations": _validate_iterations,
        "_validate_max_results": _validate_max_results,
        "_validate_query": _validate_query,
        "_validate_questions_per_iteration": _validate_questions_per_iteration,
        "_validate_search_engine": _validate_search_engine,
    }


# ---------------------------------------------------------------------------
# _classify_error
# ---------------------------------------------------------------------------


class TestClassifyError:
    """Tests for _classify_error."""

    @pytest.mark.parametrize(
        "msg, expected",
        [
            ("503 Service Unavailable", "service_unavailable"),
            ("server unavailable", "service_unavailable"),
            ("404 Not Found", "model_not_found"),
            ("resource not found", "model_not_found"),
            ("API key invalid", "auth_error"),
            ("authentication failed", "auth_error"),
            ("unauthorized access", "auth_error"),
            ("401 Unauthorized", "auth_error"),
            ("connection timeout", "timeout"),
            ("request timed out", "timeout"),
            ("rate limit exceeded", "rate_limit"),
            ("429 Too Many", "rate_limit"),
            ("connection refused", "connection_error"),
            ("validation failed", "validation_error"),
            ("invalid parameter", "validation_error"),
            ("some other error", "unknown"),
            ("", "unknown"),
        ],
    )
    def test_classification(self, msg, expected):
        srv = _import_server()
        assert srv["_classify_error"](msg) == expected


# ---------------------------------------------------------------------------
# _validate_query
# ---------------------------------------------------------------------------


class TestValidateQuery:
    """Tests for _validate_query."""

    def test_empty_string_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="empty"):
            srv["_validate_query"]("")

    def test_whitespace_only_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="empty"):
            srv["_validate_query"]("   ")

    def test_none_raises(self):
        srv = _import_server()
        with pytest.raises((AttributeError, srv["ValidationError"])):
            srv["_validate_query"](None)

    def test_valid_query_returned(self):
        srv = _import_server()
        assert srv["_validate_query"]("valid query") == "valid query"

    def test_strips_whitespace(self):
        srv = _import_server()
        assert srv["_validate_query"]("  padded  ") == "padded"

    def test_exceeds_max_length(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="exceeds maximum"):
            srv["_validate_query"]("x" * 10001)

    def test_exactly_at_max_length(self):
        srv = _import_server()
        query = "x" * 10000
        assert srv["_validate_query"](query) == query


# ---------------------------------------------------------------------------
# _validate_iterations
# ---------------------------------------------------------------------------


class TestValidateIterations:
    """Tests for _validate_iterations."""

    def test_none_returns_none(self):
        srv = _import_server()
        assert srv["_validate_iterations"](None) is None

    def test_valid_value(self):
        srv = _import_server()
        assert srv["_validate_iterations"](5) == 5

    def test_zero_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="positive integer"):
            srv["_validate_iterations"](0)

    def test_negative_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="positive integer"):
            srv["_validate_iterations"](-1)

    def test_exceeds_default_max(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="cannot exceed 20"):
            srv["_validate_iterations"](21)

    def test_at_default_max(self):
        srv = _import_server()
        assert srv["_validate_iterations"](20) == 20

    def test_minimum_value(self):
        srv = _import_server()
        assert srv["_validate_iterations"](1) == 1

    def test_string_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"]):
            srv["_validate_iterations"]("5")

    def test_float_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"]):
            srv["_validate_iterations"](1.5)

    def test_custom_max_val(self):
        srv = _import_server()
        assert srv["_validate_iterations"](10, max_val=10) == 10
        with pytest.raises(srv["ValidationError"], match="cannot exceed 10"):
            srv["_validate_iterations"](11, max_val=10)


# ---------------------------------------------------------------------------
# _validate_questions_per_iteration
# ---------------------------------------------------------------------------


class TestValidateQuestionsPerIteration:
    """Tests for _validate_questions_per_iteration."""

    def test_none_returns_none(self):
        srv = _import_server()
        assert srv["_validate_questions_per_iteration"](None) is None

    def test_valid_value(self):
        srv = _import_server()
        assert srv["_validate_questions_per_iteration"](5) == 5

    def test_zero_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"]):
            srv["_validate_questions_per_iteration"](0)

    def test_exceeds_max(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="cannot exceed 10"):
            srv["_validate_questions_per_iteration"](11)

    def test_at_max(self):
        srv = _import_server()
        assert srv["_validate_questions_per_iteration"](10) == 10


# ---------------------------------------------------------------------------
# _validate_max_results
# ---------------------------------------------------------------------------


class TestValidateMaxResults:
    """Tests for _validate_max_results."""

    def test_valid_value(self):
        srv = _import_server()
        assert srv["_validate_max_results"](10) == 10

    def test_zero_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"]):
            srv["_validate_max_results"](0)

    def test_negative_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"]):
            srv["_validate_max_results"](-1)

    def test_exceeds_max(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"], match="cannot exceed 100"):
            srv["_validate_max_results"](101)

    def test_at_max(self):
        srv = _import_server()
        assert srv["_validate_max_results"](100) == 100

    def test_string_raises(self):
        srv = _import_server()
        with pytest.raises(srv["ValidationError"]):
            srv["_validate_max_results"]("10")


# ---------------------------------------------------------------------------
# _build_settings_overrides
# ---------------------------------------------------------------------------


class TestBuildSettingsOverrides:
    """Tests for _build_settings_overrides."""

    @patch(
        "local_deep_research.mcp.server._validate_strategy",
        return_value="standard",
    )
    @patch(
        "local_deep_research.mcp.server._validate_search_engine",
        return_value="google",
    )
    def test_all_overrides(self, mock_engine, mock_strategy):
        srv = _import_server()
        result = srv["_build_settings_overrides"](
            search_engine="google",
            strategy="standard",
            iterations=5,
            questions_per_iteration=3,
            temperature=0.7,
        )
        assert result["search.tool"] == "google"
        assert result["search.search_strategy"] == "standard"
        assert result["search.iterations"] == 5
        assert result["search.questions_per_iteration"] == 3
        assert result["llm.temperature"] == 0.7

    @patch(
        "local_deep_research.mcp.server._validate_strategy",
        return_value=None,
    )
    @patch(
        "local_deep_research.mcp.server._validate_search_engine",
        return_value=None,
    )
    def test_all_none_returns_empty(self, mock_engine, mock_strategy):
        srv = _import_server()
        result = srv["_build_settings_overrides"]()
        assert result == {}

    @patch(
        "local_deep_research.mcp.server._validate_strategy",
        return_value=None,
    )
    @patch(
        "local_deep_research.mcp.server._validate_search_engine",
        return_value=None,
    )
    def test_only_some_params(self, mock_engine, mock_strategy):
        srv = _import_server()
        result = srv["_build_settings_overrides"](iterations=3, temperature=0.5)
        assert result == {
            "search.iterations": 3,
            "llm.temperature": 0.5,
        }
        assert "search.tool" not in result
        assert "search.search_strategy" not in result
        assert "search.questions_per_iteration" not in result

    @patch(
        "local_deep_research.mcp.server._validate_strategy",
        return_value=None,
    )
    @patch(
        "local_deep_research.mcp.server._validate_search_engine",
        return_value=None,
    )
    def test_engine_validates_to_none_not_added(
        self, mock_engine, mock_strategy
    ):
        srv = _import_server()
        result = srv["_build_settings_overrides"](search_engine="")
        assert "search.tool" not in result


# ---------------------------------------------------------------------------
# _COLLECTION_NAME_RE
# ---------------------------------------------------------------------------


class TestCollectionNameRegex:
    """Tests for _COLLECTION_NAME_RE collection name validation."""

    def test_path_traversal_rejected(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("../etc/passwd") is None

    def test_too_long_rejected(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("a" * 101) is None

    def test_valid_name_with_hyphens_and_underscores(self):
        srv = _import_server()
        assert (
            srv["_COLLECTION_NAME_RE"].match("my-collection_name") is not None
        )

    def test_valid_alphanumeric_with_spaces(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("My Collection 42") is not None

    def test_exactly_100_chars_accepted(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("a" * 100) is not None

    def test_empty_string_rejected(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("") is None

    def test_special_chars_rejected(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("col;drop table") is None

    def test_tab_rejected(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("col\tname") is None

    def test_newline_rejected(self):
        srv = _import_server()
        assert srv["_COLLECTION_NAME_RE"].match("col\nname") is None


# ---------------------------------------------------------------------------
# _validate_search_engine
# ---------------------------------------------------------------------------


class TestValidateSearchEngine:
    """Tests for _validate_search_engine."""

    def test_none_returns_none(self):
        srv = _import_server()
        assert srv["_validate_search_engine"](None) is None

    def test_empty_string_returns_none(self):
        srv = _import_server()
        assert srv["_validate_search_engine"]("") is None

    def test_whitespace_only_returns_none(self):
        srv = _import_server()
        assert srv["_validate_search_engine"]("   ") is None

    @patch(
        "local_deep_research.web_search_engines.search_engines_config.search_config",
        side_effect=RuntimeError("config unavailable"),
    )
    @patch("local_deep_research.mcp.server.create_settings_snapshot")
    def test_config_load_failure_raises_validation_error(
        self, mock_settings, mock_config
    ):
        srv = _import_server()
        with pytest.raises(
            srv["ValidationError"], match="engine configuration unavailable"
        ):
            srv["_validate_search_engine"]("some_engine")