modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: drop test classes (#1924)

Browse source
This commit is contained in:
Marcelo Trylesinski 2026-01-22 12:37:52 +01:00 • committed by GitHub
commit d77292fb06
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1611 additions and 1562 deletions
Download patch file Download diff file Expand all files Collapse all files

107
tests/server/auth/middleware/test_auth_context.py
View file

@ -45,76 +45,75 @@ def valid_access_token() -> AccessToken:
@pytest.mark.anyio
class TestAuthContextMiddleware:
"""Tests for the AuthContextMiddleware class."""
async def test_auth_context_middleware_with_authenticated_user(valid_access_token: AccessToken):
"""Test middleware with an authenticated user in scope."""
app = MockApp()
middleware = AuthContextMiddleware(app)
async def test_with_authenticated_user(self, valid_access_token: AccessToken):
"""Test middleware with an authenticated user in scope."""
app = MockApp()
middleware = AuthContextMiddleware(app)
# Create an authenticated user
user = AuthenticatedUser(valid_access_token)
# Create an authenticated user
user = AuthenticatedUser(valid_access_token)
scope: Scope = {"type": "http", "user": user}
scope: Scope = {"type": "http", "user": user}
# Create dummy async functions for receive and send
async def receive() -> Message: # pragma: no cover
return {"type": "http.request"}
# Create dummy async functions for receive and send
async def receive() -> Message: # pragma: no cover
return {"type": "http.request"}
async def send(message: Message) -> None: # pragma: no cover
pass
async def send(message: Message) -> None: # pragma: no cover
pass
# Verify context is empty before middleware
assert auth_context_var.get() is None
assert get_access_token() is None
# Verify context is empty before middleware
assert auth_context_var.get() is None
assert get_access_token() is None
# Run the middleware
await middleware(scope, receive, send)
# Run the middleware
await middleware(scope, receive, send)
# Verify the app was called
assert app.called
assert app.scope == scope
assert app.receive == receive
assert app.send == send
# Verify the app was called
assert app.called
assert app.scope == scope
assert app.receive == receive
assert app.send == send
# Verify the access token was available during the call
assert app.access_token_during_call == valid_access_token
# Verify the access token was available during the call
assert app.access_token_during_call == valid_access_token
# Verify context is reset after middleware
assert auth_context_var.get() is None
assert get_access_token() is None
# Verify context is reset after middleware
assert auth_context_var.get() is None
assert get_access_token() is None
async def test_with_no_user(self):
"""Test middleware with no user in scope."""
app = MockApp()
middleware = AuthContextMiddleware(app)
@pytest.mark.anyio
async def test_auth_context_middleware_with_no_user():
"""Test middleware with no user in scope."""
app = MockApp()
middleware = AuthContextMiddleware(app)
scope: Scope = {"type": "http"} # No user
scope: Scope = {"type": "http"} # No user
# Create dummy async functions for receive and send
async def receive() -> Message: # pragma: no cover
return {"type": "http.request"}
# Create dummy async functions for receive and send
async def receive() -> Message: # pragma: no cover
return {"type": "http.request"}
async def send(message: Message) -> None: # pragma: no cover
pass
async def send(message: Message) -> None: # pragma: no cover
pass
# Verify context is empty before middleware
assert auth_context_var.get() is None
assert get_access_token() is None
# Verify context is empty before middleware
assert auth_context_var.get() is None
assert get_access_token() is None
# Run the middleware
await middleware(scope, receive, send)
# Run the middleware
await middleware(scope, receive, send)
# Verify the app was called
assert app.called
assert app.scope == scope
assert app.receive == receive
assert app.send == send
# Verify the app was called
assert app.called
assert app.scope == scope
assert app.receive == receive
assert app.send == send
# Verify the access token was not available during the call
assert app.access_token_during_call is None
# Verify the access token was not available during the call
assert app.access_token_during_call is None
# Verify context is still empty after middleware
assert auth_context_var.get() is None
assert get_access_token() is None
# Verify context is still empty after middleware
assert auth_context_var.get() is None
assert get_access_token() is None

359
tests/server/auth/test_error_handling.py
View file

@ -83,176 +83,120 @@ async def registered_client(client: httpx.AsyncClient) -> dict[str, Any]:
return client_info
class TestRegistrationErrorHandling:
@pytest.mark.anyio
async def test_registration_error_handling(self, client: httpx.AsyncClient, oauth_provider: MockOAuthProvider):
# Mock the register_client method to raise a registration error
with unittest.mock.patch.object(
oauth_provider,
"register_client",
side_effect=RegistrationError(
error="invalid_redirect_uri",
error_description="The redirect URI is invalid",
),
):
# Prepare a client registration request
client_data = {
"redirect_uris": ["https://client.example.com/callback"],
"token_endpoint_auth_method": "client_secret_post",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"client_name": "Test Client",
}
# Send the registration request
response = await client.post(
"/register",
json=client_data,
)
# Verify the response
assert response.status_code == 400, response.content
data = response.json()
assert data["error"] == "invalid_redirect_uri"
assert data["error_description"] == "The redirect URI is invalid"
class TestAuthorizeErrorHandling:
@pytest.mark.anyio
async def test_authorize_error_handling(
self,
client: httpx.AsyncClient,
oauth_provider: MockOAuthProvider,
registered_client: dict[str, Any],
pkce_challenge: dict[str, str],
@pytest.mark.anyio
async def test_registration_error_handling(client: httpx.AsyncClient, oauth_provider: MockOAuthProvider):
# Mock the register_client method to raise a registration error
with unittest.mock.patch.object(
oauth_provider,
"register_client",
side_effect=RegistrationError(
error="invalid_redirect_uri",
error_description="The redirect URI is invalid",
),
):
# Mock the authorize method to raise an authorize error
with unittest.mock.patch.object(
oauth_provider,
"authorize",
side_effect=AuthorizeError(error="access_denied", error_description="The user denied the request"),
):
# Register the client
client_id = registered_client["client_id"]
redirect_uri = registered_client["redirect_uris"][0]
# Prepare a client registration request
client_data = {
"redirect_uris": ["https://client.example.com/callback"],
"token_endpoint_auth_method": "client_secret_post",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"client_name": "Test Client",
}
# Prepare an authorization request
params = {
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"code_challenge": pkce_challenge["code_challenge"],
"code_challenge_method": "S256",
"state": "test_state",
}
# Send the authorization request
response = await client.get("/authorize", params=params)
# Verify the response is a redirect with error parameters
assert response.status_code == 302
redirect_url = response.headers["location"]
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url.query)
assert query_params["error"][0] == "access_denied"
assert "error_description" in query_params
assert query_params["state"][0] == "test_state"
class TestTokenErrorHandling:
@pytest.mark.anyio
async def test_token_error_handling_auth_code(
self,
client: httpx.AsyncClient,
oauth_provider: MockOAuthProvider,
registered_client: dict[str, Any],
pkce_challenge: dict[str, str],
):
# Register the client and get an auth code
client_id = registered_client["client_id"]
client_secret = registered_client["client_secret"]
redirect_uri = registered_client["redirect_uris"][0]
# First get an authorization code
auth_response = await client.get(
"/authorize",
params={
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"code_challenge": pkce_challenge["code_challenge"],
"code_challenge_method": "S256",
"state": "test_state",
},
# Send the registration request
response = await client.post(
"/register",
json=client_data,
)
redirect_url = auth_response.headers["location"]
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url.query)
code = query_params["code"][0]
# Verify the response
assert response.status_code == 400, response.content
data = response.json()
assert data["error"] == "invalid_redirect_uri"
assert data["error_description"] == "The redirect URI is invalid"
# Mock the exchange_authorization_code method to raise a token error
with unittest.mock.patch.object(
oauth_provider,
"exchange_authorization_code",
side_effect=TokenError(
error="invalid_grant",
error_description="The authorization code is invalid",
),
):
# Try to exchange the code for tokens
token_response = await client.post(
"/token",
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri,
"client_id": client_id,
"client_secret": client_secret,
"code_verifier": pkce_challenge["code_verifier"],
},
)
# Verify the response
assert token_response.status_code == 400
data = token_response.json()
assert data["error"] == "invalid_grant"
assert data["error_description"] == "The authorization code is invalid"
@pytest.mark.anyio
async def test_token_error_handling_refresh_token(
self,
client: httpx.AsyncClient,
oauth_provider: MockOAuthProvider,
registered_client: dict[str, Any],
pkce_challenge: dict[str, str],
@pytest.mark.anyio
async def test_authorize_error_handling(
client: httpx.AsyncClient,
oauth_provider: MockOAuthProvider,
registered_client: dict[str, Any],
pkce_challenge: dict[str, str],
):
# Mock the authorize method to raise an authorize error
with unittest.mock.patch.object(
oauth_provider,
"authorize",
side_effect=AuthorizeError(error="access_denied", error_description="The user denied the request"),
):
# Register the client and get tokens
# Register the client
client_id = registered_client["client_id"]
client_secret = registered_client["client_secret"]
redirect_uri = registered_client["redirect_uris"][0]
# First get an authorization code
auth_response = await client.get(
"/authorize",
params={
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"code_challenge": pkce_challenge["code_challenge"],
"code_challenge_method": "S256",
"state": "test_state",
},
)
assert auth_response.status_code == 302, auth_response.content
# Prepare an authorization request
params = {
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"code_challenge": pkce_challenge["code_challenge"],
"code_challenge_method": "S256",
"state": "test_state",
}
redirect_url = auth_response.headers["location"]
# Send the authorization request
response = await client.get("/authorize", params=params)
# Verify the response is a redirect with error parameters
assert response.status_code == 302
redirect_url = response.headers["location"]
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url.query)
code = query_params["code"][0]
# Exchange the code for tokens
assert query_params["error"][0] == "access_denied"
assert "error_description" in query_params
assert query_params["state"][0] == "test_state"
@pytest.mark.anyio
async def test_token_error_handling_auth_code(
client: httpx.AsyncClient,
oauth_provider: MockOAuthProvider,
registered_client: dict[str, Any],
pkce_challenge: dict[str, str],
):
# Register the client and get an auth code
client_id = registered_client["client_id"]
client_secret = registered_client["client_secret"]
redirect_uri = registered_client["redirect_uris"][0]
# First get an authorization code
auth_response = await client.get(
"/authorize",
params={
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"code_challenge": pkce_challenge["code_challenge"],
"code_challenge_method": "S256",
"state": "test_state",
},
)
redirect_url = auth_response.headers["location"]
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url.query)
code = query_params["code"][0]
# Mock the exchange_authorization_code method to raise a token error
with unittest.mock.patch.object(
oauth_provider,
"exchange_authorization_code",
side_effect=TokenError(
error="invalid_grant",
error_description="The authorization code is invalid",
),
):
# Try to exchange the code for tokens
token_response = await client.post(
"/token",
data={
@ -265,31 +209,82 @@ class TestTokenErrorHandling:
},
)
tokens = token_response.json()
refresh_token = tokens["refresh_token"]
# Verify the response
assert token_response.status_code == 400
data = token_response.json()
assert data["error"] == "invalid_grant"
assert data["error_description"] == "The authorization code is invalid"
# Mock the exchange_refresh_token method to raise a token error
with unittest.mock.patch.object(
oauth_provider,
"exchange_refresh_token",
side_effect=TokenError(
error="invalid_scope",
error_description="The requested scope is invalid",
),
):
# Try to use the refresh token
refresh_response = await client.post(
"/token",
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
)
# Verify the response
assert refresh_response.status_code == 400
data = refresh_response.json()
assert data["error"] == "invalid_scope"
assert data["error_description"] == "The requested scope is invalid"
@pytest.mark.anyio
async def test_token_error_handling_refresh_token(
client: httpx.AsyncClient,
oauth_provider: MockOAuthProvider,
registered_client: dict[str, Any],
pkce_challenge: dict[str, str],
):
# Register the client and get tokens
client_id = registered_client["client_id"]
client_secret = registered_client["client_secret"]
redirect_uri = registered_client["redirect_uris"][0]
# First get an authorization code
auth_response = await client.get(
"/authorize",
params={
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"code_challenge": pkce_challenge["code_challenge"],
"code_challenge_method": "S256",
"state": "test_state",
},
)
assert auth_response.status_code == 302, auth_response.content
redirect_url = auth_response.headers["location"]
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url.query)
code = query_params["code"][0]
# Exchange the code for tokens
token_response = await client.post(
"/token",
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri,
"client_id": client_id,
"client_secret": client_secret,
"code_verifier": pkce_challenge["code_verifier"],
},
)
tokens = token_response.json()
refresh_token = tokens["refresh_token"]
# Mock the exchange_refresh_token method to raise a token error
with unittest.mock.patch.object(
oauth_provider,
"exchange_refresh_token",
side_effect=TokenError(
error="invalid_scope",
error_description="The requested scope is invalid",
),
):
# Try to use the refresh token
refresh_response = await client.post(
"/token",
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
)
# Verify the response
assert refresh_response.status_code == 400
data = refresh_response.json()
assert data["error"] == "invalid_scope"
assert data["error_description"] == "The requested scope is invalid"

