Skip to content

tests: eliminate port-allocation races in SSE/StreamableHTTP tests #2264

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
maxisbey wants to merge 1 commit into
base: main
Choose a base branch
from
+495 −907
Draft

tests: eliminate port-allocation races in SSE/StreamableHTTP tests #2264

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
265 changes: 99 additions & 166 deletions tests/client/test_http_unicode.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,10 @@
(server→client and client→server) using the streamable HTTP transport.
"""

import multiprocessing
import socket
from collections.abc import AsyncGenerator, Generator
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

import httpx
import pytest
from starlette.applications import Starlette
from starlette.routing import Mount
Expand All @@ -19,7 +18,6 @@
from mcp.server import Server, ServerRequestContext
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.types import TextContent, Tool
from tests.test_helpers import wait_for_server

# Test constants with various Unicode characters
UNICODE_TEST_STRINGS = {
Expand All @@ -41,197 +39,132 @@
}


def run_unicode_server(port: int) -> None: # pragma: no cover
"""Run the Unicode test server in a separate process."""
import uvicorn

# Need to recreate the server setup in this process
async def handle_list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(
tools=[
Tool(
name="echo_unicode",
description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
input_schema={
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to echo back"},
},
"required": ["text"],
},
),
]
)
async def _handle_list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(
tools=[
Tool(
name="echo_unicode",
description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
input_schema={
"type": "object",
"properties": {"text": {"type": "string", "description": "Text to echo back"}},
"required": ["text"],
},
),
]
)


async def _handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
if params.name == "echo_unicode":
text = params.arguments.get("text", "") if params.arguments else ""
return types.CallToolResult(content=[TextContent(type="text", text=f"Echo: {text}")])
raise ValueError(f"Unknown tool: {params.name}") # pragma: no cover


async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
if params.name == "echo_unicode":
text = params.arguments.get("text", "") if params.arguments else ""
return types.CallToolResult(
content=[
TextContent(
type="text",
text=f"Echo: {text}",
)
]
async def _handle_list_prompts(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListPromptsResult:
return types.ListPromptsResult(
prompts=[
types.Prompt(
name="unicode_prompt",
description="Unicode prompt - Слой хранилища, где располагаются",
arguments=[],
)
else:
raise ValueError(f"Unknown tool: {params.name}")

async def handle_list_prompts(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListPromptsResult:
return types.ListPromptsResult(
prompts=[
types.Prompt(
name="unicode_prompt",
description="Unicode prompt - Слой хранилища, где располагаются",
arguments=[],
]
)


async def _handle_get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> types.GetPromptResult:
if params.name == "unicode_prompt":
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text="Hello世界🌍Привет안녕مرحباשלום"),
)
]
)
raise ValueError(f"Unknown prompt: {params.name}") # pragma: no cover

async def handle_get_prompt(
ctx: ServerRequestContext, params: types.GetPromptRequestParams
) -> types.GetPromptResult:
if params.name == "unicode_prompt":
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text="Hello世界🌍Привет안녕مرحباשלום",
),
)
]
)
raise ValueError(f"Unknown prompt: {params.name}")

def _make_unicode_app() -> Starlette:
server = Server(
name="unicode_test_server",
on_list_tools=handle_list_tools,
on_call_tool=handle_call_tool,
on_list_prompts=handle_list_prompts,
on_get_prompt=handle_get_prompt,
)

# Create the session manager
session_manager = StreamableHTTPSessionManager(
app=server,
json_response=False, # Use SSE for testing
on_list_tools=_handle_list_tools,
on_call_tool=_handle_call_tool,
on_list_prompts=_handle_list_prompts,
on_get_prompt=_handle_get_prompt,
)
session_manager = StreamableHTTPSessionManager(app=server, json_response=False)

@asynccontextmanager
async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
async with session_manager.run():
yield

# Create an ASGI application
app = Starlette(
return Starlette(
debug=True,
routes=[
Mount("/mcp", app=session_manager.handle_request),
],
routes=[Mount("/mcp", app=session_manager.handle_request)],
lifespan=lifespan,
)

# Run the server
config = uvicorn.Config(
app=app,
host="127.0.0.1",
port=port,
log_level="error",
)
uvicorn_server = uvicorn.Server(config)
uvicorn_server.run()


@pytest.fixture
def unicode_server_port() -> int:
"""Find an available port for the Unicode test server."""
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]


@pytest.fixture
def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]:
"""Start a Unicode test server in a separate process."""
proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True)
proc.start()

# Wait for server to be ready
wait_for_server(unicode_server_port)

try:
yield f"http://127.0.0.1:{unicode_server_port}"
finally:
# Clean up - try graceful termination first
proc.terminate()
proc.join(timeout=2)
if proc.is_alive(): # pragma: no cover
proc.kill()
proc.join(timeout=1)
async def unicode_session() -> AsyncGenerator[ClientSession, None]:
"""Create an initialized client session connected to the in-process unicode server."""
app = _make_unicode_app()
async with app.router.lifespan_context(app):
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, follow_redirects=True) as http_client:
async with streamable_http_client("http://testserver/mcp", http_client=http_client) as (rs, ws):
async with ClientSession(rs, ws) as session:
await session.initialize()
yield session


@pytest.mark.anyio
async def test_streamable_http_client_unicode_tool_call(running_unicode_server: str) -> None:
async def test_streamable_http_client_unicode_tool_call(unicode_session: ClientSession) -> None:
"""Test that Unicode text is correctly handled in tool calls via streamable HTTP."""
base_url = running_unicode_server
endpoint_url = f"{base_url}/mcp"

async with streamable_http_client(endpoint_url) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()

# Test 1: List tools (server→client Unicode in descriptions)
tools = await session.list_tools()
assert len(tools.tools) == 1

# Check Unicode in tool descriptions
echo_tool = tools.tools[0]
assert echo_tool.name == "echo_unicode"
assert echo_tool.description is not None
assert "🔤" in echo_tool.description
assert "👋" in echo_tool.description
# Test 1: List tools (server→client Unicode in descriptions)
tools = await unicode_session.list_tools()
assert len(tools.tools) == 1

# Test 2: Send Unicode text in tool call (client→server→client)
for test_name, test_string in UNICODE_TEST_STRINGS.items():
result = await session.call_tool("echo_unicode", arguments={"text": test_string})
echo_tool = tools.tools[0]
assert echo_tool.name == "echo_unicode"
assert echo_tool.description is not None
assert "🔤" in echo_tool.description
assert "👋" in echo_tool.description

# Verify server correctly received and echoed back Unicode
assert len(result.content) == 1
content = result.content[0]
assert content.type == "text"
assert f"Echo: {test_string}" == content.text, f"Failed for {test_name}"
# Test 2: Send Unicode text in tool call (client→server→client)
for test_name, test_string in UNICODE_TEST_STRINGS.items():
result = await unicode_session.call_tool("echo_unicode", arguments={"text": test_string})
assert len(result.content) == 1
content = result.content[0]
assert content.type == "text"
assert f"Echo: {test_string}" == content.text, f"Failed for {test_name}"


@pytest.mark.anyio
async def test_streamable_http_client_unicode_prompts(running_unicode_server: str) -> None:
async def test_streamable_http_client_unicode_prompts(unicode_session: ClientSession) -> None:
"""Test that Unicode text is correctly handled in prompts via streamable HTTP."""
base_url = running_unicode_server
endpoint_url = f"{base_url}/mcp"

async with streamable_http_client(endpoint_url) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()

# Test 1: List prompts (server→client Unicode in descriptions)
prompts = await session.list_prompts()
assert len(prompts.prompts) == 1

prompt = prompts.prompts[0]
assert prompt.name == "unicode_prompt"
assert prompt.description is not None
assert "Слой хранилища, где располагаются" in prompt.description

# Test 2: Get prompt with Unicode content (server→client)
result = await session.get_prompt("unicode_prompt", arguments={})
assert len(result.messages) == 1

message = result.messages[0]
assert message.role == "user"
assert message.content.type == "text"
assert message.content.text == "Hello世界🌍Привет안녕مرحباשלום"
# Test 1: List prompts (server→client Unicode in descriptions)
prompts = await unicode_session.list_prompts()
assert len(prompts.prompts) == 1

prompt = prompts.prompts[0]
assert prompt.name == "unicode_prompt"
assert prompt.description is not None
assert "Слой хранилища, где располагаются" in prompt.description

# Test 2: Get prompt with Unicode content (server→client)
result = await unicode_session.get_prompt("unicode_prompt", arguments={})
assert len(result.messages) == 1

message = result.messages[0]
assert message.role == "user"
assert message.content.type == "text"
assert message.content.text == "Hello世界🌍Привет안녕مرحباשלום"
Loading
Loading