Spaces:
Running
Running
| """Tests for rate limiting middleware.""" | |
| import asyncio | |
| from unittest.mock import AsyncMock, MagicMock | |
| import pytest | |
| from fastmcp import FastMCP | |
| from fastmcp.client import Client | |
| from fastmcp.exceptions import ToolError | |
| from fastmcp.server.middleware.middleware import MiddlewareContext | |
| from fastmcp.server.middleware.rate_limiting import ( | |
| RateLimitError, | |
| RateLimitingMiddleware, | |
| SlidingWindowRateLimiter, | |
| SlidingWindowRateLimitingMiddleware, | |
| TokenBucketRateLimiter, | |
| ) | |
| def mock_context(): | |
| """Create a mock middleware context.""" | |
| context = MagicMock(spec=MiddlewareContext) | |
| context.method = "test_method" | |
| return context | |
| def mock_call_next(): | |
| """Create a mock call_next function.""" | |
| return AsyncMock(return_value="test_result") | |
| class TestTokenBucketRateLimiter: | |
| """Test token bucket rate limiter.""" | |
| def test_init(self): | |
| """Test initialization.""" | |
| limiter = TokenBucketRateLimiter(capacity=10, refill_rate=5.0) | |
| assert limiter.capacity == 10 | |
| assert limiter.refill_rate == 5.0 | |
| assert limiter.tokens == 10 | |
| async def test_consume_success(self): | |
| """Test successful token consumption.""" | |
| limiter = TokenBucketRateLimiter(capacity=10, refill_rate=5.0) | |
| # Should be able to consume tokens initially | |
| assert await limiter.consume(5) is True | |
| assert await limiter.consume(3) is True | |
| async def test_consume_failure(self): | |
| """Test failed token consumption.""" | |
| limiter = TokenBucketRateLimiter(capacity=5, refill_rate=1.0) | |
| # Consume all tokens | |
| assert await limiter.consume(5) is True | |
| # Should fail to consume more | |
| assert await limiter.consume(1) is False | |
| async def test_refill(self): | |
| """Test token refill over time.""" | |
| limiter = TokenBucketRateLimiter( | |
| capacity=10, refill_rate=10.0 | |
| ) # 10 tokens per second | |
| # Consume all tokens | |
| assert await limiter.consume(10) is True | |
| assert await limiter.consume(1) is False | |
| # Wait for refill (0.2 seconds = 2 tokens at 10/sec) | |
| await asyncio.sleep(0.2) | |
| assert await limiter.consume(2) is True | |
| class TestSlidingWindowRateLimiter: | |
| """Test sliding window rate limiter.""" | |
| def test_init(self): | |
| """Test initialization.""" | |
| limiter = SlidingWindowRateLimiter(max_requests=10, window_seconds=60) | |
| assert limiter.max_requests == 10 | |
| assert limiter.window_seconds == 60 | |
| assert len(limiter.requests) == 0 | |
| async def test_is_allowed_success(self): | |
| """Test allowing requests within limit.""" | |
| limiter = SlidingWindowRateLimiter(max_requests=3, window_seconds=60) | |
| # Should allow requests up to the limit | |
| assert await limiter.is_allowed() is True | |
| assert await limiter.is_allowed() is True | |
| assert await limiter.is_allowed() is True | |
| async def test_is_allowed_failure(self): | |
| """Test rejecting requests over limit.""" | |
| limiter = SlidingWindowRateLimiter(max_requests=2, window_seconds=60) | |
| # Should allow up to limit | |
| assert await limiter.is_allowed() is True | |
| assert await limiter.is_allowed() is True | |
| # Should reject over limit | |
| assert await limiter.is_allowed() is False | |
| async def test_sliding_window(self): | |
| """Test sliding window behavior.""" | |
| limiter = SlidingWindowRateLimiter(max_requests=2, window_seconds=1) | |
| # Use up requests | |
| assert await limiter.is_allowed() is True | |
| assert await limiter.is_allowed() is True | |
| assert await limiter.is_allowed() is False | |
| # Wait for window to pass | |
| await asyncio.sleep(1.1) | |
| # Should be able to make requests again | |
| assert await limiter.is_allowed() is True | |
| class TestRateLimitingMiddleware: | |
| """Test rate limiting middleware.""" | |
| def test_init_default(self): | |
| """Test default initialization.""" | |
| middleware = RateLimitingMiddleware() | |
| assert middleware.max_requests_per_second == 10.0 | |
| assert middleware.burst_capacity == 20 | |
| assert middleware.get_client_id is None | |
| assert middleware.global_limit is False | |
| def test_init_custom(self): | |
| """Test custom initialization.""" | |
| def get_client_id(ctx): | |
| return "test_client" | |
| middleware = RateLimitingMiddleware( | |
| max_requests_per_second=5.0, | |
| burst_capacity=10, | |
| get_client_id=get_client_id, | |
| global_limit=True, | |
| ) | |
| assert middleware.max_requests_per_second == 5.0 | |
| assert middleware.burst_capacity == 10 | |
| assert middleware.get_client_id is get_client_id | |
| assert middleware.global_limit is True | |
| def test_get_client_identifier_default(self, mock_context): | |
| """Test default client identifier.""" | |
| middleware = RateLimitingMiddleware() | |
| assert middleware._get_client_identifier(mock_context) == "global" | |
| def test_get_client_identifier_custom(self, mock_context): | |
| """Test custom client identifier.""" | |
| def get_client_id(ctx): | |
| return "custom_client" | |
| middleware = RateLimitingMiddleware(get_client_id=get_client_id) | |
| assert middleware._get_client_identifier(mock_context) == "custom_client" | |
| async def test_on_request_success(self, mock_context, mock_call_next): | |
| """Test successful request within rate limit.""" | |
| middleware = RateLimitingMiddleware(max_requests_per_second=100.0) # High limit | |
| result = await middleware.on_request(mock_context, mock_call_next) | |
| assert result == "test_result" | |
| assert mock_call_next.called | |
| async def test_on_request_rate_limited(self, mock_context, mock_call_next): | |
| """Test request rejection due to rate limiting.""" | |
| middleware = RateLimitingMiddleware( | |
| max_requests_per_second=1.0, burst_capacity=1 | |
| ) | |
| # First request should succeed | |
| await middleware.on_request(mock_context, mock_call_next) | |
| # Second request should be rate limited | |
| with pytest.raises(RateLimitError, match="Rate limit exceeded"): | |
| await middleware.on_request(mock_context, mock_call_next) | |
| async def test_global_rate_limiting(self, mock_context, mock_call_next): | |
| """Test global rate limiting.""" | |
| middleware = RateLimitingMiddleware( | |
| max_requests_per_second=1.0, burst_capacity=1, global_limit=True | |
| ) | |
| # First request should succeed | |
| await middleware.on_request(mock_context, mock_call_next) | |
| # Second request should be rate limited | |
| with pytest.raises(RateLimitError, match="Global rate limit exceeded"): | |
| await middleware.on_request(mock_context, mock_call_next) | |
| class TestSlidingWindowRateLimitingMiddleware: | |
| """Test sliding window rate limiting middleware.""" | |
| def test_init_default(self): | |
| """Test default initialization.""" | |
| middleware = SlidingWindowRateLimitingMiddleware(max_requests=100) | |
| assert middleware.max_requests == 100 | |
| assert middleware.window_seconds == 60 | |
| assert middleware.get_client_id is None | |
| def test_init_custom(self): | |
| """Test custom initialization.""" | |
| def get_client_id(ctx): | |
| return "test_client" | |
| middleware = SlidingWindowRateLimitingMiddleware( | |
| max_requests=50, window_minutes=5, get_client_id=get_client_id | |
| ) | |
| assert middleware.max_requests == 50 | |
| assert middleware.window_seconds == 300 # 5 minutes | |
| assert middleware.get_client_id is get_client_id | |
| async def test_on_request_success(self, mock_context, mock_call_next): | |
| """Test successful request within rate limit.""" | |
| middleware = SlidingWindowRateLimitingMiddleware(max_requests=100) | |
| result = await middleware.on_request(mock_context, mock_call_next) | |
| assert result == "test_result" | |
| assert mock_call_next.called | |
| async def test_on_request_rate_limited(self, mock_context, mock_call_next): | |
| """Test request rejection due to rate limiting.""" | |
| middleware = SlidingWindowRateLimitingMiddleware(max_requests=1) | |
| # First request should succeed | |
| await middleware.on_request(mock_context, mock_call_next) | |
| # Second request should be rate limited | |
| with pytest.raises(RateLimitError, match="Rate limit exceeded"): | |
| await middleware.on_request(mock_context, mock_call_next) | |
| class TestRateLimitError: | |
| """Test rate limit error.""" | |
| def test_init_default(self): | |
| """Test default initialization.""" | |
| error = RateLimitError() | |
| assert error.error.code == -32000 | |
| assert error.error.message == "Rate limit exceeded" | |
| def test_init_custom(self): | |
| """Test custom initialization.""" | |
| error = RateLimitError("Custom message") | |
| assert error.error.code == -32000 | |
| assert error.error.message == "Custom message" | |
| def rate_limit_server(): | |
| """Create a FastMCP server specifically for rate limiting tests.""" | |
| mcp = FastMCP("RateLimitTestServer") | |
| def quick_action(message: str) -> str: | |
| """A quick action for testing rate limits.""" | |
| return f"Processed: {message}" | |
| def batch_process(items: list[str]) -> str: | |
| """Process multiple items.""" | |
| return f"Processed {len(items)} items" | |
| def heavy_computation() -> str: | |
| """A heavy computation that might need rate limiting.""" | |
| # Simulate some work | |
| import time | |
| time.sleep(0.01) # Very short delay | |
| return "Heavy computation complete" | |
| return mcp | |
| class TestRateLimitingMiddlewareIntegration: | |
| """Integration tests for rate limiting middleware with real FastMCP server.""" | |
| async def test_rate_limiting_allows_normal_usage(self, rate_limit_server): | |
| """Test that normal usage patterns are allowed through rate limiting.""" | |
| # Generous rate limit | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware(max_requests_per_second=50.0, burst_capacity=10) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # Normal usage should be fine | |
| for i in range(5): | |
| result = await client.call_tool( | |
| "quick_action", {"message": f"task_{i}"} | |
| ) | |
| assert f"Processed: task_{i}" in str(result) | |
| async def test_rate_limiting_blocks_rapid_requests(self, rate_limit_server): | |
| """Test that rate limiting blocks rapid successive requests.""" | |
| # Very restrictive rate limit (accounting for extra list_tools calls per tool call) | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware(max_requests_per_second=10.0, burst_capacity=5) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # First few should succeed (within burst capacity) | |
| await client.call_tool("quick_action", {"message": "1"}) | |
| await client.call_tool("quick_action", {"message": "2"}) | |
| await client.call_tool("quick_action", {"message": "3"}) | |
| # Next should be rate limited | |
| with pytest.raises(ToolError, match="Rate limit exceeded"): | |
| await client.call_tool("quick_action", {"message": "4"}) | |
| async def test_rate_limiting_with_concurrent_requests(self, rate_limit_server): | |
| """Test rate limiting behavior with concurrent requests.""" | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware(max_requests_per_second=15.0, burst_capacity=8) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # Fire off many concurrent requests | |
| tasks = [] | |
| for i in range(8): | |
| task = asyncio.create_task( | |
| client.call_tool("quick_action", {"message": f"concurrent_{i}"}) | |
| ) | |
| tasks.append(task) | |
| # Gather results, allowing exceptions | |
| results = await asyncio.gather(*tasks, return_exceptions=True) | |
| # With extra list_tools calls, the exact behavior is unpredictable | |
| # Just verify that rate limiting is working (not all succeed) | |
| successes = [r for r in results if not isinstance(r, Exception)] | |
| failures = [r for r in results if isinstance(r, Exception)] | |
| total_results = len(successes) + len(failures) | |
| assert total_results == 8, f"Expected 8 results, got {total_results}" | |
| # With the unpredictable list_tools calls, we just verify that the system | |
| # is working (all requests should either succeed or fail with some exception) | |
| assert 0 <= len(successes) <= 8, "Should have between 0-8 successes" | |
| assert 0 <= len(failures) <= 8, "Should have between 0-8 failures" | |
| async def test_sliding_window_rate_limiting(self, rate_limit_server): | |
| """Test sliding window rate limiting implementation.""" | |
| rate_limit_server.add_middleware( | |
| SlidingWindowRateLimitingMiddleware( | |
| max_requests=5, # Accounting for extra list_tools calls | |
| window_minutes=1, # 1 minute window | |
| ) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # Should allow up to the limit | |
| await client.call_tool("quick_action", {"message": "1"}) | |
| await client.call_tool("quick_action", {"message": "2"}) | |
| await client.call_tool("quick_action", {"message": "3"}) | |
| # Fourth should be blocked | |
| with pytest.raises(ToolError, match="Rate limit exceeded"): | |
| await client.call_tool("quick_action", {"message": "4"}) | |
| async def test_rate_limiting_with_different_operations(self, rate_limit_server): | |
| """Test that rate limiting applies to all types of operations.""" | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware(max_requests_per_second=9.0, burst_capacity=4) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # Mix different operations | |
| await client.call_tool("quick_action", {"message": "test"}) | |
| await client.call_tool("heavy_computation") | |
| # Should be rate limited regardless of operation type | |
| with pytest.raises(ToolError, match="Rate limit exceeded"): | |
| await client.call_tool("batch_process", {"items": ["a", "b", "c"]}) | |
| async def test_custom_client_identification(self, rate_limit_server): | |
| """Test rate limiting with custom client identification.""" | |
| def get_client_id(context): | |
| # In a real scenario, this might extract from headers or context | |
| return "test_client_123" | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware( | |
| max_requests_per_second=6.0, # Accounting for extra list_tools calls | |
| burst_capacity=3, | |
| get_client_id=get_client_id, | |
| ) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # First request should succeed | |
| await client.call_tool("quick_action", {"message": "first"}) | |
| # Second should be rate limited for this specific client | |
| with pytest.raises( | |
| ToolError, match="Rate limit exceeded for client: test_client_123" | |
| ): | |
| await client.call_tool("quick_action", {"message": "second"}) | |
| async def test_global_rate_limiting(self, rate_limit_server): | |
| """Test global rate limiting across all clients.""" | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware( | |
| max_requests_per_second=6.0, | |
| burst_capacity=4, | |
| global_limit=True, # Accounting for extra list_tools calls | |
| ) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # Use up the global capacity | |
| await client.call_tool("quick_action", {"message": "1"}) | |
| await client.call_tool("quick_action", {"message": "2"}) | |
| # Should be globally rate limited | |
| with pytest.raises(ToolError, match="Global rate limit exceeded"): | |
| await client.call_tool("quick_action", {"message": "3"}) | |
| async def test_rate_limiting_recovery_over_time(self, rate_limit_server): | |
| """Test that rate limiting allows requests again after time passes.""" | |
| rate_limit_server.add_middleware( | |
| RateLimitingMiddleware( | |
| max_requests_per_second=10.0, # 10 per second = 1 every 100ms | |
| burst_capacity=3, | |
| ) | |
| ) | |
| async with Client(rate_limit_server) as client: | |
| # Use up capacity | |
| await client.call_tool("quick_action", {"message": "first"}) | |
| # Should be rate limited immediately | |
| with pytest.raises(ToolError): | |
| await client.call_tool("quick_action", {"message": "second"}) | |
| # Wait for token bucket to refill (150ms should be enough for ~1.5 tokens) | |
| await asyncio.sleep(0.15) | |
| # Should be able to make another request | |
| result = await client.call_tool("quick_action", {"message": "after_wait"}) | |
| assert "after_wait" in str(result) | |