Skip to content

Commit 3e4407d

Browse files
dogacancolakdogacancolak
committed
reorg
1 parent 1517e4b commit 3e4407d

File tree

1 file changed

+129
-177
lines changed

1 file changed

+129
-177
lines changed

‎tests/client/test_auth.py

Lines changed: 129 additions & 177 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,11 @@ async def callback_handler() -> tuple[str, str | None]:
7878
callback_handler=callback_handler,
7979
)
8080

81+
@pytest.fixture
82+
def oauth_provider_without_scope(oauth_provider: OAuthClientProvider) -> OAuthClientProvider:
83+
"""Create OAuth provider without predefined scope."""
84+
oauth_provider.context.client_metadata.scope = None
85+
return oauth_provider
8186

8287
class TestPKCEParameters:
8388
"""Test PKCE parameter generation."""
@@ -391,6 +396,130 @@ async def test_handle_metadata_response_success(self, oauth_provider: OAuthClien
391396
assert oauth_provider.context.oauth_metadata is not None
392397
assert str(oauth_provider.context.oauth_metadata.issuer) == "https://auth.example.com/"
393398

399+
@pytest.mark.anyio
400+
async def test_prioritize_prm_scopes_over_oauth_metadata(self, oauth_provider_without_scope: OAuthClientProvider):
401+
"""Test that PRM scopes are prioritized over auth server metadata scopes."""
402+
provider = oauth_provider_without_scope
403+
404+
# Set up PRM metadata with specific scopes
405+
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
406+
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
407+
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
408+
scopes_supported=["resource:read", "resource:write"],
409+
)
410+
411+
# Create OAuth metadata response with different scopes
412+
oauth_metadata_response = httpx.Response(
413+
200,
414+
content=(
415+
b'{"issuer": "https://auth.example.com", '
416+
b'"authorization_endpoint": "https://auth.example.com/authorize", '
417+
b'"token_endpoint": "https://auth.example.com/token", '
418+
b'"registration_endpoint": "https://auth.example.com/register", '
419+
b'"scopes_supported": ["read", "write", "admin"]}'
420+
),
421+
)
422+
423+
# Process the OAuth metadata
424+
await provider._handle_oauth_metadata_response(oauth_metadata_response)
425+
426+
# Verify that PRM scopes are used (not OAuth metadata scopes)
427+
assert provider.context.client_metadata.scope == "resource:read resource:write"
428+
429+
@pytest.mark.anyio
430+
async def test_fallback_to_oauth_metadata_scopes_when_no_prm_scopes(
431+
self, oauth_provider_without_scope: OAuthClientProvider
432+
):
433+
"""Test fallback to OAuth metadata scopes when PRM has no scopes."""
434+
provider = oauth_provider_without_scope
435+
436+
# Set up PRM metadata without scopes
437+
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
438+
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
439+
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
440+
scopes_supported=None, # No scopes in PRM
441+
)
442+
443+
# Create OAuth metadata response with scopes
444+
oauth_metadata_response = httpx.Response(
445+
200,
446+
content=(
447+
b'{"issuer": "https://auth.example.com", '
448+
b'"authorization_endpoint": "https://auth.example.com/authorize", '
449+
b'"token_endpoint": "https://auth.example.com/token", '
450+
b'"registration_endpoint": "https://auth.example.com/register", '
451+
b'"scopes_supported": ["read", "write", "admin"]}'
452+
),
453+
)
454+
455+
# Process the OAuth metadata
456+
await provider._handle_oauth_metadata_response(oauth_metadata_response)
457+
458+
# Verify that OAuth metadata scopes are used as fallback
459+
assert provider.context.client_metadata.scope == "read write admin"
460+
461+
@pytest.mark.anyio
462+
async def test_no_scope_changes_when_both_missing(self, oauth_provider_without_scope: OAuthClientProvider):
463+
"""Test that no scope changes occur when both PRM and OAuth metadata lack scopes."""
464+
provider = oauth_provider_without_scope
465+
466+
# Set up PRM metadata without scopes
467+
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
468+
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
469+
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
470+
scopes_supported=None, # No scopes in PRM
471+
)
472+
473+
# Create OAuth metadata response without scopes
474+
oauth_metadata_response = httpx.Response(
475+
200,
476+
content=(
477+
b'{"issuer": "https://auth.example.com", '
478+
b'"authorization_endpoint": "https://auth.example.com/authorize", '
479+
b'"token_endpoint": "https://auth.example.com/token", '
480+
b'"registration_endpoint": "https://auth.example.com/register"}'
481+
# No scopes_supported field
482+
),
483+
)
484+
485+
# Process the OAuth metadata
486+
await provider._handle_oauth_metadata_response(oauth_metadata_response)
487+
488+
# Verify that scope remains None
489+
assert provider.context.client_metadata.scope is None
490+
491+
@pytest.mark.anyio
492+
async def test_preserve_existing_client_scope(
493+
self, oauth_provider: OAuthClientProvider
494+
):
495+
"""Test that existing client scope is preserved regardless of metadata."""
496+
provider = oauth_provider
497+
498+
# Set up PRM metadata with scopes
499+
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
500+
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
501+
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
502+
scopes_supported=["resource:read", "resource:write"],
503+
)
504+
505+
# Create OAuth metadata response with scopes
506+
oauth_metadata_response = httpx.Response(
507+
200,
508+
content=(
509+
b'{"issuer": "https://auth.example.com", '
510+
b'"authorization_endpoint": "https://auth.example.com/authorize", '
511+
b'"token_endpoint": "https://auth.example.com/token", '
512+
b'"registration_endpoint": "https://auth.example.com/register", '
513+
b'"scopes_supported": ["read", "write", "admin"]}'
514+
),
515+
)
516+
517+
# Process the OAuth metadata
518+
await provider._handle_oauth_metadata_response(oauth_metadata_response)
519+
520+
# Verify that predefined scope is preserved
521+
assert provider.context.client_metadata.scope == "read write"
522+
394523
@pytest.mark.anyio
395524
async def test_register_client_request(self, oauth_provider: OAuthClientProvider):
396525
"""Test client registration request building."""
@@ -960,180 +1089,3 @@ async def callback_handler() -> tuple[str, str | None]:
9601089
result = provider._extract_resource_metadata_from_www_auth(init_response)
9611090
assert result is None, f"Should return None for {description}"
9621091

