protobuf 6.33.6: FieldMask manages camelCase JSON transformations, descriptor validation, set operations, and selective message merging with configurable sub-message and repeated field replacement

# Protobuf FieldMask Operations and Behavior

## Prior Solutions and Investigation

A search of the network cache identified existing samples for Protobuf:
- `sha256:8e083a37f459935c492bf187243b0535ebe71e4fd74b8a0c3d44f0bbf50cd63c`: Proves `MessageToDict` default value serialization across presence and non-presence scalar fields.
- `sha256:db90b829f35bd858faf7529ca149fcf58bfe9c65c18c586e7237ddb5084c0b1a`: Proves dynamic proto3 message class generation via `descriptor_pb2` and 64-bit JSON/binary round-trips.
- `sha256:45bb69bf251a6f037374e5373c90a8b7ada4e709ed3c0558a13a7f718836d1ed`: Proves message merge, clone, equality, and reset semantics across presence types and maps.

This sample proves a distinct set of behaviors in Protobuf's `FieldMask` utility (`google.protobuf.field_mask_pb2.FieldMask`), covering camelCase/snake_case JSON transformations, descriptor validation, set operations, and mask-directed selective message merging.

## Exact Release Pinned

- `protobuf==6.33.6` (Python runtime)

## Observed Behavior

1. **JSON Parsing and Casing Rules (`FromJsonString` / `ToJsonString`)**:
   - `FromJsonString` parses comma-separated `lowerCamelCase` path strings (the official Protobuf JSON representation for FieldMasks) and converts them into `snake_case` path names stored in `mask.paths`.
   - `FromJsonString` enforces strict adherence to camelCase format: attempting to parse a string containing an underscore (`_`) raises a `ValueError` (`Path name ... must not contain "_"`).
   - `ToJsonString` converts internal `snake_case` path names back to a comma-separated `lowerCamelCase` string. An empty mask returns an empty string `""`.
   - `google.protobuf.json_format.MessageToDict` serializes `FieldMask` messages directly into a camelCase string rather than a dictionary.

2. **Descriptor Path Validation (`IsValidForDescriptor`)**:
   - `IsValidForDescriptor` checks whether all paths declared in `mask.paths` correspond to existing fields on the message's `Descriptor`.
   - Returns `True` when all paths (including dot-delimited nested paths such as `profile.display_name`) exist.
   - Returns `False` when encountering non-existent field names, invalid subfields, or attempts to traverse below a scalar field (e.g., `account_id.sub`).

3. **Canonical Normalization (`CanonicalFormFromMask`)**:
   - `CanonicalFormFromMask` sorts paths alphabetically and removes redundant descendant paths that are already covered by an ancestor prefix (e.g. if `profile` is in the mask, `profile.display_name` and `profile.bio` are redundant and pruned).

4. **Set Operations (`Union` / `Intersect`)**:
   - `Union` computes the canonical union of paths from two masks, pruning redundant sub-paths.
   - `Intersect` computes the intersection of paths from two masks. When one mask specifies a parent message (e.g., `profile`) and the other specifies a subfield (e.g., `profile.display_name`), the intersection correctly yields the narrower path (`profile.display_name`).

5. **Selective Message Merging (`MergeMessage`)**:
   - `MergeMessage(source, destination)` copies only the fields specified by the mask from the source message to the destination message. Unmasked fields on the destination message remain untouched.
   - **Sub-messages (`replace_message_field`)**: By default (`replace_message_field=False`), nested sub-messages are recursively merged, preserving unmasked sibling fields in the destination sub-message. When `replace_message_field=True`, the entire destination sub-message is replaced, clearing unpopulated fields.
   - **Repeated fields (`replace_repeated_field`)**: By default (`replace_repeated_field=False`), elements from the source repeated field are appended to the destination repeated field. When `replace_repeated_field=True`, the destination list is cleared before copying source elements.

6. **All Fields Extraction (`AllFieldsFromDescriptor`)**:
   - `AllFieldsFromDescriptor` inspects the message descriptor and populates `mask.paths` with all top-level field names defined on the message.

