prefecthq--fastmcp/tests/server/providers/openapi/test_performance_comparison.py

"""Performance tests for OpenAPIProvider implementation."""

import gc
import time

import httpx
import pytest

from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider


def create_openapi_server(
    openapi_spec: dict,
    client,
    name: str = "OpenAPI Server",
) -> FastMCP:
    """Helper to create a FastMCP server with OpenAPIProvider."""
    provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
    mcp = FastMCP(name)
    mcp.add_provider(provider)
    return mcp


class TestPerformance:
    """Test performance of OpenAPIProvider implementation."""

    @pytest.fixture
    def comprehensive_spec(self):
        """Comprehensive OpenAPI spec for performance testing."""
        return {
            "openapi": "3.0.0",
            "info": {"title": "Performance Test API", "version": "1.0.0"},
            "paths": {
                "/users": {
                    "get": {
                        "operationId": "list_users",
                        "summary": "List users",
                        "parameters": [
                            {
                                "name": "limit",
                                "in": "query",
                                "required": False,
                                "schema": {"type": "integer", "default": 10},
                            },
                            {
                                "name": "offset",
                                "in": "query",
                                "required": False,
                                "schema": {"type": "integer", "default": 0},
                            },
                        ],
                        "responses": {"200": {"description": "Users listed"}},
                    },
                    "post": {
                        "operationId": "create_user",
                        "summary": "Create user",
                        "requestBody": {
                            "required": True,
                            "content": {
                                "application/json": {
                                    "schema": {
                                        "type": "object",
                                        "properties": {
                                            "name": {"type": "string"},
                                            "email": {"type": "string"},
                                            "age": {"type": "integer"},
                                        },
                                        "required": ["name", "email"],
                                    }
                                }
                            },
                        },
                        "responses": {"201": {"description": "User created"}},
                    },
                },
                "/users/{id}": {
                    "get": {
                        "operationId": "get_user",
                        "summary": "Get user",
                        "parameters": [
                            {
                                "name": "id",
                                "in": "path",
                                "required": True,
                                "schema": {"type": "integer"},
                            }
                        ],
                        "responses": {"200": {"description": "User found"}},
                    },
                    "put": {
                        "operationId": "update_user",
                        "summary": "Update user",
                        "parameters": [
                            {
                                "name": "id",
                                "in": "path",
                                "required": True,
                                "schema": {"type": "integer"},
                            }
                        ],
                        "requestBody": {
                            "required": True,
                            "content": {
                                "application/json": {
                                    "schema": {
                                        "type": "object",
                                        "properties": {
                                            "name": {"type": "string"},
                                            "email": {"type": "string"},
                                            "age": {"type": "integer"},
                                        },
                                    }
                                }
                            },
                        },
                        "responses": {"200": {"description": "User updated"}},
                    },
                    "delete": {
                        "operationId": "delete_user",
                        "summary": "Delete user",
                        "parameters": [
                            {
                                "name": "id",
                                "in": "path",
                                "required": True,
                                "schema": {"type": "integer"},
                            }
                        ],
                        "responses": {"204": {"description": "User deleted"}},
                    },
                },
                "/search": {
                    "get": {
                        "operationId": "search_users",
                        "summary": "Search users",
                        "parameters": [
                            {
                                "name": "q",
                                "in": "query",
                                "required": True,
                                "schema": {"type": "string"},
                            },
                            {
                                "name": "filters",
                                "in": "query",
                                "required": False,
                                "style": "deepObject",
                                "explode": True,
                                "schema": {
                                    "type": "object",
                                    "properties": {
                                        "age_min": {"type": "integer"},
                                        "age_max": {"type": "integer"},
                                        "status": {
                                            "type": "string",
                                            "enum": ["active", "inactive"],
                                        },
                                    },
                                },
                            },
                        ],
                        "responses": {"200": {"description": "Search results"}},
                    }
                },
            },
        }

    def test_provider_initialization_performance(self, comprehensive_spec):
        """Test that provider initialization is fast (serverless requirement)."""
        num_iterations = 5

        # Measure provider initialization
        times = []
        for _ in range(num_iterations):
            client = httpx.AsyncClient(base_url="https://api.example.com")
            start_time = time.time()
            provider = OpenAPIProvider(
                openapi_spec=comprehensive_spec,
                client=client,
            )
            # Ensure provider is fully initialized
            assert provider is not None
            end_time = time.time()
            times.append(end_time - start_time)

        avg_time = sum(times) / len(times)
        max_acceptable_time = 0.2  # 200ms (Windows CI runners regularly clip 100ms)

        print(f"Average initialization time: {avg_time:.4f}s")
        print(f"Performance: {'✓' if avg_time < max_acceptable_time else '✗'}")

        # Should initialize in under 200ms for serverless requirements
        assert avg_time < max_acceptable_time, (
            f"Provider should initialize in under 200ms, got {avg_time:.4f}s"
        )

    def test_server_initialization_performance(self, comprehensive_spec):
        """Test that full server initialization is fast."""
        num_iterations = 5

        times = []
        for _ in range(num_iterations):
            client = httpx.AsyncClient(base_url="https://api.example.com")
            start_time = time.time()
            server = create_openapi_server(
                openapi_spec=comprehensive_spec,
                client=client,
                name="Performance Test",
            )
            # Ensure server is fully initialized
            assert server is not None
            end_time = time.time()
            times.append(end_time - start_time)

        avg_time = sum(times) / len(times)
        max_acceptable_time = 0.2  # 200ms (Windows CI runners regularly clip 100ms)

        print(f"Average server initialization time: {avg_time:.4f}s")

        assert avg_time < max_acceptable_time, (
            f"Server should initialize in under 200ms, got {avg_time:.4f}s"
        )

    async def test_functionality_after_optimization(self, comprehensive_spec):
        """Verify that performance optimization doesn't break functionality."""
        client = httpx.AsyncClient(base_url="https://api.example.com")

        server = create_openapi_server(
            openapi_spec=comprehensive_spec,
            client=client,
            name="Test Server",
        )

        # Get tools from the server via public API
        tools = await server.list_tools()

        # Should have 6 operations in the spec
        assert len(tools) == 6

        # Expected operations
        expected_operations = {
            "list_users",
            "create_user",
            "get_user",
            "update_user",
            "delete_user",
            "search_users",
        }
        assert {t.name for t in tools} == expected_operations

    def test_memory_efficiency(self, comprehensive_spec):
        """Test that implementation doesn't significantly increase memory usage."""

        # Helper to count total tools across all providers
        def count_provider_tools(server):
            total = 0
            for provider in server.providers:
                if hasattr(provider, "_tools"):
                    total += len(provider._tools)
            return total

        gc.collect()  # Clean up before baseline
        baseline_refs = len(gc.get_objects())

        servers = []
        for i in range(10):
            client = httpx.AsyncClient(base_url="https://api.example.com")
            server = create_openapi_server(
                openapi_spec=comprehensive_spec,
                client=client,
                name=f"Memory Test Server {i}",
            )
            servers.append(server)

        # Servers should all be functional
        assert len(servers) == 10
        assert all(count_provider_tools(s) == 6 for s in servers)

        # Memory usage shouldn't explode
        gc.collect()
        current_refs = len(gc.get_objects())
        growth_ratio = current_refs / max(baseline_refs, 1)
        assert growth_ratio < 3.0, (
            f"Memory usage grew by {growth_ratio}x, which seems excessive"
        )