Add test for ProtectedResourceMetadataParsing (#1236)
Co-authored-by: Paul Carleton <paulcarletonjr@gmail.com> Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com> Co-authored-by: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com>
This commit is contained in:
@@ -0,0 +1,52 @@
|
||||
"""
|
||||
Integration tests for MCP Oauth Protected Resource.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from inline_snapshot import snapshot
|
||||
from pydantic import AnyHttpUrl
|
||||
from starlette.applications import Starlette
|
||||
|
||||
from mcp.server.auth.routes import create_protected_resource_routes
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_app():
|
||||
"""Fixture to create protected resource routes for testing."""
|
||||
|
||||
# Create the protected resource routes
|
||||
protected_resource_routes = create_protected_resource_routes(
|
||||
resource_url=AnyHttpUrl("https://example.com/resource"),
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com/authorization")],
|
||||
scopes_supported=["read", "write"],
|
||||
resource_name="Example Resource",
|
||||
resource_documentation=AnyHttpUrl("https://docs.example.com/resource"),
|
||||
)
|
||||
|
||||
app = Starlette(routes=protected_resource_routes)
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def test_client(test_app: Starlette):
|
||||
"""Fixture to create an HTTP client for the protected resource app."""
|
||||
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=test_app), base_url="https://mcptest.com") as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_endpoint(test_client: httpx.AsyncClient):
|
||||
"""Test the OAuth 2.0 Protected Resource metadata endpoint."""
|
||||
|
||||
response = await test_client.get("/.well-known/oauth-protected-resource")
|
||||
assert response.json() == snapshot(
|
||||
{
|
||||
"resource": "https://example.com/resource",
|
||||
"authorization_servers": ["https://auth.example.com/authorization"],
|
||||
"scopes_supported": ["read", "write"],
|
||||
"resource_name": "Example Resource",
|
||||
"resource_documentation": "https://docs.example.com/resource",
|
||||
"bearer_methods_supported": ["header"],
|
||||
}
|
||||
)
|
||||
@@ -342,11 +342,8 @@ class TestAuthEndpoints:
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_endpoint(self, test_client: httpx.AsyncClient):
|
||||
"""Test the OAuth 2.0 metadata endpoint."""
|
||||
print("Sending request to metadata endpoint")
|
||||
|
||||
response = await test_client.get("/.well-known/oauth-authorization-server")
|
||||
print(f"Got response: {response.status_code}")
|
||||
if response.status_code != 200:
|
||||
print(f"Response content: {response.content}")
|
||||
assert response.status_code == 200
|
||||
|
||||
metadata = response.json()
|
||||
@@ -399,9 +396,7 @@ class TestAuthEndpoints:
|
||||
"redirect_uri": "https://client.example.com/callback",
|
||||
},
|
||||
)
|
||||
print(f"Status code: {response.status_code}")
|
||||
print(f"Response body: {response.content}")
|
||||
print(f"Response JSON: {response.json()}")
|
||||
|
||||
assert response.status_code == 400
|
||||
error_response = response.json()
|
||||
assert error_response["error"] == "invalid_grant"
|
||||
|
||||
Reference in New Issue
Block a user