Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "sap-cloud-sdk"
version = "0.43.0"
version = "0.43.1"
description = "SAP Cloud SDK for Python"
readme = "README.md"
license = "Apache-2.0"
Expand Down
11 changes: 9 additions & 2 deletions src/sap_cloud_sdk/destination/_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -413,13 +413,15 @@ class AuthToken:
type: Token type (e.g., "Bearer", "Basic")
value: Base64 encoded token binary content
http_header: Dictionary with 'key' and 'value' for the prepared HTTP header
error: Error message returned by the Destination Service when token retrieval fails
refresh_token: Optional base64 encoded refresh token
scope: Optional token scopes as space-delimited string
"""

type: str
value: str
http_header: Dict[str, str]
error: Optional[str] = None
refresh_token: Optional[str] = None
scope: Optional[str] = None

Expand All @@ -434,23 +436,28 @@ def from_dict(cls, obj: Dict[str, Any]) -> "AuthToken":
AuthToken: Parsed auth token dataclass.

Raises:
DestinationOperationError: If required fields are missing.
DestinationOperationError: If required fields are missing, or if the token
carries an error from the Destination Service.
"""
token_type = obj.get("type") or ""
value = obj.get("value") or ""
http_header = obj.get("http_header") or {}
error = obj.get("error") or None
refresh_token = obj.get("refresh_token")
scope = obj.get("scope")

if not token_type or not value or not http_header:
if not error and (not token_type or not value or not http_header):
raise DestinationOperationError(
"auth token is missing required fields (type/value/http_header)"
)
if error and (not token_type or not value or not http_header):
raise DestinationOperationError(f"auth token retrieval failed: {error}")

return cls(
type=token_type,
value=value,
http_header=http_header,
error=error,
refresh_token=refresh_token,
scope=scope,
)
Expand Down
4 changes: 4 additions & 0 deletions tests/agentgateway/unit/test_lob.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ def test_fetches_and_decodes_token_and_url(self):
header_value = "Bearer my-raw-jwt-token-123"
mock_dest = MagicMock()
mock_dest.auth_tokens = [MagicMock()]
mock_dest.auth_tokens[0].error = None
mock_dest.auth_tokens[0].http_header = {"value": header_value}
mock_dest.url = "https://agw.example.com/"

Expand All @@ -112,6 +113,7 @@ def test_strips_trailing_slashes_from_url(self):
header_value = "Bearer token"
mock_dest = MagicMock()
mock_dest.auth_tokens = [MagicMock()]
mock_dest.auth_tokens[0].error = None
mock_dest.auth_tokens[0].http_header = {"value": header_value}
mock_dest.url = "https://agw.example.com/v1/mcp///"

Expand Down Expand Up @@ -149,6 +151,7 @@ def test_raises_when_empty_token_value(self):
"""Raise MCPServerNotFoundError when http_header value is empty."""
mock_dest = MagicMock()
mock_dest.auth_tokens = [MagicMock()]
mock_dest.auth_tokens[0].error = None
mock_dest.auth_tokens[0].http_header = {"value": ""}

with patch(
Expand All @@ -163,6 +166,7 @@ def test_passes_options_to_destination(self):
"""Pass consumption options to get_destination."""
mock_dest = MagicMock()
mock_dest.auth_tokens = [MagicMock()]
mock_dest.auth_tokens[0].error = None
mock_dest.auth_tokens[0].http_header = {"value": "Bearer token"}
mock_dest.url = "https://agw.example.com"
mock_options = MagicMock()
Expand Down
33 changes: 33 additions & 0 deletions tests/destination/unit/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -293,6 +293,39 @@ def test_get_headers_includes_auth_tokens(self):
assert dest.get_headers()["Authorization"] == "Bearer eyJ123"


class TestAuthTokenModel:
"""Tests for AuthToken dataclass."""

def test_from_dict_valid(self):
"""Parse a valid auth token dict."""
token = AuthToken.from_dict({
"type": "Bearer",
"value": "eyJ123",
"http_header": {"key": "Authorization", "value": "Bearer eyJ123"},
})
assert token.type == "Bearer"
assert token.value == "eyJ123"
assert token.error is None

def test_from_dict_with_error_field(self):
"""Raise DestinationOperationError with the Destination Service error message."""
with pytest.raises(
DestinationOperationError,
match="No consumed apis matching provided resource parameter found.",
):
AuthToken.from_dict({
"type": "",
"value": "",
"error": "No consumed apis matching provided resource parameter found.",
"expires_in": "0",
})

def test_from_dict_missing_fields_without_error_raises(self):
"""Raise DestinationOperationError when required fields are missing and no error is set."""
with pytest.raises(DestinationOperationError, match="missing required fields"):
AuthToken.from_dict({"type": "", "value": "", "http_header": {}})


class TestFragmentModel:
"""Tests for Fragment dataclass."""

Expand Down
Loading