Python Testing¶

# tests/test_tools.py
import pytest
from unittest.mock import AsyncMock, patch
from mcp.types import TextContent

from my_mcp_server.tools.database import _execute_query

@pytest.mark.asyncio
async def test_execute_query():
    """Test database query execution."""
    args = {"query": "SELECT * FROM users", "limit": 10}
    
    with patch("my_mcp_server.tools.database.db_client") as mock_db:
        mock_db.execute.return_value = [{"id": 1, "name": "Alice"}]
        
        result = await _execute_query(args)
        
        assert len(result) == 1
        assert isinstance(result[0], TextContent)
        assert "Alice" in result[0].text

# tests/test_server.py
import pytest
from mcp.server import Server
from my_mcp_server.server import MCPServer

@pytest.fixture
def server():
    """Create test server instance."""
    return MCPServer("test-server")

@pytest.mark.asyncio
async def test_server_initialization(server):
    """Test server initializes correctly."""
    assert server.server.name == "test-server"
    # Test that handlers are registered
    tools = await server.server._list_tools_handler()
    assert len(tools) > 0

# tests/test_integration.py
import asyncio
import json
from io import StringIO
from mcp.server.stdio import stdio_server

@pytest.mark.asyncio
async def test_mcp_protocol_integration():
    """Test full MCP protocol integration."""
    from my_mcp_server.server import MCPServer
    
    server = MCPServer("test-server")
    
    # Simulate stdin/stdout
    stdin_data = json.dumps({
        "jsonrpc": "2.0",
        "method": "tools/list",
        "id": 1
    }) + "\n"
    
    stdin = StringIO(stdin_data)
    stdout = StringIO()
    
    # This would require custom stdio handling for testing
    # In practice, use MCP test utilities when available

# tests/conftest.py
import pytest
from unittest.mock import AsyncMock

@pytest.fixture
def mock_database():
    """Mock database client."""
    mock_client = AsyncMock()
    mock_client.execute.return_value = [{"id": 1, "name": "test"}]
    mock_client.connect.return_value = True
    return mock_client

@pytest.fixture(autouse=True)
def patch_database(mock_database, monkeypatch):
    """Automatically patch database in all tests."""
    monkeypatch.setattr(
        "my_mcp_server.tools.database.get_db_client",
        lambda: mock_database
    )

# tests/test_api_tools.py
import pytest
import aiohttp
from aioresponses import aioresponses

@pytest.mark.asyncio
async def test_api_call():
    """Test external API calls."""
    with aioresponses() as m:
        m.get(
            "https://api.example.com/users/1",
            payload={"id": 1, "name": "Alice"}
        )
        
        result = await fetch_user_data(1)
        assert result["name"] == "Alice"

# tests/fixtures.py
import pytest
from pathlib import Path

@pytest.fixture
def sample_data():
    """Load sample test data."""
    fixture_path = Path(__file__).parent / "fixtures" / "sample_data.json"
    with open(fixture_path) as f:
        return json.load(f)

@pytest.fixture
def temp_file(tmp_path):
    """Create temporary test file."""
    test_file = tmp_path / "test.txt"
    test_file.write_text("test content")
    return test_file

# tests/assertions.py
def assert_valid_mcp_response(response):
    """Assert response is valid MCP format."""
    assert isinstance(response, list)
    for item in response:
        assert hasattr(item, 'type')
        if item.type == 'text':
            assert hasattr(item, 'text')
            assert isinstance(item.text, str)

@pytest.mark.asyncio
async def test_invalid_tool_name():
    """Test error handling for invalid tool names."""
    server = MCPServer("test-server")
    
    with pytest.raises(ValueError, match="Unknown tool"):
        await server.server._call_tool_handler(
            "nonexistent_tool",
            {"arg": "value"}
        )

@pytest.mark.asyncio
async def test_missing_required_argument():
    """Test error handling for missing arguments."""
    with pytest.raises(ValueError, match="required"):
        await call_tool_with_validation("query_database", {})

@pytest.mark.parametrize("invalid_input,expected_error", [
    ("", "Query cannot be empty"),
    (None, "Query is required"),
    ("SELECT * FROM users; DROP TABLE users;", "Invalid query"),
])
@pytest.mark.asyncio
async def test_query_validation(invalid_input, expected_error):
    """Test query validation."""
    with pytest.raises(ValueError, match=expected_error):
        await validate_query(invalid_input)

import time

@pytest.mark.asyncio
async def test_response_time():
    """Test response time requirements."""
    start_time = time.time()
    
    result = await execute_expensive_operation()
    
    elapsed = time.time() - start_time
    assert elapsed < 1.0  # Should complete within 1 second

@pytest.mark.asyncio
async def test_concurrent_requests():
    """Test handling multiple concurrent requests."""
    server = MCPServer("test-server")
    
    # Create 10 concurrent requests
    tasks = [
        server.call_tool("simple_tool", {"id": i})
        for i in range(10)
    ]
    
    results = await asyncio.gather(*tasks)
    assert len(results) == 10