diff --git a/fixed_test_mcp_server.py b/fixed_test_mcp_server.py deleted file mode 100644 index 07de342..0000000 --- a/fixed_test_mcp_server.py +++ /dev/null @@ -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__])