diff --git a/src/a2a/utils/proto_utils.py b/src/a2a/utils/proto_utils.py index e619cd72..79f248b3 100644 --- a/src/a2a/utils/proto_utils.py +++ b/src/a2a/utils/proto_utils.py @@ -366,6 +366,37 @@ def task_push_notification_config( ), ) + @classmethod + def task_push_notification_config_params( + cls, + request: ( + a2a_pb2.GetTaskPushNotificationConfigRequest + | a2a_pb2.DeleteTaskPushNotificationConfigRequest + ), + ) -> tuple[str, str]: + """Parses the task ID and push notification config ID from the request. + + Args: + request: The request to parse. + + Returns: + A tuple containing the task ID and push notification config ID. + + Raises: + ServerError: If the request name is invalid. + """ + m = _TASK_PUSH_CONFIG_NAME_MATCH.match(request.name) + if not m: + raise ServerError( + error=types.InvalidParamsError( + message=f'No task or config id for {request.name}' + ) + ) + return ( + m.group(1), + m.group(2), + ) + @classmethod def agent_card( cls, diff --git a/tests/utils/test_proto_utils.py b/tests/utils/test_proto_utils.py index da54f833..9e0fdbec 100644 --- a/tests/utils/test_proto_utils.py +++ b/tests/utils/test_proto_utils.py @@ -244,6 +244,62 @@ def test_task_push_config_from_proto_invalid_parent(self): proto_utils.FromProto.task_push_notification_config(request) assert isinstance(exc_info.value.error, types.InvalidParamsError) + @pytest.mark.parametrize( + 'request_type, name, expected_task_id, expected_config_id', + [ + ( + a2a_pb2.GetTaskPushNotificationConfigRequest, + 'tasks/task-123/pushNotificationConfigs/config-456', + 'task-123', + 'config-456', + ), + ( + a2a_pb2.DeleteTaskPushNotificationConfigRequest, + 'tasks/task-abc/pushNotificationConfigs/config-def', + 'task-abc', + 'config-def', + ), + ], + ) + def test_task_push_notification_config_params_valid( + self, request_type, name, expected_task_id, expected_config_id + ): + """Test valid name in task_push_notification_config_params.""" + request = request_type(name=name) + task_id, config_id = ( + proto_utils.ToProto.task_push_notification_config_params(request) + ) + assert task_id == expected_task_id + assert config_id == expected_config_id + + @pytest.mark.parametrize( + 'request_type', + [ + a2a_pb2.GetTaskPushNotificationConfigRequest, + a2a_pb2.DeleteTaskPushNotificationConfigRequest, + ], + ) + @pytest.mark.parametrize( + 'name', + [ + 'invalid-name-format', + 'tasks/task-123', + 'tasks/task-123/pushNotificationConfigs', + 'tasks//pushNotificationConfigs/config-456', + 'tasks/task-123/pushNotificationConfigs/', + 'foo/task-123/bar/config-456', + ], + ) + def test_task_push_notification_config_params_invalid( + self, request_type, name + ): + """Test invalid names in task_push_notification_config_params.""" + request = request_type(name=name) + with pytest.raises(ServerError) as exc_info: + proto_utils.ToProto.task_push_notification_config_params(request) + assert isinstance(exc_info.value.error, types.InvalidParamsError) + assert 'No task or config id for' in exc_info.value.error.message + def test_none_handling(self): """Test that None inputs are handled gracefully.""" assert proto_utils.ToProto.message(None) is None