ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 21:14:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix Windows test timeouts in OAuth proxy provider tests (#3123)

Browse source 
Co-authored-by: Bill Easton <strawgate@users.noreply.github.com>
Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
Bill Easton 2026-02-09 19:54:16 -06:00 • committed by GitHub
commit cdad99583e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 209 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

170
tests/server/auth/providers/test_azure.py
View file

@ -2,6 +2,8 @@
from urllib.parse import parse_qs, urlparse
import pytest
from key_value.aio.stores.memory import MemoryStore
from mcp.server.auth.provider import AuthorizationParams
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
@ -14,10 +16,16 @@ from fastmcp.server.auth.providers.azure import (
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
@pytest.fixture
def memory_storage() -> MemoryStore:
"""Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
return MemoryStore()
class TestAzureProvider:
"""Test Azure OAuth provider functionality."""
def test_init_with_explicit_params(self):
def test_init_with_explicit_params(self, memory_storage: MemoryStore):
"""Test AzureProvider initialization with explicit parameters."""
provider = AzureProvider(
client_id="12345678-1234-1234-1234-123456789012",
@ -26,6 +34,7 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read", "write"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._upstream_client_id == "12345678-1234-1234-1234-123456789012"
@ -37,7 +46,7 @@ class TestAzureProvider:
parsed_token = urlparse(provider._upstream_token_endpoint)
assert "87654321-4321-4321-4321-210987654321" in parsed_token.path
def test_init_defaults(self):
def test_init_defaults(self, memory_storage: MemoryStore):
"""Test that default values are applied correctly."""
provider = AzureProvider(
client_id="test_client",
@ -46,13 +55,14 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check defaults
assert provider._redirect_path == "/auth/callback"
# Azure provider defaults are set but we can't easily verify them without accessing internals
def test_offline_access_automatically_included(self):
def test_offline_access_automatically_included(self, memory_storage: MemoryStore):
"""Test that offline_access is automatically added to get refresh tokens."""
# Without specifying offline_access
provider = AzureProvider(
@ -62,11 +72,12 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert "offline_access" in provider.additional_authorize_scopes
def test_offline_access_not_duplicated(self):
def test_offline_access_not_duplicated(self, memory_storage: MemoryStore):
"""Test that offline_access is not duplicated if already specified."""
provider = AzureProvider(
client_id="test_client",
@ -76,13 +87,14 @@ class TestAzureProvider:
required_scopes=["read"],
additional_authorize_scopes=["User.Read", "offline_access"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Should appear exactly once
assert provider.additional_authorize_scopes.count("offline_access") == 1
assert "User.Read" in provider.additional_authorize_scopes
def test_oauth_endpoints_configured_correctly(self):
def test_oauth_endpoints_configured_correctly(self, memory_storage: MemoryStore):
"""Test that OAuth endpoints are configured correctly."""
provider = AzureProvider(
client_id="test_client",
@ -91,6 +103,7 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test_secret",
client_storage=memory_storage,
)
# Check that endpoints use the correct Azure OAuth2 v2.0 endpoints with tenant
@ -106,7 +119,7 @@ class TestAzureProvider:
provider._upstream_revocation_endpoint is None
) # Azure doesn't support revocation
def test_special_tenant_values(self):
def test_special_tenant_values(self, memory_storage: MemoryStore):
"""Test that special tenant values are accepted."""
# Test with "organizations"
provider1 = AzureProvider(
@ -116,6 +129,7 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
parsed = urlparse(provider1._upstream_authorization_endpoint)
assert "/organizations/" in parsed.path
@ -128,11 +142,12 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
parsed = urlparse(provider2._upstream_authorization_endpoint)
assert "/consumers/" in parsed.path
def test_azure_specific_scopes(self):
def test_azure_specific_scopes(self, memory_storage: MemoryStore):
"""Test handling of custom API scope formats."""
# Test that the provider accepts custom API scopes without error
provider = AzureProvider(
@ -146,6 +161,7 @@ class TestAzureProvider:
"admin",
],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Provider should initialize successfully with these scopes
@ -158,7 +174,9 @@ class TestAzureProvider:
"admin",
]
def test_init_does_not_require_api_client_id_anymore(self):
def test_init_does_not_require_api_client_id_anymore(
self, memory_storage: MemoryStore
):
"""API client ID is no longer required; audience is client_id."""
provider = AzureProvider(
client_id="test_client",
@ -167,10 +185,13 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider is not None
def test_init_with_custom_audience_uses_jwt_verifier(self):
def test_init_with_custom_audience_uses_jwt_verifier(
self, memory_storage: MemoryStore
):
"""When audience is provided, JWTVerifier is configured with JWKS and issuer."""
from fastmcp.server.auth.providers.jwt import JWTVerifier
@ -182,6 +203,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=[".default"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._token_validator is not None
@ -197,7 +219,9 @@ class TestAzureProvider:
# (Azure returns unprefixed scopes like ".default" in JWT tokens)
assert verifier.required_scopes == [".default"]
async def test_authorize_filters_resource_and_stores_unprefixed_scopes(self):
async def test_authorize_filters_resource_and_stores_unprefixed_scopes(
self, memory_storage: MemoryStore
):
"""authorize() should drop resource parameter and store unprefixed scopes for MCP clients."""
provider = AzureProvider(
client_id="test_client",
@ -207,6 +231,7 @@ class TestAzureProvider:
required_scopes=["read", "write"],
base_url="https://srv.example",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
await provider.register_client(
@ -264,7 +289,9 @@ class TestAzureProvider:
or "api://my-api/write" in upstream_url
)
async def test_authorize_appends_additional_scopes(self):
async def test_authorize_appends_additional_scopes(
self, memory_storage: MemoryStore
):
"""authorize() should append additional_authorize_scopes to the authorization request."""
provider = AzureProvider(
client_id="test_client",
@ -275,6 +302,7 @@ class TestAzureProvider:
base_url="https://srv.example",
additional_authorize_scopes=["Mail.Read", "User.Read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
await provider.register_client(
@ -326,7 +354,7 @@ class TestAzureProvider:
assert "Mail.Read" in upstream_url
assert "User.Read" in upstream_url
def test_base_authority_defaults_to_public_cloud(self):
def test_base_authority_defaults_to_public_cloud(self, memory_storage: MemoryStore):
"""Test that base_authority defaults to login.microsoftonline.com."""
provider = AzureProvider(
client_id="test_client",
@ -335,6 +363,7 @@ class TestAzureProvider:
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert (
@ -355,7 +384,7 @@ class TestAzureProvider:
== "https://login.microsoftonline.com/test-tenant/discovery/v2.0/keys"
)
def test_base_authority_azure_government(self):
def test_base_authority_azure_government(self, memory_storage: MemoryStore):
"""Test Azure Government endpoints with login.microsoftonline.us."""
provider = AzureProvider(
client_id="test_client",
@ -365,6 +394,7 @@ class TestAzureProvider:
required_scopes=["read"],
base_authority="login.microsoftonline.us",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert (
@ -385,7 +415,7 @@ class TestAzureProvider:
== "https://login.microsoftonline.us/gov-tenant-id/discovery/v2.0/keys"
)
def test_base_authority_from_parameter(self):
def test_base_authority_from_parameter(self, memory_storage: MemoryStore):
"""Test that base_authority can be set via parameter."""
provider = AzureProvider(
client_id="env-client-id",
@ -395,6 +425,7 @@ class TestAzureProvider:
required_scopes=["read"],
base_authority="login.microsoftonline.us",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert (
@ -415,7 +446,9 @@ class TestAzureProvider:
== "https://login.microsoftonline.us/env-tenant-id/discovery/v2.0/keys"
)
def test_base_authority_with_special_tenant_values(self):
def test_base_authority_with_special_tenant_values(
self, memory_storage: MemoryStore
):
"""Test that base_authority works with special tenant values like 'organizations'."""
provider = AzureProvider(
client_id="test_client",
@ -425,13 +458,16 @@ class TestAzureProvider:
required_scopes=["read"],
base_authority="login.microsoftonline.us",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
parsed = urlparse(provider._upstream_authorization_endpoint)
assert parsed.netloc == "login.microsoftonline.us"
assert "/organizations/" in parsed.path
def test_prepare_scopes_for_upstream_refresh_basic_prefixing(self):
def test_prepare_scopes_for_upstream_refresh_basic_prefixing(
self, memory_storage: MemoryStore
):
"""Test that unprefixed scopes are correctly prefixed for Azure token refresh."""
provider = AzureProvider(
client_id="test_client",
@ -441,6 +477,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=["read", "write"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Unprefixed scopes from storage should be prefixed
@ -451,7 +488,9 @@ class TestAzureProvider:
assert "offline_access" in result # Auto-included for refresh tokens
assert len(result) == 3
def test_prepare_scopes_for_upstream_refresh_already_prefixed(self):
def test_prepare_scopes_for_upstream_refresh_already_prefixed(
self, memory_storage: MemoryStore
):
"""Test that already-prefixed scopes remain unchanged."""
provider = AzureProvider(
client_id="test_client",
@ -461,6 +500,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Already prefixed scopes should pass through unchanged
@ -473,7 +513,9 @@ class TestAzureProvider:
assert "offline_access" in result # Auto-included for refresh tokens
assert len(result) == 3
def test_prepare_scopes_for_upstream_refresh_with_additional_scopes(self):
def test_prepare_scopes_for_upstream_refresh_with_additional_scopes(
self, memory_storage: MemoryStore
):
"""Test that only OIDC scopes from additional_authorize_scopes are added.
Azure only allows ONE resource per token request (AADSTS28000), so
@ -493,6 +535,7 @@ class TestAzureProvider:
"offline_access",
],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Base scopes should be prefixed, only OIDC scopes appended
@ -508,6 +551,7 @@ class TestAzureProvider:
def test_prepare_scopes_for_upstream_refresh_filters_duplicate_additional_scopes(
self,
memory_storage: MemoryStore,
):
"""Test that accidentally stored additional_authorize_scopes are filtered out."""
provider = AzureProvider(
@ -519,6 +563,7 @@ class TestAzureProvider:
required_scopes=["read"],
additional_authorize_scopes=["User.Read", "openid"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# If additional scopes were accidentally stored, they should be filtered
@ -535,7 +580,9 @@ class TestAzureProvider:
assert "offline_access" in result # Auto-included and is OIDC
assert len(result) == 3
def test_prepare_scopes_for_upstream_refresh_mixed_scopes(self):
def test_prepare_scopes_for_upstream_refresh_mixed_scopes(
self, memory_storage: MemoryStore
):
"""Test mixed scenario with both prefixed and unprefixed scopes."""
provider = AzureProvider(
client_id="test_client",
@ -546,6 +593,7 @@ class TestAzureProvider:
required_scopes=["read"],
additional_authorize_scopes=["openid"], # OIDC scope
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Mix of prefixed and unprefixed scopes
@ -560,7 +608,9 @@ class TestAzureProvider:
assert "offline_access" in result # Auto-included
assert len(result) == 5
def test_prepare_scopes_for_upstream_refresh_scope_with_slash(self):
def test_prepare_scopes_for_upstream_refresh_scope_with_slash(
self, memory_storage: MemoryStore
):
"""Test that scopes containing '/' are not prefixed."""
provider = AzureProvider(
client_id="test_client",
@ -570,6 +620,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Scopes with "/" should not be prefixed (already fully qualified)
@ -582,7 +633,9 @@ class TestAzureProvider:
"https://graph.microsoft.com/.default" in result
) # Not prefixed (contains ://)
def test_prepare_scopes_for_upstream_refresh_empty_scopes(self):
def test_prepare_scopes_for_upstream_refresh_empty_scopes(
self, memory_storage: MemoryStore
):
"""Test behavior with empty scopes list."""
provider = AzureProvider(
client_id="test_client",
@ -593,6 +646,7 @@ class TestAzureProvider:
required_scopes=["read"],
additional_authorize_scopes=["User.Read", "openid"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Empty scopes should still add OIDC scopes (not User.Read)
@ -603,7 +657,9 @@ class TestAzureProvider:
assert "offline_access" in result # Auto-included
assert len(result) == 2 # Only OIDC scopes: openid + offline_access
def test_prepare_scopes_for_upstream_refresh_no_additional_scopes(self):
def test_prepare_scopes_for_upstream_refresh_no_additional_scopes(
self, memory_storage: MemoryStore
):
"""Test behavior when no additional_authorize_scopes are configured."""
provider = AzureProvider(
client_id="test_client",
@ -613,6 +669,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Should prefix base scopes, plus auto-added offline_access
@ -623,7 +680,9 @@ class TestAzureProvider:
assert "offline_access" in result # Auto-included
assert len(result) == 3
def test_prepare_scopes_for_upstream_refresh_deduplicates_scopes(self):
def test_prepare_scopes_for_upstream_refresh_deduplicates_scopes(
self, memory_storage: MemoryStore
):
"""Test that duplicate scopes are deduplicated while preserving order."""
provider = AzureProvider(
client_id="test_client",
@ -634,6 +693,7 @@ class TestAzureProvider:
required_scopes=["read"],
additional_authorize_scopes=["openid", "profile"], # OIDC scopes only
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Test with duplicate base scopes
@ -651,7 +711,9 @@ class TestAzureProvider:
]
assert len(result) == 5
def test_prepare_scopes_for_upstream_refresh_deduplicates_prefixed_variants(self):
def test_prepare_scopes_for_upstream_refresh_deduplicates_prefixed_variants(
self, memory_storage: MemoryStore
):
"""Test that both prefixed and unprefixed variants are deduplicated."""
provider = AzureProvider(
client_id="test_client",
@ -661,6 +723,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Test with both prefixed and unprefixed variants of same scope
@ -688,11 +751,13 @@ class TestOIDCScopeHandling:
3. OIDC scopes are still advertised to clients via valid_scopes
"""
def test_oidc_scopes_constant(self):
def test_oidc_scopes_constant(self, memory_storage: MemoryStore):
"""Verify OIDC_SCOPES contains the standard OIDC scopes."""
assert OIDC_SCOPES == {"openid", "profile", "email", "offline_access"}
def test_prefix_scopes_does_not_prefix_oidc_scopes(self):
def test_prefix_scopes_does_not_prefix_oidc_scopes(
self, memory_storage: MemoryStore
):
"""Test that _prefix_scopes_for_azure never prefixes OIDC scopes."""
provider = AzureProvider(
client_id="test_client",
@ -702,6 +767,7 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# All OIDC scopes should pass through unchanged
@ -711,7 +777,7 @@ class TestOIDCScopeHandling:
assert result == ["openid", "profile", "email", "offline_access"]
def test_prefix_scopes_mixed_oidc_and_custom(self):
def test_prefix_scopes_mixed_oidc_and_custom(self, memory_storage: MemoryStore):
"""Test prefixing with a mix of OIDC and custom scopes."""
provider = AzureProvider(
client_id="test_client",
@ -721,6 +787,7 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
result = provider._prefix_scopes_for_azure(
@ -736,7 +803,9 @@ class TestOIDCScopeHandling:
assert "api://my-api/openid" not in result
assert "api://my-api/profile" not in result
def test_prefix_scopes_dot_notation_gets_prefixed(self):
def test_prefix_scopes_dot_notation_gets_prefixed(
self, memory_storage: MemoryStore
):
"""Test that dot-notation scopes get prefixed (use additional_authorize_scopes for Graph)."""
provider = AzureProvider(
client_id="test_client",
@ -746,6 +815,7 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Dot-notation scopes ARE prefixed - use additional_authorize_scopes for Graph
@ -754,7 +824,9 @@ class TestOIDCScopeHandling:
assert result == ["api://my-api/my.scope", "api://my-api/admin.read"]
def test_prefix_scopes_fully_qualified_graph_not_prefixed(self):
def test_prefix_scopes_fully_qualified_graph_not_prefixed(
self, memory_storage: MemoryStore
):
"""Test that fully-qualified Graph scopes are not prefixed."""
provider = AzureProvider(
client_id="test_client",
@ -764,6 +836,7 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
result = provider._prefix_scopes_for_azure(
@ -779,7 +852,9 @@ class TestOIDCScopeHandling:
"https://graph.microsoft.com/Mail.Send",
]
def test_required_scopes_with_oidc_filters_validation(self):
def test_required_scopes_with_oidc_filters_validation(
self, memory_storage: MemoryStore
):
"""Test that OIDC scopes in required_scopes are filtered from token validation."""
provider = AzureProvider(
client_id="test_client",
@ -789,12 +864,15 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read", "openid", "profile"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Token validator should only require non-OIDC scopes
assert provider._token_validator.required_scopes == ["read"]
def test_required_scopes_all_oidc_results_in_no_validation(self):
def test_required_scopes_all_oidc_results_in_no_validation(
self, memory_storage: MemoryStore
):
"""Test that if all required_scopes are OIDC, no scope validation occurs."""
provider = AzureProvider(
client_id="test_client",
@ -804,12 +882,13 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["openid", "profile"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Token validator should have empty required scopes (all were OIDC)
assert provider._token_validator.required_scopes == []
def test_valid_scopes_includes_oidc_scopes(self):
def test_valid_scopes_includes_oidc_scopes(self, memory_storage: MemoryStore):
"""Test that valid_scopes advertises OIDC scopes to clients."""
provider = AzureProvider(
client_id="test_client",
@ -819,6 +898,7 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read", "openid", "profile"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# required_scopes (used for validation) excludes OIDC scopes
@ -831,7 +911,9 @@ class TestOIDCScopeHandling:
"profile",
]
def test_prepare_scopes_for_refresh_handles_oidc_scopes(self):
def test_prepare_scopes_for_refresh_handles_oidc_scopes(
self, memory_storage: MemoryStore
):
"""Test that token refresh correctly handles OIDC scopes."""
provider = AzureProvider(
client_id="test_client",
@ -841,6 +923,7 @@ class TestOIDCScopeHandling:
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Simulate stored scopes that include OIDC scopes
@ -864,7 +947,7 @@ class TestAzureTokenExchangeScopes:
properly prefixed scopes.
"""
def test_prepare_scopes_returns_prefixed_scopes(self):
def test_prepare_scopes_returns_prefixed_scopes(self, memory_storage: MemoryStore):
"""Test that _prepare_scopes_for_token_exchange returns prefixed scopes."""
provider = AzureProvider(
client_id="test_client",
@ -874,6 +957,7 @@ class TestAzureTokenExchangeScopes:
identifier_uri="api://my-api",
required_scopes=["read", "write"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
scopes = provider._prepare_scopes_for_token_exchange(["read", "write"])
@ -881,7 +965,9 @@ class TestAzureTokenExchangeScopes:
assert "api://my-api/read" in scopes
assert "api://my-api/write" in scopes
def test_prepare_scopes_includes_additional_oidc_scopes(self):
def test_prepare_scopes_includes_additional_oidc_scopes(
self, memory_storage: MemoryStore
):
"""Test that _prepare_scopes_for_token_exchange includes OIDC scopes."""
provider = AzureProvider(
client_id="test_client",
@ -892,6 +978,7 @@ class TestAzureTokenExchangeScopes:
required_scopes=["read"],
additional_authorize_scopes=["openid", "profile", "offline_access"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
scopes = provider._prepare_scopes_for_token_exchange(["read"])
@ -901,7 +988,9 @@ class TestAzureTokenExchangeScopes:
assert "profile" in scopes
assert "offline_access" in scopes
def test_prepare_scopes_excludes_other_api_scopes(self):
def test_prepare_scopes_excludes_other_api_scopes(
self, memory_storage: MemoryStore
):
"""Test token exchange excludes other API scopes (Azure AADSTS28000).
Azure only allows ONE resource per token exchange. Other API scopes
@ -921,6 +1010,7 @@ class TestAzureTokenExchangeScopes:
"api://11111111-2222-3333-4444-555555555555/user_impersonation",
],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
scopes = provider._prepare_scopes_for_token_exchange(["user_impersonation"])
@ -935,7 +1025,7 @@ class TestAzureTokenExchangeScopes:
assert not any("api://aaaaaaaa" in s for s in scopes)
assert not any("api://11111111" in s for s in scopes)
def test_prepare_scopes_deduplicates_scopes(self):
def test_prepare_scopes_deduplicates_scopes(self, memory_storage: MemoryStore):
"""Test that duplicate scopes are deduplicated."""
provider = AzureProvider(
client_id="test_client",
@ -946,6 +1036,7 @@ class TestAzureTokenExchangeScopes:
required_scopes=["read"],
additional_authorize_scopes=["api://my-api/read", "openid"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Pass a scope that will be prefixed to match one in additional_authorize_scopes
@ -955,7 +1046,9 @@ class TestAzureTokenExchangeScopes:
assert scopes.count("api://my-api/read") == 1
assert "openid" in scopes
def test_extra_token_params_does_not_contain_scope(self):
def test_extra_token_params_does_not_contain_scope(
self, memory_storage: MemoryStore
):
"""Test that extra_token_params doesn't contain scope to avoid TypeError.
Previously, Azure provider set extra_token_params={"scope": ...} during init.
@ -974,6 +1067,7 @@ class TestAzureTokenExchangeScopes:
required_scopes=["read", "write"],
additional_authorize_scopes=["openid", "profile", "offline_access"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# extra_token_params should NOT contain "scope" to avoid TypeError during refresh

21
tests/server/auth/providers/test_discord.py
View file

@ -1,12 +1,21 @@
"""Tests for Discord OAuth provider."""
import pytest
from key_value.aio.stores.memory import MemoryStore
from fastmcp.server.auth.providers.discord import DiscordProvider
@pytest.fixture
def memory_storage() -> MemoryStore:
"""Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
return MemoryStore()
class TestDiscordProvider:
"""Test Discord OAuth provider functionality."""
def test_init_with_explicit_params(self):
def test_init_with_explicit_params(self, memory_storage: MemoryStore):
"""Test DiscordProvider initialization with explicit parameters."""
provider = DiscordProvider(
client_id="env_client_id",
@ -14,31 +23,34 @@ class TestDiscordProvider:
base_url="https://myserver.com",
required_scopes=["email", "identify"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._upstream_client_id == "env_client_id"
assert provider._upstream_client_secret.get_secret_value() == "GOCSPX-test123"
assert str(provider.base_url) == "https://myserver.com/"
def test_init_defaults(self):
def test_init_defaults(self, memory_storage: MemoryStore):
"""Test that default values are applied correctly."""
provider = DiscordProvider(
client_id="env_client_id",
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check defaults
assert provider._redirect_path == "/auth/callback"
def test_oauth_endpoints_configured_correctly(self):
def test_oauth_endpoints_configured_correctly(self, memory_storage: MemoryStore):
"""Test that OAuth endpoints are configured correctly."""
provider = DiscordProvider(
client_id="env_client_id",
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check that endpoints use Discord's OAuth2 endpoints
@ -52,7 +64,7 @@ class TestDiscordProvider:
# Discord provider doesn't currently set a revocation endpoint
assert provider._upstream_revocation_endpoint is None
def test_discord_specific_scopes(self):
def test_discord_specific_scopes(self, memory_storage: MemoryStore):
"""Test handling of Discord-specific scope formats."""
# Just test that the provider accepts Discord-specific scopes without error
provider = DiscordProvider(
@ -64,6 +76,7 @@ class TestDiscordProvider:
"email",
],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Provider should initialize successfully with these scopes

19
tests/server/auth/providers/test_github.py
View file

@ -2,16 +2,25 @@
from unittest.mock import MagicMock, patch
import pytest
from key_value.aio.stores.memory import MemoryStore
from fastmcp.server.auth.providers.github import (
GitHubProvider,
GitHubTokenVerifier,
)
@pytest.fixture
def memory_storage() -> MemoryStore:
"""Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
return MemoryStore()
class TestGitHubProvider:
"""Test GitHubProvider initialization."""
def test_init_with_explicit_params(self):
def test_init_with_explicit_params(self, memory_storage: MemoryStore):
"""Test initialization with explicit parameters."""
provider = GitHubProvider(
client_id="test_client",
@ -21,6 +30,7 @@ class TestGitHubProvider:
required_scopes=["user", "repo"],
timeout_seconds=30,
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check that the provider was initialized correctly
@ -31,13 +41,14 @@ class TestGitHubProvider:
) # URLs get normalized with trailing slash
assert provider._redirect_path == "/custom/callback"
def test_init_defaults(self):
def test_init_defaults(self, memory_storage: MemoryStore):
"""Test that default values are applied correctly."""
provider = GitHubProvider(
client_id="test_client",
client_secret="test_secret",
base_url="https://example.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check defaults
@ -49,7 +60,7 @@ class TestGitHubProvider:
class TestGitHubTokenVerifier:
"""Test GitHubTokenVerifier."""
def test_init_with_custom_scopes(self):
def test_init_with_custom_scopes(self, memory_storage: MemoryStore):
"""Test initialization with custom required scopes."""
verifier = GitHubTokenVerifier(
required_scopes=["user", "repo"],
@ -59,7 +70,7 @@ class TestGitHubTokenVerifier:
assert verifier.required_scopes == ["user", "repo"]
assert verifier.timeout_seconds == 30
def test_init_defaults(self):
def test_init_defaults(self, memory_storage: MemoryStore):
"""Test initialization with defaults."""
verifier = GitHubTokenVerifier()

32
tests/server/auth/providers/test_google.py
View file

@ -1,12 +1,21 @@
"""Tests for Google OAuth provider."""
import pytest
from key_value.aio.stores.memory import MemoryStore
from fastmcp.server.auth.providers.google import GoogleProvider
@pytest.fixture
def memory_storage() -> MemoryStore:
"""Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
return MemoryStore()
class TestGoogleProvider:
"""Test Google OAuth provider functionality."""
def test_init_with_explicit_params(self):
def test_init_with_explicit_params(self, memory_storage: MemoryStore):
"""Test GoogleProvider initialization with explicit parameters."""
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
@ -14,32 +23,35 @@ class TestGoogleProvider:
base_url="https://myserver.com",
required_scopes=["openid", "email", "profile"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._upstream_client_id == "123456789.apps.googleusercontent.com"
assert provider._upstream_client_secret.get_secret_value() == "GOCSPX-test123"
assert str(provider.base_url) == "https://myserver.com/"
def test_init_defaults(self):
def test_init_defaults(self, memory_storage: MemoryStore):
"""Test that default values are applied correctly."""
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check defaults
assert provider._redirect_path == "/auth/callback"
# Google provider has ["openid"] as default but we can't easily verify without accessing internals
def test_oauth_endpoints_configured_correctly(self):
def test_oauth_endpoints_configured_correctly(self, memory_storage: MemoryStore):
"""Test that OAuth endpoints are configured correctly."""
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check that endpoints use Google's OAuth2 endpoints
@ -53,7 +65,7 @@ class TestGoogleProvider:
# Google provider doesn't currently set a revocation endpoint
assert provider._upstream_revocation_endpoint is None
def test_google_specific_scopes(self):
def test_google_specific_scopes(self, memory_storage: MemoryStore):
"""Test handling of Google-specific scope formats."""
# Just test that the provider accepts Google-specific scopes without error
provider = GoogleProvider(
@ -66,18 +78,20 @@ class TestGoogleProvider:
"https://www.googleapis.com/auth/userinfo.profile",
],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Provider should initialize successfully with these scopes
assert provider is not None
def test_extra_authorize_params_defaults(self):
def test_extra_authorize_params_defaults(self, memory_storage: MemoryStore):
"""Test that Google-specific defaults are set for refresh token support."""
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Should have Google-specific defaults for refresh token support
@ -86,7 +100,9 @@ class TestGoogleProvider:
"prompt": "consent",
}
def test_extra_authorize_params_override_defaults(self):
def test_extra_authorize_params_override_defaults(
self, memory_storage: MemoryStore
):
"""Test that user can override default extra authorize params."""
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
@ -94,6 +110,7 @@ class TestGoogleProvider:
base_url="https://myserver.com",
jwt_signing_key="test-secret",
extra_authorize_params={"prompt": "select_account"},
client_storage=memory_storage,
)
# User override should replace the default
@ -101,7 +118,7 @@ class TestGoogleProvider:
# But other defaults should remain
assert provider._extra_authorize_params["access_type"] == "offline"
def test_extra_authorize_params_add_new_params(self):
def test_extra_authorize_params_add_new_params(self, memory_storage: MemoryStore):
"""Test that user can add additional authorize params."""
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
@ -109,6 +126,7 @@ class TestGoogleProvider:
base_url="https://myserver.com",
jwt_signing_key="test-secret",
extra_authorize_params={"login_hint": "user@example.com"},
client_storage=memory_storage,
)
# New param should be added

25
tests/server/auth/providers/test_workos.py
View file

@ -4,6 +4,7 @@ from urllib.parse import urlparse
import httpx
import pytest
from key_value.aio.stores.memory import MemoryStore
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
@ -11,10 +12,16 @@ from fastmcp.server.auth.providers.workos import AuthKitProvider, WorkOSProvider
from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
@pytest.fixture
def memory_storage() -> MemoryStore:
"""Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
return MemoryStore()
class TestWorkOSProvider:
"""Test WorkOS OAuth provider functionality."""
def test_init_with_explicit_params(self):
def test_init_with_explicit_params(self, memory_storage: MemoryStore):
"""Test WorkOSProvider initialization with explicit parameters."""
provider = WorkOSProvider(
client_id="client_test123",
@ -23,13 +30,14 @@ class TestWorkOSProvider:
base_url="https://myserver.com",
required_scopes=["openid", "profile"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._upstream_client_id == "client_test123"
assert provider._upstream_client_secret.get_secret_value() == "secret_test456"
assert str(provider.base_url) == "https://myserver.com/"
def test_authkit_domain_https_prefix_handling(self):
def test_authkit_domain_https_prefix_handling(self, memory_storage: MemoryStore):
"""Test that authkit_domain handles missing https:// prefix."""
# Without https:// - should add it
provider1 = WorkOSProvider(
@ -38,6 +46,7 @@ class TestWorkOSProvider:
authkit_domain="test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
parsed = urlparse(provider1._upstream_authorization_endpoint)
assert parsed.scheme == "https"
@ -51,6 +60,7 @@ class TestWorkOSProvider:
authkit_domain="https://test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
parsed = urlparse(provider2._upstream_authorization_endpoint)
assert parsed.scheme == "https"
@ -64,13 +74,14 @@ class TestWorkOSProvider:
authkit_domain="http://localhost:8080",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
parsed = urlparse(provider3._upstream_authorization_endpoint)
assert parsed.scheme == "http"
assert parsed.netloc == "localhost:8080"
assert parsed.path == "/oauth2/authorize"
def test_init_defaults(self):
def test_init_defaults(self, memory_storage: MemoryStore):
"""Test that default values are applied correctly."""
provider = WorkOSProvider(
client_id="test_client",
@ -78,13 +89,14 @@ class TestWorkOSProvider:
authkit_domain="https://test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check defaults
assert provider._redirect_path == "/auth/callback"
# WorkOS provider has no default scopes but we can't easily verify without accessing internals
def test_oauth_endpoints_configured_correctly(self):
def test_oauth_endpoints_configured_correctly(self, memory_storage: MemoryStore):
"""Test that OAuth endpoints are configured correctly."""
provider = WorkOSProvider(
client_id="test_client",
@ -92,6 +104,7 @@ class TestWorkOSProvider:
authkit_domain="https://test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
# Check that endpoints use the authkit domain
@ -135,7 +148,9 @@ def client_with_headless_oauth(mcp_server_url: str) -> Client:
class TestAuthKitProvider:
async def test_unauthorized_access(self, mcp_server_url: str):
async def test_unauthorized_access(
self, memory_storage: MemoryStore, mcp_server_url: str
):
with pytest.raises(httpx.HTTPStatusError) as exc_info:
async with Client(mcp_server_url) as client:
tools = await client.list_tools() # noqa: F841