forked from rsp2k/mcp-mailu
Ryan Malloy
66d1c0732a
- Complete FastMCP server with OpenAPI integration and fallback tools - Automatic tool generation from Mailu REST API endpoints - Bearer token authentication support - Comprehensive test suite and documentation - PyPI-ready package configuration with proper metadata - Environment-based configuration support - Production-ready error handling and logging - Examples and publishing scripts included Features: - User management (list, create, update, delete) - Domain management (list, create, update, delete) - Alias management and email forwarding - DKIM key generation - Manager assignment for domains - Graceful fallback when OpenAPI validation fails Ready for Claude Desktop integration and PyPI distribution.
231 lines
8.5 KiB
Python
231 lines
8.5 KiB
Python
"""Tests for the MCP Mailu server."""
|
|
|
|
import pytest
|
|
import httpx
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
import os
|
|
|
|
from mcp_mailu.server import create_mailu_client, create_mcp_server, MAILU_OPENAPI_SPEC
|
|
|
|
|
|
class TestMailuClient:
|
|
"""Test cases for Mailu HTTP client creation."""
|
|
|
|
def test_create_mailu_client(self):
|
|
"""Test creation of authenticated HTTP client."""
|
|
base_url = "https://mail.example.com/api/v1"
|
|
api_token = "test-token"
|
|
|
|
client = create_mailu_client(base_url, api_token)
|
|
|
|
assert isinstance(client, httpx.AsyncClient)
|
|
assert client.base_url == base_url
|
|
assert client.headers["Authorization"] == "Bearer test-token"
|
|
assert client.headers["Content-Type"] == "application/json"
|
|
assert client.headers["Accept"] == "application/json"
|
|
|
|
|
|
class TestMCPServer:
|
|
"""Test cases for the MCP server initialization."""
|
|
|
|
@pytest.mark.asyncio
|
|
@patch.dict(os.environ, {
|
|
"MAILU_BASE_URL": "https://test.example.com",
|
|
"MAILU_API_TOKEN": "test-token"
|
|
})
|
|
async def test_create_mcp_server_with_env_vars(self):
|
|
"""Test MCP server creation with environment variables."""
|
|
with patch('mcp_mailu.server.FastMCP') as mock_fastmcp:
|
|
mock_server = Mock()
|
|
mock_fastmcp.from_openapi.return_value = mock_server
|
|
|
|
server = await create_mcp_server()
|
|
|
|
# Verify FastMCP.from_openapi was called
|
|
mock_fastmcp.from_openapi.assert_called_once()
|
|
|
|
# Get the call arguments
|
|
call_args = mock_fastmcp.from_openapi.call_args
|
|
|
|
# Verify client is an httpx.AsyncClient
|
|
assert isinstance(call_args.kwargs['client'], httpx.AsyncClient)
|
|
|
|
# Verify OpenAPI spec is passed
|
|
assert call_args.kwargs['openapi_spec'] is not None
|
|
|
|
# Verify route maps are configured
|
|
assert 'route_maps' in call_args.kwargs
|
|
assert len(call_args.kwargs['route_maps']) > 0
|
|
|
|
# Verify server info
|
|
assert call_args.kwargs['name'] == "Mailu MCP Server"
|
|
assert call_args.kwargs['version'] == "1.0.0"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch.dict(os.environ, {}, clear=True)
|
|
async def test_create_mcp_server_without_env_vars(self):
|
|
"""Test MCP server creation with default values."""
|
|
with patch('mcp_mailu.server.FastMCP') as mock_fastmcp:
|
|
mock_server = Mock()
|
|
mock_fastmcp.from_openapi.return_value = mock_server
|
|
|
|
server = await create_mcp_server()
|
|
|
|
# Should still create server with defaults
|
|
mock_fastmcp.from_openapi.assert_called_once()
|
|
|
|
call_args = mock_fastmcp.from_openapi.call_args
|
|
client = call_args.kwargs['client']
|
|
|
|
# Should use default base URL
|
|
assert "mail.example.com" in str(client.base_url)
|
|
|
|
def test_openapi_spec_structure(self):
|
|
"""Test that the OpenAPI spec has required structure."""
|
|
spec = MAILU_OPENAPI_SPEC
|
|
|
|
# Basic OpenAPI structure
|
|
assert spec["swagger"] == "2.0"
|
|
assert "info" in spec
|
|
assert "paths" in spec
|
|
assert "definitions" in spec
|
|
|
|
# Required API info
|
|
assert spec["info"]["title"] == "Mailu API"
|
|
assert spec["basePath"] == "/api/v1"
|
|
|
|
# Security configuration
|
|
assert "securityDefinitions" in spec
|
|
assert "Bearer" in spec["securityDefinitions"]
|
|
|
|
# Verify key endpoints exist
|
|
assert "/user" in spec["paths"]
|
|
assert "/domain" in spec["paths"]
|
|
assert "/alias" in spec["paths"]
|
|
|
|
# Verify user endpoints have CRUD operations
|
|
user_path = spec["paths"]["/user"]
|
|
assert "get" in user_path # List users
|
|
assert "post" in user_path # Create user
|
|
|
|
user_detail_path = spec["paths"]["/user/{email}"]
|
|
assert "get" in user_detail_path # Get user
|
|
assert "patch" in user_detail_path # Update user
|
|
assert "delete" in user_detail_path # Delete user
|
|
|
|
# Verify definitions exist
|
|
assert "UserCreate" in spec["definitions"]
|
|
assert "UserGet" in spec["definitions"]
|
|
assert "UserUpdate" in spec["definitions"]
|
|
assert "Domain" in spec["definitions"]
|
|
assert "Alias" in spec["definitions"]
|
|
|
|
def test_user_create_schema(self):
|
|
"""Test UserCreate schema has required fields."""
|
|
spec = MAILU_OPENAPI_SPEC
|
|
user_create = spec["definitions"]["UserCreate"]
|
|
|
|
# Required fields
|
|
assert "email" in user_create["required"]
|
|
assert "raw_password" in user_create["required"]
|
|
|
|
# Properties
|
|
properties = user_create["properties"]
|
|
assert "email" in properties
|
|
assert "raw_password" in properties
|
|
assert "quota_bytes" in properties
|
|
assert "enabled" in properties
|
|
assert "displayed_name" in properties
|
|
|
|
# Field types
|
|
assert properties["email"]["type"] == "string"
|
|
assert properties["raw_password"]["type"] == "string"
|
|
assert properties["quota_bytes"]["type"] == "integer"
|
|
assert properties["enabled"]["type"] == "boolean"
|
|
|
|
def test_domain_schema(self):
|
|
"""Test Domain schema structure."""
|
|
spec = MAILU_OPENAPI_SPEC
|
|
domain = spec["definitions"]["Domain"]
|
|
|
|
# Required fields
|
|
assert "name" in domain["required"]
|
|
|
|
# Properties
|
|
properties = domain["properties"]
|
|
assert "name" in properties
|
|
assert "max_users" in properties
|
|
assert "max_aliases" in properties
|
|
assert "signup_enabled" in properties
|
|
|
|
# Field types
|
|
assert properties["name"]["type"] == "string"
|
|
assert properties["max_users"]["type"] == "integer"
|
|
assert properties["signup_enabled"]["type"] == "boolean"
|
|
|
|
|
|
class TestEnvironmentConfiguration:
|
|
"""Test environment variable handling."""
|
|
|
|
@patch.dict(os.environ, {
|
|
"MAILU_BASE_URL": "https://custom.mail.com",
|
|
"MAILU_API_TOKEN": "custom-token"
|
|
})
|
|
def test_environment_variables_loaded(self):
|
|
"""Test that environment variables are properly loaded."""
|
|
# These would be used in create_mcp_server
|
|
assert os.getenv("MAILU_BASE_URL") == "https://custom.mail.com"
|
|
assert os.getenv("MAILU_API_TOKEN") == "custom-token"
|
|
|
|
@patch.dict(os.environ, {}, clear=True)
|
|
def test_default_values_when_no_env_vars(self):
|
|
"""Test default values when environment variables are not set."""
|
|
base_url = os.getenv("MAILU_BASE_URL", "https://mail.example.com")
|
|
api_token = os.getenv("MAILU_API_TOKEN", "")
|
|
|
|
assert base_url == "https://mail.example.com"
|
|
assert api_token == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestIntegration:
|
|
"""Integration tests for the complete server."""
|
|
|
|
@patch('mcp_mailu.server.FastMCP')
|
|
async def test_server_initialization_flow(self, mock_fastmcp):
|
|
"""Test the complete server initialization flow."""
|
|
mock_server = Mock()
|
|
mock_fastmcp.from_openapi.return_value = mock_server
|
|
|
|
with patch.dict(os.environ, {
|
|
"MAILU_BASE_URL": "https://test.mail.com",
|
|
"MAILU_API_TOKEN": "integration-token"
|
|
}):
|
|
server = await create_mcp_server()
|
|
|
|
# Verify server was created
|
|
assert server == mock_server
|
|
|
|
# Verify from_openapi was called with correct parameters
|
|
call_args = mock_fastmcp.from_openapi.call_args
|
|
|
|
# Check client configuration
|
|
client = call_args.kwargs['client']
|
|
assert isinstance(client, httpx.AsyncClient)
|
|
assert "test.mail.com" in str(client.base_url)
|
|
assert client.headers["Authorization"] == "Bearer integration-token"
|
|
|
|
# Check OpenAPI spec
|
|
spec = call_args.kwargs['openapi_spec']
|
|
assert spec['host'] == 'test.mail.com'
|
|
assert spec['schemes'] == ['https']
|
|
|
|
# Check route mappings
|
|
route_maps = call_args.kwargs['route_maps']
|
|
assert len(route_maps) > 0
|
|
|
|
# Verify some expected route patterns
|
|
patterns = [rm.pattern for rm in route_maps]
|
|
assert any("GET /user$" in pattern for pattern in patterns)
|
|
assert any("GET /domain$" in pattern for pattern in patterns)
|