"""Tests for MCP tools (search and fetch)."""
import pytest
from mcp_server_builder.tools.docs import fetch_mcp_doc, search_mcp_docs
from mcp_server_builder.utils import cache
@pytest.fixture(scope="module", autouse=True)
def ensure_cache() -> None:
"""Ensure cache is initialized before tests."""
cache.ensure_ready()
class TestSearchMcpDocs:
"""Tests for search_mcp_docs tool."""
def test_returns_list(self) -> None:
"""Test that search returns a list."""
results = search_mcp_docs("tools")
assert isinstance(results, list)
def test_results_have_required_fields(self) -> None:
"""Test that results contain required fields."""
results = search_mcp_docs("mcp protocol", k=1)
if results:
result = results[0]
assert "url" in result
assert "title" in result
assert "score" in result
assert "snippet" in result
def test_respects_k_parameter(self) -> None:
"""Test that k parameter limits results."""
results = search_mcp_docs("documentation", k=3)
assert len(results) <= 3
def test_score_is_numeric(self) -> None:
"""Test that score is a number."""
results = search_mcp_docs("transport", k=1)
if results:
assert isinstance(results[0]["score"], (int, float))
def test_empty_query(self) -> None:
"""Test handling of empty query."""
results = search_mcp_docs("")
assert isinstance(results, list)
def test_returns_relevant_results(self) -> None:
"""Test that results are relevant to query."""
results = search_mcp_docs("FastMCP decorator", k=5)
# At least one result should mention FastMCP or decorator
if results:
combined = " ".join(r["title"].lower() + " " + r["snippet"].lower() for r in results)
assert "fastmcp" in combined or "decor" in combined or "tool" in combined
class TestFetchMcpDoc:
"""Tests for fetch_mcp_doc tool."""
def test_fetch_valid_url(self) -> None:
"""Test fetching a valid documentation URL."""
result = fetch_mcp_doc("https://modelcontextprotocol.io/docs/concepts/tools")
assert "url" in result
assert "title" in result
# Should have content or error
assert "content" in result or "error" in result
def test_fetch_returns_content(self) -> None:
"""Test that fetch returns actual content."""
result = fetch_mcp_doc("https://modelcontextprotocol.io/docs/concepts/tools")
if "content" in result:
assert len(result["content"]) > 0
def test_fetch_invalid_url_returns_error(self) -> None:
"""Test that invalid URLs return error response."""
result = fetch_mcp_doc("https://invalid-domain-xyz.com/page")
assert "error" in result
assert "url" in result
def test_fetch_disallowed_domain(self) -> None:
"""Test that disallowed domains are rejected."""
result = fetch_mcp_doc("https://evil.com/malware")
assert "error" in result
class TestCacheIntegration:
"""Tests for cache integration with tools."""
def test_index_is_populated(self) -> None:
"""Test that the index has documents after initialization."""
index = cache.get_index()
assert index is not None
assert len(index.docs) > 0
def test_url_titles_populated(self) -> None:
"""Test that URL titles are populated from llms.txt."""
titles = cache.get_url_titles()
assert len(titles) > 0
def test_multiple_ensure_ready_calls(self) -> None:
"""Test that ensure_ready is idempotent."""
cache.ensure_ready()
index1 = cache.get_index()
cache.ensure_ready()
index2 = cache.get_index()
assert index1 is index2 # Same instance