Skip to content

Add test for ProtectedResourceMetadataParsing #1236

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.

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

Open
wants to merge 19 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
52 changes: 52 additions & 0 deletions tests/server/auth/test_protected_resource.py
Original file line number Diff line number Diff line change
@@ -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"],
}
)
13 changes: 4 additions & 9 deletions tests/server/fastmcp/auth/test_auth_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ def mock_oauth_provider():


@pytest.fixture
def auth_app(mock_oauth_provider):
def auth_app(mock_oauth_provider: MockOAuthProvider):
# Create auth router
auth_routes = create_auth_routes(
mock_oauth_provider,
Expand All @@ -217,7 +217,7 @@ def auth_app(mock_oauth_provider):


@pytest.fixture
async def test_client(auth_app):
async def test_client(auth_app: Starlette):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=auth_app), base_url="https://mcptest.com") as client:
yield client

Expand Down Expand Up @@ -335,11 +335,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()
Expand Down Expand Up @@ -387,9 +384,7 @@ async def test_token_invalid_auth_code(self, test_client, registered_client, pkc
"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"
Expand Down
Loading