"""
pytestmark = pytest.mark.skip(reason="Needs refactoring after module restructure")
Comprehensive unit tests for MCP tools in crawl4ai_mcp.py
This module tests all @mcp.tool decorated functions with proper mocking
of external dependencies including Crawl4AI, databases, and Neo4j.
Target coverage: >90% for all MCP tool functions
"""
import os
import pytest
npytestmark = pytest.mark.skip(reason="Test needs refactoring after module restructure")
import sys # noqa: E402
from pathlib import Path # noqa: E402
from unittest.mock import AsyncMock, Mock, mock_open, patch # noqa: E402
import pytest # noqa: E402
# Add src to path for imports
src_path = Path(__file__).parent.parent / "src"
sys.path.insert(0, str(src_path))
# Add tests to path for test helpers
tests_path = Path(__file__).parent
sys.path.insert(0, str(tests_path))
# Mock the FastMCP Context before importing the main module
class MockContext:
"""Mock FastMCP Context for testing"""
def __init__(self):
self.request_context = Mock()
self.request_context.lifespan_context = Mock()
self.request_context.lifespan_context.database_client = AsyncMock()
# Create a properly mocked AsyncWebCrawler that supports async context manager
mock_crawler = AsyncMock()
mock_crawler.__aenter__ = AsyncMock(return_value=mock_crawler)
mock_crawler.__aexit__ = AsyncMock(return_value=None)
# Mock both single and batch crawling methods
mock_crawler.arun = AsyncMock()
mock_crawler.arun_many = AsyncMock()
self.request_context.lifespan_context.crawler = mock_crawler
# Create a properly mocked Neo4j repository extractor with session context manager
mock_repo_extractor = Mock()
mock_session = AsyncMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
# Mock session methods for Neo4j operations
mock_session.run = AsyncMock()
mock_session.single = AsyncMock()
# Mock the driver and session creation
mock_driver = Mock()
mock_driver.session = Mock(return_value=mock_session)
mock_repo_extractor.driver = mock_driver
mock_repo_extractor.analyze_repository = AsyncMock()
self.request_context.lifespan_context.repo_extractor = mock_repo_extractor
# Add knowledge validator for hallucination testing
mock_knowledge_validator = AsyncMock()
mock_knowledge_validator.validate_script = AsyncMock()
self.request_context.lifespan_context.knowledge_validator = (
mock_knowledge_validator
)
@pytest.fixture
def mock_context():
"""Provide a mock FastMCP context for testing"""
return MockContext()
# Import mock helper
from mock_openai_helper import patch_openai_embeddings # noqa: E402
# Mock all external dependencies before importing
@pytest.fixture(autouse=True)
def mock_external_dependencies():
"""Mock all external dependencies used by MCP tools"""
# Set up environment variables for all tests
env_vars = {
# Search and Web Scraping
"SEARXNG_URL": "http://localhost:8888",
"SEARXNG_USER_AGENT": "MCP-Crawl4AI-RAG-Server/1.0",
"SEARXNG_TIMEOUT": "30",
"SEARXNG_DEFAULT_ENGINES": "",
# Knowledge Graph / Neo4j
"USE_KNOWLEDGE_GRAPH": "true",
"NEO4J_URI": "bolt://localhost:7687",
"NEO4J_USER": "neo4j",
"NEO4J_USERNAME": "neo4j",
"NEO4J_PASSWORD": "password",
# RAG Features
"USE_AGENTIC_RAG": "true",
"USE_RERANKING": "true",
"USE_HYBRID_SEARCH": "true",
# Vector Database
"VECTOR_DATABASE": "qdrant",
"QDRANT_URL": "http://localhost:6333",
"QDRANT_API_KEY": "test-key",
# API Keys
"OPENAI_API_KEY": "test-openai-key",
# General Configuration
"MCP_DEBUG": "false",
"USE_TEST_ENV": "true",
"HOST": "0.0.0.0",
"PORT": "8051",
"TRANSPORT": "http",
}
# Get OpenAI patches
openai_patches = patch_openai_embeddings()
# Create a context manager list
context_managers = (
[patch.dict(os.environ, env_vars)]
+ openai_patches
+ [
patch.dict(
"sys.modules",
{
"crawl4ai": Mock(),
"fastmcp": Mock(),
"sentence_transformers": Mock(),
"database.factory": Mock(),
"utils": Mock(),
"knowledge_graph_validator": Mock(),
"parse_repo_into_neo4j": Mock(),
"ai_script_analyzer": Mock(),
"hallucination_reporter": Mock(),
},
),
]
)
# Apply all context managers using ExitStack
from contextlib import ExitStack
with ExitStack() as stack:
for cm in context_managers:
stack.enter_context(cm)
# Mock specific classes and functions
mock_crawler = stack.enter_context(patch("crawl4ai.AsyncWebCrawler"))
mock_browser_config = stack.enter_context(patch("crawl4ai.BrowserConfig"))
mock_crawler_config = stack.enter_context(patch("crawl4ai.CrawlerRunConfig"))
mock_requests_get = stack.enter_context(patch("requests.get"))
mock_db_factory = stack.enter_context(
patch("database.factory.create_and_initialize_database"),
)
mock_add_docs = stack.enter_context(
patch("utils.add_documents_to_database"),
)
mock_search_docs = stack.enter_context(
patch("utils.search_documents"),
)
mock_extract_code = stack.enter_context(
patch("utils.extract_code_blocks"),
)
mock_code_summary = stack.enter_context(
patch("utils.generate_code_example_summary"),
)
mock_kg_validator = stack.enter_context(
patch("knowledge_graph_validator.KnowledgeGraphValidator"),
)
mock_neo4j_extractor = stack.enter_context(
patch("parse_repo_into_neo4j.DirectNeo4jExtractor"),
)
mock_ai_analyzer = stack.enter_context(
patch("ai_script_analyzer.AIScriptAnalyzer"),
)
mock_hallucination_reporter = stack.enter_context(
patch("hallucination_reporter.HallucinationReporter"),
)
mock_path_exists = stack.enter_context(
patch("os.path.exists", return_value=True),
)
mock_file_open = stack.enter_context(
patch("builtins.open", mock_open(read_data="script content")),
)
yield {
"mock_crawler": mock_crawler,
"mock_requests_get": mock_requests_get,
"mock_db_factory": mock_db_factory,
"mock_add_docs": mock_add_docs,
"mock_search_docs": mock_search_docs,
"mock_extract_code": mock_extract_code,
"mock_code_summary": mock_code_summary,
"mock_kg_validator": mock_kg_validator,
"mock_neo4j_extractor": mock_neo4j_extractor,
"mock_ai_analyzer": mock_ai_analyzer,
"mock_hallucination_reporter": mock_hallucination_reporter,
"mock_path_exists": mock_path_exists,
"mock_file_open": mock_file_open,
}
# Import the module under test after mocking
# Instead of importing wrapped functions, import the module and access the underlying functions
import crawl4ai_mcp # noqa: E402
# Access the underlying functions from the FastMCP tool wrappers
def get_tool_function(tool_name: str):
"""Get the underlying function from FastMCP tool wrapper"""
tool_attr = getattr(crawl4ai_mcp, tool_name, None)
if hasattr(tool_attr, "fn"):
return tool_attr.fn
if callable(tool_attr):
return tool_attr
raise AttributeError(f"Cannot find callable function for {tool_name}")
# Get the actual functions
# DISABLED: Need refactoring after module restructure
search = None
scrape_urls = None
smart_crawl_url = None
get_available_sources = None
perform_rag_query = None
search_code_examples = None
check_ai_script_hallucinations = None
