"""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")