42Crunch MCP Server

"""Combined integration tests comparing HTTP and MCP stdio servers.

These tests assume both servers are running:
- HTTP server: python http_main.py (http://localhost:8000)
- MCP stdio server: python main.py
"""

import pytest
import requests
import subprocess
import json
import time
from typing import Dict, Any


class HTTPMCPClient:
    """HTTP client for testing."""
    
    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url.rstrip('/')
        self.jsonrpc_url = f"{self.base_url}/jsonrpc"
        self.request_id = 1
    
    def call_tool(self, name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
        """Call a tool."""
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "tools/call",
            "params": {
                "name": name,
                "arguments": arguments or {}
            }
        }
        self.request_id += 1
        
        response = requests.post(
            self.jsonrpc_url,
            json=request,
            headers={"Content-Type": "application/json"},
            timeout=30
        )
        response.raise_for_status()
        return response.json()


class MCPStdioClient:
    """Stdio client for testing."""
    
    def __init__(self, server_command: list[str] = None):
        self.server_command = server_command or ["python", "main.py"]
        self.process = None
        self.request_id = 1
    
    def start(self):
        """Start the server."""
        self.process = subprocess.Popen(
            self.server_command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1
        )
        time.sleep(0.5)
    
    def stop(self):
        """Stop the server."""
        if self.process:
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait()
    
    def call_tool(self, name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
        """Call a tool."""
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "tools/call",
            "params": {
                "name": name,
                "arguments": arguments or {}
            }
        }
        self.request_id += 1
        
        request_json = json.dumps(request) + "\n"
        self.process.stdin.write(request_json)
        self.process.stdin.flush()
        
        response_line = self.process.stdout.readline()
        return json.loads(response_line.strip())
    
    def __enter__(self):
        self.start()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stop()


@pytest.fixture(scope="module")
def http_client():
    """Create HTTP client."""
    client = HTTPMCPClient()
    try:
        response = requests.get(f"{client.base_url}/health", timeout=5)
        response.raise_for_status()
    except requests.exceptions.ConnectionError:
        pytest.skip("HTTP server is not running. Start it with: python http_main.py")
    return client


@pytest.fixture(scope="function")
def mcp_client():
    """Create MCP stdio client."""
    client = MCPStdioClient()
    try:
        client.start()
        # Test connection with real API
        try:
            client.call_tool("list_collections", {"per_page": 1})
        except Exception as e:
            pytest.skip(f"MCP server not responding properly: {e}")
        yield client
    finally:
        client.stop()


@pytest.fixture(scope="function")
def collection_id(http_client):
    """Get a collection ID from real API for comparison tests."""
    try:
        response = http_client.call_tool("list_collections", {"per_page": 10})
        if response["result"]["success"]:
            data = response["result"]["data"]
            if "collections" in data and len(data["collections"]) > 0:
                collection = data["collections"][0]
                return collection.get("id") or collection.get("collection_id")
            elif isinstance(data, list) and len(data) > 0:
                collection = data[0]
                return collection.get("id") or collection.get("collection_id")
    except Exception:
        pass
    pytest.skip("No collection ID available from real API")


@pytest.fixture(scope="function")
def api_id(http_client, collection_id):
    """Get an API ID from real API for comparison tests."""
    try:
        response = http_client.call_tool("get_collection_apis", {
            "collection_id": collection_id,
            "with_tags": True
        })
        if response["result"]["success"]:
            data = response["result"]["data"]
            if "apis" in data and len(data["apis"]) > 0:
                api = data["apis"][0]
                return api.get("id") or api.get("api_id")
            elif isinstance(data, list) and len(data) > 0:
                api = data[0]
                return api.get("id") or api.get("api_id")
    except Exception:
        pass
    pytest.skip("No API ID available from real API")


