|
| 1 | +import pytest |
| 2 | +import anyio |
| 3 | +from starlette.applications import Starlette |
| 4 | +from starlette.routing import Mount, Route |
| 5 | +import uvicorn |
| 6 | +from mcp.client.sse import sse_client |
| 7 | +from exceptiongroup import ExceptionGroup |
| 8 | +import asyncio |
| 9 | +import httpx |
| 10 | +from httpx import ReadTimeout |
| 11 | + |
| 12 | +from mcp.server.sse import SseServerTransport |
| 13 | + |
| 14 | +@pytest.fixture |
| 15 | +async def sse_server(): |
| 16 | + |
| 17 | + # Create an SSE transport at an endpoint |
| 18 | + sse = SseServerTransport("/messages/") |
| 19 | + |
| 20 | + # Create Starlette routes for SSE and message handling |
| 21 | + routes = [ |
| 22 | + Route("/sse", endpoint=handle_sse), |
| 23 | + Mount("/messages/", app=sse.handle_post_message), |
| 24 | + ] |
| 25 | + # |
| 26 | + # Create and run Starlette app |
| 27 | + app = Starlette(routes=routes) |
| 28 | + |
| 29 | + # Define handler functions |
| 30 | + async def handle_sse(request): |
| 31 | + async with sse.connect_sse( |
| 32 | + request.scope, request.receive, request._send |
| 33 | + ) as streams: |
| 34 | + await app.run( |
| 35 | + streams[0], streams[1], app.create_initialization_options() |
| 36 | + ) |
| 37 | + |
| 38 | + uvicorn.run(app, host="127.0.0.1", port=34891) |
| 39 | + |
| 40 | + async def sse_handler(request): |
| 41 | + response = httpx.Response(200, content_type="text/event-stream") |
| 42 | + response.send_headers() |
| 43 | + response.write("data: test\n\n") |
| 44 | + await response.aclose() |
| 45 | + |
| 46 | + async with httpx.AsyncServer(sse_handler) as server: |
| 47 | + yield server.url |
| 48 | + |
| 49 | + |
| 50 | +@pytest.fixture |
| 51 | +async def sse_client(): |
| 52 | + async with sse_client("http://test/sse") as (read_stream, write_stream): |
| 53 | + async with read_stream: |
| 54 | + async for message in read_stream: |
| 55 | + if isinstance(message, Exception): |
| 56 | + raise message |
| 57 | + |
| 58 | + return read_stream, write_stream |
| 59 | + |
| 60 | +@pytest.mark.anyio |
| 61 | +async def test_sse_happy_path(monkeypatch): |
| 62 | + # Mock httpx.AsyncClient to return our mock response |
| 63 | + monkeypatch.setattr(httpx, "AsyncClient", MockClient) |
| 64 | + |
| 65 | + with pytest.raises(ReadTimeout) as exc_info: |
| 66 | + async with sse_client( |
| 67 | + "http://test/sse", |
| 68 | + timeout=5, # Connection timeout - make this longer |
| 69 | + sse_read_timeout=1 # Read timeout - this should trigger |
| 70 | + ) as (read_stream, write_stream): |
| 71 | + async with read_stream: |
| 72 | + async for message in read_stream: |
| 73 | + if isinstance(message, Exception): |
| 74 | + raise message |
| 75 | + |
| 76 | + error = exc_info.value |
| 77 | + assert isinstance(error, ReadTimeout) |
| 78 | + assert str(error) == "Read timeout" |
| 79 | + |
| 80 | +@pytest.mark.anyio |
| 81 | +async def test_sse_read_timeouts(monkeypatch): |
| 82 | + """Test that the SSE client properly handles read timeouts between SSE messages.""" |
0 commit comments