test_server.py•9.86 kB
#!/usr/bin/env python3
"""
Test script for MCP Domain Availability Server
"""
import asyncio
import json
import os
import sys
import time
from unittest.mock import AsyncMock, MagicMock, patch
# Add the project root to the Python path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from domain_availability.server import DomainAvailabilityServer, RateLimiter
from mcp.types import CallToolRequest, ListToolsRequest, CallToolRequestParams
async def test_rate_limiter():
"""Test rate limiter functionality"""
print("Testing rate limiter...")
# Create a rate limiter with very low limits for testing
rate_limiter = RateLimiter(max_calls=2, window_seconds=1)
# First two calls should be allowed
assert rate_limiter.is_allowed() == True
assert rate_limiter.is_allowed() == True
# Third call should be denied
assert rate_limiter.is_allowed() == False
# Wait for window to expire
await asyncio.sleep(1.1)
# Should be allowed again
assert rate_limiter.is_allowed() == True
print("✅ Rate limiter test passed")
async def test_list_tools():
"""Test listing available tools"""
print("Testing list_tools...")
# Mock environment variables
with patch.dict(
os.environ,
{"DOMAIN_API_KEY": "test-key", "DOMAIN_API_URL": "https://api.test.com"},
):
server = DomainAvailabilityServer()
# Test list_tools
request = ListToolsRequest(method="tools/list")
result = await server.list_tools(request)
assert len(result.tools) == 1
assert result.tools[0].name == "check_domain_availability"
assert "Rate limited: 60 calls per minute" in result.tools[0].description
print("✅ list_tools test passed")
async def test_domain_availability_success():
"""Test successful domain availability check"""
print("Testing domain availability check (success)...")
# Mock response data
mock_response_data = {
"domains": [
{
"domain": "example.com",
"available": True,
"price": 12.99,
"currency": "USD",
"definitive": True,
"period": 1,
},
{
"domain": "example.org",
"available": False,
"price": 0,
"currency": "USD",
"definitive": True,
"period": 0,
},
],
"errors": [],
}
with patch.dict(
os.environ,
{"DOMAIN_API_KEY": "test-key", "DOMAIN_API_URL": "https://api.test.com"},
):
server = DomainAvailabilityServer()
# Mock the HTTP client
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_response_data
with patch.object(
server.client, "post", new_callable=AsyncMock, return_value=mock_response
):
# Test with domains list
request = CallToolRequest(
method="tools/call",
params=CallToolRequestParams(
name="check_domain_availability",
arguments={
"domains": ["example.com", "example.org"],
"checkType": "FAST",
},
),
)
result = await server.call_tool(request)
assert not result.isError
assert len(result.content) == 1 # Only text content now
assert "✅ AVAILABLE DOMAINS:" in result.content[0].text
assert "❌ UNAVAILABLE DOMAINS:" in result.content[0].text
assert "Raw API Response:" in result.content[0].text
print("✅ Domain availability check (success) test passed")
async def test_domain_availability_with_suffixes():
"""Test domain availability check with base name and suffixes"""
print("Testing domain availability check (with suffixes)...")
# Mock response data
mock_response_data = {
"domains": [
{
"domain": "testsite.com",
"available": True,
"price": 12.99,
"currency": "USD",
"definitive": True,
"period": 1,
},
{
"domain": "testsite.org",
"available": True,
"price": 8.99,
"currency": "USD",
"definitive": False,
"period": 1,
},
{
"domain": "testsite.net",
"available": False,
"price": 0,
"currency": "USD",
"definitive": True,
"period": 0,
},
],
"errors": [],
}
with patch.dict(
os.environ,
{"DOMAIN_API_KEY": "test-key", "DOMAIN_API_URL": "https://api.test.com"},
):
server = DomainAvailabilityServer()
# Mock the HTTP client
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_response_data
with patch.object(
server.client, "post", new_callable=AsyncMock, return_value=mock_response
):
# Test with base name and suffixes
request = CallToolRequest(
method="tools/call",
params=CallToolRequestParams(
name="check_domain_availability",
arguments={
"base_name": "testsite",
"tld_suffixes": [".com", ".org", ".net"],
"checkType": "FULL",
},
),
)
result = await server.call_tool(request)
assert not result.isError
assert "testsite.com" in result.content[0].text
assert "testsite.org" in result.content[0].text
assert "testsite.net" in result.content[0].text
print("✅ Domain availability check (with suffixes) test passed")
async def test_rate_limit_exceeded():
"""Test rate limit exceeded scenario"""
print("Testing rate limit exceeded...")
with patch.dict(
os.environ,
{"DOMAIN_API_KEY": "test-key", "DOMAIN_API_URL": "https://api.test.com"},
):
server = DomainAvailabilityServer()
# Mock the rate limiter to return False (rate limit exceeded)
with patch.object(server.rate_limiter, "is_allowed", return_value=False):
with patch.object(server.rate_limiter, "get_retry_after", return_value=30):
request = CallToolRequest(
method="tools/call",
params=CallToolRequestParams(
name="check_domain_availability",
arguments={"domains": ["example.com"]},
),
)
result = await server.call_tool(request)
assert result.isError
assert "Rate limit exceeded" in result.content[0].text
assert "60 calls per minute" in result.content[0].text
print("✅ Rate limit exceeded test passed")
async def test_error_handling():
"""Test error handling"""
print("Testing error handling...")
with patch.dict(
os.environ,
{"DOMAIN_API_KEY": "test-key", "DOMAIN_API_URL": "https://api.test.com"},
):
server = DomainAvailabilityServer()
# Mock HTTP error
mock_response = MagicMock()
mock_response.status_code = 429
mock_response.json.return_value = {"message": "Rate limit exceeded"}
with patch.object(
server.client, "post", new_callable=AsyncMock, return_value=mock_response
):
request = CallToolRequest(
method="tools/call",
params=CallToolRequestParams(
name="check_domain_availability",
arguments={"domains": ["example.com"]},
),
)
result = await server.call_tool(request)
assert result.isError
assert "API rate limit exceeded" in result.content[0].text
print("✅ Error handling test passed")
async def test_invalid_arguments():
"""Test invalid arguments handling"""
print("Testing invalid arguments...")
with patch.dict(
os.environ,
{"DOMAIN_API_KEY": "test-key", "DOMAIN_API_URL": "https://api.test.com"},
):
server = DomainAvailabilityServer()
# Test with missing required arguments
request = CallToolRequest(
method="tools/call",
params=CallToolRequestParams(
name="check_domain_availability", arguments={}
),
)
result = await server.call_tool(request)
assert result.isError
assert (
"base_name" in result.content[0].text or "domains" in result.content[0].text
)
print("✅ Invalid arguments test passed")
async def main():
"""Run all tests"""
print("Running MCP Domain Availability Server tests...\n")
try:
await test_rate_limiter()
await test_list_tools()
await test_domain_availability_success()
await test_domain_availability_with_suffixes()
await test_rate_limit_exceeded()
await test_error_handling()
await test_invalid_arguments()
print("\n🎉 All tests passed!")
return 0
except Exception as e:
print(f"\n❌ Test failed with error: {e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)