Skip to content

Commit 6bd3960

Browse files
committed
Move scope step-up test to top-level function
Per AGENTS.md, new tests should be plain top-level test_* functions, not methods of legacy Test* classes. test_403_step_up_preserves_scope_from_stored_token landed inside TestAuthFlow in #2931; move it out to match the convention and its sibling test_union_scopes.
1 parent 1331131 commit 6bd3960

1 file changed

Lines changed: 56 additions & 55 deletions

File tree

tests/client/test_auth.py

Lines changed: 56 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -1447,64 +1447,65 @@ async def mock_callback() -> AuthorizationCodeResult:
14471447
except StopAsyncIteration:
14481448
pass # Expected
14491449

1450-
@pytest.mark.anyio
1451-
async def test_403_step_up_preserves_scope_from_stored_token(
1452-
self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
1453-
):
1454-
"""SEP-2350: a restart-loaded token's scope is folded into the step-up union.
1455-
1456-
On restart only the token is reloaded (not client_metadata.scope), so the stored token's
1457-
granted scope must seed the union, or the challenge would re-authorize for less.
1458-
"""
1459-
client_info = OAuthClientInformationFull(
1460-
client_id="test_client_id",
1461-
client_secret="test_client_secret",
1462-
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
1463-
)
1464-
# Simulate a restart: a token granted "read" is loaded, but client_metadata carries no scope.
1465-
oauth_provider.context.current_tokens = OAuthToken(access_token="t", scope="read")
1466-
oauth_provider.context.token_expiry_time = time.time() + 1800
1467-
oauth_provider.context.client_info = client_info
1468-
oauth_provider.context.client_metadata.scope = None
1469-
oauth_provider._initialized = True
1470-
1471-
captured_state: str | None = None
1472-
reauthorize_scope: str | None = None
1473-
1474-
async def capture_redirect(url: str) -> None:
1475-
nonlocal captured_state, reauthorize_scope
1476-
params = parse_qs(urlparse(url).query)
1477-
reauthorize_scope = params["scope"][0]
1478-
captured_state = params.get("state", [None])[0]
1479-
1480-
async def mock_callback() -> AuthorizationCodeResult:
1481-
return AuthorizationCodeResult(code="auth_code", state=captured_state)
1482-
1483-
oauth_provider.context.redirect_handler = capture_redirect
1484-
oauth_provider.context.callback_handler = mock_callback
14851450

1486-
auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/mcp"))
1487-
request = await auth_flow.__anext__()
1488-
response_403 = httpx.Response(
1489-
403,
1490-
headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
1491-
request=request,
1492-
)
1493-
token_exchange_request = await auth_flow.asend(response_403)
1451+
@pytest.mark.anyio
1452+
async def test_403_step_up_preserves_scope_from_stored_token(
1453+
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
1454+
):
1455+
"""SEP-2350: a restart-loaded token's scope is folded into the step-up union.
1456+
1457+
On restart only the token is reloaded (not client_metadata.scope), so the stored token's
1458+
granted scope must seed the union, or the challenge would re-authorize for less.
1459+
"""
1460+
client_info = OAuthClientInformationFull(
1461+
client_id="test_client_id",
1462+
client_secret="test_client_secret",
1463+
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
1464+
)
1465+
# Simulate a restart: a token granted "read" is loaded, but client_metadata carries no scope.
1466+
oauth_provider.context.current_tokens = OAuthToken(access_token="t", scope="read")
1467+
oauth_provider.context.token_expiry_time = time.time() + 1800
1468+
oauth_provider.context.client_info = client_info
1469+
oauth_provider.context.client_metadata.scope = None
1470+
oauth_provider._initialized = True
1471+
1472+
captured_state: str | None = None
1473+
reauthorize_scope: str | None = None
1474+
1475+
async def capture_redirect(url: str) -> None:
1476+
nonlocal captured_state, reauthorize_scope
1477+
params = parse_qs(urlparse(url).query)
1478+
reauthorize_scope = params["scope"][0]
1479+
captured_state = params.get("state", [None])[0]
1480+
1481+
async def mock_callback() -> AuthorizationCodeResult:
1482+
return AuthorizationCodeResult(code="auth_code", state=captured_state)
1483+
1484+
oauth_provider.context.redirect_handler = capture_redirect
1485+
oauth_provider.context.callback_handler = mock_callback
1486+
1487+
auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/mcp"))
1488+
request = await auth_flow.__anext__()
1489+
response_403 = httpx.Response(
1490+
403,
1491+
headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
1492+
request=request,
1493+
)
1494+
token_exchange_request = await auth_flow.asend(response_403)
14941495

1495-
assert reauthorize_scope == "read write"
1496+
assert reauthorize_scope == "read write"
14961497

1497-
# Drive the flow to completion so the context lock is released cleanly
1498-
token_response = httpx.Response(
1499-
200,
1500-
json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
1501-
request=token_exchange_request,
1502-
)
1503-
final_request = await auth_flow.asend(token_response)
1504-
try:
1505-
await auth_flow.asend(httpx.Response(200, request=final_request))
1506-
except StopAsyncIteration:
1507-
pass
1498+
# Drive the flow to completion so the context lock is released cleanly
1499+
token_response = httpx.Response(
1500+
200,
1501+
json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
1502+
request=token_exchange_request,
1503+
)
1504+
final_request = await auth_flow.asend(token_response)
1505+
try:
1506+
await auth_flow.asend(httpx.Response(200, request=final_request))
1507+
except StopAsyncIteration:
1508+
pass
15081509

15091510

15101511
@pytest.mark.parametrize(

0 commit comments

Comments
 (0)