chore: Make UT of a2a consistent about how tests should be skipped when python verison < 3.10

PiperOrigin-RevId: 801040421
This commit is contained in:
Xiang (Sean) Zhou
2025-08-29 14:59:27 -07:00
committed by Copybara-Service
parent 2eddc5e4d3
commit 98b0426cd2
11 changed files with 197 additions and 323 deletions
@@ -49,34 +49,10 @@ try:
from google.adk.events.event_actions import EventActions
except ImportError as e:
if sys.version_info < (3, 10):
# Create dummy classes to prevent NameError during test collection
# Tests will be skipped anyway due to pytestmark
class DummyTypes:
pass
DataPart = DummyTypes()
Message = DummyTypes()
Role = DummyTypes()
Task = DummyTypes()
TaskState = DummyTypes()
TaskStatusUpdateEvent = DummyTypes()
_create_artifact_id = lambda *args: None
_create_error_status_event = lambda *args: None
_create_status_update_event = lambda *args: None
_get_adk_metadata_key = lambda *args: None
_get_context_metadata = lambda *args: None
_process_long_running_tool = lambda *args: None
_serialize_metadata_value = lambda *args: None
ADK_METADATA_KEY_PREFIX = "adk_"
ARTIFACT_ID_SEPARATOR = "_"
convert_event_to_a2a_events = lambda *args: None
convert_event_to_a2a_message = lambda *args: None
convert_a2a_task_to_event = lambda *args: None
DEFAULT_ERROR_MESSAGE = "error"
InvocationContext = DummyTypes()
Event = DummyTypes()
EventActions = DummyTypes()
types = DummyTypes()
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
@@ -38,21 +38,10 @@ try:
from google.genai import types as genai_types
except ImportError as e:
if sys.version_info < (3, 10):
# Create dummy classes to prevent NameError during test collection
# Tests will be skipped anyway due to pytestmark
class DummyTypes:
pass
a2a_types = DummyTypes()
genai_types = DummyTypes()
A2A_DATA_PART_METADATA_TYPE_FUNCTION_CALL = "function_call"
A2A_DATA_PART_METADATA_TYPE_FUNCTION_RESPONSE = "function_response"
A2A_DATA_PART_METADATA_TYPE_CODE_EXECUTION_RESULT = "code_execution_result"
A2A_DATA_PART_METADATA_TYPE_EXECUTABLE_CODE = "executable_code"
A2A_DATA_PART_METADATA_TYPE_KEY = "type"
convert_a2a_part_to_genai_part = lambda x: None
convert_genai_part_to_a2a_part = lambda x: None
_get_adk_metadata_key = lambda x: f"adk_{x}"
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
@@ -20,7 +20,7 @@ import pytest
# Skip all tests in this module if Python version is less than 3.10
pytestmark = pytest.mark.skipif(
sys.version_info < (3, 10), reason="A2A tool requires Python 3.10+"
sys.version_info < (3, 10), reason="A2A requires Python 3.10+"
)
# Import dependencies with version checking
@@ -32,17 +32,10 @@ try:
from google.genai import types as genai_types
except ImportError as e:
if sys.version_info < (3, 10):
# Create dummy classes to prevent NameError during test collection
# Tests will be skipped anyway due to pytestmark
class DummyTypes:
pass
a2a_types = DummyTypes()
genai_types = DummyTypes()
RequestContext = DummyTypes()
RunConfig = DummyTypes()
_get_user_id = lambda x: None
convert_a2a_request_to_adk_run_args = lambda x: None
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
+15 -6
View File
@@ -21,12 +21,21 @@ pytestmark = pytest.mark.skipif(
sys.version_info < (3, 10), reason="A2A requires Python 3.10+"
)
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
# Import dependencies with version checking
try:
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
except ImportError as e:
if sys.version_info < (3, 10):
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
class TestUtilsFunctions:
@@ -21,7 +21,7 @@ import pytest
# Skip all tests in this module if Python version is less than 3.10
pytestmark = pytest.mark.skipif(
sys.version_info < (3, 10), reason="A2A tool requires Python 3.10+"
sys.version_info < (3, 10), reason="A2A requires Python 3.10+"
)
# Import dependencies with version checking
@@ -37,23 +37,10 @@ try:
from google.adk.runners import Runner
except ImportError as e:
if sys.version_info < (3, 10):
# Create dummy classes to prevent NameError during test collection
# Tests will be skipped anyway due to pytestmark
class DummyTypes:
pass
RequestContext = DummyTypes()
EventQueue = DummyTypes()
Message = DummyTypes()
Role = DummyTypes()
TaskState = DummyTypes()
TaskStatus = DummyTypes()
TaskStatusUpdateEvent = DummyTypes()
TextPart = DummyTypes()
A2aAgentExecutor = DummyTypes()
A2aAgentExecutorConfig = DummyTypes()
Event = DummyTypes()
Runner = DummyTypes()
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
@@ -17,22 +17,32 @@ from unittest.mock import Mock
import pytest
# Skip entire module if Python < 3.10
if sys.version_info < (3, 10):
pytest.skip("A2A requires Python 3.10+", allow_module_level=True)
# Skip all tests in this module if Python version is less than 3.10
pytestmark = pytest.mark.skipif(
sys.version_info < (3, 10), reason="A2A requires Python 3.10+"
)
# Normal imports after the skip
from a2a.types import Message
from a2a.types import Part
from a2a.types import Role
from a2a.types import TaskState
from a2a.types import TaskStatus
from a2a.types import TaskStatusUpdateEvent
from a2a.types import TextPart
from google.adk.a2a.executor.task_result_aggregator import TaskResultAggregator
# Import dependencies with version checking
try:
from a2a.types import Message
from a2a.types import Part
from a2a.types import Role
from a2a.types import TaskState
from a2a.types import TaskStatus
from a2a.types import TaskStatusUpdateEvent
from a2a.types import TextPart
from google.adk.a2a.executor.task_result_aggregator import TaskResultAggregator
except ImportError as e:
if sys.version_info < (3, 10):
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
def create_test_message(text: str) -> Message:
def create_test_message(text: str):
"""Helper function to create a test Message object."""
return Message(
message_id="test-msg",
+28 -33
View File
@@ -21,22 +21,35 @@ from unittest.mock import patch
import pytest
# Skip entire module if Python < 3.10
if sys.version_info < (3, 10):
pytest.skip("A2A requires Python 3.10+", allow_module_level=True)
# Skip all tests in this module if Python version is less than 3.10
pytestmark = pytest.mark.skipif(
sys.version_info < (3, 10), reason="A2A requires Python 3.10+"
)
# Normal imports after the skip
from a2a.types import DataPart as A2ADataPart
from a2a.types import Message as A2AMessage
from a2a.types import MessageSendConfiguration
from a2a.types import MessageSendParams
from a2a.types import Part as A2APart
from a2a.types import Role
from a2a.types import SendMessageRequest
from a2a.types import Task as A2ATask
from a2a.types import TaskState
from a2a.types import TaskStatus
from a2a.types import TextPart as A2ATextPart
# Import dependencies with version checking
try:
from a2a.types import DataPart as A2ADataPart
from a2a.types import Message as A2AMessage
from a2a.types import MessageSendConfiguration
from a2a.types import MessageSendParams
from a2a.types import Part as A2APart
from a2a.types import Role
from a2a.types import SendMessageRequest
from a2a.types import Task as A2ATask
from a2a.types import TaskState
from a2a.types import TaskStatus
from a2a.types import TextPart as A2ATextPart
from google.adk.a2a.logs.log_utils import build_a2a_request_log
from google.adk.a2a.logs.log_utils import build_a2a_response_log
from google.adk.a2a.logs.log_utils import build_message_part_log
except ImportError as e:
if sys.version_info < (3, 10):
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
class TestBuildMessagePartLog:
@@ -44,7 +57,6 @@ class TestBuildMessagePartLog:
def test_text_part_short_text(self):
"""Test TextPart with short text."""
from google.adk.a2a.logs.log_utils import build_message_part_log
# Create real A2A objects
text_part = A2ATextPart(text="Hello, world!")
@@ -56,7 +68,6 @@ class TestBuildMessagePartLog:
def test_text_part_long_text(self):
"""Test TextPart with long text that gets truncated."""
from google.adk.a2a.logs.log_utils import build_message_part_log
long_text = "x" * 150 # Long text that should be truncated
text_part = A2ATextPart(text=long_text)
@@ -69,7 +80,6 @@ class TestBuildMessagePartLog:
def test_data_part_simple_data(self):
"""Test DataPart with simple data."""
from google.adk.a2a.logs.log_utils import build_message_part_log
data_part = A2ADataPart(data={"key1": "value1", "key2": 42})
part = A2APart(root=data_part)
@@ -82,7 +92,6 @@ class TestBuildMessagePartLog:
def test_data_part_large_values(self):
"""Test DataPart with large values that get summarized."""
from google.adk.a2a.logs.log_utils import build_message_part_log
large_dict = {f"key{i}": f"value{i}" for i in range(50)}
large_list = list(range(100))
@@ -109,7 +118,6 @@ class TestBuildMessagePartLog:
def test_other_part_type(self):
"""Test handling of other part types (not Text or Data)."""
from google.adk.a2a.logs.log_utils import build_message_part_log
# Create a mock part that will fall through to the else case
mock_root = Mock()
@@ -132,7 +140,6 @@ class TestBuildA2ARequestLog:
def test_request_with_parts_and_config(self):
"""Test request logging with message parts and configuration."""
from google.adk.a2a.logs.log_utils import build_a2a_request_log
# Create mock request with all components
req = SendMessageRequest(
@@ -184,7 +191,6 @@ class TestBuildA2ARequestLog:
def test_request_without_parts(self):
"""Test request logging without message parts."""
from google.adk.a2a.logs.log_utils import build_a2a_request_log
req = Mock()
req.id = "req-123"
@@ -210,7 +216,6 @@ class TestBuildA2ARequestLog:
def test_request_with_empty_parts_list(self):
"""Test request logging with empty parts list."""
from google.adk.a2a.logs.log_utils import build_a2a_request_log
req = Mock()
req.id = "req-123"
@@ -237,7 +242,6 @@ class TestBuildA2AResponseLog:
def test_error_response(self):
"""Test error response logging."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
resp = Mock()
resp.root.error.code = 500
@@ -257,7 +261,6 @@ class TestBuildA2AResponseLog:
def test_error_response_no_data(self):
"""Test error response logging without error data."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
resp = Mock()
resp.root.error.code = 404
@@ -276,7 +279,6 @@ class TestBuildA2AResponseLog:
def test_success_response_with_task(self):
"""Test success response logging with Task result."""
# Use module-level imported types consistently
from google.adk.a2a.logs.log_utils import build_a2a_response_log
task_status = TaskStatus(state=TaskState.working)
task = A2ATask(id="task-123", context_id="ctx-456", status=task_status)
@@ -305,7 +307,6 @@ class TestBuildA2AResponseLog:
def test_success_response_with_task_and_status_message(self):
"""Test success response with Task that has status message."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
# Create status message using module-level imported types
status_message = A2AMessage(
@@ -348,7 +349,6 @@ class TestBuildA2AResponseLog:
def test_success_response_with_message(self):
"""Test success response logging with Message result."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
# Use module-level imported types consistently
message = A2AMessage(
@@ -384,7 +384,6 @@ class TestBuildA2AResponseLog:
def test_success_response_with_message_no_parts(self):
"""Test success response with Message that has no parts."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
# Use mock for this case since we want to test empty parts handling
message = Mock()
@@ -411,7 +410,6 @@ class TestBuildA2AResponseLog:
def test_success_response_with_other_result_type(self):
"""Test success response with result type that's not Task or Message."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
other_result = Mock()
other_result.__class__.__name__ = "OtherResult"
@@ -434,7 +432,6 @@ class TestBuildA2AResponseLog:
def test_success_response_without_model_dump_json(self):
"""Test success response with result that doesn't have model_dump_json."""
from google.adk.a2a.logs.log_utils import build_a2a_response_log
other_result = Mock()
other_result.__class__.__name__ = "SimpleResult"
@@ -456,7 +453,6 @@ class TestBuildA2AResponseLog:
def test_build_message_part_log_with_metadata(self):
"""Test build_message_part_log with metadata in the part."""
from google.adk.a2a.logs.log_utils import build_message_part_log
mock_root = Mock()
mock_root.__class__.__name__ = "MockPartWithMetadata"
@@ -475,7 +471,6 @@ class TestBuildA2AResponseLog:
def test_build_a2a_request_log_with_message_metadata(self):
"""Test request logging with message metadata."""
from google.adk.a2a.logs.log_utils import build_a2a_request_log
req = Mock()
req.id = "req-with-metadata"
@@ -54,39 +54,10 @@ try:
from google.adk.tools.example_tool import ExampleTool
except ImportError as e:
if sys.version_info < (3, 10):
# Create dummy classes to prevent NameError during test collection
# Tests will be skipped anyway due to pytestmark
class DummyTypes:
pass
AgentCapabilities = DummyTypes()
AgentCard = DummyTypes()
AgentProvider = DummyTypes()
AgentSkill = DummyTypes()
SecurityScheme = DummyTypes()
AgentCardBuilder = DummyTypes()
BaseAgent = DummyTypes()
LlmAgent = DummyTypes()
LoopAgent = DummyTypes()
ParallelAgent = DummyTypes()
SequentialAgent = DummyTypes()
ExampleTool = DummyTypes()
# Dummy functions
_build_agent_description = lambda x: ""
_build_llm_agent_description_with_instructions = lambda x: ""
_build_orchestration_skill = lambda x, y: None
_build_parallel_description = lambda x: ""
_build_sequential_description = lambda x: ""
_build_loop_description = lambda x: ""
_convert_example_tool_examples = lambda x: []
_extract_examples_from_instruction = lambda x: None
_get_agent_skill_name = lambda x: ""
_get_agent_type = lambda x: ""
_get_default_description = lambda x: ""
_get_input_modes = lambda x: None
_get_output_modes = lambda x: None
_get_workflow_description = lambda x: None
_replace_pronouns = lambda x: ""
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
+8 -132
View File
@@ -42,25 +42,10 @@ try:
from starlette.applications import Starlette
except ImportError as e:
if sys.version_info < (3, 10):
# Create dummy classes to prevent NameError during test collection
# Tests will be skipped anyway due to pytestmark
class DummyTypes:
pass
A2AStarletteApplication = DummyTypes()
DefaultRequestHandler = DummyTypes()
InMemoryTaskStore = DummyTypes()
AgentCard = DummyTypes()
Starlette = DummyTypes()
BaseAgent = DummyTypes()
InMemoryArtifactService = DummyTypes()
InMemoryCredentialService = DummyTypes()
InMemoryMemoryService = DummyTypes()
Runner = DummyTypes()
InMemorySessionService = DummyTypes()
A2aAgentExecutor = DummyTypes()
AgentCardBuilder = DummyTypes()
to_a2a = lambda x, **kwargs: None
# Imports are not needed since tests will be skipped due to pytestmark.
# The imported names are only used within test methods, not at module level,
# so no NameError occurs during module compilation.
pass
else:
raise e
@@ -123,7 +108,7 @@ class TestToA2A:
@patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
@patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
@patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
def test_to_a2a_custom_host_port_protocol(
def test_to_a2a_custom_host_port(
self,
mock_starlette_class,
mock_card_builder_class,
@@ -131,7 +116,7 @@ class TestToA2A:
mock_request_handler_class,
mock_agent_executor_class,
):
"""Test to_a2a with custom host, port, and protocol."""
"""Test to_a2a with custom host and port."""
# Arrange
mock_app = Mock(spec=Starlette)
mock_starlette_class.return_value = mock_app
@@ -145,14 +130,12 @@ class TestToA2A:
mock_card_builder_class.return_value = mock_card_builder
# Act
result = to_a2a(
self.mock_agent, host="example.com", port=9000, protocol="https"
)
result = to_a2a(self.mock_agent, host="example.com", port=9000)
# Assert
assert result == mock_app
mock_card_builder_class.assert_called_once_with(
agent=self.mock_agent, rpc_url="https://example.com:9000/"
agent=self.mock_agent, rpc_url="http://example.com:9000/"
)
@patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
@@ -706,110 +689,3 @@ class TestToA2A:
mock_card_builder_class.assert_called_once_with(
agent=self.mock_agent, rpc_url="http://192.168.1.1:8000/"
)
@patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
@patch("google.adk.a2a.utils.agent_to_a2a.DefaultRequestHandler")
@patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
@patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
@patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
def test_to_a2a_with_https_protocol(
self,
mock_starlette_class,
mock_card_builder_class,
mock_task_store_class,
mock_request_handler_class,
mock_agent_executor_class,
):
"""Test to_a2a with HTTPS protocol."""
# Arrange
mock_app = Mock(spec=Starlette)
mock_starlette_class.return_value = mock_app
mock_task_store = Mock(spec=InMemoryTaskStore)
mock_task_store_class.return_value = mock_task_store
mock_agent_executor = Mock(spec=A2aAgentExecutor)
mock_agent_executor_class.return_value = mock_agent_executor
mock_request_handler = Mock(spec=DefaultRequestHandler)
mock_request_handler_class.return_value = mock_request_handler
mock_card_builder = Mock(spec=AgentCardBuilder)
mock_card_builder_class.return_value = mock_card_builder
# Act
result = to_a2a(self.mock_agent, protocol="https")
# Assert
assert result == mock_app
mock_card_builder_class.assert_called_once_with(
agent=self.mock_agent, rpc_url="https://localhost:8000/"
)
@patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
@patch("google.adk.a2a.utils.agent_to_a2a.DefaultRequestHandler")
@patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
@patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
@patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
def test_to_a2a_with_custom_protocol(
self,
mock_starlette_class,
mock_card_builder_class,
mock_task_store_class,
mock_request_handler_class,
mock_agent_executor_class,
):
"""Test to_a2a with custom protocol."""
# Arrange
mock_app = Mock(spec=Starlette)
mock_starlette_class.return_value = mock_app
mock_task_store = Mock(spec=InMemoryTaskStore)
mock_task_store_class.return_value = mock_task_store
mock_agent_executor = Mock(spec=A2aAgentExecutor)
mock_agent_executor_class.return_value = mock_agent_executor
mock_request_handler = Mock(spec=DefaultRequestHandler)
mock_request_handler_class.return_value = mock_request_handler
mock_card_builder = Mock(spec=AgentCardBuilder)
mock_card_builder_class.return_value = mock_card_builder
# Act
result = to_a2a(self.mock_agent, protocol="ws")
# Assert
assert result == mock_app
mock_card_builder_class.assert_called_once_with(
agent=self.mock_agent, rpc_url="ws://localhost:8000/"
)
@patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
@patch("google.adk.a2a.utils.agent_to_a2a.DefaultRequestHandler")
@patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
@patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
@patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
def test_to_a2a_with_all_custom_parameters(
self,
mock_starlette_class,
mock_card_builder_class,
mock_task_store_class,
mock_request_handler_class,
mock_agent_executor_class,
):
"""Test to_a2a with all custom parameters."""
# Arrange
mock_app = Mock(spec=Starlette)
mock_starlette_class.return_value = mock_app
mock_task_store = Mock(spec=InMemoryTaskStore)
mock_task_store_class.return_value = mock_task_store
mock_agent_executor = Mock(spec=A2aAgentExecutor)
mock_agent_executor_class.return_value = mock_agent_executor
mock_request_handler = Mock(spec=DefaultRequestHandler)
mock_request_handler_class.return_value = mock_request_handler
mock_card_builder = Mock(spec=AgentCardBuilder)
mock_card_builder_class.return_value = mock_card_builder
# Act
result = to_a2a(
self.mock_agent, host="api.example.com", port=443, protocol="https"
)
# Assert
assert result == mock_app
mock_card_builder_class.assert_called_once_with(
agent=self.mock_agent, rpc_url="https://api.example.com:443/"
)