156
tests/server/auth/test_protected_resource.py
View file

@ -105,90 +105,94 @@ async def test_metadata_endpoint_without_path(root_resource_client: httpx.AsyncC
)
class TestMetadataUrlConstruction:
"""Test URL construction utility function."""
# Tests for URL construction utility function
def test_url_without_path(self):
"""Test URL construction for resource without path component."""
resource_url = AnyHttpUrl("https://example.com")
result = build_resource_metadata_url(resource_url)
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
def test_url_with_path_component(self):
"""Test URL construction for resource with path component."""
resource_url = AnyHttpUrl("https://example.com/mcp")
result = build_resource_metadata_url(resource_url)
assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp"
def test_metadata_url_construction_url_without_path():
"""Test URL construction for resource without path component."""
resource_url = AnyHttpUrl("https://example.com")
result = build_resource_metadata_url(resource_url)
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
def test_url_with_trailing_slash_only(self):
"""Test URL construction for resource with trailing slash only."""
resource_url = AnyHttpUrl("https://example.com/")
result = build_resource_metadata_url(resource_url)
# Trailing slash should be treated as empty path
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
@pytest.mark.parametrize(
"resource_url,expected_url",
[
("https://example.com", "https://example.com/.well-known/oauth-protected-resource"),
("https://example.com/", "https://example.com/.well-known/oauth-protected-resource"),
("https://example.com/mcp", "https://example.com/.well-known/oauth-protected-resource/mcp"),
("http://localhost:8001/mcp", "http://localhost:8001/.well-known/oauth-protected-resource/mcp"),
],
def test_metadata_url_construction_url_with_path_component():
"""Test URL construction for resource with path component."""
resource_url = AnyHttpUrl("https://example.com/mcp")
result = build_resource_metadata_url(resource_url)
assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp"
def test_metadata_url_construction_url_with_trailing_slash_only():
"""Test URL construction for resource with trailing slash only."""
resource_url = AnyHttpUrl("https://example.com/")
result = build_resource_metadata_url(resource_url)
# Trailing slash should be treated as empty path
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
@pytest.mark.parametrize(
"resource_url,expected_url",
[
("https://example.com", "https://example.com/.well-known/oauth-protected-resource"),
("https://example.com/", "https://example.com/.well-known/oauth-protected-resource"),
("https://example.com/mcp", "https://example.com/.well-known/oauth-protected-resource/mcp"),
("http://localhost:8001/mcp", "http://localhost:8001/.well-known/oauth-protected-resource/mcp"),
],
)
def test_metadata_url_construction_various_resource_configurations(resource_url: str, expected_url: str):
"""Test URL construction with various resource configurations."""
result = build_resource_metadata_url(AnyHttpUrl(resource_url))
assert str(result) == expected_url
# Tests for consistency between URL generation and route registration
def test_route_consistency_route_path_matches_metadata_url():
"""Test that route path matches the generated metadata URL."""
resource_url = AnyHttpUrl("https://example.com/mcp")
# Generate metadata URL
metadata_url = build_resource_metadata_url(resource_url)
# Create routes
routes = create_protected_resource_routes(
resource_url=resource_url,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
def test_various_resource_configurations(self, resource_url: str, expected_url: str):
"""Test URL construction with various resource configurations."""
result = build_resource_metadata_url(AnyHttpUrl(resource_url))
assert str(result) == expected_url
# Extract path from metadata URL
metadata_path = urlparse(str(metadata_url)).path
# Verify consistency
assert len(routes) == 1
assert routes[0].path == metadata_path
class TestRouteConsistency:
"""Test consistency between URL generation and route registration."""
@pytest.mark.parametrize(
"resource_url,expected_path",
[
("https://example.com", "/.well-known/oauth-protected-resource"),
("https://example.com/", "/.well-known/oauth-protected-resource"),
("https://example.com/mcp", "/.well-known/oauth-protected-resource/mcp"),
],
)
def test_route_consistency_consistent_paths_for_various_resources(resource_url: str, expected_path: str):
"""Test that URL generation and route creation are consistent."""
resource_url_obj = AnyHttpUrl(resource_url)
def test_route_path_matches_metadata_url(self):
"""Test that route path matches the generated metadata URL."""
resource_url = AnyHttpUrl("https://example.com/mcp")
# Test URL generation
metadata_url = build_resource_metadata_url(resource_url_obj)
url_path = urlparse(str(metadata_url)).path
# Generate metadata URL
metadata_url = build_resource_metadata_url(resource_url)
# Create routes
routes = create_protected_resource_routes(
resource_url=resource_url,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
# Extract path from metadata URL
metadata_path = urlparse(str(metadata_url)).path
# Verify consistency
assert len(routes) == 1
assert routes[0].path == metadata_path
@pytest.mark.parametrize(
"resource_url,expected_path",
[
("https://example.com", "/.well-known/oauth-protected-resource"),
("https://example.com/", "/.well-known/oauth-protected-resource"),
("https://example.com/mcp", "/.well-known/oauth-protected-resource/mcp"),
],
# Test route creation
routes = create_protected_resource_routes(
resource_url=resource_url_obj,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
def test_consistent_paths_for_various_resources(self, resource_url: str, expected_path: str):
"""Test that URL generation and route creation are consistent."""
resource_url_obj = AnyHttpUrl(resource_url)
route_path = routes[0].path
# Test URL generation
metadata_url = build_resource_metadata_url(resource_url_obj)
url_path = urlparse(str(metadata_url)).path
# Test route creation
routes = create_protected_resource_routes(
resource_url=resource_url_obj,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
route_path = routes[0].path
# Both should match expected path
assert url_path == expected_path
assert route_path == expected_path
assert url_path == route_path
# Both should match expected path
assert url_path == expected_path
assert route_path == expected_path
assert url_path == route_path

112
tests/server/auth/test_provider.py
View file

@ -3,73 +3,77 @@
from mcp.server.auth.provider import construct_redirect_uri
class TestConstructRedirectUri:
"""Tests for the construct_redirect_uri function."""
def test_construct_redirect_uri_no_existing_params():
"""Test construct_redirect_uri with no existing query parameters."""
base_uri = "http://localhost:8000/callback"
result = construct_redirect_uri(base_uri, code="auth_code", state="test_state")
def test_construct_redirect_uri_no_existing_params(self):
"""Test construct_redirect_uri with no existing query parameters."""
base_uri = "http://localhost:8000/callback"
result = construct_redirect_uri(base_uri, code="auth_code", state="test_state")
assert "http://localhost:8000/callback?code=auth_code&state=test_state" == result
assert "http://localhost:8000/callback?code=auth_code&state=test_state" == result
def test_construct_redirect_uri_with_existing_params(self):
"""Test construct_redirect_uri with existing query parameters (regression test for #1279)."""
base_uri = "http://localhost:8000/callback?session_id=1234"
result = construct_redirect_uri(base_uri, code="auth_code", state="test_state")
def test_construct_redirect_uri_with_existing_params():
"""Test construct_redirect_uri with existing query parameters (regression test for #1279)."""
base_uri = "http://localhost:8000/callback?session_id=1234"
result = construct_redirect_uri(base_uri, code="auth_code", state="test_state")
# Should preserve existing params and add new ones
assert "session_id=1234" in result
assert "code=auth_code" in result
assert "state=test_state" in result
assert result.startswith("http://localhost:8000/callback?")
# Should preserve existing params and add new ones
assert "session_id=1234" in result
assert "code=auth_code" in result
assert "state=test_state" in result
assert result.startswith("http://localhost:8000/callback?")
def test_construct_redirect_uri_multiple_existing_params(self):
"""Test construct_redirect_uri with multiple existing query parameters."""
base_uri = "http://localhost:8000/callback?session_id=1234&user=test"
result = construct_redirect_uri(base_uri, code="auth_code")
assert "session_id=1234" in result
assert "user=test" in result
assert "code=auth_code" in result
def test_construct_redirect_uri_multiple_existing_params():
"""Test construct_redirect_uri with multiple existing query parameters."""
base_uri = "http://localhost:8000/callback?session_id=1234&user=test"
result = construct_redirect_uri(base_uri, code="auth_code")
def test_construct_redirect_uri_with_none_values(self):
"""Test construct_redirect_uri filters out None values."""
base_uri = "http://localhost:8000/callback"
result = construct_redirect_uri(base_uri, code="auth_code", state=None)
assert "session_id=1234" in result
assert "user=test" in result
assert "code=auth_code" in result
assert result == "http://localhost:8000/callback?code=auth_code"
assert "state" not in result
def test_construct_redirect_uri_empty_params(self):
"""Test construct_redirect_uri with no additional parameters."""
base_uri = "http://localhost:8000/callback?existing=param"
result = construct_redirect_uri(base_uri)
def test_construct_redirect_uri_with_none_values():
"""Test construct_redirect_uri filters out None values."""
base_uri = "http://localhost:8000/callback"
result = construct_redirect_uri(base_uri, code="auth_code", state=None)
assert result == "http://localhost:8000/callback?existing=param"
assert result == "http://localhost:8000/callback?code=auth_code"
assert "state" not in result
def test_construct_redirect_uri_duplicate_param_names(self):
"""Test construct_redirect_uri when adding param that already exists."""
base_uri = "http://localhost:8000/callback?code=existing"
result = construct_redirect_uri(base_uri, code="new_code")
# Should contain both values (this is expected behavior of parse_qs/urlencode)
assert "code=existing" in result
assert "code=new_code" in result
def test_construct_redirect_uri_empty_params():
"""Test construct_redirect_uri with no additional parameters."""
base_uri = "http://localhost:8000/callback?existing=param"
result = construct_redirect_uri(base_uri)
def test_construct_redirect_uri_multivalued_existing_params(self):
"""Test construct_redirect_uri with existing multi-valued parameters."""
base_uri = "http://localhost:8000/callback?scope=read&scope=write"
result = construct_redirect_uri(base_uri, code="auth_code")
assert result == "http://localhost:8000/callback?existing=param"
assert "scope=read" in result
assert "scope=write" in result
assert "code=auth_code" in result
def test_construct_redirect_uri_encoded_values(self):
"""Test construct_redirect_uri handles URL encoding properly."""
base_uri = "http://localhost:8000/callback"
result = construct_redirect_uri(base_uri, state="test state with spaces")
def test_construct_redirect_uri_duplicate_param_names():
"""Test construct_redirect_uri when adding param that already exists."""
base_uri = "http://localhost:8000/callback?code=existing"
result = construct_redirect_uri(base_uri, code="new_code")
# urlencode uses + for spaces by default
assert "state=test+state+with+spaces" in result
# Should contain both values (this is expected behavior of parse_qs/urlencode)
assert "code=existing" in result
assert "code=new_code" in result
def test_construct_redirect_uri_multivalued_existing_params():
"""Test construct_redirect_uri with existing multi-valued parameters."""
base_uri = "http://localhost:8000/callback?scope=read&scope=write"
result = construct_redirect_uri(base_uri, code="auth_code")
assert "scope=read" in result
assert "scope=write" in result
assert "code=auth_code" in result
def test_construct_redirect_uri_encoded_values():
"""Test construct_redirect_uri handles URL encoding properly."""
base_uri = "http://localhost:8000/callback"
result = construct_redirect_uri(base_uri, state="test state with spaces")
# urlencode uses + for spaces by default
assert "state=test+state+with+spaces" in result

67
tests/server/lowlevel/test_helper_types.py
View file

@ -10,51 +10,50 @@ by the low-level server to package resource content before sending it over the M
from mcp.server.lowlevel.helper_types import ReadResourceContents
class TestReadResourceContentsMetadata:
"""Test ReadResourceContents meta field.
def test_read_resource_contents_with_metadata():
"""Test that ReadResourceContents accepts meta parameter.
ReadResourceContents is an internal helper type used by the low-level MCP server.
When a resource is read, the server creates a ReadResourceContents instance that
contains the content, mime type, and now metadata. The low-level server then
extracts the meta field and includes it in the protocol response as _meta.
"""
# Bridge between Resource.meta and MCP protocol _meta field (helper_types.py:11)
metadata = {"version": "1.0", "cached": True}
def test_read_resource_contents_with_metadata(self):
"""Test that ReadResourceContents accepts meta parameter."""
# Bridge between Resource.meta and MCP protocol _meta field (helper_types.py:11)
metadata = {"version": "1.0", "cached": True}
contents = ReadResourceContents(
content="test content",
mime_type="text/plain",
meta=metadata,
)
contents = ReadResourceContents(
content="test content",
mime_type="text/plain",
meta=metadata,
)
assert contents.meta is not None
assert contents.meta == metadata
assert contents.meta["version"] == "1.0"
assert contents.meta["cached"] is True
assert contents.meta is not None
assert contents.meta == metadata
assert contents.meta["version"] == "1.0"
assert contents.meta["cached"] is True
def test_read_resource_contents_without_metadata(self):
"""Test that ReadResourceContents meta defaults to None."""
# Ensures backward compatibility - meta defaults to None, _meta omitted from protocol (helper_types.py:11)
contents = ReadResourceContents(
content="test content",
mime_type="text/plain",
)
def test_read_resource_contents_without_metadata():
"""Test that ReadResourceContents meta defaults to None."""
# Ensures backward compatibility - meta defaults to None, _meta omitted from protocol (helper_types.py:11)
contents = ReadResourceContents(
content="test content",
mime_type="text/plain",
)
assert contents.meta is None
assert contents.meta is None
def test_read_resource_contents_with_bytes(self):
"""Test that ReadResourceContents works with bytes content and meta."""
# Verifies meta works with both str and bytes content (binary resources like images, PDFs)
metadata = {"encoding": "utf-8"}
contents = ReadResourceContents(
content=b"binary content",
mime_type="application/octet-stream",
meta=metadata,
)
def test_read_resource_contents_with_bytes():
"""Test that ReadResourceContents works with bytes content and meta."""
# Verifies meta works with both str and bytes content (binary resources like images, PDFs)
metadata = {"encoding": "utf-8"}
assert contents.content == b"binary content"
assert contents.meta == metadata
contents = ReadResourceContents(
content=b"binary content",
mime_type="application/octet-stream",
meta=metadata,
)
assert contents.content == b"binary content"
assert contents.meta == metadata

228
tests/server/test_validation.py
View file

@ -20,122 +20,132 @@ from mcp.types import (
ToolUseContent,
)
class TestCheckSamplingToolsCapability:
"""Tests for check_sampling_tools_capability function."""
def test_returns_false_when_caps_none(self) -> None:
"""Returns False when client_caps is None."""
assert check_sampling_tools_capability(None) is False
def test_returns_false_when_sampling_none(self) -> None:
"""Returns False when client_caps.sampling is None."""
caps = ClientCapabilities()
assert check_sampling_tools_capability(caps) is False
def test_returns_false_when_tools_none(self) -> None:
"""Returns False when client_caps.sampling.tools is None."""
caps = ClientCapabilities(sampling=SamplingCapability())
assert check_sampling_tools_capability(caps) is False
def test_returns_true_when_tools_present(self) -> None:
"""Returns True when sampling.tools is present."""
caps = ClientCapabilities(sampling=SamplingCapability(tools=SamplingToolsCapability()))
assert check_sampling_tools_capability(caps) is True
# Tests for check_sampling_tools_capability function
class TestValidateSamplingTools:
"""Tests for validate_sampling_tools function."""
def test_no_error_when_tools_none(self) -> None:
"""No error when tools and tool_choice are None."""
validate_sampling_tools(None, None, None) # Should not raise
def test_raises_when_tools_provided_but_no_capability(self) -> None:
"""Raises McpError when tools provided but client doesn't support."""
tool = Tool(name="test", input_schema={"type": "object"})
with pytest.raises(McpError) as exc_info:
validate_sampling_tools(None, [tool], None)
assert "sampling tools capability" in str(exc_info.value)
def test_raises_when_tool_choice_provided_but_no_capability(self) -> None:
"""Raises McpError when tool_choice provided but client doesn't support."""
with pytest.raises(McpError) as exc_info:
validate_sampling_tools(None, None, ToolChoice(mode="auto"))
assert "sampling tools capability" in str(exc_info.value)
def test_no_error_when_capability_present(self) -> None:
"""No error when client has sampling.tools capability."""
caps = ClientCapabilities(sampling=SamplingCapability(tools=SamplingToolsCapability()))
tool = Tool(name="test", input_schema={"type": "object"})
validate_sampling_tools(caps, [tool], ToolChoice(mode="auto")) # Should not raise
def test_check_sampling_tools_capability_returns_false_when_caps_none() -> None:
"""Returns False when client_caps is None."""
assert check_sampling_tools_capability(None) is False
class TestValidateToolUseResultMessages:
"""Tests for validate_tool_use_result_messages function."""
def test_check_sampling_tools_capability_returns_false_when_sampling_none() -> None:
"""Returns False when client_caps.sampling is None."""
caps = ClientCapabilities()
assert check_sampling_tools_capability(caps) is False
def test_no_error_for_empty_messages(self) -> None:
"""No error when messages list is empty."""
validate_tool_use_result_messages([]) # Should not raise
def test_no_error_for_simple_text_messages(self) -> None:
"""No error for simple text messages."""
messages = [
SamplingMessage(role="user", content=TextContent(type="text", text="Hello")),
SamplingMessage(role="assistant", content=TextContent(type="text", text="Hi")),
]
validate_tool_use_result_messages(messages) # Should not raise
def test_check_sampling_tools_capability_returns_false_when_tools_none() -> None:
"""Returns False when client_caps.sampling.tools is None."""
caps = ClientCapabilities(sampling=SamplingCapability())
assert check_sampling_tools_capability(caps) is False
def test_raises_when_tool_result_mixed_with_other_content(self) -> None:
"""Raises when tool_result is mixed with other content types."""
messages = [
SamplingMessage(
role="user",
content=[
ToolResultContent(type="tool_result", tool_use_id="123"),
TextContent(type="text", text="also this"),
],
),
]
with pytest.raises(ValueError, match="only tool_result content"):
validate_tool_use_result_messages(messages)
def test_raises_when_tool_result_without_previous_tool_use(self) -> None:
"""Raises when tool_result appears without preceding tool_use."""
messages = [
SamplingMessage(
role="user",
content=ToolResultContent(type="tool_result", tool_use_id="123"),
),
]
with pytest.raises(ValueError, match="previous message containing tool_use"):
validate_tool_use_result_messages(messages)
def test_check_sampling_tools_capability_returns_true_when_tools_present() -> None:
"""Returns True when sampling.tools is present."""
caps = ClientCapabilities(sampling=SamplingCapability(tools=SamplingToolsCapability()))
assert check_sampling_tools_capability(caps) is True
def test_raises_when_tool_result_ids_dont_match_tool_use(self) -> None:
"""Raises when tool_result IDs don't match tool_use IDs."""
messages = [
SamplingMessage(
role="assistant",
content=ToolUseContent(type="tool_use", id="tool-1", name="test", input={}),
),
SamplingMessage(
role="user",
content=ToolResultContent(type="tool_result", tool_use_id="tool-2"),
),
]
with pytest.raises(ValueError, match="do not match"):
validate_tool_use_result_messages(messages)
def test_no_error_when_tool_result_matches_tool_use(self) -> None:
"""No error when tool_result IDs match tool_use IDs."""
messages = [
SamplingMessage(
role="assistant",
content=ToolUseContent(type="tool_use", id="tool-1", name="test", input={}),
),
SamplingMessage(
role="user",
content=ToolResultContent(type="tool_result", tool_use_id="tool-1"),
),
]
validate_tool_use_result_messages(messages) # Should not raise
# Tests for validate_sampling_tools function
def test_validate_sampling_tools_no_error_when_tools_none() -> None:
"""No error when tools and tool_choice are None."""
validate_sampling_tools(None, None, None) # Should not raise
def test_validate_sampling_tools_raises_when_tools_provided_but_no_capability() -> None:
"""Raises McpError when tools provided but client doesn't support."""
tool = Tool(name="test", input_schema={"type": "object"})
with pytest.raises(McpError) as exc_info:
validate_sampling_tools(None, [tool], None)
assert "sampling tools capability" in str(exc_info.value)
def test_validate_sampling_tools_raises_when_tool_choice_provided_but_no_capability() -> None:
"""Raises McpError when tool_choice provided but client doesn't support."""
with pytest.raises(McpError) as exc_info:
validate_sampling_tools(None, None, ToolChoice(mode="auto"))
assert "sampling tools capability" in str(exc_info.value)
def test_validate_sampling_tools_no_error_when_capability_present() -> None:
"""No error when client has sampling.tools capability."""
caps = ClientCapabilities(sampling=SamplingCapability(tools=SamplingToolsCapability()))
tool = Tool(name="test", input_schema={"type": "object"})
validate_sampling_tools(caps, [tool], ToolChoice(mode="auto")) # Should not raise
# Tests for validate_tool_use_result_messages function
def test_validate_tool_use_result_messages_no_error_for_empty_messages() -> None:
"""No error when messages list is empty."""
validate_tool_use_result_messages([]) # Should not raise
def test_validate_tool_use_result_messages_no_error_for_simple_text_messages() -> None:
"""No error for simple text messages."""
messages = [
SamplingMessage(role="user", content=TextContent(type="text", text="Hello")),
SamplingMessage(role="assistant", content=TextContent(type="text", text="Hi")),
]
validate_tool_use_result_messages(messages) # Should not raise
def test_validate_tool_use_result_messages_raises_when_tool_result_mixed_with_other_content() -> None:
"""Raises when tool_result is mixed with other content types."""
messages = [
SamplingMessage(
role="user",
content=[
ToolResultContent(type="tool_result", tool_use_id="123"),
TextContent(type="text", text="also this"),
],
),
]
with pytest.raises(ValueError, match="only tool_result content"):
validate_tool_use_result_messages(messages)
def test_validate_tool_use_result_messages_raises_when_tool_result_without_previous_tool_use() -> None:
"""Raises when tool_result appears without preceding tool_use."""
messages = [
SamplingMessage(
role="user",
content=ToolResultContent(type="tool_result", tool_use_id="123"),
),
]
with pytest.raises(ValueError, match="previous message containing tool_use"):
validate_tool_use_result_messages(messages)
def test_validate_tool_use_result_messages_raises_when_tool_result_ids_dont_match_tool_use() -> None:
"""Raises when tool_result IDs don't match tool_use IDs."""
messages = [
SamplingMessage(
role="assistant",
content=ToolUseContent(type="tool_use", id="tool-1", name="test", input={}),
),
SamplingMessage(
role="user",
content=ToolResultContent(type="tool_result", tool_use_id="tool-2"),
),
]
with pytest.raises(ValueError, match="do not match"):
validate_tool_use_result_messages(messages)
def test_validate_tool_use_result_messages_no_error_when_tool_result_matches_tool_use() -> None:
"""No error when tool_result IDs match tool_use IDs."""
messages = [
SamplingMessage(
role="assistant",
content=ToolUseContent(type="tool_use", id="tool-1", name="test", input={}),
),
SamplingMessage(
role="user",
content=ToolResultContent(type="tool_result", tool_use_id="tool-1"),
),
]
validate_tool_use_result_messages(messages) # Should not raise