Spaces:
Running
Running
File size: 2,837 Bytes
9a6f944 c219134 9a6f944 3959209 9a6f944 b6176a5 9a6f944 b6176a5 9a6f944 c219134 9a6f944 c219134 9a6f944 c219134 9a6f944 b6176a5 9a6f944 b6176a5 9a6f944 c219134 9a6f944 c219134 9a6f944 c219134 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import pytest
from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from starlette.routing import Route
from fastmcp.server import FastMCP
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
from fastmcp.server.http import create_streamable_http_app
class TestStreamableHTTPAppResourceMetadataURL:
"""Test resource_metadata_url logic in create_streamable_http_app."""
@pytest.fixture
def rsa_key_pair(self) -> RSAKeyPair:
"""Generate RSA key pair for testing."""
return RSAKeyPair.generate()
@pytest.fixture
def bearer_auth_provider(self, rsa_key_pair):
provider = JWTVerifier(
public_key=rsa_key_pair.public_key,
issuer="https://issuer",
audience="https://audience",
resource_server_url="https://resource.example.com",
)
return provider
def test_require_auth_middleware_receives_resource_metadata_url(
self, bearer_auth_provider
):
server = FastMCP(name="TestServer")
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
auth=bearer_auth_provider,
)
route = next(r for r in app.routes if isinstance(r, Route) and r.path == "/mcp")
assert isinstance(route.endpoint, RequireAuthMiddleware)
assert (
str(route.endpoint.resource_metadata_url)
== "https://resource.example.com/.well-known/oauth-protected-resource"
)
def test_trailing_slash_handling_in_resource_server_url(self, rsa_key_pair):
provider = JWTVerifier(
public_key=rsa_key_pair.public_key,
issuer="https://issuer",
audience="https://audience",
resource_server_url="https://resource.example.com/",
)
server = FastMCP(name="TestServer")
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
auth=provider,
)
route = next(r for r in app.routes if isinstance(r, Route) and r.path == "/mcp")
assert isinstance(route.endpoint, RequireAuthMiddleware)
# Should not have double slash
assert (
str(route.endpoint.resource_metadata_url)
== "https://resource.example.com/.well-known/oauth-protected-resource"
)
def test_no_auth_provider_mounts_without_require_auth_middleware(
self, rsa_key_pair
):
server = FastMCP(name="TestServer")
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
auth=None,
)
route = next(r for r in app.routes if isinstance(r, Route) and r.path == "/mcp")
assert not isinstance(route.endpoint, RequireAuthMiddleware)
|