Delete fixed_test_mcp_server.py
This commit is contained in:
Ryan Malloy
GitHub
parent
5077778388
commit
2228dce982
@ -1,399 +0,0 @@
|
||||
"""Tests for MCP server functionality using FastMCP testing patterns - FIXED VERSION."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import patch, AsyncMock
|
||||
from fastmcp import Client
|
||||
from vultr_dns_mcp.server import VultrDNSServer, create_mcp_server
|
||||
|
||||
|
||||
class TestMCPServerBasics:
|
||||
"""Test basic MCP server functionality."""
|
||||
|
||||
def test_server_creation(self, mock_api_key):
|
||||
"""Test that MCP server can be created successfully."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
assert server is not None
|
||||
assert hasattr(server, '_tools')
|
||||
assert hasattr(server, '_resources')
|
||||
|
||||
def test_server_creation_without_api_key(self):
|
||||
"""Test that server creation fails without API key."""
|
||||
with pytest.raises(ValueError, match="VULTR_API_KEY must be provided"):
|
||||
create_mcp_server(None)
|
||||
|
||||
@patch.dict('os.environ', {'VULTR_API_KEY': 'env-test-key'})
|
||||
def test_server_creation_from_env(self):
|
||||
"""Test server creation using environment variable."""
|
||||
server = create_mcp_server()
|
||||
assert server is not None
|
||||
|
||||
|
||||
@pytest.mark.mcp
|
||||
class TestMCPTools:
|
||||
"""Test MCP tools through in-memory client connection."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_dns_domains_tool(self, mock_vultr_client):
|
||||
"""Test the list_dns_domains MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("list_dns_domains", {})
|
||||
|
||||
assert result is not None
|
||||
# Check if we got a response (could be list or wrapped response)
|
||||
if isinstance(result, list):
|
||||
# Direct list response
|
||||
mock_vultr_client.list_domains.assert_called_once()
|
||||
elif hasattr(result, 'content') and isinstance(result.content, list):
|
||||
# Wrapped response format
|
||||
mock_vultr_client.list_domains.assert_called_once()
|
||||
else:
|
||||
# Handle other response formats
|
||||
mock_vultr_client.list_domains.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_dns_domain_tool(self, mock_vultr_client):
|
||||
"""Test the get_dns_domain MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("get_dns_domain", {"domain": "example.com"})
|
||||
|
||||
assert result is not None
|
||||
mock_vultr_client.get_domain.assert_called_once_with("example.com")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_dns_domain_tool(self, mock_vultr_client):
|
||||
"""Test the create_dns_domain MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("create_dns_domain", {
|
||||
"domain": "newdomain.com",
|
||||
"ip": "192.168.1.100"
|
||||
})
|
||||
|
||||
assert result is not None
|
||||
mock_vultr_client.create_domain.assert_called_once_with("newdomain.com", "192.168.1.100")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_dns_domain_tool(self, mock_vultr_client):
|
||||
"""Test the delete_dns_domain MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("delete_dns_domain", {"domain": "example.com"})
|
||||
|
||||
assert result is not None
|
||||
mock_vultr_client.delete_domain.assert_called_once_with("example.com")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_dns_records_tool(self, mock_vultr_client):
|
||||
"""Test the list_dns_records MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("list_dns_records", {"domain": "example.com"})
|
||||
|
||||
assert result is not None
|
||||
mock_vultr_client.list_records.assert_called_once_with("example.com")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_dns_record_tool(self, mock_vultr_client):
|
||||
"""Test the create_dns_record MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("create_dns_record", {
|
||||
"domain": "example.com",
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "192.168.1.100",
|
||||
"ttl": 300
|
||||
})
|
||||
|
||||
assert result is not None
|
||||
mock_vultr_client.create_record.assert_called_once_with(
|
||||
"example.com", "A", "www", "192.168.1.100", 300, None
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_dns_record_tool(self, mock_api_key):
|
||||
"""Test the validate_dns_record MCP tool."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
# Test valid A record
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "192.168.1.100",
|
||||
"ttl": 300
|
||||
})
|
||||
|
||||
assert result is not None
|
||||
# Check validation result structure
|
||||
if isinstance(result, dict):
|
||||
assert "validation" in result or "valid" in result or "record_type" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_dns_record_invalid(self, mock_api_key):
|
||||
"""Test the validate_dns_record tool with invalid data."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
# Test invalid A record (bad IP)
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "invalid-ip-address"
|
||||
})
|
||||
|
||||
assert result is not None
|
||||
# Should detect the invalid IP address
|
||||
if isinstance(result, dict) and "validation" in result:
|
||||
validation = result["validation"]
|
||||
assert "valid" in validation
|
||||
# For invalid IP, should be False or have errors
|
||||
if validation.get("valid") is not False:
|
||||
assert len(validation.get("errors", [])) > 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analyze_dns_records_tool(self, mock_vultr_client):
|
||||
"""Test the analyze_dns_records MCP tool."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("analyze_dns_records", {"domain": "example.com"})
|
||||
|
||||
assert result is not None
|
||||
mock_vultr_client.list_records.assert_called_once_with("example.com")
|
||||
|
||||
|
||||
@pytest.mark.mcp
|
||||
class TestMCPResources:
|
||||
"""Test MCP resources through in-memory client connection."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_domains_resource(self, mock_vultr_client):
|
||||
"""Test the vultr://domains resource."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
# Get available resources
|
||||
resources = await client.list_resources()
|
||||
|
||||
# Check that domains resource is available
|
||||
resource_uris = [r.uri for r in resources]
|
||||
assert "vultr://domains" in resource_uris
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_capabilities_resource(self, mock_api_key):
|
||||
"""Test the vultr://capabilities resource."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
resources = await client.list_resources()
|
||||
resource_uris = [r.uri for r in resources]
|
||||
assert "vultr://capabilities" in resource_uris
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_read_domains_resource(self, mock_vultr_client):
|
||||
"""Test reading the domains resource content."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
try:
|
||||
result = await client.read_resource("vultr://domains")
|
||||
assert result is not None
|
||||
mock_vultr_client.list_domains.assert_called_once()
|
||||
except Exception:
|
||||
# Resource reading might not be available in all FastMCP versions
|
||||
# This is acceptable for now
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.mcp
|
||||
class TestMCPToolErrors:
|
||||
"""Test MCP tool error handling."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_with_api_error(self):
|
||||
"""Test tool behavior when API returns an error."""
|
||||
mock_client = AsyncMock()
|
||||
mock_client.list_domains.side_effect = Exception("API Error")
|
||||
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("list_dns_domains", {})
|
||||
|
||||
# Should handle the error gracefully
|
||||
assert result is not None
|
||||
# Check if error is properly handled
|
||||
if isinstance(result, list) and len(result) > 0:
|
||||
if isinstance(result[0], dict) and "error" in result[0]:
|
||||
assert "API Error" in str(result[0]["error"])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_required_parameters(self, mock_api_key):
|
||||
"""Test tool behavior with missing required parameters."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
# Try to call tool without required parameter
|
||||
try:
|
||||
# This should fail due to missing required 'domain' parameter
|
||||
result = await client.call_tool("get_dns_domain", {})
|
||||
# If it doesn't raise an exception, check if error is in result
|
||||
if isinstance(result, dict) and "error" in result:
|
||||
assert "domain" in str(result["error"]).lower()
|
||||
else:
|
||||
# Should have failed in some way
|
||||
assert False, "Expected error for missing domain parameter"
|
||||
except Exception as e:
|
||||
# Expected to raise an exception
|
||||
assert "domain" in str(e).lower() or "required" in str(e).lower()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestMCPIntegration:
|
||||
"""Integration tests for the complete MCP workflow."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_complete_domain_workflow(self, mock_vultr_client):
|
||||
"""Test a complete domain management workflow."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
# 1. List domains
|
||||
domains = await client.call_tool("list_dns_domains", {})
|
||||
assert domains is not None
|
||||
|
||||
# 2. Get domain details
|
||||
domain_info = await client.call_tool("get_dns_domain", {"domain": "example.com"})
|
||||
assert domain_info is not None
|
||||
|
||||
# 3. List records
|
||||
records = await client.call_tool("list_dns_records", {"domain": "example.com"})
|
||||
assert records is not None
|
||||
|
||||
# 4. Analyze configuration
|
||||
analysis = await client.call_tool("analyze_dns_records", {"domain": "example.com"})
|
||||
assert analysis is not None
|
||||
|
||||
# Verify all expected API calls were made
|
||||
mock_vultr_client.list_domains.assert_called()
|
||||
mock_vultr_client.get_domain.assert_called_with("example.com")
|
||||
mock_vultr_client.list_records.assert_called_with("example.com")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_record_management_workflow(self, mock_vultr_client):
|
||||
"""Test record creation and management workflow."""
|
||||
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
|
||||
server = create_mcp_server("test-api-key")
|
||||
|
||||
async with Client(server) as client:
|
||||
# 1. Validate record before creation
|
||||
validation = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "192.168.1.100"
|
||||
})
|
||||
assert validation is not None
|
||||
|
||||
# 2. Create the record
|
||||
create_result = await client.call_tool("create_dns_record", {
|
||||
"domain": "example.com",
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "192.168.1.100",
|
||||
"ttl": 300
|
||||
})
|
||||
assert create_result is not None
|
||||
|
||||
# 3. Verify the record was created
|
||||
mock_vultr_client.create_record.assert_called_with(
|
||||
"example.com", "A", "www", "192.168.1.100", 300, None
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestValidationLogic:
|
||||
"""Test DNS record validation logic in isolation."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_record_validation(self, mock_api_key):
|
||||
"""Test A record validation logic."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
# Valid IPv4
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "192.168.1.1"
|
||||
})
|
||||
assert result is not None
|
||||
|
||||
# Invalid IPv4
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "A",
|
||||
"name": "www",
|
||||
"data": "999.999.999.999"
|
||||
})
|
||||
assert result is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cname_validation(self, mock_api_key):
|
||||
"""Test CNAME record validation logic."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
# Invalid: CNAME on root domain
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "CNAME",
|
||||
"name": "@",
|
||||
"data": "example.com"
|
||||
})
|
||||
assert result is not None
|
||||
|
||||
# Valid: CNAME on subdomain
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "CNAME",
|
||||
"name": "www",
|
||||
"data": "example.com"
|
||||
})
|
||||
assert result is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mx_validation(self, mock_api_key):
|
||||
"""Test MX record validation logic."""
|
||||
server = create_mcp_server(mock_api_key)
|
||||
async with Client(server) as client:
|
||||
# Invalid: Missing priority
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "MX",
|
||||
"name": "@",
|
||||
"data": "mail.example.com"
|
||||
})
|
||||
assert result is not None
|
||||
|
||||
# Valid: With priority
|
||||
result = await client.call_tool("validate_dns_record", {
|
||||
"record_type": "MX",
|
||||
"name": "@",
|
||||
"data": "mail.example.com",
|
||||
"priority": 10
|
||||
})
|
||||
assert result is not None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__])
|
||||
Loading…
x
Reference in New Issue
Block a user