Skip to content
Open
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
57 changes: 0 additions & 57 deletions src/google/adk/a2a/converters/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,6 @@
from __future__ import annotations

ADK_METADATA_KEY_PREFIX = "adk_"
ADK_CONTEXT_ID_PREFIX = "ADK"
ADK_CONTEXT_ID_SEPARATOR = "/"


def _get_adk_metadata_key(key: str) -> str:
Expand All @@ -34,58 +32,3 @@ def _get_adk_metadata_key(key: str) -> str:
if not key:
raise ValueError("Metadata key cannot be empty or None")
return f"{ADK_METADATA_KEY_PREFIX}{key}"


def _to_a2a_context_id(app_name: str, user_id: str, session_id: str) -> str:
"""Converts app name, user id and session id to an A2A context id.

Args:
app_name: The app name.
user_id: The user id.
session_id: The session id.

Returns:
The A2A context id.

Raises:
ValueError: If any of the input parameters are empty or None.
"""
if not all([app_name, user_id, session_id]):
raise ValueError(
"All parameters (app_name, user_id, session_id) must be non-empty"
)
return ADK_CONTEXT_ID_SEPARATOR.join(
[ADK_CONTEXT_ID_PREFIX, app_name, user_id, session_id]
)


def _from_a2a_context_id(
context_id: str | None,
) -> tuple[str, str, str] | tuple[None, None, None]:
"""Converts an A2A context id to app name, user id and session id.
if context_id is None, return None, None, None
if context_id is not None, but not in the format of
ADK$app_name$user_id$session_id, return None, None, None

Args:
context_id: The A2A context id.

Returns:
The app name, user id and session id, or (None, None, None) if invalid.
"""
if not context_id:
return None, None, None

try:
parts = context_id.split(ADK_CONTEXT_ID_SEPARATOR)
if len(parts) != 4:
return None, None, None

prefix, app_name, user_id, session_id = parts
if prefix == ADK_CONTEXT_ID_PREFIX and app_name and user_id and session_id:
return app_name, user_id, session_id
except ValueError:
# Handle any split errors gracefully
pass

return None, None, None
157 changes: 0 additions & 157 deletions tests/unittests/a2a/converters/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from google.adk.a2a.converters.utils import _from_a2a_context_id
from google.adk.a2a.converters.utils import _get_adk_metadata_key
from google.adk.a2a.converters.utils import _to_a2a_context_id
from google.adk.a2a.converters.utils import ADK_CONTEXT_ID_PREFIX
from google.adk.a2a.converters.utils import ADK_METADATA_KEY_PREFIX
import pytest

Expand Down Expand Up @@ -48,157 +45,3 @@ def test_get_adk_metadata_key_whitespace(self):
key = " "
result = _get_adk_metadata_key(key)
assert result == f"{ADK_METADATA_KEY_PREFIX}{key}"

def test_to_a2a_context_id_success(self):
"""Test successful context ID generation."""
app_name = "test-app"
user_id = "test-user"
session_id = "test-session"

result = _to_a2a_context_id(app_name, user_id, session_id)

expected = f"{ADK_CONTEXT_ID_PREFIX}/test-app/test-user/test-session"
assert result == expected

def test_to_a2a_context_id_empty_app_name(self):
"""Test context ID generation with empty app name."""
with pytest.raises(
ValueError,
match=(
"All parameters \\(app_name, user_id, session_id\\) must be"
" non-empty"
),
):
_to_a2a_context_id("", "user", "session")

def test_to_a2a_context_id_empty_user_id(self):
"""Test context ID generation with empty user ID."""
with pytest.raises(
ValueError,
match=(
"All parameters \\(app_name, user_id, session_id\\) must be"
" non-empty"
),
):
_to_a2a_context_id("app", "", "session")

def test_to_a2a_context_id_empty_session_id(self):
"""Test context ID generation with empty session ID."""
with pytest.raises(
ValueError,
match=(
"All parameters \\(app_name, user_id, session_id\\) must be"
" non-empty"
),
):
_to_a2a_context_id("app", "user", "")

def test_to_a2a_context_id_none_values(self):
"""Test context ID generation with None values."""
with pytest.raises(
ValueError,
match=(
"All parameters \\(app_name, user_id, session_id\\) must be"
" non-empty"
),
):
_to_a2a_context_id(None, "user", "session")

def test_to_a2a_context_id_special_characters(self):
"""Test context ID generation with special characters."""
app_name = "test-app@2024"
user_id = "user_123"
session_id = "session-456"

result = _to_a2a_context_id(app_name, user_id, session_id)

expected = f"{ADK_CONTEXT_ID_PREFIX}/test-app@2024/user_123/session-456"
assert result == expected

def test_from_a2a_context_id_success(self):
"""Test successful context ID parsing."""
context_id = f"{ADK_CONTEXT_ID_PREFIX}/test-app/test-user/test-session"

app_name, user_id, session_id = _from_a2a_context_id(context_id)

assert app_name == "test-app"
assert user_id == "test-user"
assert session_id == "test-session"

def test_from_a2a_context_id_none_input(self):
"""Test context ID parsing with None input."""
result = _from_a2a_context_id(None)
assert result == (None, None, None)

def test_from_a2a_context_id_empty_string(self):
"""Test context ID parsing with empty string."""
result = _from_a2a_context_id("")
assert result == (None, None, None)

def test_from_a2a_context_id_invalid_prefix(self):
"""Test context ID parsing with invalid prefix."""
context_id = "INVALID/test-app/test-user/test-session"

result = _from_a2a_context_id(context_id)

assert result == (None, None, None)

def test_from_a2a_context_id_too_few_parts(self):
"""Test context ID parsing with too few parts."""
context_id = f"{ADK_CONTEXT_ID_PREFIX}/test-app/test-user"

result = _from_a2a_context_id(context_id)

assert result == (None, None, None)

def test_from_a2a_context_id_too_many_parts(self):
"""Test context ID parsing with too many parts."""
context_id = (
f"{ADK_CONTEXT_ID_PREFIX}/test-app/test-user/test-session/extra"
)

result = _from_a2a_context_id(context_id)

assert result == (None, None, None)

def test_from_a2a_context_id_empty_components(self):
"""Test context ID parsing with empty components."""
context_id = f"{ADK_CONTEXT_ID_PREFIX}//test-user/test-session"

result = _from_a2a_context_id(context_id)

assert result == (None, None, None)

def test_from_a2a_context_id_no_dollar_separator(self):
"""Test context ID parsing without dollar separators."""
context_id = f"{ADK_CONTEXT_ID_PREFIX}-test-app-test-user-test-session"

result = _from_a2a_context_id(context_id)

assert result == (None, None, None)

def test_roundtrip_context_id(self):
"""Test roundtrip conversion: to -> from."""
app_name = "test-app"
user_id = "test-user"
session_id = "test-session"

# Convert to context ID
context_id = _to_a2a_context_id(app_name, user_id, session_id)

# Convert back
parsed_app, parsed_user, parsed_session = _from_a2a_context_id(context_id)

assert parsed_app == app_name
assert parsed_user == user_id
assert parsed_session == session_id

def test_from_a2a_context_id_special_characters(self):
"""Test context ID parsing with special characters."""
context_id = f"{ADK_CONTEXT_ID_PREFIX}/test-app@2024/user_123/session-456"

app_name, user_id, session_id = _from_a2a_context_id(context_id)

assert app_name == "test-app@2024"
assert user_id == "user_123"
assert session_id == "session-456"
Loading