如何对 MCP 进行测试？有哪些测试策略和最佳实践？

MCP 的测试策略对于确保系统质量和可靠性至关重要。以下是详细的测试方法和最佳实践：

测试层次结构

MCP 测试应涵盖以下层次：

1. 单元测试：测试单个函数和组件
2. 集成测试：测试组件之间的交互
3. 端到端测试：测试完整的用户场景
4. 性能测试：测试系统性能和可扩展性
5. 安全测试：测试安全漏洞和防护机制

1. 单元测试

python
import pytest
from unittest.mock import Mock, AsyncMock
from mcp.server import Server

class TestMCPTools:
    @pytest.fixture
    def server(self):
        """创建测试服务器实例"""
        return Server("test-server")
    
    @pytest.mark.asyncio
    async def test_tool_registration(self, server):
        """测试工具注册"""
        @server.tool(
            name="test_tool",
            description="测试工具"
        )
        async def test_tool(param: str) -> str:
            return f"Result: {param}"
        
        # 验证工具已注册
        tools = await server.list_tools()
        assert any(t["name"] == "test_tool" for t in tools)
    
    @pytest.mark.asyncio
    async def test_tool_execution(self, server):
        """测试工具执行"""
        @server.tool(
            name="calculate",
            description="计算工具"
        )
        async def calculate(a: int, b: int) -> int:
            return a + b
        
        # 执行工具
        result = await server.call_tool("calculate", {"a": 2, "b": 3})
        assert result == 5
    
    @pytest.mark.asyncio
    async def test_parameter_validation(self, server):
        """测试参数验证"""
        @server.tool(
            name="validate",
            description="参数验证工具",
            inputSchema={
                "type": "object",
                "properties": {
                    "email": {
                        "type": "string",
                        "format": "email"
                    }
                },
                "required": ["email"]
            }
        )
        async def validate(email: str) -> str:
            return f"Valid: {email}"
        
        # 测试有效参数
        result = await server.call_tool(
            "validate",
            {"email": "test@example.com"}
        )
        assert "Valid" in result
        
        # 测试无效参数
        with pytest.raises(ValueError):
            await server.call_tool("validate", {"email": "invalid"})

2. 集成测试

python
class TestMCPIntegration:
    @pytest.mark.asyncio
    async def test_client_server_communication(self):
        """测试客户端-服务器通信"""
        from mcp.client import Client
        from mcp.server import Server
        
        # 创建服务器
        server = Server("integration-test-server")
        
        @server.tool(name="echo", description="回显工具")
        async def echo(message: str) -> str:
            return message
        
        # 启动服务器
        await server.start()
        
        try:
            # 创建客户端
            client = Client("http://localhost:8000")
            
            # 测试通信
            result = await client.call_tool("echo", {"message": "Hello"})
            assert result == "Hello"
        
        finally:
            await server.stop()
    
    @pytest.mark.asyncio
    async def test_resource_access(self):
        """测试资源访问"""
        server = Server("resource-test-server")
        
        @server.resource(
            uri="file:///test.txt",
            name="测试文件",
            description="测试资源"
        )
        async def test_resource() -> str:
            return "Test content"
        
        await server.start()
        
        try:
            client = Client("http://localhost:8000")
            
            # 读取资源
            content = await client.read_resource("file:///test.txt")
            assert content == "Test content"
        
        finally:
            await server.stop()

3. 端到端测试

python
class TestMCPEndToEnd:
    @pytest.mark.asyncio
    async def test_complete_workflow(self):
        """测试完整工作流"""
        # 模拟用户场景：查询数据库并生成报告
        
        server = Server("e2e-test-server")
        
        @server.tool(name="query_db", description="查询数据库")
        async def query_db(query: str) -> list:
            return [{"id": 1, "name": "Test"}]
        
        @server.tool(name="generate_report", description="生成报告")
        async def generate_report(data: list) -> str:
            return f"Report: {len(data)} items"
        
        await server.start()
        
        try:
            client = Client("http://localhost:8000")
            
            # 执行工作流
            data = await client.call_tool("query_db", {"query": "SELECT *"})
            report = await client.call_tool("generate_report", {"data": data})
            
            assert "1 items" in report
        
        finally:
            await server.stop()

4. 性能测试

python
import asyncio
import time
from locust import HttpUser, task, between