963-
964-
@pytest.fixture
965-
def client_metadata_no_scope() -> OAuthClientMetadata:
966-
"""Client metadata without a predefined scope."""
967-
return OAuthClientMetadata(
968-
client_name="Test Client",
969-
client_uri=AnyHttpUrl("https://example.com"),
970-
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
971-
# No scope defined
972-
scope=None,
973-
)
974-
975-
976-
@pytest.fixture
977-
def oauth_provider_without_scope(
978-
client_metadata_no_scope: OAuthClientMetadata, mock_storage: MockTokenStorage
979-
) -> OAuthClientProvider:
980-
"""Create OAuth provider without predefined scope."""
981-
982-
async def redirect_handler(url: str) -> None:
983-
pass
984-
985-
async def callback_handler() -> tuple[str, str | None]:
986-
return "test_auth_code", "test_state"
987-
988-
return OAuthClientProvider(
989-
server_url="https://api.example.com/v1/mcp",
990-
client_metadata=client_metadata_no_scope,
991-
storage=mock_storage,
992-
redirect_handler=redirect_handler,
993-
callback_handler=callback_handler,
994-
)
995-
996-
997-
class TestScopeHandlingPriority:
998-
"""Test OAuth scope handling priority between PRM and auth metadata."""
999-
1000-
@pytest.mark.anyio
1001-
async def test_prioritize_prm_scopes_over_oauth_metadata(self, oauth_provider_without_scope: OAuthClientProvider):
1002-
"""Test that PRM scopes are prioritized over auth server metadata scopes."""
1003-
provider = oauth_provider_without_scope
1004-
1005-
# Set up PRM metadata with specific scopes
1006-
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
1007-
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
1008-
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
1009-
scopes_supported=["resource:read", "resource:write"],
1010-
)
1011-
1012-
# Create OAuth metadata response with different scopes
1013-
oauth_metadata_response = httpx.Response(
1014-
200,
1015-
content=(
1016-
b'{"issuer": "https://auth.example.com", '
1017-
b'"authorization_endpoint": "https://auth.example.com/authorize", '
1018-
b'"token_endpoint": "https://auth.example.com/token", '
1019-
b'"registration_endpoint": "https://auth.example.com/register", '
1020-
b'"scopes_supported": ["read", "write", "admin"]}'
1021-
),
1022-
)
1023-
1024-
# Process the OAuth metadata
1025-
await provider._handle_oauth_metadata_response(oauth_metadata_response)
1026-
1027-
# Verify that PRM scopes are used (not OAuth metadata scopes)
1028-
assert provider.context.client_metadata.scope == "resource:read resource:write"
1029-
1030-
@pytest.mark.anyio
1031-
async def test_fallback_to_oauth_metadata_scopes_when_no_prm_scopes(
1032-
self, oauth_provider_without_scope: OAuthClientProvider
1033-
):
1034-
"""Test fallback to OAuth metadata scopes when PRM has no scopes."""
1035-
provider = oauth_provider_without_scope
1036-
1037-
# Set up PRM metadata without scopes
1038-
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
1039-
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
1040-
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
1041-
scopes_supported=None, # No scopes in PRM
1042-
)
1043-
1044-
# Create OAuth metadata response with scopes
1045-
oauth_metadata_response = httpx.Response(
1046-
200,
1047-
content=(
1048-
b'{"issuer": "https://auth.example.com", '
1049-
b'"authorization_endpoint": "https://auth.example.com/authorize", '
1050-
b'"token_endpoint": "https://auth.example.com/token", '
1051-
b'"registration_endpoint": "https://auth.example.com/register", '
1052-
b'"scopes_supported": ["read", "write", "admin"]}'
1053-
),
1054-
)
1055-
1056-
# Process the OAuth metadata
1057-
await provider._handle_oauth_metadata_response(oauth_metadata_response)
1058-
1059-
# Verify that OAuth metadata scopes are used as fallback
1060-
assert provider.context.client_metadata.scope == "read write admin"
1061-
1062-
@pytest.mark.anyio
1063-
async def test_no_scope_changes_when_both_missing(self, oauth_provider_without_scope: OAuthClientProvider):
1064-
"""Test that no scope changes occur when both PRM and OAuth metadata lack scopes."""
1065-
provider = oauth_provider_without_scope
1066-
1067-
# Set up PRM metadata without scopes
1068-
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
1069-
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
1070-
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
1071-
scopes_supported=None, # No scopes in PRM
1072-
)
1073-
1074-
# Create OAuth metadata response without scopes
1075-
oauth_metadata_response = httpx.Response(
1076-
200,
1077-
content=(
1078-
b'{"issuer": "https://auth.example.com", '
1079-
b'"authorization_endpoint": "https://auth.example.com/authorize", '
1080-
b'"token_endpoint": "https://auth.example.com/token", '
1081-
b'"registration_endpoint": "https://auth.example.com/register"}'
1082-
# No scopes_supported field
1083-
),
1084-
)
1085-
1086-
# Process the OAuth metadata
1087-
await provider._handle_oauth_metadata_response(oauth_metadata_response)
1088-
1089-
# Verify that scope remains None
1090-
assert provider.context.client_metadata.scope is None
1091-
1092-
@pytest.mark.anyio
1093-
async def test_preserve_existing_client_scope(
1094-
self, client_metadata_no_scope: OAuthClientMetadata, mock_storage: MockTokenStorage
1095-
):
1096-
"""Test that existing client scope is preserved regardless of metadata."""
1097-
# Create client with predefined scope
1098-
client_metadata = client_metadata_no_scope
1099-
client_metadata.scope = "predefined:scope"
1100-
1101-
# Create provider
1102-
async def redirect_handler(url: str) -> None:
1103-
pass
1104-
1105-
async def callback_handler() -> tuple[str, str | None]:
1106-
return "test_auth_code", "test_state"
1107-
1108-
provider = OAuthClientProvider(
1109-
server_url="https://api.example.com/v1/mcp",
1110-
client_metadata=client_metadata,
1111-
storage=mock_storage,
1112-
redirect_handler=redirect_handler,
1113-
callback_handler=callback_handler,
1114-
)
1115-
1116-
# Set up PRM metadata with scopes
1117-
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
1118-
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
1119-
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
1120-
scopes_supported=["resource:read", "resource:write"],
1121-
)
1122-
1123-
# Create OAuth metadata response with scopes
1124-
oauth_metadata_response = httpx.Response(
1125-
200,
1126-
content=(
1127-
b'{"issuer": "https://auth.example.com", '
1128-
b'"authorization_endpoint": "https://auth.example.com/authorize", '
1129-
b'"token_endpoint": "https://auth.example.com/token", '
1130-
b'"registration_endpoint": "https://auth.example.com/register", '
1131-
b'"scopes_supported": ["read", "write", "admin"]}'
1132-
),
1133-
)
1134-
1135-
# Process the OAuth metadata
1136-
await provider._handle_oauth_metadata_response(oauth_metadata_response)
1137-
1138-
# Verify that predefined scope is preserved
1139-
assert provider.context.client_metadata.scope == "predefined:scope"

0 commit comments

Comments
 (0)