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
23 changes: 11 additions & 12 deletions src/a2a/utils/proto_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,10 +174,7 @@ def parse_params(params: QueryParams, message: ProtobufMessage) -> None:
field = fields[k]
v_list = params.getlist(k)

# TODO(https://github.com/a2aproject/a2a-python/issues/1011): Replace
# deprecated `field.label` with `field.is_repeated` once the minimum
# protobuf version requirement is bumped.
if field.label == FieldDescriptor.LABEL_REPEATED:
if _is_field_repeated(field):
accumulated: list[Any] = []
for v in v_list:
if not v:
Expand Down Expand Up @@ -206,15 +203,20 @@ class ValidationDetail(TypedDict):
message: str


def _is_field_repeated(field: FieldDescriptor) -> bool:
"""Return whether a protobuf field is repeated across protobuf versions."""
try:
return field.is_repeated
except AttributeError:
return field.label == FieldDescriptor.LABEL_REPEATED


def _check_required_field_violation(
msg: ProtobufMessage, field: FieldDescriptor
) -> ValidationDetail | None:
"""Check if a required field is missing or invalid."""
val = getattr(msg, field.name)
# TODO(https://github.com/a2aproject/a2a-python/issues/1011): Replace
# deprecated `field.label` with `field.is_repeated` once the minimum
# protobuf version requirement is bumped.
if field.label == FieldDescriptor.LABEL_REPEATED:
if _is_field_repeated(field):
if not val:
return ValidationDetail(
field=field.name,
Expand Down Expand Up @@ -255,10 +257,7 @@ def _recurse_validation(
return errors

val = getattr(msg, field.name)
# TODO(https://github.com/a2aproject/a2a-python/issues/1011): Replace
# deprecated `field.label` with `field.is_repeated` once the minimum
# protobuf version requirement is bumped.
if field.label != FieldDescriptor.LABEL_REPEATED:
if not _is_field_repeated(field):
if msg.HasField(field.name):
sub_errs = _validate_proto_required_fields_internal(val)
_append_nested_errors(errors, field.name, sub_errs)
Expand Down
25 changes: 25 additions & 0 deletions tests/utils/test_proto_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,6 +241,31 @@ def _message_to_rest_params(self, message: ProtobufMessage) -> QueryParams:
).url.params


class TestIsFieldRepeated:
"""Tests for _is_field_repeated helper."""

def test_scalar_field_not_repeated(self):
field = Message.DESCRIPTOR.fields_by_name['message_id']
assert proto_utils._is_field_repeated(field) is False

def test_repeated_field(self):
field = AgentSkill.DESCRIPTOR.fields_by_name['tags']
assert proto_utils._is_field_repeated(field) is True

def test_fallback_when_is_repeated_missing(self, monkeypatch):
field = AgentSkill.DESCRIPTOR.fields_by_name['tags']

def raise_attribute_error(_self):
raise AttributeError('is_repeated')

monkeypatch.setattr(
type(field),
'is_repeated',
property(raise_attribute_error),
)
assert proto_utils._is_field_repeated(field) is True
Comment on lines +255 to +266

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Modifying the attributes or properties of FieldDescriptor (or its type) using monkeypatch will fail with a TypeError in environments where the fast C++ implementation of Protobuf is used (which is the default in many environments, as FieldDescriptor is a C-extension type and its attributes/type are read-only).

Instead of monkeypatching the built-in FieldDescriptor class, you can use a simple mock object or a custom class that lacks the is_repeated attribute to safely test the fallback path.

Suggested change
def test_fallback_when_is_repeated_missing(self, monkeypatch):
field = AgentSkill.DESCRIPTOR.fields_by_name['tags']
def raise_attribute_error(_self):
raise AttributeError('is_repeated')
monkeypatch.setattr(
type(field),
'is_repeated',
property(raise_attribute_error),
)
assert proto_utils._is_field_repeated(field) is True
def test_fallback_when_is_repeated_missing(self):
class MockField:
label = proto_utils.FieldDescriptor.LABEL_REPEATED
field = MockField()
assert proto_utils._is_field_repeated(field) is True
References
  1. When checking for repeated fields in Protobuf, access descriptor constants via the FieldDescriptor class (e.g., FieldDescriptor.LABEL_REPEATED) rather than directly from the field instance (e.g., field.is_repeated).



class TestValidateProtoRequiredFields:
"""Tests for validate_proto_required_fields function."""

Expand Down
Loading