class TestConsistency:
    """Test consistency between HTTP and MCP stdio servers."""
    
    def test_list_collections_consistency(self, http_client, mcp_client):
        """Test that both servers return consistent results for list_collections from real API."""
        http_response = http_client.call_tool("list_collections", {
            "page": 1,
            "per_page": 5
        })
        
        mcp_response = mcp_client.call_tool("list_collections", {
            "page": 1,
            "per_page": 5
        })
        
        # Both should succeed with real API
        assert http_response["result"]["success"] is True, "HTTP server failed to get real collections"
        assert mcp_response["result"]["success"] is True, "MCP server failed to get real collections"
        
        # Both should have data from real API
        assert "data" in http_response["result"]
        assert "data" in mcp_response["result"]
        
        # Data structure should be similar (may not be identical due to timing)
        http_data = http_response["result"]["data"]
        mcp_data = mcp_response["result"]["data"]
        
        # Both should have similar keys or structure
        assert type(http_data) == type(mcp_data) or (
            isinstance(http_data, dict) and isinstance(mcp_data, dict)
        )
        
        # Both should return non-empty data from real API
        if isinstance(http_data, dict):
            assert len(http_data) > 0
        elif isinstance(http_data, list):
            assert len(http_data) >= 0  # May be empty but should be a list
    
    def test_get_collection_apis_consistency(self, http_client, mcp_client):
        """Test consistency for get_collection_apis with real API data."""
        # First get a collection ID from real API
        http_list = http_client.call_tool("list_collections", {"per_page": 10})
        
        assert http_list["result"]["success"] is True, "Failed to get collections from real API"
        
        # Extract collection ID (structure may vary)
        data = http_list["result"]["data"]
        collection_id = None
        
        if "collections" in data and len(data["collections"]) > 0:
            collection = data["collections"][0]
            collection_id = collection.get("id") or collection.get("collection_id")
        elif isinstance(data, list) and len(data) > 0:
            collection = data[0]
            collection_id = collection.get("id") or collection.get("collection_id")
        elif isinstance(data, dict):
            for key in ["items", "results", "data"]:
                if key in data and isinstance(data[key], list) and len(data[key]) > 0:
                    collection = data[key][0]
                    collection_id = collection.get("id") or collection.get("collection_id")
                    break
        
        assert collection_id is not None, "No collection ID found in real API response"
        
        # Test both servers with real collection ID
        http_response = http_client.call_tool("get_collection_apis", {
            "collection_id": collection_id,
            "with_tags": True
        })
        
        mcp_response = mcp_client.call_tool("get_collection_apis", {
            "collection_id": collection_id,
            "with_tags": True
        })
        
        # Both should succeed or fail consistently
        assert http_response["result"]["success"] == mcp_response["result"]["success"]
        
        if http_response["result"]["success"]:
            assert "data" in http_response["result"]
            assert "data" in mcp_response["result"]
            
            # Both should return similar data structures
            http_data = http_response["result"]["data"]
            mcp_data = mcp_response["result"]["data"]
            assert type(http_data) == type(mcp_data) or (
                isinstance(http_data, dict) and isinstance(mcp_data, dict)
            )
    
    def test_error_handling_consistency(self, http_client, mcp_client):
        """Test that both servers handle errors consistently."""
        # Test missing required parameter
        http_response = http_client.call_tool("get_collection_apis", {})
        mcp_response = mcp_client.call_tool("get_collection_apis", {})
        
        # Both should return error or success=False
        http_success = http_response.get("result", {}).get("success", True)
        mcp_success = mcp_response.get("result", {}).get("success", True)
        
        # Both should fail
        assert http_success is False or "error" in http_response
        assert mcp_success is False or "error" in mcp_response
    
    def test_response_format_consistency(self, http_client, mcp_client):
        """Test that both servers return consistent response formats."""
        http_response = http_client.call_tool("list_collections")
        mcp_response = mcp_client.call_tool("list_collections")
        
        # Both should have JSON-RPC structure
        assert "jsonrpc" in http_response
        assert "jsonrpc" in mcp_response
        assert http_response["jsonrpc"] == "2.0"
        assert mcp_response["jsonrpc"] == "2.0"
        
        # Both should have result
        assert "result" in http_response
        assert "result" in mcp_response
        
        # Result structure should be consistent
        http_result = http_response["result"]
        mcp_result = mcp_response["result"]
        
        assert "success" in http_result
        assert "success" in mcp_result
        assert isinstance(http_result["success"], bool)
        assert isinstance(mcp_result["success"], bool)


