|
19 | 19 |
|
20 | 20 | import celpy |
21 | 21 | from celpy import celtypes |
22 | | -from google.protobuf import any_pb2, descriptor, duration_pb2, message, message_factory, timestamp_pb2 |
| 22 | +from google.protobuf import any_pb2, descriptor, duration_pb2, message, message_factory, timestamp_pb2, unknown_fields # type: ignore[attr-defined] |
23 | 23 |
|
24 | 24 | from buf.validate import validate_pb2 |
25 | 25 | from protovalidate.internal.cel_field_presence import InterpretedRunner, in_has |
@@ -565,6 +565,9 @@ def __init__( |
565 | 565 | self._required = field_level.required |
566 | 566 | if type_case is not None: |
567 | 567 | rules: message.Message = getattr(field_level, type_case) |
| 568 | + if len(unknown_fields.UnknownFieldSet(rules)) > 0: |
| 569 | + msg = f"unknown rules in {rules.DESCRIPTOR.full_name}" |
| 570 | + raise CompilationError(msg) |
568 | 571 | # For each set field in the message, look for the private rule |
569 | 572 | # extension. |
570 | 573 | for list_field, _ in rules.ListFields(): |
@@ -1031,6 +1034,9 @@ def _new_scalar_field_rule( |
1031 | 1034 | check_field_type(field, 0, "google.protobuf.Any") |
1032 | 1035 | result = AnyRules(self._env, self._funcs, field, field_level) |
1033 | 1036 | return result |
| 1037 | + else: |
| 1038 | + msg = f"unknown rule type {type_case!r}" |
| 1039 | + raise CompilationError(msg) |
1034 | 1040 |
|
1035 | 1041 | def _new_field_rule( |
1036 | 1042 | self, |
|
0 commit comments