conftest.pyā¢5.57 kB
"""
Pytest configuration and shared fixtures for Crawl4AI MCP Server tests.
This module provides common test fixtures, configuration, and utilities
for testing the MCP server functionality.
"""
import asyncio
import json
import pytest
from unittest.mock import MagicMock, patch
from typing import Dict, Any
import fastmcp
from crawl4ai_mcp_server import mcp
from tests.mocks import (
create_mock_crawler,
create_mock_extraction_strategy,
generate_test_html,
generate_test_schema,
generate_extracted_data
)
@pytest.fixture(scope="session")
def event_loop():
"""Create an event loop for the entire test session."""
policy = asyncio.get_event_loop_policy()
loop = policy.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function")
async def mcp_client():
"""Create an in-memory FastMCP client for testing."""
async with fastmcp.Client(mcp) as client:
yield client
@pytest.fixture
def mock_crawler():
"""Create a mock AsyncWebCrawler instance using our mock infrastructure."""
return create_mock_crawler(success=True, screenshot_enabled=True)
@pytest.fixture
def mock_extraction_result():
"""Mock extraction result for schema-based crawling."""
return generate_extracted_data()[0]
@pytest.fixture
def sample_schema():
"""Sample extraction schema for testing."""
return generate_test_schema()
@pytest.fixture
def mock_crawler_failure():
"""Create a mock crawler that simulates failures."""
return create_mock_crawler(success=False)
@pytest.fixture
def mock_extraction_strategy():
"""Create a mock extraction strategy."""
return create_mock_extraction_strategy()
@pytest.fixture
def test_html_content():
"""Generate test HTML content."""
return generate_test_html()
@pytest.fixture
def crawl4ai_patches():
"""Patch Crawl4AI imports for isolated testing."""
with patch('crawl4ai.AsyncWebCrawler') as mock_crawler_class, \
patch('crawl4ai.extraction_strategy.JsonCssExtractionStrategy') as mock_strategy_class, \
patch('crawl4ai.CrawlerRunConfig') as mock_config_class:
# Configure the mock classes to return our mock instances
mock_crawler_class.return_value = create_mock_crawler()
mock_strategy_class.return_value = create_mock_extraction_strategy()
mock_config_class.return_value = MagicMock()
yield {
'crawler': mock_crawler_class,
'strategy': mock_strategy_class,
'config': mock_config_class
}
@pytest.fixture
def mock_context():
"""Create a mock MCP context for testing."""
context = MagicMock()
context.info = MagicMock()
context.warning = MagicMock()
context.error = MagicMock()
return context
@pytest.fixture
def httpx_mock():
"""Mock httpx responses for web requests."""
with patch('httpx.AsyncClient') as mock_client:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = "<html><body><h1>Test</h1></body></html>"
mock_response.json.return_value = {"status": "ok"}
mock_client.return_value.__aenter__.return_value.get.return_value = mock_response
yield mock_client
@pytest.fixture
def test_urls():
"""Provide test URLs for various scenarios."""
return {
"valid": "https://httpbin.org/html",
"invalid": "https://invalid-url-that-does-not-exist.com",
"timeout": "https://httpbin.org/delay/10",
"error": "https://httpbin.org/status/500"
}
@pytest.fixture
async def server_instance():
"""Create a test server instance."""
# This would be used for integration testing
# For now, we'll use the global mcp instance
yield mcp
# Test data fixtures
@pytest.fixture
def sample_html():
"""Sample HTML content for testing."""
return """
<html>
<head>
<title>Test Page</title>
</head>
<body>
<h1>Main Title</h1>
<div class="content">
<p class="description">This is a test page</p>
<span class="price">$29.99</span>
<ul class="features">
<li>Feature 1</li>
<li>Feature 2</li>
</ul>
</div>
</body>
</html>
"""
@pytest.fixture
def sample_markdown():
"""Sample markdown content for testing."""
return """
# Main Title
This is a test page
**Price:** $29.99
## Features
- Feature 1
- Feature 2
"""
# Async test utilities
async def wait_for_condition(condition, timeout=5, interval=0.1):
"""Wait for a condition to become true."""
elapsed = 0
while elapsed < timeout:
if await condition():
return True
await asyncio.sleep(interval)
elapsed += interval
return False
def assert_valid_json(json_string: str) -> Dict[str, Any]:
"""Assert that a string is valid JSON and return parsed data."""
try:
return json.loads(json_string)
except json.JSONDecodeError as e:
pytest.fail(f"Invalid JSON: {e}")
def assert_has_keys(data: Dict[str, Any], required_keys: list):
"""Assert that a dictionary has all required keys."""
missing_keys = [key for key in required_keys if key not in data]
if missing_keys:
pytest.fail(f"Missing required keys: {missing_keys}")
# Markers for test categorization
pytest.mark.unit = pytest.mark.unit
pytest.mark.integration = pytest.mark.integration
pytest.mark.slow = pytest.mark.slow