{"case":{"caseId":"case:sha256:ee62d64e25506733e42ca497b51e1bfe406f5f58e7758ad20977599facf90dff","contract":["FromJsonString parses comma-separated lowerCamelCase field paths into snake_case paths and strictly rejects paths containing underscore characters with ValueError","ToJsonString converts stored snake_case paths back into a comma-separated lowerCamelCase string, returning an empty string when the mask contains no paths","IsValidForDescriptor verifies that all paths in the mask resolve to existing message fields and returns False for nonexistent fields or traversal into scalar fields","CanonicalFormFromMask sorts paths alphabetically and removes redundant descendant paths covered by an ancestor prefix","Union and Intersect perform canonical set operations across masks while maintaining prefix pruning and alphabetical ordering","MergeMessage copies only masked fields from source to destination, recursively merging sub-messages and appending repeated elements by default","MergeMessage replaces sub-messages when replace_message_field is True and clears repeated lists before copying when replace_repeated_field is True","AllFieldsFromDescriptor populates the mask with all top-level field names declared on the message descriptor"],"goal":"FieldMask manages camelCase JSON transformations, descriptor validation, set operations, and selective message merging with configurable sub-message and repeated field replacement","kind":"HOW","packages":["pkg:pypi/protobuf@6.33.6"],"schemaVersion":1,"symbols":["google.protobuf.field_mask_pb2.FieldMask","google.protobuf.field_mask_pb2.FieldMask.FromJsonString","google.protobuf.field_mask_pb2.FieldMask.ToJsonString","google.protobuf.field_mask_pb2.FieldMask.IsValidForDescriptor","google.protobuf.field_mask_pb2.FieldMask.CanonicalFormFromMask","google.protobuf.field_mask_pb2.FieldMask.Union","google.protobuf.field_mask_pb2.FieldMask.Intersect","google.protobuf.field_mask_pb2.FieldMask.MergeMessage","google.protobuf.field_mask_pb2.FieldMask.AllFieldsFromDescriptor"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/protobuf@6.33.6"],"schemaVersion":1,"symbols":["google.protobuf.field_mask_pb2.FieldMask","google.protobuf.field_mask_pb2.FieldMask.FromJsonString","google.protobuf.field_mask_pb2.FieldMask.ToJsonString","google.protobuf.field_mask_pb2.FieldMask.IsValidForDescriptor","google.protobuf.field_mask_pb2.FieldMask.CanonicalFormFromMask","google.protobuf.field_mask_pb2.FieldMask.Union","google.protobuf.field_mask_pb2.FieldMask.Intersect","google.protobuf.field_mask_pb2.FieldMask.MergeMessage","google.protobuf.field_mask_pb2.FieldMask.AllFieldsFromDescriptor"],"verifierAdapter":"python@1"}

protobuf==6.33.6

protobuf==6.33.6

from google.protobuf import descriptor_pb2, descriptor_pool, message_factory, field_mask_pb2, json_format


def _build_test_message_classes():
    pool = descriptor_pool.DescriptorPool()
    file_proto = descriptor_pb2.FileDescriptorProto(
        name="user_account.proto",
        package="example",
        syntax="proto3",
        message_type=[
            descriptor_pb2.DescriptorProto(
                name="Profile",
                field=[
                    descriptor_pb2.FieldDescriptorProto(
                        name="display_name",
                        number=1,
                        type=descriptor_pb2.FieldDescriptorProto.TYPE_STRING,
                        label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                    ),
                    descriptor_pb2.FieldDescriptorProto(
                        name="bio",
                        number=2,
                        type=descriptor_pb2.FieldDescriptorProto.TYPE_STRING,
                        label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                    ),
                ],
            ),
            descriptor_pb2.DescriptorProto(
                name="Account",
                field=[
                    descriptor_pb2.FieldDescriptorProto(
                        name="account_id",
                        number=1,
                        type=descriptor_pb2.FieldDescriptorProto.TYPE_INT64,
                        label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                    ),
                    descriptor_pb2.FieldDescriptorProto(
                        name="profile",
                        number=2,
                        type=descriptor_pb2.FieldDescriptorProto.TYPE_MESSAGE,
                        type_name=".example.Profile",
                        label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                    ),
                    descriptor_pb2.FieldDescriptorProto(
                        name="roles",
                        number=3,
                        type=descriptor_pb2.FieldDescriptorProto.TYPE_STRING,
                        label=descriptor_pb2.FieldDescriptorProto.LABEL_REPEATED,
                    ),
                    descriptor_pb2.FieldDescriptorProto(
                        name="is_active",
                        number=4,
                        type=descriptor_pb2.FieldDescriptorProto.TYPE_BOOL,
                        label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                    ),
                ],
            ),
        ],
    )
    pool.Add(file_proto)
    account_cls = message_factory.GetMessageClass(
        pool.FindMessageTypeByName("example.Account")
    )
    profile_cls = message_factory.GetMessageClass(
        pool.FindMessageTypeByName("example.Profile")
    )
    return account_cls, profile_cls


Account, Profile = _build_test_message_classes()


def test_from_json_string_and_underscore_rejection():
    # FromJsonString converts lowerCamelCase paths to snake_case paths
    mask = field_mask_pb2.FieldMask()
    mask.FromJsonString("accountId,profile.displayName,isActive")
    assert list(mask.paths) == [
        "account_id",
        "profile.display_name",
        "is_active",
    ]

    # Reject paths with underscore in JSON string format
    invalid_mask = field_mask_pb2.FieldMask()
    try:
        invalid_mask.FromJsonString("account_id")
        raise AssertionError("Expected ValueError when parsing path with underscore")
    except ValueError as err:
        assert 'Path name account_id must not contain "_"' in str(err)


def test_to_json_string_and_empty_mask():
    # ToJsonString converts internal snake_case paths back to lowerCamelCase
    mask = field_mask_pb2.FieldMask(
        paths=["account_id", "profile.display_name", "is_active"]
    )
    assert mask.ToJsonString() == "accountId,profile.displayName,isActive"

    # An empty mask returns an empty string
    empty_mask = field_mask_pb2.FieldMask()
    assert empty_mask.ToJsonString() == ""

    # json_format.MessageToDict on a FieldMask serializes to string directly
    assert json_format.MessageToDict(mask) == "accountId,profile.displayName,isActive"


def test_is_valid_for_descriptor():
    valid_mask = field_mask_pb2.FieldMask(
        paths=["account_id", "profile.display_name", "roles"]
    )
    assert valid_mask.IsValidForDescriptor(Account.DESCRIPTOR) is True

    # Invalid: non-existent top-level field
    invalid_field_mask = field_mask_pb2.FieldMask(paths=["nonexistent_field"])
    assert invalid_field_mask.IsValidForDescriptor(Account.DESCRIPTOR) is False

    # Invalid: non-existent subfield on nested message
    invalid_subfield_mask = field_mask_pb2.FieldMask(
        paths=["profile.unknown_field"]
    )
    assert invalid_subfield_mask.IsValidForDescriptor(Account.DESCRIPTOR) is False

    # Invalid: traversal into a scalar field
    invalid_scalar_child = field_mask_pb2.FieldMask(paths=["account_id.sub"])
    assert invalid_scalar_child.IsValidForDescriptor(Account.DESCRIPTOR) is False


def test_canonical_form_from_mask():
    # CanonicalFormFromMask eliminates redundant descendant paths and sorts alphabetically
    redundant_mask = field_mask_pb2.FieldMask(
        paths=[
            "profile.display_name",
            "account_id",
            "profile",
            "profile.bio",
            "roles",
        ]
    )
    canonical_mask = field_mask_pb2.FieldMask()
    canonical_mask.CanonicalFormFromMask(redundant_mask)
    assert list(canonical_mask.paths) == ["account_id", "profile", "roles"]


def test_union_and_intersect():
    mask1 = field_mask_pb2.FieldMask(paths=["account_id", "profile.display_name"])
    mask2 = field_mask_pb2.FieldMask(paths=["profile", "roles"])

    union_mask = field_mask_pb2.FieldMask()
    union_mask.Union(mask1, mask2)
    # Since mask2 has 'profile', the descendant 'profile.display_name' is pruned in the canonical union
    assert list(union_mask.paths) == ["account_id", "profile", "roles"]

    intersect_mask = field_mask_pb2.FieldMask()
    intersect_mask.Intersect(mask1, mask2)
    # The intersection of 'profile.display_name' and 'profile' is 'profile.display_name'
    assert list(intersect_mask.paths) == ["profile.display_name"]


def test_merge_message_selective_updates_and_default_behavior():
    src = Account(
        account_id=999,
        profile=Profile(display_name="NewName", bio="NewBio"),
        roles=["admin", "editor"],
        is_active=True,
    )
    dst = Account(
        account_id=100,
        profile=Profile(display_name="OldName", bio="OldBio"),
        roles=["viewer"],
        is_active=False,
    )

    mask = field_mask_pb2.FieldMask(paths=["profile.display_name", "roles"])
    mask.MergeMessage(src, dst)

    # Unmasked fields remain untouched
    assert dst.account_id == 100
    assert dst.is_active is False
    # Nested message field is selectively updated while sibling subfields are preserved
    assert dst.profile.display_name == "NewName"
    assert dst.profile.bio == "OldBio"
    # By default, repeated fields are appended rather than replaced
    assert list(dst.roles) == ["viewer", "admin", "editor"]


def test_merge_message_replace_message_and_repeated_flags():
    src = Account(
        profile=Profile(display_name="OnlyNewName"),
        roles=["admin"],
    )

    # Test replace_message_field=True: replaces entire sub-message, clearing unspecified subfields
    dst_msg_replace = Account(
        profile=Profile(display_name="OldName", bio="PreservedBio")
    )
    mask_msg = field_mask_pb2.FieldMask(paths=["profile"])
    mask_msg.MergeMessage(src, dst_msg_replace, replace_message_field=True)
    assert dst_msg_replace.profile.display_name == "OnlyNewName"
    assert dst_msg_replace.profile.bio == ""

    # Test replace_repeated_field=True: clears destination list before copying source items
    dst_rep_replace = Account(roles=["old_role_1", "old_role_2"])
    mask_rep = field_mask_pb2.FieldMask(paths=["roles"])
    mask_rep.MergeMessage(src, dst_rep_replace, replace_repeated_field=True)
    assert list(dst_rep_replace.roles) == ["admin"]


def test_all_fields_from_descriptor():
    mask = field_mask_pb2.FieldMask()
    mask.AllFieldsFromDescriptor(Account.DESCRIPTOR)
    assert list(mask.paths) == ["account_id", "profile", "roles", "is_active"]


if __name__ == "__main__":
    test_from_json_string_and_underscore_rejection()
    test_to_json_string_and_empty_mask()
    test_is_valid_for_descriptor()
    test_canonical_form_from_mask()
    test_union_and_intersect()
    test_merge_message_selective_updates_and_default_behavior()
    test_merge_message_replace_message_and_repeated_flags()
    test_all_fields_from_descriptor()
    print("Contract assertions passed.")