Testing Guidelines - iowarp/iowarp-mcps GitHub Wiki

Unit Tests

Create comprehensive unit tests for each tool to ensure reliability and correctness.

Test Structure

import pytest
import asyncio
from unittest.mock import Mock, patch
from your_mcp.server import app

class TestYourMCP:
    @pytest.fixture
    async def server(self):
        """Setup test server instance"""
        return app
    
    @pytest.mark.asyncio
    async def test_tool_normal_operation(self, server):
        """Test tool with valid inputs"""
        result = await server.call_tool("tool_name", {"param": "valid_value"})
        assert result[0].text == "expected_output"
    
    @pytest.mark.asyncio
    async def test_tool_invalid_input(self, server):
        """Test tool with invalid inputs"""
        result = await server.call_tool("tool_name", {"param": ""})
        assert "Error" in result[0].text
    
    @pytest.mark.asyncio
    async def test_tool_missing_parameter(self, server):
        """Test tool with missing required parameters"""
        result = await server.call_tool("tool_name", {})
        assert "required" in result[0].text.lower()

Testing Best Practices

Normal Operation Testing

• Test with typical, valid inputs
• Verify expected outputs and formats
• Check return types and structures
• Validate tool schema compliance

Error Handling Testing

• Test with invalid input types
• Test with missing required parameters
• Test with malformed data
• Test with boundary conditions
• Test with extremely large inputs

Edge Cases

@pytest.mark.asyncio
async def test_empty_file(self, server):
    """Test handling of empty files"""
    with patch('builtins.open', mock_open(read_data="")):
        result = await server.call_tool("read_file", {"path": "empty.txt"})
        assert "empty" in result[0].text.lower()

@pytest.mark.asyncio
async def test_large_dataset(self, server):
    """Test with large dataset"""
    large_data = ["data"] * 10000
    result = await server.call_tool("process_data", {"data": large_data})
    assert result is not None

Async Testing Support

import pytest_asyncio

@pytest_asyncio.fixture
async def async_setup():
    """Setup for async tests"""
    # Async setup code
    yield
    # Async cleanup code

@pytest.mark.asyncio
async def test_async_operation(async_setup):
    """Test async operations"""
    result = await some_async_function()
    assert result == expected_value

Integration Tests

Test with actual MCP clients to ensure end-to-end functionality.

Universal Client Testing

# Install client dependencies
uv pip install -r bin/requirements.txt

# Test with different LLM providers
python bin/wrp.py --conf=bin/confs/Gemini.yaml --test-mode

# Test specific tool
python bin/wrp.py --tool="your_tool" --params='{"param": "value"}'

Unified Launcher Testing

# Test server launch
uvx iowarp-mcps your-server

# Test with specific configuration
uvx iowarp-mcps your-server --config=test-config.json

Process Testing

import subprocess
import json
import asyncio

async def test_server_process():
    """Test server as separate process"""
    # Start server process
    process = await asyncio.create_subprocess_exec(
        "python", "-m", "your_mcp.server",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    # Wait for server to start
    await asyncio.sleep(2)
    
    # Test communication
    # ... MCP protocol testing
    
    # Cleanup
    process.terminate()
    await process.wait()

Protocol Verification

import json
from mcp.client import ClientSession

async def test_mcp_protocol():
    """Test MCP protocol compliance"""
    # Connect to server
    session = ClientSession()
    
    # Test tool listing
    tools = await session.list_tools()
    assert len(tools) > 0
    
    # Test tool execution
    result = await session.call_tool("tool_name", {"param": "value"})
    assert result.content[0].text is not None
    
    # Test error handling
    try:
        await session.call_tool("nonexistent_tool", {})
        assert False, "Should have raised exception"
    except Exception as e:
        assert "not found" in str(e).lower()

Test Coverage

Coverage Requirements

• Aim for >90% code coverage
• Cover all public functions and methods
• Test all error paths
• Include integration test coverage

Coverage Tools

# Install coverage tools
pip install pytest-cov coverage

# Run tests with coverage
pytest --cov=your_mcp tests/

# Generate coverage report
coverage html
coverage report --show-missing

Coverage Configuration

# .coveragerc
[run]
source = your_mcp
omit = 
    */tests/*
    */venv/*
    */build/*

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    raise AssertionError
    raise NotImplementedError

Performance Testing

Load Testing

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

async def performance_test():
    """Test tool performance under load"""
    start_time = time.time()
    
    # Run multiple concurrent requests
    tasks = []
    for i in range(100):
        task = asyncio.create_task(
            server.call_tool("tool_name", {"param": f"value_{i}"})
        )
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    end_time = time.time()
    
    # Verify all requests succeeded
    assert all(result is not None for result in results)
    
    # Check performance metrics
    total_time = end_time - start_time
    requests_per_second = len(tasks) / total_time
    
    print(f"Processed {len(tasks)} requests in {total_time:.2f}s")
    print(f"Rate: {requests_per_second:.2f} requests/second")
    
    # Assert performance requirements
    assert requests_per_second > 10  # Minimum performance threshold

Memory Testing

import psutil
import gc

def test_memory_usage():
    """Test memory usage during operations"""
    process = psutil.Process()
    initial_memory = process.memory_info().rss
    
    # Perform memory-intensive operation
    large_data = generate_large_dataset()
    result = process_large_data(large_data)
    
    # Force garbage collection
    del large_data
    gc.collect()
    
    final_memory = process.memory_info().rss
    memory_increase = final_memory - initial_memory
    
    # Assert memory usage is reasonable
    assert memory_increase < 100 * 1024 * 1024  # Less than 100MB increase

Continuous Integration

GitHub Actions Configuration

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: [3.10, 3.11, 3.12]
    
    steps:
    - uses: actions/checkout@v4
    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install pytest pytest-cov pytest-asyncio
        pip install -e .
    
    - name: Run tests
      run: |
        pytest --cov=your_mcp --cov-report=xml tests/
    
    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml

Test Organization

Directory Structure

tests/
├── unit/
│   ├── test_tools.py
│   ├── test_resources.py
│   └── test_utils.py
├── integration/
│   ├── test_client_interaction.py
│   └── test_protocol_compliance.py
├── performance/
│   ├── test_load.py
│   └── test_memory.py
└── fixtures/
    ├── sample_data.json
    └── test_files/

Test Configuration

# conftest.py
import pytest
import asyncio
from pathlib import Path

@pytest.fixture(scope="session")
def event_loop():
    """Create event loop for async tests"""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

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

@pytest.fixture
def temp_file():
    """Create temporary file for testing"""
    import tempfile
    with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
        f.write("test content")
        temp_path = f.name
    
    yield temp_path
    
    # Cleanup
    import os
    os.unlink(temp_path)