Skip to content

Commit 359e159

Browse files
dogacancolakdogacancolak
committed
test: add scope priority tests for OAuth flow
1 parent 07ae8c0 commit 359e159

File tree

1 file changed

+196
-0
lines changed

1 file changed

+196
-0
lines changed

‎tests/client/test_auth.py‎

Lines changed: 196 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -959,3 +959,199 @@ async def callback_handler() -> tuple[str, str | None]:
959959

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

0 commit comments

Comments
 (0)