"""Tests for MCP server handlers."""
import json
from local_deepwiki.handlers import (
handle_ask_question,
handle_export_wiki_html,
handle_index_repository,
handle_read_wiki_page,
handle_read_wiki_structure,
handle_search_code,
)
class TestHandleIndexRepository:
"""Tests for handle_index_repository handler."""
async def test_returns_error_for_nonexistent_path(self, tmp_path):
"""Test error returned for non-existent repository path."""
nonexistent = tmp_path / "does_not_exist"
result = await handle_index_repository({"repo_path": str(nonexistent)})
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "does not exist" in result[0].text
async def test_returns_error_for_file_path(self, tmp_path):
"""Test error returned when path is a file, not directory."""
file_path = tmp_path / "test.txt"
file_path.write_text("content")
result = await handle_index_repository({"repo_path": str(file_path)})
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "not a directory" in result[0].text
async def test_returns_error_for_invalid_language(self, tmp_path):
"""Test error returned for invalid language filter."""
result = await handle_index_repository(
{
"repo_path": str(tmp_path),
"languages": ["python", "invalid_lang"],
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "Invalid languages" in result[0].text
async def test_returns_error_for_invalid_llm_provider(self, tmp_path):
"""Test error returned for invalid LLM provider."""
result = await handle_index_repository(
{
"repo_path": str(tmp_path),
"llm_provider": "invalid_provider",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
# Pydantic enum validation error
assert "llm_provider" in result[0].text or "Input should be" in result[0].text
async def test_returns_error_for_invalid_embedding_provider(self, tmp_path):
"""Test error returned for invalid embedding provider."""
result = await handle_index_repository(
{
"repo_path": str(tmp_path),
"embedding_provider": "invalid_provider",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
# Pydantic enum validation error
assert (
"embedding_provider" in result[0].text
or "Input should be" in result[0].text
)
class TestHandleAskQuestion:
"""Tests for handle_ask_question handler."""
async def test_returns_error_for_empty_question(self):
"""Test error returned for empty question."""
result = await handle_ask_question(
{
"repo_path": "/some/path",
"question": "",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert (
"at least 1 character" in result[0].text
or "string_too_short" in result[0].text
)
async def test_returns_error_for_whitespace_question(self, tmp_path):
"""Test error returned for whitespace-only question (fails on not indexed)."""
result = await handle_ask_question(
{
"repo_path": str(tmp_path),
"question": " ",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
# Whitespace passes min_length but fails at repo check
assert "not indexed" in result[0].text
async def test_returns_error_for_unindexed_repo(self, tmp_path):
"""Test error returned when repository is not indexed."""
result = await handle_ask_question(
{
"repo_path": str(tmp_path),
"question": "What does this code do?",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "not indexed" in result[0].text
async def test_rejects_max_context_out_of_range(self, tmp_path):
"""Test that max_context out of range is rejected."""
result = await handle_ask_question(
{
"repo_path": str(tmp_path),
"question": "Test question",
"max_context": 1000, # Above max (50), now rejected by Pydantic
}
)
# Pydantic rejects out-of-range values
assert "error" in result[0].text.lower()
assert "less_than_equal" in result[0].text or "50" in result[0].text
class TestHandleSearchCode:
"""Tests for handle_search_code handler."""
async def test_returns_error_for_empty_query(self):
"""Test error returned for empty query."""
result = await handle_search_code(
{
"repo_path": "/some/path",
"query": "",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert (
"at least 1 character" in result[0].text
or "string_too_short" in result[0].text
)
async def test_returns_error_for_invalid_language_filter(self, tmp_path):
"""Test error returned for invalid language filter."""
result = await handle_search_code(
{
"repo_path": str(tmp_path),
"query": "test query",
"language": "invalid_lang",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "Invalid language" in result[0].text
async def test_returns_error_for_unindexed_repo(self, tmp_path):
"""Test error returned when repository is not indexed."""
result = await handle_search_code(
{
"repo_path": str(tmp_path),
"query": "find something",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "not indexed" in result[0].text
async def test_rejects_limit_out_of_range(self, tmp_path):
"""Test that limit out of range is rejected."""
result = await handle_search_code(
{
"repo_path": str(tmp_path),
"query": "test query",
"limit": 1000, # Above max (100), now rejected by Pydantic
}
)
# Pydantic rejects out-of-range values
assert "error" in result[0].text.lower()
assert "less_than_equal" in result[0].text or "100" in result[0].text
class TestHandleReadWikiStructure:
"""Tests for handle_read_wiki_structure handler."""
async def test_returns_error_for_nonexistent_path(self, tmp_path):
"""Test error returned for non-existent wiki path."""
nonexistent = tmp_path / "does_not_exist"
result = await handle_read_wiki_structure({"wiki_path": str(nonexistent)})
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "does not exist" in result[0].text
async def test_returns_structure_for_empty_wiki(self, tmp_path):
"""Test returns empty structure for wiki with no pages."""
result = await handle_read_wiki_structure({"wiki_path": str(tmp_path)})
assert len(result) == 1
data = json.loads(result[0].text)
assert "pages" in data
assert "sections" in data
async def test_returns_toc_json_if_exists(self, tmp_path):
"""Test returns toc.json content if file exists."""
toc_data = {"title": "Test Wiki", "sections": []}
(tmp_path / "toc.json").write_text(json.dumps(toc_data))
result = await handle_read_wiki_structure({"wiki_path": str(tmp_path)})
assert len(result) == 1
data = json.loads(result[0].text)
assert data["title"] == "Test Wiki"
async def test_builds_structure_from_markdown_files(self, tmp_path):
"""Test builds structure from markdown files when no toc.json."""
# Create some markdown files
(tmp_path / "index.md").write_text("# Home\nWelcome")
modules_dir = tmp_path / "modules"
modules_dir.mkdir()
(modules_dir / "core.md").write_text("# Core Module\nDescription")
result = await handle_read_wiki_structure({"wiki_path": str(tmp_path)})
assert len(result) == 1
data = json.loads(result[0].text)
# Check that pages were found
all_pages = data.get("pages", [])
all_sections = data.get("sections", {})
assert len(all_pages) > 0 or len(all_sections) > 0
class TestHandleReadWikiPage:
"""Tests for handle_read_wiki_page handler."""
async def test_returns_error_for_nonexistent_wiki(self, tmp_path):
"""Test error when wiki path doesn't exist."""
nonexistent = tmp_path / "does_not_exist"
result = await handle_read_wiki_page(
{
"wiki_path": str(nonexistent),
"page": "index.md",
}
)
# The implementation checks page existence, not wiki
# Let's just verify it handles missing page
assert len(result) == 1
async def test_returns_error_for_nonexistent_page(self, tmp_path):
"""Test error returned for non-existent page."""
result = await handle_read_wiki_page(
{
"wiki_path": str(tmp_path),
"page": "nonexistent.md",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
# Error message now includes hints and may say "does not exist" instead of "not found"
assert (
"not found" in result[0].text.lower()
or "does not exist" in result[0].text.lower()
)
async def test_returns_page_content(self, tmp_path):
"""Test returns page content successfully."""
page_content = "# Test Page\n\nThis is test content."
(tmp_path / "test.md").write_text(page_content)
result = await handle_read_wiki_page(
{
"wiki_path": str(tmp_path),
"page": "test.md",
}
)
assert len(result) == 1
assert result[0].text == page_content
async def test_blocks_path_traversal(self, tmp_path):
"""Test that path traversal attacks are blocked."""
# Create a file outside the wiki directory
parent_file = tmp_path.parent / "secret.txt"
parent_file.write_text("secret content")
result = await handle_read_wiki_page(
{
"wiki_path": str(tmp_path),
"page": "../secret.txt",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "traversal" in result[0].text.lower()
async def test_returns_nested_page(self, tmp_path):
"""Test returns nested page content."""
modules_dir = tmp_path / "modules"
modules_dir.mkdir()
page_content = "# Module Doc"
(modules_dir / "core.md").write_text(page_content)
result = await handle_read_wiki_page(
{
"wiki_path": str(tmp_path),
"page": "modules/core.md",
}
)
assert len(result) == 1
assert result[0].text == page_content
async def test_rejects_oversized_file(self, tmp_path):
"""Test that oversized files are rejected to prevent memory exhaustion."""
from local_deepwiki.validation import MAX_WIKI_PAGE_SIZE
# Create a file that exceeds the size limit
large_file = tmp_path / "large.md"
# Write just over the limit (10 MB + 1 byte)
large_file.write_bytes(b"x" * (MAX_WIKI_PAGE_SIZE + 1))
result = await handle_read_wiki_page(
{
"wiki_path": str(tmp_path),
"page": "large.md",
}
)
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "too large" in result[0].text
class TestHandleExportWikiHtml:
"""Tests for handle_export_wiki_html handler."""
async def test_returns_error_for_nonexistent_wiki(self, tmp_path):
"""Test error returned for non-existent wiki path."""
nonexistent = tmp_path / "does_not_exist"
result = await handle_export_wiki_html({"wiki_path": str(nonexistent)})
assert len(result) == 1
assert "error" in result[0].text.lower()
assert "does not exist" in result[0].text
async def test_exports_wiki_successfully(self, tmp_path):
"""Test successful wiki export."""
# Create a minimal wiki
(tmp_path / "index.md").write_text("# Test Wiki\n\nWelcome!")
output_path = tmp_path / "html_output"
result = await handle_export_wiki_html(
{
"wiki_path": str(tmp_path),
"output_path": str(output_path),
}
)
assert len(result) == 1
data = json.loads(result[0].text)
assert data["status"] == "success"
assert output_path.exists()
