mcp-protocol-compliance-testing

What this skill does

# MCP Protocol Compliance Testing Skill

## Metadata (Tier 1)

**Keywords**: testing, protocol, json-rpc, validation, pytest, compliance

**File Patterns**: **/test_*.py, **/tests/*.py

**Modes**: testing_backend

---

## Instructions (Tier 2)

### JSON-RPC 2.0 Message Validation

```python
import pytest
from pydantic import ValidationError
from server.protocol import JsonRpcRequest, JsonRpcResponse

class TestJsonRpcCompliance:
    """Test JSON-RPC 2.0 protocol compliance."""

    def test_valid_request(self):
        """Valid JSON-RPC 2.0 request passes validation."""
        request = JsonRpcRequest(
            jsonrpc="2.0",
            method="tools/call",
            params={"name": "search", "arguments": {}},
            id=1
        )
        assert request.jsonrpc == "2.0"
        assert request.method == "tools/call"
        assert request.id == 1

    def test_invalid_jsonrpc_version(self):
        """Invalid jsonrpc version raises ValidationError."""
        with pytest.raises(ValidationError) as exc_info:
            JsonRpcRequest(
                jsonrpc="1.0",  # Invalid
                method="tools/call",
                id=1
            )
        assert "jsonrpc" in str(exc_info.value).lower()

    def test_missing_required_fields(self):
        """Missing required fields raise ValidationError."""
        with pytest.raises(ValidationError):
            JsonRpcRequest(
                jsonrpc="2.0"
                # Missing method
            )

    def test_valid_response(self):
        """Valid JSON-RPC 2.0 response structure."""
        response = JsonRpcResponse(
            jsonrpc="2.0",
            result={"data": "test"},
            id=1
        )
        assert response.result == {"data": "test"}
        assert response.error is None

    def test_error_response(self):
        """Error response has proper structure."""
        error = JsonRpcError(
            code=-32601,
            message="Method not found",
            data={"method": "unknown"}
        )
        response = JsonRpcResponse(
            jsonrpc="2.0",
            error=error,
            id=1
        )
        assert response.error.code == -32601
        assert response.result is None
```

### Tool Schema Validation

```python
from tools.schemas import SearchInput, SearchOutput

class TestToolSchemas:
    """Test Pydantic tool schema compliance."""

    def test_valid_input(self):
        """Valid input passes strict mode validation."""
        input_data = SearchInput(
            query="test",
            limit=10,
            filter_type="code"
        )
        assert input_data.query == "test"
        assert input_data.limit == 10

    def test_strict_mode_prevents_coercion(self):
        """Strict mode blocks type coercion."""
        with pytest.raises(ValidationError) as exc_info:
            SearchInput(
                query="test",
                limit="10"  # String instead of int
            )
        # Verify it's a type error, not coercion
        assert "int" in str(exc_info.value).lower()

    def test_field_constraints(self):
        """Field constraints are enforced (ge, le, etc.)."""
        # Below minimum
        with pytest.raises(ValidationError):
            SearchInput(query="test", limit=0)

        # Above maximum
        with pytest.raises(ValidationError):
            SearchInput(query="test", limit=101)

    def test_literal_enum_validation(self):
        """Literal types only accept specified values."""
        # Valid literal
        input1 = SearchInput(query="test", filter_type="code")
        assert input1.filter_type == "code"

        # Invalid literal
        with pytest.raises(ValidationError):
            SearchInput(query="test", filter_type="invalid")

    def test_schema_generation(self):
        """Pydantic generates valid JSON Schema."""
        schema = SearchInput.model_json_schema()

        assert schema["type"] == "object"
        assert "query" in schema["properties"]
        assert "limit" in schema["properties"]

        # Verify constraints in schema
        limit_schema = schema["properties"]["limit"]
        assert limit_schema["minimum"] == 1
        assert limit_schema["maximum"] == 100

    def test_output_serialization(self):
        """Output models serialize correctly."""
        output = SearchOutput(
            results=[{"file": "test.py", "line": 10}],
            total_count=1,
            execution_time_ms=42
        )

        dumped = output.model_dump()
        assert isinstance(dumped, dict)
        assert dumped["total_count"] == 1
```

### Async Tool Execution Tests

```python
import pytest
import asyncio

class TestAsyncToolExecution:
    """Test async tool execution patterns."""

    @pytest.mark.asyncio
    async def test_tool_executes_async(self, mcp_server):
        """Tools execute without blocking."""
        result = await mcp_server.call_tool(
            "search_code",
            {"query": "async def", "limit": 5}
        )
        assert result is not None
        assert "results" in result

    @pytest.mark.asyncio
    async def test_concurrent_execution(self, mcp_server):
        """Multiple tools execute concurrently."""
        start = asyncio.get_event_loop().time()

        # Execute concurrently using TaskGroup
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(
                mcp_server.call_tool("search", {"query": "test1"})
            )
            t2 = tg.create_task(
                mcp_server.call_tool("search", {"query": "test2"})
            )

        duration = asyncio.get_event_loop().time() - start

        # Concurrent execution should be faster than sequential
        assert duration < 2.0
        assert t1.result() is not None
        assert t2.result() is not None

    @pytest.mark.asyncio
    async def test_tool_timeout(self, mcp_server):
        """Tool execution respects timeout."""
        with pytest.raises(asyncio.TimeoutError):
            await asyncio.wait_for(
                mcp_server.call_tool("slow_tool", {}),
                timeout=1.0
            )

    @pytest.mark.asyncio
    async def test_tool_cancellation(self, mcp_server):
        """Tool execution can be cancelled."""
        task = asyncio.create_task(
            mcp_server.call_tool("long_running_tool", {})
        )

        # Cancel after short delay
        await asyncio.sleep(0.1)
        task.cancel()

        with pytest.raises(asyncio.CancelledError):
            await task
```

### Integration Tests

```python
class TestMcpIntegration:
    """Full request/response cycle tests."""

    @pytest.mark.asyncio
    async def test_list_tools(self, mcp_server):
        """list_tools returns valid tool descriptors."""
        tools = await mcp_server.list_tools()

        assert len(tools) > 0
        for tool in tools:
            assert "name" in tool
            assert "description" in tool
            assert "inputSchema" in tool

            # Validate schema structure
            schema = tool["inputSchema"]
            assert schema["type"] == "object"
            assert "properties" in schema

    @pytest.mark.asyncio
    async def test_call_tool_success(self, mcp_server):
        """Successful tool call returns result."""
        result = await mcp_server.call_tool(
            "search_code",
            {"query": "test", "limit": 5}
        )

        assert "results" in result
        assert isinstance(result["results"], list)

    @pytest.mark.asyncio
    async def test_call_tool_invalid_name(self, mcp_server):
        """Invalid tool name returns error."""
        with pytest.raises(ValueError) as exc_info:
            await mcp_server.call_tool("nonexistent_tool", {})

        assert "unknown" in str(exc_info.value).lower()

    @pytest.mark.asyncio
    async def test_call_tool_invalid_arguments(self, mcp_server):
        """Invalid arguments raise ValidationError."""
        with pytest.raises(ValidationError):