the-nexus/mcp_servers/test_servers.py

#!/usr/bin/env python3
"""
Test script for MCP servers.
Validates that both desktop-control and steam-info servers respond correctly to MCP requests.
"""

import json
import subprocess
import sys
from typing import Dict, Any, Tuple, List


def send_request(server_script: str, request: Dict[str, Any]) -> Tuple[bool, Dict[str, Any], str]:
    """Send a JSON-RPC request to an MCP server and return the response."""
    try:
        proc = subprocess.run(
            ["python3", server_script],
            input=json.dumps(request) + "\n",
            capture_output=True,
            text=True,
            timeout=10
        )
        
        # Parse stdout for JSON-RPC response
        for line in proc.stdout.strip().split("\n"):
            line = line.strip()
            if line and line.startswith("{"):
                try:
                    response = json.loads(line)
                    if "jsonrpc" in response:
                        return True, response, ""
                except json.JSONDecodeError:
                    continue
        
        return False, {}, f"No valid JSON-RPC response found. stderr: {proc.stderr}"
    
    except subprocess.TimeoutExpired:
        return False, {}, "Server timed out"
    except Exception as e:
        return False, {}, str(e)


def test_desktop_control_server() -> List[str]:
    """Test the desktop control MCP server."""
    errors = []
    server = "mcp_servers/desktop_control_server.py"
    
    print("\n=== Testing Desktop Control Server ===")
    
    # Test initialize
    print("  Testing initialize...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {}
    })
    if not success:
        errors.append(f"initialize failed: {error}")
    elif "error" in response:
        errors.append(f"initialize returned error: {response['error']}")
    else:
        print("    ✓ initialize works")
    
    # Test tools/list
    print("  Testing tools/list...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/list",
        "params": {}
    })
    if not success:
        errors.append(f"tools/list failed: {error}")
    elif "error" in response:
        errors.append(f"tools/list returned error: {response['error']}")
    else:
        tools = response.get("result", {}).get("tools", [])
        expected_tools = [
            "take_screenshot", "get_screen_size", "get_mouse_position",
            "pixel_color", "click", "right_click", "move_to", "drag_to",
            "type_text", "press_key", "hotkey", "scroll", "get_os"
        ]
        tool_names = [t["name"] for t in tools]
        missing = [t for t in expected_tools if t not in tool_names]
        if missing:
            errors.append(f"Missing tools: {missing}")
        else:
            print(f"    ✓ tools/list works ({len(tools)} tools available)")
    
    # Test get_os (works without display)
    print("  Testing tools/call get_os...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 3,
        "method": "tools/call",
        "params": {"name": "get_os", "arguments": {}}
    })
    if not success:
        errors.append(f"get_os failed: {error}")
    elif "error" in response:
        errors.append(f"get_os returned error: {response['error']}")
    else:
        content = response.get("result", {}).get("content", [])
        if content and not response["result"].get("isError"):
            result_data = json.loads(content[0]["text"])
            if "system" in result_data:
                print(f"    ✓ get_os works (system: {result_data['system']})")
            else:
                errors.append("get_os response missing system info")
        else:
            errors.append("get_os returned error content")
    
    return errors


def test_steam_info_server() -> List[str]:
    """Test the Steam info MCP server."""
    errors = []
    server = "mcp_servers/steam_info_server.py"
    
    print("\n=== Testing Steam Info Server ===")
    
    # Test initialize
    print("  Testing initialize...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {}
    })
    if not success:
        errors.append(f"initialize failed: {error}")
    elif "error" in response:
        errors.append(f"initialize returned error: {response['error']}")
    else:
        print("    ✓ initialize works")
    
    # Test tools/list
    print("  Testing tools/list...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/list",
        "params": {}
    })
    if not success:
        errors.append(f"tools/list failed: {error}")
    elif "error" in response:
        errors.append(f"tools/list returned error: {response['error']}")
    else:
        tools = response.get("result", {}).get("tools", [])
        expected_tools = [
            "steam_recently_played", "steam_player_achievements",
            "steam_user_stats", "steam_current_players", "steam_news",
            "steam_app_details"
        ]
        tool_names = [t["name"] for t in tools]
        missing = [t for t in expected_tools if t not in tool_names]
        if missing:
            errors.append(f"Missing tools: {missing}")
        else:
            print(f"    ✓ tools/list works ({len(tools)} tools available)")
    
    # Test steam_current_players (mock mode)
    print("  Testing tools/call steam_current_players...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 3,
        "method": "tools/call",
        "params": {"name": "steam_current_players", "arguments": {"app_id": "261550"}}
    })
    if not success:
        errors.append(f"steam_current_players failed: {error}")
    elif "error" in response:
        errors.append(f"steam_current_players returned error: {response['error']}")
    else:
        content = response.get("result", {}).get("content", [])
        if content and not response["result"].get("isError"):
            result_data = json.loads(content[0]["text"])
            if "player_count" in result_data:
                mode = "mock" if result_data.get("mock") else "live"
                print(f"    ✓ steam_current_players works ({mode} mode, {result_data['player_count']} players)")
            else:
                errors.append("steam_current_players response missing player_count")
        else:
            errors.append("steam_current_players returned error content")
    
    # Test steam_recently_played (mock mode)
    print("  Testing tools/call steam_recently_played...")
    success, response, error = send_request(server, {
        "jsonrpc": "2.0",
        "id": 4,
        "method": "tools/call",
        "params": {"name": "steam_recently_played", "arguments": {"user_id": "12345"}}
    })
    if not success:
        errors.append(f"steam_recently_played failed: {error}")
    elif "error" in response:
        errors.append(f"steam_recently_played returned error: {response['error']}")
    else:
        content = response.get("result", {}).get("content", [])
        if content and not response["result"].get("isError"):
            result_data = json.loads(content[0]["text"])
            if "games" in result_data:
                print(f"    ✓ steam_recently_played works ({len(result_data['games'])} games)")
            else:
                errors.append("steam_recently_played response missing games")
        else:
            errors.append("steam_recently_played returned error content")
    
    return errors


def main():
    """Run all tests."""
    print("=" * 60)
    print("MCP Server Test Suite")
    print("=" * 60)
    
    all_errors = []
    
    all_errors.extend(test_desktop_control_server())
    all_errors.extend(test_steam_info_server())
    
    print("\n" + "=" * 60)
    if all_errors:
        print(f"FAILED: {len(all_errors)} error(s)")
        for err in all_errors:
            print(f"  - {err}")
        sys.exit(1)
    else:
        print("ALL TESTS PASSED")
        print("=" * 60)
        sys.exit(0)


if __name__ == "__main__":
    main()