class TestPerformance:
    """Test performance comparison between HTTP and MCP."""
    
    def test_response_time_comparison(self, http_client, mcp_client):
        """Compare response times (informational only)."""
        import time
        
        # HTTP request
        start = time.time()
        http_response = http_client.call_tool("list_collections")
        http_time = time.time() - start
        
        # MCP request
        start = time.time()
        mcp_response = mcp_client.call_tool("list_collections")
        mcp_time = time.time() - start
        
        # Both should succeed
        assert http_response["result"]["success"] is True
        assert mcp_response["result"]["success"] is True
        
        # Log times (not asserting, just informational)
        print(f"\nHTTP response time: {http_time:.3f}s")
        print(f"MCP response time: {mcp_time:.3f}s")
    
    def test_concurrent_requests_http(self, http_client):
        """Test concurrent HTTP requests."""
        import concurrent.futures
        
        def make_request():
            return http_client.call_tool("list_collections", {"per_page": 1})
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(make_request) for _ in range(5)]
            results = [f.result() for f in concurrent.futures.as_completed(futures)]
        
        # All should succeed
        assert len(results) == 5
        for result in results:
            assert result["result"]["success"] is True


class TestFeatureParity:
    """Test that both servers have feature parity."""
    
    def test_tools_available(self, http_client, mcp_client):
        """Test that both servers expose the same tools."""
        # Get tools from HTTP
        http_tools_response = requests.post(
            http_client.jsonrpc_url,
            json={
                "jsonrpc": "2.0",
                "id": 1,
                "method": "tools/list"
            },
            timeout=5
        ).json()
        
        # Get tools from MCP
        mcp_tools_request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list"
        }
        mcp_tools_json = json.dumps(mcp_tools_request) + "\n"
        mcp_client.process.stdin.write(mcp_tools_json)
        mcp_client.process.stdin.flush()
        mcp_tools_line = mcp_client.process.stdout.readline()
        mcp_tools_response = json.loads(mcp_tools_line.strip())
        
        # Both should have tools
        http_tools = http_tools_response["result"]["tools"]
        mcp_tools = mcp_tools_response["result"]["tools"]
        
        # Should have same number of tools
        assert len(http_tools) == len(mcp_tools)
        
        # Should have same tool names
        http_names = {tool["name"] for tool in http_tools}
        mcp_names = {tool["name"] for tool in mcp_tools}
        assert http_names == mcp_names
    
    def test_error_message_consistency(self, http_client, mcp_client):
        """Test that both servers return consistent error messages."""
        # Test missing required parameter
        http_response = http_client.call_tool("get_collection_apis", {})
        mcp_response = mcp_client.call_tool("get_collection_apis", {})
        
        # Both should fail
        http_success = http_response.get("result", {}).get("success", True)
        mcp_success = mcp_response.get("result", {}).get("success", True)
        
        assert http_success is False
        assert mcp_success is False
        
        # Both should have error messages
        http_error = http_response.get("result", {}).get("error", http_response.get("error", {}).get("message", ""))
        mcp_error = mcp_response.get("result", {}).get("error", "")
        
        # Error messages should exist (may not be identical but should be present)
        assert http_error or http_response.get("error")
        assert mcp_error
    
    def test_response_time_consistency(self, http_client, mcp_client):
        """Test that both servers respond in reasonable time."""
        import time
        
        # HTTP request
        start = time.time()
        http_response = http_client.call_tool("list_collections")
        http_time = time.time() - start
        
        # MCP request
        start = time.time()
        mcp_response = mcp_client.call_tool("list_collections")
        mcp_time = time.time() - start
        
        # Both should succeed
        assert http_response["result"]["success"] is True
        assert mcp_response["result"]["success"] is True
        
        # Both should respond within reasonable time (less than 30 seconds)
        assert http_time < 30
        assert mcp_time < 30
        
        # Log times for comparison
        print(f"\nHTTP response time: {http_time:.3f}s")
        print(f"MCP response time: {mcp_time:.3f}s")