"""Unit tests for fetch tool."""
import asyncio
import sys
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# Mock OpenAI before any imports
mock_openai_module = MagicMock()
mock_openai_client = MagicMock()
mock_openai_module.OpenAI = mock_openai_client
# Mock the response from OpenAI
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = '{"title": "Test", "summary": "Test summary"}'
mock_response.usage.prompt_tokens = 100
mock_response.usage.completion_tokens = 50
mock_response.usage.total_tokens = 150
# Set up mock OpenAI client instance
mock_openai_instance = MagicMock()
mock_openai_instance.chat.completions.create.return_value = mock_response
mock_openai_client.return_value = mock_openai_instance
# Add mock to sys.modules
sys.modules['openai'] = mock_openai_module
# Mock Crawl4AI
mock_crawl4ai = MagicMock()
mock_browser_config = MagicMock()
mock_crawler_run_config = MagicMock()
mock_crawler = MagicMock()
mock_result = MagicMock()
mock_crawl4ai.BrowserConfig = mock_browser_config
mock_crawl4ai.CrawlerRunConfig = mock_crawler_run_config
mock_crawl4ai.AsyncWebCrawler = mock_crawler
# Setup mock result
mock_result.success = True
mock_result.url = "https://example.com"
mock_result.html = "<html><body>Test</body></html>"
mock_result.markdown = MagicMock()
mock_result.markdown.raw_markdown = "# Test\n\nThis is a test."
mock_result.markdown.fit_markdown = "This is a test."
mock_result.metadata = {"title": "Test Page"}
mock_result.status_code = 200
mock_result.error_message = None
# Setup mock crawler as async context manager
async_mock_crawler = AsyncMock()
async_mock_crawler.arun = AsyncMock(return_value=mock_result)
async def mock_aenter(*args, **kwargs):
return async_mock_crawler
async def mock_aexit(*args, **kwargs):
return None
async_mock_crawler.__aenter__ = mock_aenter
async_mock_crawler.__aexit__ = mock_aexit
mock_crawler.return_value = async_mock_crawler
# Add mock to sys.modules
sys.modules['crawl4ai'] = mock_crawl4ai
sys.modules['crawl4ai.async_configs'] = mock_crawl4ai
# Now import the module under test
from mcp_webscout.tools.fetch import (
CRAWL4AI_AVAILABLE,
DeepSeekClient,
fetch,
fetch_with_crawl4ai,
)
class TestFetchV2:
"""Test cases for fetch function."""
@pytest.mark.asyncio
async def test_fetch_simple_mode_success(self):
"""Test fetch in simple mode with successful response."""
result = await fetch(
url="https://example.com",
mode="simple",
)
assert result["success"] is True
assert result["url"] == "https://example.com"
assert result["title"] == "Test Page"
assert result["markdown"] is not None
assert result["fit_markdown"] is not None
@pytest.mark.asyncio
async def test_fetch_simple_mode_failure(self):
"""Test fetch in simple mode with failed response."""
# Mock failed result
mock_result.success = False
mock_result.error_message = "Connection timeout"
result = await fetch(
url="https://example.com",
mode="simple",
)
assert result["success"] is False
assert result["error"] is not None
# Reset mock for other tests
mock_result.success = True
mock_result.error_message = None
@pytest.mark.asyncio
async def test_fetch_llm_mode_success(self):
"""Test fetch in LLM mode with successful extraction."""
result = await fetch(
url="https://example.com",
mode="llm",
prompt="提取标题和摘要",
api_key="test-api-key",
)
assert result["success"] is True
assert result["extracted"] is not None
assert result["model"] == "deepseek-chat"
assert result["usage"] is not None
@pytest.mark.asyncio
async def test_fetch_invalid_mode(self):
"""Test fetch with invalid mode."""
with pytest.raises(ValueError) as exc_info:
await fetch(
url="https://example.com",
mode="invalid_mode",
)
assert "Invalid mode" in str(exc_info.value)
@pytest.mark.asyncio
async def test_fetch_llm_mode_without_api_key(self):
"""Test fetch in LLM mode without API key."""
with pytest.raises(ValueError) as exc_info:
await fetch(
url="https://example.com",
mode="llm",
api_key=None,
)
assert "api_key is required" in str(exc_info.value)
def test_deep_seek_client_import(self):
"""Test that DeepSeekClient can be imported."""
assert DeepSeekClient is not None
def test_crawl4ai_availability(self):
"""Test Crawl4AI availability flag."""
assert isinstance(CRAWL4AI_AVAILABLE, bool)
@pytest.mark.asyncio
async def test_fetch_with_crawl4ai(self):
"""Test the fetch_with_crawl4ai function directly."""
result = await fetch_with_crawl4ai(
url="https://example.com",
js_render=True,
use_proxy=True,
timeout=30,
)
assert result["success"] is True
assert result["url"] == "https://example.com"
assert result["markdown"] is not None
