cover strict RFC 9728/8414 discovery parse paths

The existing discovery tests all use partial payloads that exercise the
lenient fallback; add tests for fully spec-compliant payloads so the strict
ProtectedResourceMetadata/OAuthMetadata branches (and their URL
normalization) are pinned, plus the non-object payload guards
This commit is contained in:
mateo-berri 2026-06-10 17:29:54 +00:00
parent 37c75ff43b
commit 72d31f351c
No known key found for this signature in database

View file

@ -911,6 +911,100 @@ class TestMCPServerManager:
]
assert scopes == ["read", "write"]
@pytest.mark.asyncio
async def test_fetch_oauth_metadata_from_resource_parses_strict_rfc9728_payload(
self,
):
"""A fully spec-compliant RFC 9728 payload (with ``resource``) takes the
strict ProtectedResourceMetadata path, which normalizes the URLs."""
manager = MCPServerManager()
mock_response = MagicMock()
mock_response.json.return_value = {
"resource": "https://protected.example.com/mcp",
"authorization_servers": ["https://auth.example.com"],
"scopes_supported": ["read", "write"],
}
mock_response.raise_for_status = MagicMock()
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
servers, scopes = await manager._fetch_oauth_metadata_from_resource(
"https://protected.example.com/.well-known/oauth-protected-resource/mcp",
"https://protected.example.com/mcp",
)
assert servers == ["https://auth.example.com/"]
assert scopes == ["read", "write"]
@pytest.mark.asyncio
async def test_discovery_payloads_that_are_not_json_objects_yield_no_metadata(
self,
):
manager = MCPServerManager()
mock_response = MagicMock()
mock_response.json.return_value = ["not", "an", "object"]
mock_response.raise_for_status = MagicMock()
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
servers, scopes = await manager._fetch_oauth_metadata_from_resource(
"https://protected.example.com/.well-known/oauth-protected-resource",
"https://protected.example.com/mcp",
)
assert servers == []
assert scopes is None
assert manager._oauth_metadata_from_asm_payload(["not", "an", "object"]) is None
@pytest.mark.asyncio
async def test_fetch_single_authorization_server_metadata_parses_strict_rfc8414_payload(
self,
):
"""A fully spec-compliant RFC 8414 payload (with ``issuer``) takes the
strict OAuthMetadata path and maps all endpoint fields."""
manager = MCPServerManager()
mock_response = MagicMock()
mock_response.json.return_value = {
"issuer": "https://provider.example.com",
"authorization_endpoint": "https://provider.example.com/authorize",
"token_endpoint": "https://provider.example.com/token",
"registration_endpoint": "https://provider.example.com/register",
"response_types_supported": ["code"],
"scopes_supported": ["mcp.read"],
}
mock_response.raise_for_status = MagicMock()
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
result = await manager._fetch_single_authorization_server_metadata(
"https://provider.example.com",
"https://provider.example.com",
)
assert result is not None
assert result.authorization_url == "https://provider.example.com/authorize"
assert result.token_url == "https://provider.example.com/token"
assert result.registration_url == "https://provider.example.com/register"
assert result.scopes == ["mcp.read"]
@pytest.mark.asyncio
async def test_descovery_metadata_probes_well_known_when_server_does_not_challenge(
self,