class MCPPerformanceTest(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        """测试开始时的初始化"""
        self.client = Client(self.host)
    
    @task
    def tool_call_performance(self):
        """测试工具调用性能"""
        start_time = time.time()
        
        result = self.client.call_tool("test_tool", {"param": "value"})
        
        elapsed = time.time() - start_time
        
        # 断言响应时间
        assert elapsed < 1.0, f"响应时间过长: {elapsed}s"
    
    @task
    def concurrent_requests(self):
        """测试并发请求"""
        async def make_request():
            return self.client.call_tool("test_tool", {"param": "value"})
        
        # 并发执行 10 个请求
        tasks = [make_request() for _ in range(10)]
        results = asyncio.run(asyncio.gather(*tasks))
        
        # 验证所有请求都成功
        assert all(results)

5. 安全测试

python
class TestMCPSecurity:
    @pytest.mark.asyncio
    async def test_authentication(self):
        """测试认证机制"""
        server = Server("security-test-server")
        
        # 配置认证
        server.set_authenticator(lambda token: token == "valid-token")
        
        @server.tool(name="secure_tool", description="安全工具")
        async def secure_tool() -> str:
            return "Secure data"
        
        await server.start()
        
        try:
            # 测试有效令牌
            client = Client("http://localhost:8000", token="valid-token")
            result = await client.call_tool("secure_tool", {})
            assert result == "Secure data"
            
            # 测试无效令牌
            client_invalid = Client("http://localhost:8000", token="invalid-token")
            with pytest.raises(AuthenticationError):
                await client_invalid.call_tool("secure_tool", {})
        
        finally:
            await server.stop()
    
    @pytest.mark.asyncio
    async def test_sql_injection_prevention(self):
        """测试 SQL 注入防护"""
        server = Server("sql-injection-test-server")
        
        @server.tool(name="query", description="查询工具")
        async def query(sql: str) -> list:
            # 应该使用参数化查询
            return execute_safe_query(sql)
        
        # 测试 SQL 注入尝试
        malicious_sql = "SELECT * FROM users WHERE '1'='1'"
        result = await server.call_tool("query", {"sql": malicious_sql})
        
        # 验证注入被阻止
        assert result == []
    
    @pytest.mark.asyncio
    async def test_rate_limiting(self):
        """测试速率限制"""
        server = Server("rate-limit-test-server")
        
        # 配置速率限制
        server.set_rate_limit(max_requests=10, window=60)
        
        @server.tool(name="limited_tool", description="受限工具")
        async def limited_tool() -> str:
            return "Success"
        
        # 快速发送多个请求
        for i in range(15):
            try:
                await server.call_tool("limited_tool", {})
            except RateLimitError:
                # 预期的速率限制错误
                assert i >= 10
                break
        else:
            pytest.fail("未触发速率限制")

6. Mock 和 Stub

python
from unittest.mock import Mock, patch

class TestMCPWithMocks:
    @pytest.mark.asyncio
    async def test_with_external_dependency_mock(self):
        """使用 Mock 测试外部依赖"""
        server = Server("mock-test-server")
        
        @server.tool(name="fetch_data", description="获取数据")
        async def fetch_data(url: str) -> dict:
            # Mock 外部 API 调用
            with patch('requests.get') as mock_get:
                mock_get.return_value.json.return_value = {
                    "data": "mocked"
                }
                
                response = requests.get(url)
                return response.json()
        
        result = await server.call_tool(
            "fetch_data",
            {"url": "http://api.example.com"}
        )
        
        assert result == {"data": "mocked"}
        mock_get.assert_called_once()

7. 测试覆盖率

python
# 使用 pytest-cov 生成覆盖率报告
# 运行命令: pytest --cov=mcp --cov-report=html

class TestCoverage:
    @pytest.mark.asyncio
    async def test_all_code_paths(self):
        """测试所有代码路径"""
        server = Server("coverage-test-server")
        
        @server.tool(name="complex_tool", description="复杂工具")
        async def complex_tool(condition: bool) -> str:
            if condition:
                return "Branch A"
            else:
                return "Branch B"
        
        # 测试所有分支
        result_a = await server.call_tool("complex_tool", {"condition": True})
        assert result_a == "Branch A"
        
        result_b = await server.call_tool("complex_tool", {"condition": False})
        assert result_b == "Branch B"

最佳实践：

1. 测试金字塔：大量单元测试，适量集成测试，少量端到端测试
2. 独立性：每个测试应该独立运行，不依赖其他测试
3. 可重复性：测试结果应该可重复，不受环境因素影响
4. 快速反馈：单元测试应该快速执行，提供快速反馈
5. 持续集成：将测试集成到 CI/CD 流程中
6. 覆盖率目标：设定合理的代码覆盖率目标（如 80%）

通过完善的测试策略，可以确保 MCP 系统的质量和可靠性。