test_new_features.py•9.53 kB
"""
Tests for new features: configurable port and caching.
"""
import subprocess
import time
import requests
import pytest
from unittest.mock import patch, MagicMock
import functools
from wikipedia_mcp.wikipedia_client import WikipediaClient
from wikipedia_mcp.server import create_server
import sys
class TestPortConfiguration:
"""Test configurable port functionality."""
def test_cli_port_argument(self):
"""Test that --port argument is available in CLI."""
result = subprocess.run(
[sys.executable, "-m", "wikipedia_mcp", "--help"],
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0
assert "--port PORT" in result.stdout
assert "Port for SSE transport" in result.stdout
def test_port_default_value(self):
"""Test that port defaults to 8000."""
result = subprocess.run(
[sys.executable, "-m", "wikipedia_mcp", "--help"],
capture_output=True,
text=True,
timeout=10,
)
assert "default: 8000" in result.stdout
@pytest.mark.timeout(30)
def test_sse_server_starts_on_custom_port(self):
"""Test that SSE server starts on the specified port."""
port = 8082
process = None
try:
# Start server in background
process = subprocess.Popen(
[
sys.executable,
"-m",
"wikipedia_mcp",
"--transport",
"sse",
"--port",
str(port),
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
# Wait for server to start
time.sleep(3)
# Check if port is in use
try:
response = requests.get(f"http://localhost:{port}/", timeout=5)
# Server should respond (even if with an error)
assert response.status_code in [200, 404, 405, 500]
except requests.exceptions.RequestException:
# Connection error is also acceptable as the server might not have HTTP endpoints
pass
# Verify process is still running
assert process.poll() is None, "Server process should still be running"
finally:
if process:
process.terminate()
process.wait(timeout=5)
def test_stdio_transport_ignores_port(self):
"""Test that STDIO transport ignores port parameter."""
# This should not raise an error
result = subprocess.run(
[
sys.executable,
"-m",
"wikipedia_mcp",
"--transport",
"stdio",
"--port",
"9999",
"--help",
],
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0
class TestHostConfiguration:
"""Test host configuration functionality."""
def test_cli_host_argument(self):
"""Test that --host argument is present in CLI."""
result = subprocess.run(
[sys.executable, "-m", "wikipedia_mcp", "--help"],
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0
assert "--host" in result.stdout
assert "0.0.0.0 for all interfaces" in result.stdout
def test_host_default_value(self):
"""Test that host defaults to 127.0.0.1."""
result = subprocess.run(
[sys.executable, "-m", "wikipedia_mcp", "--help"],
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0
assert "default: 127.0.0.1" in result.stdout
def test_stdio_transport_ignores_host(self):
"""Test that STDIO transport ignores host parameter."""
# This should not raise an error
result = subprocess.run(
[
sys.executable,
"-m",
"wikipedia_mcp",
"--transport",
"stdio",
"--host",
"0.0.0.0",
"--help",
],
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0
class TestCachingFunctionality:
"""Test caching functionality."""
def test_cli_cache_argument(self):
"""Test that --enable-cache argument is available in CLI."""
result = subprocess.run(
[sys.executable, "-m", "wikipedia_mcp", "--help"],
capture_output=True,
text=True,
timeout=10,
)
assert result.returncode == 0
assert "--enable-cache" in result.stdout
# Just check for the word "caching" which we can see is there
assert "caching" in result.stdout
def test_wikipedia_client_without_cache(self):
"""Test WikipediaClient without caching."""
client = WikipediaClient(enable_cache=False)
assert client.enable_cache is False
# Methods should not be wrapped with lru_cache
assert not hasattr(client.search, "cache_info")
def test_wikipedia_client_with_cache(self):
"""Test WikipediaClient with caching enabled."""
client = WikipediaClient(enable_cache=True)
assert client.enable_cache is True
# Methods should be wrapped with lru_cache
assert hasattr(client.search, "cache_info")
assert hasattr(client.get_article, "cache_info")
assert hasattr(client.get_summary, "cache_info")
@patch("wikipedia_mcp.wikipedia_client.requests.get")
def test_cache_effectiveness(self, mock_get):
"""Test that caching actually works."""
# Mock the API response
mock_response = MagicMock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {
"query": {
"search": [
{
"title": "Test Article",
"snippet": "Test snippet",
"pageid": 123,
"wordcount": 100,
"timestamp": "2023-01-01T00:00:00Z",
}
]
}
}
mock_get.return_value = mock_response
# Create client with caching
client = WikipediaClient(enable_cache=True)
# Call the same search twice
result1 = client.search("test query")
result2 = client.search("test query")
# Should be the same result
assert result1 == result2
# API should only be called once due to caching
assert mock_get.call_count == 1
# Cache info should show hits
cache_info = client.search.cache_info()
assert cache_info.hits == 1
assert cache_info.misses == 1
def test_cache_methods_coverage(self):
"""Test that all expected methods are cached when caching is enabled."""
client = WikipediaClient(enable_cache=True)
cached_methods = [
"search",
"get_article",
"get_summary",
"get_sections",
"get_links",
"get_related_topics",
"summarize_for_query",
"summarize_section",
"extract_facts",
]
for method_name in cached_methods:
method = getattr(client, method_name)
assert hasattr(method, "cache_info"), f"Method {method_name} should be cached"
def test_cache_size_limit(self):
"""Test that cache has proper size limit."""
client = WikipediaClient(enable_cache=True)
# Check cache maxsize is set correctly
cache_info = client.search.cache_info()
assert cache_info.maxsize == 128
class TestIntegrationNewFeatures:
"""Integration tests for new features."""
def test_server_creation_with_all_options(self):
"""Test server creation with all new options."""
server = create_server(language="en", enable_cache=True)
assert server is not None
@pytest.mark.timeout(20)
def test_full_cli_with_new_options(self):
"""Test CLI with all new options."""
result = subprocess.run(
[
sys.executable,
"-m",
"wikipedia_mcp",
"--transport",
"stdio",
"--port",
"8083",
"--enable-cache",
"--language",
"en",
"--log-level",
"INFO",
"--help",
],
capture_output=True,
text=True,
timeout=15,
)
assert result.returncode == 0
assert "Wikipedia MCP Server" in result.stdout
def test_cache_and_port_independence(self):
"""Test that caching and port options work independently."""
# Test cache without custom port
server1 = create_server(enable_cache=True)
assert server1 is not None
# Test custom port without cache (implicitly)
server2 = create_server(language="en") # enable_cache defaults to False
assert server2 is not None