MCP/mcp-vultr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed

Browse Source
This commit is contained in:
Ryan Malloy 2025-06-11 18:14:19 -06:00
parent 5df5e9e8a0
commit 7273fe8539
1 changed files with 50 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
tests/test_mcp_server.py
View File

@ -1,8 +1,9 @@
"""Tests for MCP server functionality using FastMCP testing patterns.""" """Tests for MCP server functionality using official MCP testing patterns."""
import pytest import pytest
from unittest.mock import patch, AsyncMock from unittest.mock import patch, AsyncMock
from fastmcp import Client from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client
from vultr_dns_mcp.server import VultrDNSServer, create_mcp_server from vultr_dns_mcp.server import VultrDNSServer, create_mcp_server
@ -38,8 +39,9 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: # For the official MCP package, we need to use ClientSession
result = await client.call_tool("list_dns_domains", {}) async with ClientSession(server) as session:
result = await session.call_tool("list_dns_domains", {})
assert isinstance(result, list) assert isinstance(result, list)
# The result should be a list containing the response # The result should be a list containing the response
@ -55,8 +57,8 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("get_dns_domain", {"domain": "example.com"}) result = await session.call_tool("get_dns_domain", {"domain": "example.com"})
assert result is not None assert result is not None
mock_vultr_client.get_domain.assert_called_once_with("example.com") mock_vultr_client.get_domain.assert_called_once_with("example.com")
@ -67,8 +69,8 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("create_dns_domain", { result = await session.call_tool("create_dns_domain", {
"domain": "newdomain.com", "domain": "newdomain.com",
"ip": "192.168.1.100" "ip": "192.168.1.100"
}) })
@ -82,8 +84,8 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("delete_dns_domain", {"domain": "example.com"}) result = await session.call_tool("delete_dns_domain", {"domain": "example.com"})
assert result is not None assert result is not None
mock_vultr_client.delete_domain.assert_called_once_with("example.com") mock_vultr_client.delete_domain.assert_called_once_with("example.com")
@ -94,8 +96,8 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("list_dns_records", {"domain": "example.com"}) result = await session.call_tool("list_dns_records", {"domain": "example.com"})
assert result is not None assert result is not None
mock_vultr_client.list_records.assert_called_once_with("example.com") mock_vultr_client.list_records.assert_called_once_with("example.com")
@ -106,8 +108,8 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("create_dns_record", { result = await session.call_tool("create_dns_record", {
"domain": "example.com", "domain": "example.com",
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
@ -123,9 +125,9 @@ class TestMCPTools:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_validate_dns_record_tool(self, mcp_server): async def test_validate_dns_record_tool(self, mcp_server):
"""Test the validate_dns_record MCP tool.""" """Test the validate_dns_record MCP tool."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
# Test valid A record # Test valid A record
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
"data": "192.168.1.100", "data": "192.168.1.100",
@ -138,9 +140,9 @@ class TestMCPTools:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_validate_dns_record_invalid(self, mcp_server): async def test_validate_dns_record_invalid(self, mcp_server):
"""Test the validate_dns_record tool with invalid data.""" """Test the validate_dns_record tool with invalid data."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
# Test invalid A record (bad IP) # Test invalid A record (bad IP)
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
"data": "invalid-ip-address" "data": "invalid-ip-address"
@ -155,8 +157,8 @@ class TestMCPTools:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("analyze_dns_records", {"domain": "example.com"}) result = await session.call_tool("analyze_dns_records", {"domain": "example.com"})
assert result is not None assert result is not None
mock_vultr_client.list_records.assert_called_once_with("example.com") mock_vultr_client.list_records.assert_called_once_with("example.com")
@ -172,9 +174,9 @@ class TestMCPResources:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
# Get available resources # Get available resources
resources = await client.list_resources() resources = await session.list_resources()
# Check that domains resource is available # Check that domains resource is available
resource_uris = [r.uri for r in resources] resource_uris = [r.uri for r in resources]
@ -183,8 +185,8 @@ class TestMCPResources:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_capabilities_resource(self, mcp_server): async def test_capabilities_resource(self, mcp_server):
"""Test the vultr://capabilities resource.""" """Test the vultr://capabilities resource."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
resources = await client.list_resources() resources = await session.list_resources()
resource_uris = [r.uri for r in resources] resource_uris = [r.uri for r in resources]
assert "vultr://capabilities" in resource_uris assert "vultr://capabilities" in resource_uris
@ -194,13 +196,13 @@ class TestMCPResources:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
try: try:
result = await client.read_resource("vultr://domains") result = await session.read_resource("vultr://domains")
assert result is not None assert result is not None
mock_vultr_client.list_domains.assert_called_once() mock_vultr_client.list_domains.assert_called_once()
except Exception: except Exception:
# Resource reading might not be available in all FastMCP versions # Resource reading might not be available in all MCP versions
pass pass
@ -217,8 +219,8 @@ class TestMCPToolErrors:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
result = await client.call_tool("list_dns_domains", {}) result = await session.call_tool("list_dns_domains", {})
# Should handle the error gracefully # Should handle the error gracefully
assert result is not None assert result is not None
@ -226,10 +228,10 @@ class TestMCPToolErrors:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_missing_required_parameters(self, mcp_server): async def test_missing_required_parameters(self, mcp_server):
"""Test tool behavior with missing required parameters.""" """Test tool behavior with missing required parameters."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
with pytest.raises(Exception): with pytest.raises(Exception):
# This should fail due to missing required 'domain' parameter # This should fail due to missing required 'domain' parameter
await client.call_tool("get_dns_domain", {}) await session.call_tool("get_dns_domain", {})
@pytest.mark.integration @pytest.mark.integration
@ -242,21 +244,21 @@ class TestMCPIntegration:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
# 1. List domains # 1. List domains
domains = await client.call_tool("list_dns_domains", {}) domains = await session.call_tool("list_dns_domains", {})
assert domains is not None assert domains is not None
# 2. Get domain details # 2. Get domain details
domain_info = await client.call_tool("get_dns_domain", {"domain": "example.com"}) domain_info = await session.call_tool("get_dns_domain", {"domain": "example.com"})
assert domain_info is not None assert domain_info is not None
# 3. List records # 3. List records
records = await client.call_tool("list_dns_records", {"domain": "example.com"}) records = await session.call_tool("list_dns_records", {"domain": "example.com"})
assert records is not None assert records is not None
# 4. Analyze configuration # 4. Analyze configuration
analysis = await client.call_tool("analyze_dns_records", {"domain": "example.com"}) analysis = await session.call_tool("analyze_dns_records", {"domain": "example.com"})
assert analysis is not None assert analysis is not None
# Verify all expected API calls were made # Verify all expected API calls were made
@ -270,9 +272,9 @@ class TestMCPIntegration:
with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client): with patch('vultr_dns_mcp.server.VultrDNSServer', return_value=mock_vultr_client):
server = create_mcp_server("test-api-key") server = create_mcp_server("test-api-key")
async with Client(server) as client: async with ClientSession(server) as session:
# 1. Validate record before creation # 1. Validate record before creation
validation = await client.call_tool("validate_dns_record", { validation = await session.call_tool("validate_dns_record", {
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
"data": "192.168.1.100" "data": "192.168.1.100"
@ -280,7 +282,7 @@ class TestMCPIntegration:
assert validation is not None assert validation is not None
# 2. Create the record # 2. Create the record
create_result = await client.call_tool("create_dns_record", { create_result = await session.call_tool("create_dns_record", {
"domain": "example.com", "domain": "example.com",
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
@ -302,9 +304,9 @@ class TestValidationLogic:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_a_record_validation(self, mcp_server): async def test_a_record_validation(self, mcp_server):
"""Test A record validation logic.""" """Test A record validation logic."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
# Valid IPv4 # Valid IPv4
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
"data": "192.168.1.1" "data": "192.168.1.1"
@ -312,7 +314,7 @@ class TestValidationLogic:
assert result is not None assert result is not None
# Invalid IPv4 # Invalid IPv4
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "A", "record_type": "A",
"name": "www", "name": "www",
"data": "999.999.999.999" "data": "999.999.999.999"
@ -322,9 +324,9 @@ class TestValidationLogic:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_cname_validation(self, mcp_server): async def test_cname_validation(self, mcp_server):
"""Test CNAME record validation logic.""" """Test CNAME record validation logic."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
# Invalid: CNAME on root domain # Invalid: CNAME on root domain
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "CNAME", "record_type": "CNAME",
"name": "@", "name": "@",
"data": "example.com" "data": "example.com"
@ -332,7 +334,7 @@ class TestValidationLogic:
assert result is not None assert result is not None
# Valid: CNAME on subdomain # Valid: CNAME on subdomain
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "CNAME", "record_type": "CNAME",
"name": "www", "name": "www",
"data": "example.com" "data": "example.com"
@ -342,9 +344,9 @@ class TestValidationLogic:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_mx_validation(self, mcp_server): async def test_mx_validation(self, mcp_server):
"""Test MX record validation logic.""" """Test MX record validation logic."""
async with Client(mcp_server) as client: async with ClientSession(mcp_server) as session:
# Invalid: Missing priority # Invalid: Missing priority
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "MX", "record_type": "MX",
"name": "@", "name": "@",
"data": "mail.example.com" "data": "mail.example.com"
@ -352,7 +354,7 @@ class TestValidationLogic:
assert result is not None assert result is not None
# Valid: With priority # Valid: With priority
result = await client.call_tool("validate_dns_record", { result = await session.call_tool("validate_dns_record", {
"record_type": "MX", "record_type": "MX",
"name": "@", "name": "@",
"data": "mail.example.com", "data": "mail.example.com",