Skip to content

Commit e029139

Browse files
committed
Add dedicated tests for SSE mount_path functionality
Add tests that verify the integration between FastMCP mount paths and SSE endpoints: - Test SseServerTransport initialization with different endpoints - Test FastMCP and SseServerTransport integration with various mount paths - Test session URI construction with different mount paths and URL encoding These tests verify that the normalized message endpoint is correctly passed to SseServerTransport and used to generate proper session URIs.
1 parent 0366f68 commit e029139

File tree

1 file changed

+111
-0
lines changed

1 file changed

+111
-0
lines changed
Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
import pytest
2+
from starlette.applications import Starlette
3+
from starlette.routing import Mount, Route
4+
from urllib.parse import quote
5+
from uuid import UUID, uuid4
6+
from unittest.mock import patch
7+
8+
from mcp.server.fastmcp import FastMCP
9+
from mcp.server.sse import SseServerTransport
10+
11+
12+
class TestSSEWithMountPath:
13+
def test_sse_transport_endpoint_formation(self):
14+
"""Test that SseServerTransport correctly initializes with different endpoints.
15+
16+
This test verifies that the SSE transport correctly stores the endpoint
17+
passed to its constructor, which will be used to build the session URI.
18+
"""
19+
# Test with root endpoint
20+
transport1 = SseServerTransport("/messages/")
21+
assert transport1._endpoint == "/messages/"
22+
23+
# Test with custom endpoint
24+
transport2 = SseServerTransport("/api/messages/")
25+
assert transport2._endpoint == "/api/messages/"
26+
27+
# Test with normalized path from FastMCP
28+
mcp = FastMCP()
29+
normalized_endpoint = mcp._normalize_path("/github", "/messages/")
30+
transport3 = SseServerTransport(normalized_endpoint)
31+
assert transport3._endpoint == "/github/messages/"
32+
33+
def test_fastmcp_sse_transport_integration(self):
34+
"""Test the integration between FastMCP and SseServerTransport with mount paths.
35+
36+
Verifies that FastMCP correctly generates normalized endpoints based on mount_path
37+
settings and passes them to the SseServerTransport.
38+
"""
39+
# Test cases with different mount path configurations
40+
test_cases = [
41+
# (mount_path, expected_normalized_endpoint)
42+
("/", "/messages/"),
43+
("/api", "/api/messages/"),
44+
("/github", "/github/messages/"),
45+
("/custom/path", "/custom/path/messages/"),
46+
]
47+
48+
for mount_path, expected_endpoint in test_cases:
49+
# Create FastMCP with the specified mount path
50+
mcp = FastMCP()
51+
mcp.settings.mount_path = mount_path
52+
53+
# Generate the normalized endpoint
54+
normalized_endpoint = mcp._normalize_path(mount_path, "/messages/")
55+
56+
# Verify the normalized endpoint matches expectations
57+
assert normalized_endpoint == expected_endpoint
58+
59+
# Create app and verify mount_path setting is preserved or updated
60+
app = mcp.sse_app()
61+
assert mcp.settings.mount_path == mount_path
62+
63+
# Test with direct mount_path parameter
64+
mcp2 = FastMCP()
65+
app2 = mcp2.sse_app(mount_path=mount_path)
66+
67+
# Verify the mount_path was correctly set in settings
68+
assert mcp2.settings.mount_path == mount_path
69+
70+
def test_sse_session_uri_construction(self):
71+
"""Test that session_uri is correctly constructed for different mount paths.
72+
73+
This test verifies that the session URI is properly constructed and URL-encoded
74+
when using different mount paths. The session URI is used by clients to
75+
connect to the correct SSE endpoint.
76+
"""
77+
# Use a fixed UUID for consistent testing
78+
test_uuid = uuid4()
79+
80+
# Test cases with different endpoints and expected URL-encoded prefixes
81+
test_cases = [
82+
# (endpoint, expected_session_uri_prefix)
83+
("/messages/", quote("/messages/")),
84+
("/api/messages/", quote("/api/messages/")),
85+
("/github/messages/", quote("/github/messages/")),
86+
# Verify with normalized paths from FastMCP
87+
(
88+
FastMCP()._normalize_path("/github", "/messages/"),
89+
quote("/github/messages/"),
90+
),
91+
(
92+
FastMCP()._normalize_path("/custom/path", "/messages/"),
93+
quote("/custom/path/messages/"),
94+
),
95+
]
96+
97+
with patch("uuid.uuid4", return_value=test_uuid):
98+
for endpoint, expected_encoded_prefix in test_cases:
99+
# Create SSE transport with the endpoint
100+
transport = SseServerTransport(endpoint)
101+
102+
# Manually construct the session_uri as done in connect_sse
103+
session_uri = f"{quote(endpoint)}?session_id={test_uuid.hex}"
104+
105+
# Verify session_uri format
106+
assert session_uri.startswith(expected_encoded_prefix)
107+
assert session_uri.endswith(f"?session_id={test_uuid.hex}")
108+
109+
# Complete URI validation
110+
expected_uri = f"{quote(endpoint)}?session_id={test_uuid.hex}"
111+
assert session_uri == expected_uri

0 commit comments

Comments
 (0)