From 3c9fbb1821777a80a6355cbc7b7fdbcad9b46957 Mon Sep 17 00:00:00 2001 From: Anuraag Agrawal Date: Tue, 21 Jul 2026 15:58:09 +0900 Subject: [PATCH] Modernize patterns following protobuf-py conventions --- protovalidate/__init__.py | 25 +- .../{internal/backend.py => _backend.py} | 2 + ...eld_presence.py => _cel_field_presence.py} | 14 +- .../celexpr => _celexpr}/__init__.py | 2 + .../{internal/celexpr => _celexpr}/_utils.py | 2 + .../{internal/celexpr => _celexpr}/bridge.py | 30 +- .../celexpr => _celexpr}/extra_func.py | 160 +++++-- .../{internal/celexpr => _celexpr}/rules.py | 392 +++++++++++++----- protovalidate/{internal => }/_core.py | 30 +- .../extra_func.py => _extra_func.py} | 19 +- protovalidate/{internal => }/_funcs.py | 260 +++++++----- .../{internal/legacy.py => _legacy.py} | 16 +- .../{internal/rules.py => _rules.py} | 329 ++++++++++----- .../string_format.py => _string_format.py} | 19 +- protovalidate/{validator.py => _validator.py} | 92 ++-- pyproject.toml | 103 +++-- .../internal => scripts}/__init__.py | 0 scripts/generate_cel.py | 8 +- scripts/generate_protovalidate.py | 11 +- test/conformance/runner.py | 62 ++- test/conformance/test_conformance.py | 2 +- test/conftest.py | 16 +- test/test_benchmark.py | 71 +++- test/test_format.py | 68 +-- test/test_matches.py | 6 +- test/test_validate.py | 124 ++++-- test/versions.py | 2 + 27 files changed, 1262 insertions(+), 603 deletions(-) rename protovalidate/{internal/backend.py => _backend.py} (96%) rename protovalidate/{internal/cel_field_presence.py => _cel_field_presence.py} (82%) rename protovalidate/{internal/celexpr => _celexpr}/__init__.py (95%) rename protovalidate/{internal/celexpr => _celexpr}/_utils.py (96%) rename protovalidate/{internal/celexpr => _celexpr}/bridge.py (73%) rename protovalidate/{internal/celexpr => _celexpr}/extra_func.py (52%) rename protovalidate/{internal/celexpr => _celexpr}/rules.py (79%) rename protovalidate/{internal => }/_core.py (84%) rename protovalidate/{internal/extra_func.py => _extra_func.py} (88%) rename protovalidate/{internal => }/_funcs.py (88%) rename protovalidate/{internal/legacy.py => _legacy.py} (85%) rename protovalidate/{internal/rules.py => _rules.py} (81%) rename protovalidate/{internal/string_format.py => _string_format.py} (94%) rename protovalidate/{validator.py => _validator.py} (67%) rename {protovalidate/internal => scripts}/__init__.py (100%) diff --git a/protovalidate/__init__.py b/protovalidate/__init__.py index a530d3c..f86cccd 100644 --- a/protovalidate/__init__.py +++ b/protovalidate/__init__.py @@ -12,15 +12,28 @@ # See the License for the specific language governing permissions and # limitations under the License. -from protovalidate import validator +"""The semantic validation library for Protobuf in Python.""" -Validator = validator.Validator -CompilationError = validator.CompilationError -ValidationError = validator.ValidationError -Violations = validator.Violations +from __future__ import annotations + +from protovalidate._validator import ( + CompilationError, + ValidationError, + Validator, + Violation, + Violations, +) _default_validator = Validator() validate = _default_validator.validate collect_violations = _default_validator.collect_violations -__all__ = ["CompilationError", "ValidationError", "Validator", "Violations", "collect_violations", "validate"] +__all__ = [ + "CompilationError", + "ValidationError", + "Validator", + "Violation", + "Violations", + "collect_violations", + "validate", +] diff --git a/protovalidate/internal/backend.py b/protovalidate/_backend.py similarity index 96% rename from protovalidate/internal/backend.py rename to protovalidate/_backend.py index f8cdc28..03aeb18 100644 --- a/protovalidate/internal/backend.py +++ b/protovalidate/_backend.py @@ -14,6 +14,8 @@ """Which CEL backend is available.""" +from __future__ import annotations + def _detect() -> bool: try: diff --git a/protovalidate/internal/cel_field_presence.py b/protovalidate/_cel_field_presence.py similarity index 82% rename from protovalidate/internal/cel_field_presence.py rename to protovalidate/_cel_field_presence.py index 3788ce7..36534a9 100644 --- a/protovalidate/internal/cel_field_presence.py +++ b/protovalidate/_cel_field_presence.py @@ -12,17 +12,22 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import threading +from typing import TYPE_CHECKING import celpy import celpy.celtypes +if TYPE_CHECKING: + from lark import Tree + _has_state = threading.local() def in_has() -> bool: - """ - Returns true if inside of CEL interpreter `has` macro. + """Returns true if inside of CEL interpreter `has` macro. This enables working around an issue in cel-python where it is not possible to implement protobuf semantics around the `has` macro. @@ -35,12 +40,11 @@ def in_has() -> bool: class InterpretedRunner(celpy.InterpretedRunner): def evaluate(self, context: celpy.Context) -> celpy.celtypes.Value: class Evaluator(celpy.Evaluator): - def macro_has_eval(self, exprlist) -> celpy.celtypes.BoolType: + def macro_has_eval(self, exprlist: Tree) -> celpy.celtypes.BoolType: _has_state.in_has = True result = super().macro_has_eval(exprlist) _has_state.in_has = False return result e = Evaluator(ast=self.ast, activation=self.new_activation()) - value = e.evaluate(context) - return value + return e.evaluate(context) diff --git a/protovalidate/internal/celexpr/__init__.py b/protovalidate/_celexpr/__init__.py similarity index 95% rename from protovalidate/internal/celexpr/__init__.py rename to protovalidate/_celexpr/__init__.py index 89392ce..7d4fe78 100644 --- a/protovalidate/internal/celexpr/__init__.py +++ b/protovalidate/_celexpr/__init__.py @@ -14,6 +14,8 @@ """The cel-expr-python (cel-cpp) validation engine.""" +from __future__ import annotations + from .bridge import GoogleBridge from .extra_func import make_extension from .rules import RuleFactory diff --git a/protovalidate/internal/celexpr/_utils.py b/protovalidate/_celexpr/_utils.py similarity index 96% rename from protovalidate/internal/celexpr/_utils.py rename to protovalidate/_celexpr/_utils.py index 89bcb43..a995c52 100644 --- a/protovalidate/internal/celexpr/_utils.py +++ b/protovalidate/_celexpr/_utils.py @@ -11,6 +11,8 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + from google.protobuf import descriptor as google_descriptor # protobuf 7+ removed FieldDescriptor.label / LABEL_REPEATED in favour of is_repeated. diff --git a/protovalidate/internal/celexpr/bridge.py b/protovalidate/_celexpr/bridge.py similarity index 73% rename from protovalidate/internal/celexpr/bridge.py rename to protovalidate/_celexpr/bridge.py index 32e22cd..5128f77 100644 --- a/protovalidate/internal/celexpr/bridge.py +++ b/protovalidate/_celexpr/bridge.py @@ -16,30 +16,32 @@ from __future__ import annotations -import typing +from typing import TYPE_CHECKING -from google.protobuf import descriptor_pb2 as google_descriptor_pb2 -from google.protobuf import descriptor_pool as google_descriptor_pool -from google.protobuf import message as google_message -from google.protobuf import message_factory as google_message_factory -from protobuf import DescFile, DescMessage +from google.protobuf import ( + descriptor_pb2 as google_descriptor_pb2, + descriptor_pool as google_descriptor_pool, + message as google_message, + message_factory as google_message_factory, +) -if typing.TYPE_CHECKING: +if TYPE_CHECKING: import protobuf + from protobuf import DescFile, DescMessage class GoogleBridge: - """Lazily mirrors protobuf-py descriptors into google's pool and bridges - protobuf-py message values to google dynamic messages.""" + """Bridges from protobuf-py messages to google.protobuf for use with cel-expr-python.""" def __init__(self) -> None: - self._pool: google_descriptor_pool.DescriptorPool = google_descriptor_pool.Default() + self._pool: google_descriptor_pool.DescriptorPool = ( + google_descriptor_pool.Default() + ) self._mirrored: set[str] = set() self._classes: dict[str, type[google_message.Message]] = {} def _mirror_file(self, desc_file: DescFile) -> None: - """Registers a protobuf-py DescFile (and its transitive deps) into the - google pool, dependencies first, skipping files already present.""" + """Registers a protobuf-py DescFile and deps into the google pool.""" if desc_file.name in self._mirrored: return self._mirrored.add(desc_file.name) @@ -48,7 +50,9 @@ def _mirror_file(self, desc_file: DescFile) -> None: try: self._pool.FindFileByName(desc_file.name) except KeyError: - proto = google_descriptor_pb2.FileDescriptorProto.FromString(desc_file.proto.to_binary()) + proto = google_descriptor_pb2.FileDescriptorProto.FromString( + desc_file.proto.to_binary() + ) self._pool.Add(proto) def google_class(self, desc: DescMessage) -> type[google_message.Message]: diff --git a/protovalidate/internal/celexpr/extra_func.py b/protovalidate/_celexpr/extra_func.py similarity index 52% rename from protovalidate/internal/celexpr/extra_func.py rename to protovalidate/_celexpr/extra_func.py index fe8e944..e03c719 100644 --- a/protovalidate/internal/celexpr/extra_func.py +++ b/protovalidate/_celexpr/extra_func.py @@ -14,12 +14,16 @@ """cel-expr-python registration for protovalidate's custom CEL functions.""" +from __future__ import annotations + from cel_expr_python import cel -from google.protobuf import descriptor as google_descriptor -from google.protobuf import message as google_message -from google.protobuf import wrappers_pb2 as google_wrappers_pb2 +from google.protobuf import ( + descriptor as google_descriptor, + message as google_message, + wrappers_pb2 as google_wrappers_pb2, +) -from protovalidate.internal import _funcs +from protovalidate import _funcs from ._utils import is_repeated @@ -27,10 +31,10 @@ def cel_get_field(message: object, field_name: object) -> object: if not isinstance(message, google_message.Message): msg = "invalid argument, expected message" - raise ValueError(msg) + raise TypeError(msg) if not isinstance(field_name, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) if field_name not in message.DESCRIPTOR.fields_by_name: msg = f"no such field: {field_name}" raise ValueError(msg) @@ -50,7 +54,7 @@ def cel_get_field(message: object, field_name: object) -> object: def cel_unique(val: object) -> bool: if not isinstance(val, list): msg = "invalid argument, expected list" - raise ValueError(msg) + raise TypeError(msg) # Track seen values keyed by (type, value) so that distinct CEL types that # are equal in Python (notably bool vs int: ``True == 1``) are not treated # as duplicates, and so that bytes are never confused with strings. @@ -97,59 +101,157 @@ def make_extension() -> cel.CelExtension: return cel.CelExtension( "protovalidate", [ - cel.FunctionDecl("getField", [cel.Overload("get_field", _dyn, [_dyn, _s], impl=cel_get_field)]), cel.FunctionDecl( - "isNan", [cel.Overload("double_is_nan", _b, [_d], is_member=True, impl=_funcs.cel_is_nan)] + "getField", + [cel.Overload("get_field", _dyn, [_dyn, _s], impl=cel_get_field)], + ), + cel.FunctionDecl( + "isNan", + [ + cel.Overload( + "double_is_nan", + _b, + [_d], + is_member=True, + impl=_funcs.cel_is_nan, + ) + ], ), cel.FunctionDecl( "isInf", [ - cel.Overload("double_is_inf", _b, [_d], is_member=True, impl=_funcs.cel_is_inf), - cel.Overload("double_int_is_inf", _b, [_d, _i], is_member=True, impl=_funcs.cel_is_inf), + cel.Overload( + "double_is_inf", + _b, + [_d], + is_member=True, + impl=_funcs.cel_is_inf, + ), + cel.Overload( + "double_int_is_inf", + _b, + [_d, _i], + is_member=True, + impl=_funcs.cel_is_inf, + ), ], ), cel.FunctionDecl( "isIp", [ - cel.Overload("string_is_ip", _b, [_s], is_member=True, impl=_funcs.cel_is_ip), - cel.Overload("string_int_is_ip", _b, [_s, _i], is_member=True, impl=_funcs.cel_is_ip), + cel.Overload( + "string_is_ip", _b, [_s], is_member=True, impl=_funcs.cel_is_ip + ), + cel.Overload( + "string_int_is_ip", + _b, + [_s, _i], + is_member=True, + impl=_funcs.cel_is_ip, + ), ], ), cel.FunctionDecl( "isIpPrefix", [ - cel.Overload("string_is_ip_prefix", _b, [_s], is_member=True, impl=_funcs.cel_is_ip_prefix), - cel.Overload("string_int_is_ip_prefix", _b, [_s, _i], is_member=True, impl=_funcs.cel_is_ip_prefix), cel.Overload( - "string_bool_is_ip_prefix", _b, [_s, _b], is_member=True, impl=_funcs.cel_is_ip_prefix + "string_is_ip_prefix", + _b, + [_s], + is_member=True, + impl=_funcs.cel_is_ip_prefix, ), cel.Overload( - "string_int_bool_is_ip_prefix", _b, [_s, _i, _b], is_member=True, impl=_funcs.cel_is_ip_prefix + "string_int_is_ip_prefix", + _b, + [_s, _i], + is_member=True, + impl=_funcs.cel_is_ip_prefix, + ), + cel.Overload( + "string_bool_is_ip_prefix", + _b, + [_s, _b], + is_member=True, + impl=_funcs.cel_is_ip_prefix, + ), + cel.Overload( + "string_int_bool_is_ip_prefix", + _b, + [_s, _i, _b], + is_member=True, + impl=_funcs.cel_is_ip_prefix, ), ], ), cel.FunctionDecl( - "isEmail", [cel.Overload("string_is_email", _b, [_s], is_member=True, impl=_funcs.cel_is_email)] + "isEmail", + [ + cel.Overload( + "string_is_email", + _b, + [_s], + is_member=True, + impl=_funcs.cel_is_email, + ) + ], ), cel.FunctionDecl( - "isUri", [cel.Overload("string_is_uri", _b, [_s], is_member=True, impl=_funcs.cel_is_uri)] + "isUri", + [ + cel.Overload( + "string_is_uri", + _b, + [_s], + is_member=True, + impl=_funcs.cel_is_uri, + ) + ], ), cel.FunctionDecl( - "isUriRef", [cel.Overload("string_is_uri_ref", _b, [_s], is_member=True, impl=_funcs.cel_is_uri_ref)] + "isUriRef", + [ + cel.Overload( + "string_is_uri_ref", + _b, + [_s], + is_member=True, + impl=_funcs.cel_is_uri_ref, + ) + ], ), cel.FunctionDecl( "isHostname", - [cel.Overload("string_is_hostname", _b, [_s], is_member=True, impl=_funcs.cel_is_hostname)], + [ + cel.Overload( + "string_is_hostname", + _b, + [_s], + is_member=True, + impl=_funcs.cel_is_hostname, + ) + ], ), cel.FunctionDecl( "isHostAndPort", [ cel.Overload( - "string_bool_is_host_and_port", _b, [_s, _b], is_member=True, impl=_funcs.cel_is_host_and_port + "string_bool_is_host_and_port", + _b, + [_s, _b], + is_member=True, + impl=_funcs.cel_is_host_and_port, + ) + ], + ), + cel.FunctionDecl( + "unique", + [ + cel.Overload( + "list_unique", _b, [_l], is_member=True, impl=cel_unique ) ], ), - cel.FunctionDecl("unique", [cel.Overload("list_unique", _b, [_l], is_member=True, impl=cel_unique)]), cel.FunctionDecl( "startsWith", [ @@ -166,7 +268,11 @@ def make_extension() -> cel.CelExtension: "endsWith", [ cel.Overload( - "bytes_ends_with", _b, [cel.Type.BYTES, cel.Type.BYTES], is_member=True, impl=_bytes_ends_with + "bytes_ends_with", + _b, + [cel.Type.BYTES, cel.Type.BYTES], + is_member=True, + impl=_bytes_ends_with, ) ], ), @@ -174,7 +280,11 @@ def make_extension() -> cel.CelExtension: "contains", [ cel.Overload( - "bytes_contains", _b, [cel.Type.BYTES, cel.Type.BYTES], is_member=True, impl=_bytes_contains + "bytes_contains", + _b, + [cel.Type.BYTES, cel.Type.BYTES], + is_member=True, + impl=_bytes_contains, ) ], ), diff --git a/protovalidate/internal/celexpr/rules.py b/protovalidate/_celexpr/rules.py similarity index 79% rename from protovalidate/internal/celexpr/rules.py rename to protovalidate/_celexpr/rules.py index b34337d..7c84d55 100644 --- a/protovalidate/internal/celexpr/rules.py +++ b/protovalidate/_celexpr/rules.py @@ -14,31 +14,37 @@ """The rule engine.""" +from __future__ import annotations + import dataclasses import datetime import functools import typing -from collections.abc import Container import protobuf from cel_expr_python import cel from cel_expr_python.ext import ext_strings -from google.protobuf import any_pb2 as google_any_pb2 -from google.protobuf import descriptor as google_descriptor -from google.protobuf import descriptor_pool as google_descriptor_pool -from google.protobuf import message as google_message -from google.protobuf import wrappers_pb2 as google_wrappers_pb2 +from google.protobuf import ( + any_pb2 as google_any_pb2, + descriptor as google_descriptor, + descriptor_pool as google_descriptor_pool, + message as google_message, + wrappers_pb2 as google_wrappers_pb2, +) from protobuf import Oneof from protobuf.wkt import FieldDescriptorProto -from protovalidate._gen.buf.validate import validate_pb - # Backend-agnostic primitives shared with the celpy engine (protovalidate.internal.rules). -from protovalidate.internal._core import CompilationError, RuleContext, Rules, Violation -from protovalidate.internal.celexpr.bridge import GoogleBridge +from protovalidate._core import CompilationError, RuleContext, Rules, Violation +from protovalidate._gen.buf.validate import validate_pb from ._utils import is_repeated +if typing.TYPE_CHECKING: + from collections.abc import Container + + from protovalidate._celexpr.bridge import GoogleBridge + _FIELD_TYPE_NAMES: dict[int, str] = { google_descriptor.FieldDescriptor.TYPE_MESSAGE: "message", google_descriptor.FieldDescriptor.TYPE_GROUP: "group", @@ -65,13 +71,17 @@ def _get_type_name(fd: typing.Any) -> str: return _FIELD_TYPE_NAMES.get(fd, "unknown") -def _proto_message_has_field(msg: google_message.Message, field: google_descriptor.FieldDescriptor) -> typing.Any: +def _proto_message_has_field( + msg: google_message.Message, field: google_descriptor.FieldDescriptor +) -> typing.Any: if field.is_extension: return msg.HasExtension(field) # ty: ignore[invalid-argument-type] return msg.HasField(field.name) -def _proto_message_get_field(msg: google_message.Message, field: google_descriptor.FieldDescriptor) -> typing.Any: +def _proto_message_get_field( + msg: google_message.Message, field: google_descriptor.FieldDescriptor +) -> typing.Any: if field.is_extension: return msg.Extensions[field] # ty: ignore[invalid-argument-type] return getattr(msg, field.name) @@ -87,7 +97,9 @@ def _proto_message_get_field(msg: google_message.Message, field: google_descript ) -def _scalar_field_value_to_cel(val: typing.Any, field: google_descriptor.FieldDescriptor) -> typing.Any: +def _scalar_field_value_to_cel( + val: typing.Any, field: google_descriptor.FieldDescriptor +) -> typing.Any: # The runtime converts Python scalars and messages (including the # well-known wrapper, timestamp, and duration types) to CEL values # natively, so most values pass through unchanged. The exceptions ride in @@ -112,7 +124,9 @@ def _scalar_field_value_to_cel(val: typing.Any, field: google_descriptor.FieldDe return val -def _field_value_to_cel(val: typing.Any, field: google_descriptor.FieldDescriptor) -> typing.Any: +def _field_value_to_cel( + val: typing.Any, field: google_descriptor.FieldDescriptor +) -> typing.Any: if is_repeated(field): if field.message_type is not None and field.message_type.GetOptions().map_entry: return dict(val) @@ -120,7 +134,9 @@ def _field_value_to_cel(val: typing.Any, field: google_descriptor.FieldDescripto return _scalar_field_value_to_cel(val, field) -def _is_empty_field(msg: google_message.Message, field: google_descriptor.FieldDescriptor) -> bool: +def _is_empty_field( + msg: google_message.Message, field: google_descriptor.FieldDescriptor +) -> bool: if field.has_presence: return not _proto_message_has_field(msg, field) if is_repeated(field): @@ -128,7 +144,9 @@ def _is_empty_field(msg: google_message.Message, field: google_descriptor.FieldD return _proto_message_get_field(msg, field) == field.default_value -def field_to_cel(msg: google_message.Message, field: google_descriptor.FieldDescriptor) -> typing.Any: +def field_to_cel( + msg: google_message.Message, field: google_descriptor.FieldDescriptor +) -> typing.Any: return _field_value_to_cel(_proto_message_get_field(msg, field), field) @@ -137,7 +155,9 @@ def _ftype(google_type: int) -> FieldDescriptorProto.Type: return FieldDescriptorProto.Type(google_type) -def _field_to_element(field: google_descriptor.FieldDescriptor) -> validate_pb.FieldPathElement: +def _field_to_element( + field: google_descriptor.FieldDescriptor, +) -> validate_pb.FieldPathElement: """A FieldPathElement for a (google) field of the message being validated.""" return validate_pb.FieldPathElement( field_number=field.number, @@ -146,7 +166,9 @@ def _field_to_element(field: google_descriptor.FieldDescriptor) -> validate_pb.F ) -def _indexed_field_element(field: google_descriptor.FieldDescriptor, index: int) -> validate_pb.FieldPathElement: +def _indexed_field_element( + field: google_descriptor.FieldDescriptor, index: int +) -> validate_pb.FieldPathElement: return validate_pb.FieldPathElement( field_number=field.number, field_name=field.name if not field.is_extension else f"[{field.full_name}]", @@ -155,7 +177,9 @@ def _indexed_field_element(field: google_descriptor.FieldDescriptor, index: int) ) -def _oneof_to_element(oneof: google_descriptor.OneofDescriptor) -> validate_pb.FieldPathElement: +def _oneof_to_element( + oneof: google_descriptor.OneofDescriptor, +) -> validate_pb.FieldPathElement: return validate_pb.FieldPathElement(field_name=oneof.name) @@ -217,7 +241,9 @@ def _spec_element(pb_field: protobuf.DescField) -> validate_pb.FieldPathElement: ) -def _indexed_spec_element(pb_field: protobuf.DescField, index: int) -> validate_pb.FieldPathElement: +def _indexed_spec_element( + pb_field: protobuf.DescField, index: int +) -> validate_pb.FieldPathElement: return validate_pb.FieldPathElement( field_number=pb_field.number, field_name=pb_field.name, @@ -243,7 +269,9 @@ def _google_predefined_ext() -> typing.Any: ``buf.validate`` (and the user's files, which define any custom predefined rule extensions) when it bridges rule messages. """ - return google_descriptor_pool.Default().FindExtensionByName("buf.validate.predefined") + return google_descriptor_pool.Default().FindExtensionByName( + "buf.validate.predefined" + ) def _message_child(pb_field: protobuf.DescField) -> protobuf.DescMessage | None: @@ -251,9 +279,13 @@ def _message_child(pb_field: protobuf.DescField) -> protobuf.DescMessage | None: value = pb_field.value if isinstance(value, protobuf.DescFieldValueMessage): return value.message - if isinstance(value, protobuf.DescFieldValueList) and isinstance(value.element, protobuf.DescMessage): + if isinstance(value, protobuf.DescFieldValueList) and isinstance( + value.element, protobuf.DescMessage + ): return value.element - if isinstance(value, protobuf.DescFieldValueMap) and isinstance(value.value, protobuf.DescMessage): + if isinstance(value, protobuf.DescFieldValueMap) and isinstance( + value.value, protobuf.DescMessage + ): return value.value return None @@ -278,7 +310,7 @@ class CelRules(Rules): _rules: google_message.Message | None = None _uses_now: bool = False - def __init__(self, rules_google: google_message.Message | None): + def __init__(self, rules_google: google_message.Message | None) -> None: self._cel = [] self._rules = rules_google @@ -289,7 +321,7 @@ def _validate_cel( this_value: typing.Any | None = None, this_cel: typing.Any | None = None, for_key: bool = False, - ): + ) -> None: if not self._cel: return activation: dict[str, typing.Any] = {} @@ -315,7 +347,7 @@ def _validate_cel( rule_id=runner.rule.id, message=msg, for_key=for_key, - ), + ) ) elif result_type == cel.Type.STRING: # Formatting bytes with %s can yield a CEL string that is not @@ -333,7 +365,7 @@ def _validate_cel( rule_id=runner.rule.id, message=result_message, # ty: ignore[invalid-argument-type] for_key=for_key, - ), + ) ) elif result_type == cel.Type.ERROR: raise RuntimeError(str(result.plain_value())) @@ -345,7 +377,7 @@ def add_rule( *, rule_field: google_descriptor.FieldDescriptor | None = None, rule_path: validate_pb.FieldPath | None = None, - ): + ) -> None: if isinstance(rules, str): expression = rules rules = validate_pb.Rule(id=expression, expression=expression) @@ -372,14 +404,18 @@ def add_rule( class MessageOneofRule(Rules): - """Validates a single buf.validate.MessageOneofRule given via the message option (buf.validate.message).oneof""" + """Validates a single buf.validate.MessageOneofRule given via the message option (buf.validate.message).oneof.""" - def __init__(self, fields: list[google_descriptor.FieldDescriptor], *, required: bool): + def __init__( + self, fields: list[google_descriptor.FieldDescriptor], *, required: bool + ) -> None: self._fields = fields self._required = required - def validate(self, ctx: RuleContext, message: google_message.Message): - num_set_fields = sum(1 for field in self._fields if not _is_empty_field(message, field)) + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: + num_set_fields = sum( + 1 for field in self._fields if not _is_empty_field(message, field) + ) if num_set_fields > 1: ctx.add( Violation( @@ -401,12 +437,16 @@ class MessageRules(CelRules): _oneofs: list[MessageOneofRule] - def __init__(self, rules_google: google_message.Message | None, desc: google_descriptor.Descriptor): + def __init__( + self, + rules_google: google_message.Message | None, + desc: google_descriptor.Descriptor, + ) -> None: super().__init__(rules_google) self._oneofs = [] self._desc = desc - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: if self._cel: self._validate_cel(ctx, this_cel=message) if ctx.done: @@ -416,7 +456,7 @@ def validate(self, ctx: RuleContext, message: google_message.Message): if ctx.done: return - def add_oneof(self, rule: typing.Any): + def add_oneof(self, rule: typing.Any) -> None: fields = [] seen = set() if len(rule.fields) == 0: @@ -435,16 +475,23 @@ def add_oneof(self, rule: typing.Any): self._oneofs.append(MessageOneofRule(fields, required=rule.required)) -def check_field_type(field: google_descriptor.FieldDescriptor, expected: int, wrapper_name: str | None = None): +def check_field_type( + field: google_descriptor.FieldDescriptor, + expected: int, + wrapper_name: str | None = None, +) -> None: if field.type != expected and ( - field.type != google_descriptor.FieldDescriptor.TYPE_MESSAGE or field.message_type.full_name != wrapper_name + field.type != google_descriptor.FieldDescriptor.TYPE_MESSAGE + or field.message_type.full_name != wrapper_name ): field_type_str = _get_type_name(field.type) if expected == 0: if wrapper_name is not None: expected_type_str = wrapper_name else: - expected_type_str = _get_type_name(google_descriptor.FieldDescriptor.TYPE_MESSAGE) + expected_type_str = _get_type_name( + google_descriptor.FieldDescriptor.TYPE_MESSAGE + ) else: expected_type_str = _get_type_name(expected) msg = f"field {field.name} has type {field_type_str} but expected {expected_type_str}" @@ -470,7 +517,7 @@ def __init__( *, for_items: bool = False, force_ignore_empty: bool = False, - ): + ) -> None: type_case = _which_type(field_level) rules_pb = field_level.type.value if type_case is not None else None super().__init__(bridge.to_google(rules_pb) if rules_pb is not None else None) @@ -490,8 +537,12 @@ def __init__( # files, so custom extension rules decode (they would otherwise # remain undecoded on the relocatable protobuf-py stub). g_rules = self._rules - assert g_rules is not None # rules_pb set implies the bridged rules # noqa: S101 - assert type_case is not None # rules_pb set implies a type oneof # noqa: S101 + assert ( # noqa: S101 + g_rules is not None + ) # rules_pb set implies the bridged rules + assert ( # noqa: S101 + type_case is not None + ) # rules_pb set implies a type oneof type_field = _spec_field(validate_pb.FieldRules, type_case) predefined_ext = _google_predefined_ext() for rule_field, _value in g_rules.ListFields(): @@ -504,7 +555,10 @@ def __init__( cel_rule, rule_field=rule_field, rule_path=validate_pb.FieldPath( - elements=[_field_to_element(rule_field), _spec_element(type_field)] + elements=[ + _field_to_element(rule_field), + _spec_element(type_field), + ] ), ) cel_expression_field = _spec_field(validate_pb.FieldRules, "cel_expression") @@ -512,27 +566,33 @@ def __init__( self.add_rule( env, cel_rule, - rule_path=validate_pb.FieldPath(elements=[_indexed_spec_element(cel_expression_field, i)]), + rule_path=validate_pb.FieldPath( + elements=[_indexed_spec_element(cel_expression_field, i)] + ), ) cel_field = _spec_field(validate_pb.FieldRules, "cel") for i, cel_rule in enumerate(field_level.cel): self.add_rule( env, cel_rule, - rule_path=validate_pb.FieldPath(elements=[_indexed_spec_element(cel_field, i)]), + rule_path=validate_pb.FieldPath( + elements=[_indexed_spec_element(cel_field, i)] + ), ) - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: if _is_empty_field(message, self._field): if self._required: ctx.add( Violation( - field=validate_pb.FieldPath(elements=[_field_to_element(self._field)]), + field=validate_pb.FieldPath( + elements=[_field_to_element(self._field)] + ), rule=FieldRules._required_rule_path, rule_value=self._required, rule_id="required", message="value is required", - ), + ) ) return if self._ignore_empty: @@ -547,13 +607,20 @@ def validate(self, ctx: RuleContext, message: google_message.Message): sub_ctx.add_field_path_element(element) ctx.add_errors(sub_ctx) - def validate_item(self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False): + def validate_item( + self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False + ) -> None: self._validate_value(ctx, value, for_key=for_key) self._validate_cel( - ctx, this_value=value, this_cel=_scalar_field_value_to_cel(value, self._field), for_key=for_key + ctx, + this_value=value, + this_cel=_scalar_field_value_to_cel(value, self._field), + for_key=for_key, ) - def _validate_value(self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False): + def _validate_value( + self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False + ) -> None: pass @@ -564,14 +631,14 @@ class AnyRules(FieldRules): elements=[ _spec_element(_spec_field(validate_pb.AnyRules, "in")), _spec_element(_spec_field(validate_pb.FieldRules, "any")), - ], + ] ) _not_in_rule_path: typing.ClassVar[validate_pb.FieldPath] = validate_pb.FieldPath( elements=[ _spec_element(_spec_field(validate_pb.AnyRules, "not_in")), _spec_element(_spec_field(validate_pb.FieldRules, "any")), - ], + ] ) def __init__( @@ -580,13 +647,15 @@ def __init__( bridge: GoogleBridge, field: google_descriptor.FieldDescriptor, field_level: typing.Any, - ): + ) -> None: super().__init__(env, bridge, field, field_level) any_rules = field_level.type.value self._in = list(any_rules.in_) or [] self._not_in: Container[str] = list(any_rules.not_in) or [] - def _validate_value(self, ctx: RuleContext, value: google_any_pb2.Any, *, for_key: bool = False): + def _validate_value( + self, ctx: RuleContext, value: google_any_pb2.Any, *, for_key: bool = False + ) -> None: if len(self._in) > 0 and value.type_url not in self._in: ctx.add( Violation( @@ -614,11 +683,13 @@ class EnumRules(FieldRules): _defined_only = False - _defined_only_rule_path: typing.ClassVar[validate_pb.FieldPath] = validate_pb.FieldPath( - elements=[ - _spec_element(_spec_field(validate_pb.EnumRules, "defined_only")), - _spec_element(_spec_field(validate_pb.FieldRules, "enum")), - ], + _defined_only_rule_path: typing.ClassVar[validate_pb.FieldPath] = ( + validate_pb.FieldPath( + elements=[ + _spec_element(_spec_field(validate_pb.EnumRules, "defined_only")), + _spec_element(_spec_field(validate_pb.FieldRules, "enum")), + ] + ) ) def __init__( @@ -630,24 +701,37 @@ def __init__( *, for_items: bool = False, force_ignore_empty: bool = False, - ): - super().__init__(env, bridge, field, field_level, for_items=for_items, force_ignore_empty=force_ignore_empty) + ) -> None: + super().__init__( + env, + bridge, + field, + field_level, + for_items=for_items, + force_ignore_empty=force_ignore_empty, + ) if field_level.type.value.defined_only: self._defined_only = True - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: super().validate(ctx, message) if ctx.done: return - if self._defined_only and getattr(message, self._field.name) not in self._field.enum_type.values_by_number: + if ( + self._defined_only + and getattr(message, self._field.name) + not in self._field.enum_type.values_by_number + ): ctx.add( Violation( - field=validate_pb.FieldPath(elements=[_field_to_element(self._field)]), + field=validate_pb.FieldPath( + elements=[_field_to_element(self._field)] + ), rule=EnumRules._defined_only_rule_path, rule_value=self._defined_only, rule_id="enum.defined_only", message="value must be one of the defined enum values", - ), + ) ) @@ -668,12 +752,12 @@ def __init__( field: google_descriptor.FieldDescriptor, field_level: typing.Any, item_rules: FieldRules | None, - ): + ) -> None: super().__init__(env, bridge, field, field_level) if item_rules is not None: self._item_rules = item_rules - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: super().validate(ctx, message) if ctx.done: return @@ -685,7 +769,9 @@ def validate(self, ctx: RuleContext, message: google_message.Message): sub_ctx = ctx.sub_context() self._item_rules.validate_item(sub_ctx, item) if sub_ctx.has_errors(): - sub_ctx.add_field_path_element(_indexed_field_element(self._field, i)) + sub_ctx.add_field_path_element( + _indexed_field_element(self._field, i) + ) sub_ctx.add_rule_path_elements(RepeatedRules._items_rules_suffix) ctx.add_errors(sub_ctx) if ctx.done: @@ -716,14 +802,14 @@ def __init__( field_level: typing.Any, key_rules: FieldRules | None, value_rules: FieldRules | None, - ): + ) -> None: super().__init__(env, bridge, field, field_level) if key_rules is not None: self._key_rules = key_rules if value_rules is not None: self._value_rules = value_rules - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: super().validate(ctx, message) if ctx.done: return @@ -732,20 +818,22 @@ def validate(self, ctx: RuleContext, message: google_message.Message): value_field = self._field.message_type.fields_by_name["value"] for k, v in value.items(): key_ctx = ctx.sub_context() - if self._key_rules is not None: - if not self._key_rules._ignore_empty or k: - self._key_rules.validate_item(key_ctx, k, for_key=True) - if key_ctx.has_errors(): - key_ctx.add_rule_path_elements(MapRules._key_rules_suffix) + if self._key_rules is not None and (not self._key_rules._ignore_empty or k): + self._key_rules.validate_item(key_ctx, k, for_key=True) + if key_ctx.has_errors(): + key_ctx.add_rule_path_elements(MapRules._key_rules_suffix) map_ctx = ctx.sub_context() - if self._value_rules is not None: - if not self._value_rules._ignore_empty or v: - self._value_rules.validate_item(map_ctx, v) - if map_ctx.has_errors(): - map_ctx.add_rule_path_elements(MapRules._value_rules_suffix) + if self._value_rules is not None and ( + not self._value_rules._ignore_empty or v + ): + self._value_rules.validate_item(map_ctx, v) + if map_ctx.has_errors(): + map_ctx.add_rule_path_elements(MapRules._value_rules_suffix) map_ctx.add_errors(key_ctx) if map_ctx.has_errors(): - map_ctx.add_field_path_element(_map_key_element(self._field, k, key_field, value_field)) + map_ctx.add_field_path_element( + _map_key_element(self._field, k, key_field, value_field) + ) ctx.add_errors(map_ctx) @@ -754,17 +842,21 @@ class OneofRules(Rules): required = True - def __init__(self, oneof: google_descriptor.OneofDescriptor, rules: typing.Any): + def __init__( + self, oneof: google_descriptor.OneofDescriptor, rules: typing.Any + ) -> None: self._oneof = oneof if not rules.required: self.required = False - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: if not message.WhichOneof(self._oneof.name): if self.required: ctx.add( Violation( - field=validate_pb.FieldPath(elements=[_oneof_to_element(self._oneof)]), + field=validate_pb.FieldPath( + elements=[_oneof_to_element(self._oneof)] + ), rule_id="required", message="exactly one field is required in oneof", ) @@ -777,7 +869,7 @@ class RuleFactory: _env: cel.Env - def __init__(self, extension: cel.CelExtensionBase, bridge: GoogleBridge): + def __init__(self, extension: cel.CelExtensionBase, bridge: GoogleBridge) -> None: self._bridge = bridge # cel-expr-python evaluates against — and type-checks bindings against — # the global pool the bridge populates. @@ -798,14 +890,16 @@ def get(self, desc: protobuf.DescMessage) -> list[Rules]: if key not in self._cache: try: self._cache[key] = self._new_rules(desc) - except Exception as e: + except Exception as e: # noqa: BLE001 self._cache[key] = e result = self._cache[key] if isinstance(result, Exception): raise result return result - def _new_message_rule(self, rules: typing.Any, desc: google_descriptor.Descriptor) -> MessageRules: + def _new_message_rule( + self, rules: typing.Any, desc: google_descriptor.Descriptor + ) -> MessageRules: result = MessageRules(self._bridge.to_google(rules), desc) for oneof in rules.oneof: result.add_oneof(oneof) @@ -822,7 +916,7 @@ def _new_scalar_field_rule( *, for_items: bool = False, force_ignore_empty: bool = False, - ): + ) -> FieldRules | None: if field_level.ignore == validate_pb.Ignore.ALWAYS: return None type_case = _which_type(field_level) @@ -842,10 +936,18 @@ def _new_scalar_field_rule( check_field_type(field, google_descriptor.FieldDescriptor.TYPE_ENUM) return EnumRules(self._env, self._bridge, field, field_level, **kw) if type_case == "bool": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_BOOL, "google.protobuf.BoolValue") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_BOOL, + "google.protobuf.BoolValue", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "bytes": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_BYTES, "google.protobuf.BytesValue") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_BYTES, + "google.protobuf.BytesValue", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "fixed32": check_field_type(field, google_descriptor.FieldDescriptor.TYPE_FIXED32) @@ -854,16 +956,32 @@ def _new_scalar_field_rule( check_field_type(field, google_descriptor.FieldDescriptor.TYPE_FIXED64) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "float": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_FLOAT, "google.protobuf.FloatValue") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_FLOAT, + "google.protobuf.FloatValue", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "double": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_DOUBLE, "google.protobuf.DoubleValue") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_DOUBLE, + "google.protobuf.DoubleValue", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "int32": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_INT32, "google.protobuf.Int32Value") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_INT32, + "google.protobuf.Int32Value", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "int64": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_INT64, "google.protobuf.Int64Value") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_INT64, + "google.protobuf.Int64Value", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "sfixed32": check_field_type(field, google_descriptor.FieldDescriptor.TYPE_SFIXED32) @@ -878,13 +996,25 @@ def _new_scalar_field_rule( check_field_type(field, google_descriptor.FieldDescriptor.TYPE_SINT64) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "uint32": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_UINT32, "google.protobuf.UInt32Value") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_UINT32, + "google.protobuf.UInt32Value", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "uint64": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_UINT64, "google.protobuf.UInt64Value") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_UINT64, + "google.protobuf.UInt64Value", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "string": - check_field_type(field, google_descriptor.FieldDescriptor.TYPE_STRING, "google.protobuf.StringValue") + check_field_type( + field, + google_descriptor.FieldDescriptor.TYPE_STRING, + "google.protobuf.StringValue", + ) return FieldRules(self._env, self._bridge, field, field_level, **kw) if type_case == "any": check_field_type(field, 0, "google.protobuf.Any") @@ -898,9 +1028,11 @@ def _new_field_rule( field_level: typing.Any, *, force_ignore_empty: bool = False, - ) -> FieldRules: + ) -> FieldRules | None: if not is_repeated(field): - return self._new_scalar_field_rule(field, field_level, force_ignore_empty=force_ignore_empty) + return self._new_scalar_field_rule( + field, field_level, force_ignore_empty=force_ignore_empty + ) type_case = _which_type(field_level) if field.message_type is not None and field.message_type.GetOptions().map_entry: map_rules = field_level.type.value if type_case == "map" else None @@ -908,11 +1040,17 @@ def _new_field_rule( value_rules = None if map_rules is not None and map_rules.keys is not None: key_field = field.message_type.fields_by_name["key"] - key_rules = self._new_scalar_field_rule(key_field, map_rules.keys, for_items=True) + key_rules = self._new_scalar_field_rule( + key_field, map_rules.keys, for_items=True + ) if map_rules is not None and map_rules.values is not None: value_field = field.message_type.fields_by_name["value"] - value_rules = self._new_scalar_field_rule(value_field, map_rules.values, for_items=True) - return MapRules(self._env, self._bridge, field, field_level, key_rules, value_rules) + value_rules = self._new_scalar_field_rule( + value_field, map_rules.values, for_items=True + ) + return MapRules( + self._env, self._bridge, field, field_level, key_rules, value_rules + ) item_rule = None rep_rules = field_level.type.value if type_case == "repeated" else None if rep_rules is not None and rep_rules.items is not None: @@ -920,7 +1058,10 @@ def _new_field_rule( return RepeatedRules(self._env, self._bridge, field, field_level, item_rule) def _new_rules(self, pb_desc: protobuf.DescMessage) -> list[Rules]: - gdesc = typing.cast(google_descriptor.Descriptor, self._bridge.google_class(pb_desc).DESCRIPTOR) + gdesc = typing.cast( + "google_descriptor.Descriptor", + self._bridge.google_class(pb_desc).DESCRIPTOR, + ) result: list[Rules] = [] all_msg_oneof_fields: set[str] = set() @@ -952,13 +1093,22 @@ def _new_rules(self, pb_desc: protobuf.DescMessage) -> list[Rules]: if field_level is not None: # A field in a message-level oneof rule that does not set its own # ignore behaviour is treated as ignore-if-zero-value. - force_ignore_empty = ignore_field not in field_level and gfield.name in all_msg_oneof_fields + force_ignore_empty = ( + ignore_field not in field_level + and gfield.name in all_msg_oneof_fields + ) if field_level.ignore == validate_pb.Ignore.ALWAYS: continue - result.append(self._new_field_rule(gfield, field_level, force_ignore_empty=force_ignore_empty)) + if r := self._new_field_rule( + gfield, field_level, force_ignore_empty=force_ignore_empty + ): + result.append(r) if _which_type(field_level) == "repeated": rep: typing.Any = field_level.type.value # ty: ignore[unresolved-attribute] - if rep.items is not None and rep.items.ignore == validate_pb.Ignore.ALWAYS: + if ( + rep.items is not None + and rep.items.ignore == validate_pb.Ignore.ALWAYS + ): continue if gfield.message_type is None or pb_field is None: continue @@ -968,7 +1118,9 @@ def _new_rules(self, pb_desc: protobuf.DescMessage) -> list[Rules]: if gfield.message_type.GetOptions().map_entry: key_field = gfield.message_type.fields_by_name["key"] value_field = gfield.message_type.fields_by_name["value"] - result.append(MapValMsgRule(self, gfield, key_field, value_field, sub_desc)) + result.append( + MapValMsgRule(self, gfield, key_field, value_field, sub_desc) + ) elif is_repeated(gfield): result.append(RepeatedMsgRule(self, gfield, sub_desc)) else: @@ -977,12 +1129,17 @@ def _new_rules(self, pb_desc: protobuf.DescMessage) -> list[Rules]: class SubMsgRule(Rules): - def __init__(self, factory: RuleFactory, field: google_descriptor.FieldDescriptor, sub_desc: protobuf.DescMessage): + def __init__( + self, + factory: RuleFactory, + field: google_descriptor.FieldDescriptor, + sub_desc: protobuf.DescMessage, + ) -> None: self._factory = factory self._field = field self._sub_desc = sub_desc - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: if not message.HasField(self._field.name): return rules = self._factory.get(self._sub_desc) @@ -1005,14 +1162,14 @@ def __init__( key_field: google_descriptor.FieldDescriptor, value_field: google_descriptor.FieldDescriptor, sub_desc: protobuf.DescMessage, - ): + ) -> None: self._factory = factory self._field = field self._key_field = key_field self._value_field = value_field self._sub_desc = sub_desc - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: val = getattr(message, self._field.name) if not val: return @@ -1024,17 +1181,24 @@ def validate(self, ctx: RuleContext, message: google_message.Message): for rule in rules: rule.validate(sub_ctx, v) if sub_ctx.has_errors(): - sub_ctx.add_field_path_element(_map_key_element(self._field, k, self._key_field, self._value_field)) + sub_ctx.add_field_path_element( + _map_key_element(self._field, k, self._key_field, self._value_field) + ) ctx.add_errors(sub_ctx) class RepeatedMsgRule(Rules): - def __init__(self, factory: RuleFactory, field: google_descriptor.FieldDescriptor, sub_desc: protobuf.DescMessage): + def __init__( + self, + factory: RuleFactory, + field: google_descriptor.FieldDescriptor, + sub_desc: protobuf.DescMessage, + ) -> None: self._factory = factory self._field = field self._sub_desc = sub_desc - def validate(self, ctx: RuleContext, message: google_message.Message): + def validate(self, ctx: RuleContext, message: google_message.Message) -> None: val = getattr(message, self._field.name) if not val: return diff --git a/protovalidate/internal/_core.py b/protovalidate/_core.py similarity index 84% rename from protovalidate/internal/_core.py rename to protovalidate/_core.py index 0d21a8f..d0b615e 100644 --- a/protovalidate/internal/_core.py +++ b/protovalidate/_core.py @@ -14,6 +14,8 @@ """Backend-agnostic rule-engine primitives shared by both CEL backends.""" +from __future__ import annotations + import abc import typing @@ -40,11 +42,15 @@ def __init__( rule_id: str = "", message: str = "", for_key: bool = False, - ): + ) -> None: self.field_value = field_value self.rule_value = rule_value - self._field_elements: list[validate_pb.FieldPathElement] = list(field.elements) if field is not None else [] - self._rule_elements: list[validate_pb.FieldPathElement] = list(rule.elements) if rule is not None else [] + self._field_elements: list[validate_pb.FieldPathElement] = ( + list(field.elements) if field is not None else [] + ) + self._rule_elements: list[validate_pb.FieldPathElement] = ( + list(rule.elements) if rule is not None else [] + ) self._rule_id = rule_id self._message = message self._for_key = for_key @@ -52,7 +58,9 @@ def __init__( def append_field_element(self, element: validate_pb.FieldPathElement) -> None: self._field_elements.append(element) - def extend_rule_elements(self, elements: list[validate_pb.FieldPathElement]) -> None: + def extend_rule_elements( + self, elements: list[validate_pb.FieldPathElement] + ) -> None: self._rule_elements.extend(elements) def finalize_paths(self) -> None: @@ -78,7 +86,7 @@ class RuleContext: _violations: list[Violation] - def __init__(self, *, fail_fast: bool = False): + def __init__(self, *, fail_fast: bool = False) -> None: self._fail_fast = fail_fast self._violations = [] @@ -86,17 +94,19 @@ def __init__(self, *, fail_fast: bool = False): def violations(self) -> list[Violation]: return self._violations - def add(self, violation: Violation): + def add(self, violation: Violation) -> None: self._violations.append(violation) - def add_errors(self, other_ctx: "RuleContext"): + def add_errors(self, other_ctx: RuleContext) -> None: self._violations.extend(other_ctx.violations) - def add_field_path_element(self, element: validate_pb.FieldPathElement): + def add_field_path_element(self, element: validate_pb.FieldPathElement) -> None: for violation in self._violations: violation.append_field_element(element) - def add_rule_path_elements(self, elements: list[validate_pb.FieldPathElement]): + def add_rule_path_elements( + self, elements: list[validate_pb.FieldPathElement] + ) -> None: for violation in self._violations: violation.extend_rule_elements(elements) @@ -107,7 +117,7 @@ def done(self) -> bool: def has_errors(self) -> bool: return len(self._violations) > 0 - def sub_context(self) -> "RuleContext": + def sub_context(self) -> RuleContext: return RuleContext(fail_fast=self._fail_fast) diff --git a/protovalidate/internal/extra_func.py b/protovalidate/_extra_func.py similarity index 88% rename from protovalidate/internal/extra_func.py rename to protovalidate/_extra_func.py index 6c1f44c..5c131ad 100644 --- a/protovalidate/internal/extra_func.py +++ b/protovalidate/_extra_func.py @@ -14,21 +14,24 @@ """celpy registration for protovalidate's custom CEL functions.""" +from __future__ import annotations + import functools import typing -from collections.abc import Callable import celpy import re2 from celpy import celtypes -from protovalidate.internal import _funcs, string_format -from protovalidate.internal.rules import MessageType +from protovalidate import _funcs, _string_format +from protovalidate._rules import MessageType + +if typing.TYPE_CHECKING: + from collections.abc import Callable def _to_py(value: object) -> object: - """Convert a celpy celtypes value to the plain Python type the shared - ``_funcs`` implementations expect. + """Convert a celpy celtypes value to a plain Python type for func evaluation. Most celtypes subclass their builtin, so this would be a no-op — except ``celtypes.BoolType`` subclasses ``int`` (``bool`` cannot be subclassed), so @@ -53,9 +56,7 @@ def _to_py(value: object) -> object: def _wrap(fn: Callable[..., object]) -> celpy.CELFunction: - """Adapt a shared plain-Python ``_funcs`` implementation to celpy: normalize - celtypes arguments to plain Python, wrap a ``bool``/``str`` return in the - matching celtype, and re-raise ``ValueError`` as ``celpy.CELEvalError``.""" + """Adapt a plain Python func result to celpy.""" @functools.wraps(fn) def wrapper(*args: celtypes.Value) -> celpy.Result: @@ -108,7 +109,7 @@ def cel_matches(text: str, pattern: str) -> celpy.Result: def make_extra_funcs() -> dict[str, celpy.CELFunction]: - string_fmt = string_format.StringFormat() + string_fmt = _string_format.StringFormat() return { # Missing standard functions "format": string_fmt.format, diff --git a/protovalidate/internal/_funcs.py b/protovalidate/_funcs.py similarity index 88% rename from protovalidate/internal/_funcs.py rename to protovalidate/_funcs.py index e57ba2f..a1a63e5 100644 --- a/protovalidate/internal/_funcs.py +++ b/protovalidate/_funcs.py @@ -14,7 +14,10 @@ """Pure-Python implementations of protovalidate's custom CEL functions.""" +from __future__ import annotations + import math +from typing import overload from urllib import parse as urlparse import re2 @@ -39,21 +42,18 @@ def cel_is_ip(val: object, ver: object | None = None) -> bool: """ if not isinstance(val, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) if ver is not None and (not isinstance(ver, int) or isinstance(ver, bool)): msg = "invalid argument, expected int" - raise ValueError(msg) + raise TypeError(msg) - if ver is None: - version = 0 - else: - version = ver + version = 0 if ver is None else ver return _is_ip(val, version) def _is_ip(string: str, version: int) -> bool: - """Internal implementation""" + """Internal implementation.""" valid = False if version == 6: valid = Ipv6(string).address() @@ -65,10 +65,26 @@ def _is_ip(string: str, version: int) -> bool: return valid -def cel_is_ip_prefix(val: object, *args) -> bool: - """Return True if the string is a valid IP with prefix length, optionally - limited to a specific version (v4 or v6), and optionally requiring the host - portion to be all zeros. +@overload +def cel_is_ip_prefix(val: str, /) -> bool: ... +@overload +def cel_is_ip_prefix(val: str, strict: bool, /) -> bool: ... # noqa: FBT001 +@overload +def cel_is_ip_prefix(val: str, version: int, /) -> bool: ... +@overload +def cel_is_ip_prefix(val: str, version: int, strict: bool, /) -> bool: ... # noqa: FBT001 + + +def cel_is_ip_prefix( + val: str, + strict_or_version: bool | int | None = None, # noqa: FBT001 + strict_param: bool | None = None, # noqa: FBT001 + /, +) -> bool: + """Checks if the string is a valid IP with prefix length. + + The check may be optionally limited to a specific version (v4 or v6), and + optionally requiring the host portion to be all zeros. An address prefix divides an IP address into a network portion, and a host portion. The prefix length specifies how many bits the network portion has. @@ -84,31 +100,37 @@ def cel_is_ip_prefix(val: object, *args) -> bool: the first 24 bits of the 32-bit IPv4 as the network prefix. """ - if not isinstance(val, str): - msg = "invalid argument, expected string or bytes" - raise ValueError(msg) + msg = "invalid argument, expected string" + raise TypeError(msg) version = 0 strict = False - if len(args) == 1 and isinstance(args[0], bool): - strict = bool(args[0]) - elif len(args) == 1 and isinstance(args[0], int): - version = args[0] - elif len(args) == 1: - msg = "invalid argument, expected bool or int" - raise ValueError(msg) - elif len(args) == 2 and isinstance(args[0], int) and not isinstance(args[0], bool) and isinstance(args[1], bool): - version = args[0] - strict = bool(args[1]) - elif len(args) == 2: - msg = "invalid argument, expected int and bool" - raise ValueError(msg) + if strict_param is not None: + if not isinstance(strict_param, bool): + msg = "invalid argument, expected bool" + raise TypeError(msg) + if not isinstance(strict_or_version, int) or isinstance( + strict_or_version, bool + ): + msg = "invalid argument, expected int" + raise TypeError(msg) + strict = strict_param + version = strict_or_version + elif strict_or_version is not None: + match strict_or_version: + case bool(): + strict = strict_or_version + case int(): + version = strict_or_version + case _: + msg = "invalid argument, expected bool or int" + raise TypeError(msg) return _is_ip_prefix(val, version, strict=strict) -def _is_ip_prefix(string: str, version: int, *, strict=False) -> bool: - """Internal implementation""" +def _is_ip_prefix(string: str, version: int, *, strict: bool = False) -> bool: + """Internal implementation.""" valid = False if version == 6: v6 = Ipv6(string) @@ -117,7 +139,9 @@ def _is_ip_prefix(string: str, version: int, *, strict=False) -> bool: v4 = Ipv4(string) valid = v4.address_prefix() and (not strict or v4.is_prefix_only()) elif version == 0: - valid = _is_ip_prefix(string, 6, strict=strict) or _is_ip_prefix(string, 4, strict=strict) + valid = _is_ip_prefix(string, 6, strict=strict) or _is_ip_prefix( + string, 4, strict=strict + ) return valid @@ -133,7 +157,7 @@ def cel_is_email(string: object) -> bool: """ if not isinstance(string, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) return _email_regex.fullmatch(string) is not None @@ -146,12 +170,14 @@ def cel_is_uri(string: object) -> bool: """ if not isinstance(string, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) return Uri(str(string)).uri() def cel_is_uri_ref(string: object) -> bool: - """Return True if the string is a URI Reference - a URI such as "https://example.com/foo/bar?baz=quux#frag" or + """Checks if the string is a URI Reference. + + A URI reference is a URI such as "https://example.com/foo/bar?baz=quux#frag" or a Relative Reference such as "./foo/bar?query". URI, URI Reference, and Relative Reference are defined in the internet standard RFC 3986. @@ -160,12 +186,12 @@ def cel_is_uri_ref(string: object) -> bool: """ if not isinstance(string, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) return Uri(str(string)).uri_reference() def cel_is_hostname(val: object) -> bool: - """Returns True if the string is a valid hostname, for example "foo.example.com". + """Checks if the string is a valid hostname, for example "foo.example.com". A valid hostname follows the rules below: - The name consists of one or more labels, separated by a dot ("."). @@ -178,19 +204,16 @@ def cel_is_hostname(val: object) -> bool: """ if not isinstance(val, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) return _is_hostname(val) def _is_hostname(val: str) -> bool: - """Internal implementation""" + """Internal implementation.""" if len(val) > 253: return False - if val.endswith("."): - string = val[0 : len(val) - 1].lower() - else: - string = val.lower() + string = val[0 : len(val) - 1].lower() if val.endswith(".") else val.lower() all_digits = False parts = string.lower().split(sep=".") @@ -246,14 +269,14 @@ def cel_is_host_and_port(string: object, port_required: object) -> bool: """ if not isinstance(string, str): msg = "invalid argument, expected string" - raise ValueError(msg) + raise TypeError(msg) if not isinstance(port_required, bool): msg = "invalid argument, expected bool" - raise ValueError(msg) + raise TypeError(msg) return _is_host_and_port(string, port_required=bool(port_required)) -def _is_host_and_port(val: str, *, port_required=False) -> bool: +def _is_host_and_port(val: str, *, port_required: bool = False) -> bool: if len(val) == 0: return False @@ -265,11 +288,10 @@ def _is_host_and_port(val: str, *, port_required=False) -> bool: if end_plus == len(val): return not port_required and _is_ip(val[1:end], 6) - elif end_plus == split_idx: + if end_plus == split_idx: return _is_ip(val[1:end], 6) and _is_port(val[split_idx + 1 :]) - else: - # malformed - return False + # malformed + return False if split_idx < 0: return not port_required and (_is_hostname(val) or _is_ip(val, 4)) @@ -283,26 +305,25 @@ def _is_host_and_port(val: str, *, port_required=False) -> bool: def cel_is_nan(val: object) -> bool: if not isinstance(val, float): msg = "invalid argument, expected double" - raise ValueError(msg) + raise TypeError(msg) return math.isnan(val) def cel_is_inf(val: object, sign: object | None = None) -> bool: if not isinstance(val, float): msg = "invalid argument, expected double" - raise ValueError(msg) + raise TypeError(msg) if sign is None: return math.isinf(val) if not isinstance(sign, int) or isinstance(sign, bool): msg = "invalid argument, expected int" - raise ValueError(msg) + raise TypeError(msg) if sign > 0: return math.isinf(val) and val > 0 - elif sign < 0: + if sign < 0: return math.isinf(val) and val < 0 - else: - return math.isinf(val) + return math.isinf(val) class Ipv4: @@ -313,9 +334,8 @@ class Ipv4: _octets: bytearray _prefix_len: int - def __init__(self, string: str): + def __init__(self, string: str) -> None: """Initialize an Ipv4 validation class with a given string.""" - super().__init__() self._string = string self._index = 0 @@ -329,7 +349,10 @@ def address(self) -> bool: def address_prefix(self) -> bool: """Parses an IPv4 Address prefix.""" return ( - self.__address_part() and self.__take("/") and self.__prefix_length() and self._index == len(self._string) + self.__address_part() + and self.__take("/") + and self.__prefix_length() + and self._index == len(self._string) ) def get_bits(self) -> int: @@ -341,7 +364,12 @@ def get_bits(self) -> int: if len(self._octets) != 4: return -1 - return (self._octets[0] << 24) | (self._octets[1] << 16) | (self._octets[2] << 8) | self._octets[3] + return ( + (self._octets[0] << 24) + | (self._octets[1] << 16) + | (self._octets[2] << 8) + | self._octets[3] + ) def is_prefix_only(self) -> bool: """Return True if all bits to the right of the prefix-length are all zeros. @@ -362,8 +390,7 @@ def is_prefix_only(self) -> bool: return bits == masked def __prefix_length(self) -> bool: - """Store value in prefix_len""" - + """Store value in prefix_len.""" start = self._index while self.__digit(): @@ -388,12 +415,11 @@ def __prefix_length(self) -> bool: return False self._prefix_len = value - - return True - except ValueError: # Error converting to number return False + else: + return True def __address_part(self) -> bool: start = self._index @@ -437,12 +463,11 @@ def __dec_octet(self) -> bool: return False self._octets.append(value) - - return True - except ValueError: # Error converting to number return False + else: + return True def __digit(self) -> bool: """Report whether the current position is a digit. @@ -452,7 +477,6 @@ def __digit(self) -> bool: DIGIT = %x30-39 ; 0-9 """ - if self._index >= len(self._string): return False @@ -465,7 +489,6 @@ def __digit(self) -> bool: def __take(self, char: str) -> bool: """Take the given char at the current position, incrementing the index if necessary.""" - if self._index >= len(self._string): return False @@ -489,9 +512,8 @@ class Ipv6: _zone_id_found: bool _prefix_len: int # 0 -128 - def __init__(self, string: str): + def __init__(self, string: str) -> None: """Initialize a URI validation class with a given string.""" - super().__init__() self._string = string self._index = 0 @@ -555,19 +577,14 @@ def is_prefix_only(self) -> bool: mask = ~(0xFFFFFFFF_FFFFFFFF_FFFFFFFF_FFFFFFFF >> self._prefix_len) masked = bits & mask - if bits != masked: - return False - - return True + return bits == masked def address(self) -> bool: """Parse an IPv6 Address following RFC 4291, with optional zone id following RFC 4007.""" - return self.__address_part() and self._index == len(self._string) def address_prefix(self) -> bool: """Parse an IPv6 Address Prefix following RFC 4291. Zone id is not permitted.""" - return ( self.__address_part() and not self._zone_id_found @@ -602,16 +619,14 @@ def __prefix_length(self) -> bool: return False self._prefix_len = value - - return True - except ValueError: # Error converting to number return False + else: + return True def __address_part(self) -> bool: """Store dotted notation for right-most 32 bits in dotted_raw / dotted_addr if found.""" - while self._index < len(self._string): # dotted notation for right-most 32 bits, e.g. 0:0:0:0:0:ffff:192.1.56.10 if (self._double_colon_seen or len(self._pieces) == 6) and self.__dotted(): @@ -665,13 +680,12 @@ def __zone_id(self) -> bool: """ start = self._index - if self.__take("%"): - if len(self._string) - self._index > 0: - # permit any non-null string - self._index = len(self._string) - self._zone_id_found = True + if self.__take("%") and len(self._string) - self._index > 0: + # permit any non-null string + self._index = len(self._string) + self._zone_id_found = True - return True + return True self._index = start self._zone_id_found = False @@ -686,7 +700,6 @@ def __dotted(self) -> bool: Stores match in _dotted_raw. """ - start = self._index self._dotted_raw = "" @@ -713,7 +726,6 @@ def __h16(self) -> bool: If more than 4 hex digits are found or the found hex digits cannot be converted to an int, a ValueError is raised. """ - start = self._index while self.__hex_dig(): @@ -802,7 +814,7 @@ class Uri: _index: int _pct_encoded_found: bool - def __init__(self, string: str): + def __init__(self, string: str) -> None: """Initialize a URI validation class with a given string.""" super().__init__() self._string = string @@ -855,7 +867,12 @@ def __hier_part(self) -> bool: """ start = self._index - if self.__take("/") and self.__take("/") and self.__authority() and self.__path_abempty(): + if ( + self.__take("/") + and self.__take("/") + and self.__authority() + and self.__path_abempty() + ): return True self._index = start @@ -899,7 +916,12 @@ def __relative_part(self) -> bool: """ start = self._index - if self.__take("/") and self.__take("/") and self.__authority() and self.__path_abempty(): + if ( + self.__take("/") + and self.__take("/") + and self.__authority() + and self.__path_abempty() + ): return True self._index = start @@ -917,7 +939,13 @@ def __scheme(self) -> bool: """ start = self._index if self.__alpha(): - while self.__alpha() or self.__digit() or self.__take("+") or self.__take("-") or self.__take("."): + while ( + self.__alpha() + or self.__digit() + or self.__take("+") + or self.__take("-") + or self.__take(".") + ): pass if self.__peek(":"): @@ -937,19 +965,17 @@ def __authority(self) -> bool: """ start = self._index - if self.__userinfo(): - if not self.__take("@"): - self._index = start - return False + if self.__userinfo() and not self.__take("@"): + self._index = start + return False if not self.__host(): self._index = start return False - if self.__take(":"): - if not self.__port(): - self._index = start - return False + if self.__take(":") and not self.__port(): + self._index = start + return False if not self.__is_authority_end(): self._index = start @@ -983,7 +1009,12 @@ def __userinfo(self) -> bool: """ start = self._index - while self.__unreserved() or self.__pct_encoded() or self.__sub_delims() or self.__take(":"): + while ( + self.__unreserved() + or self.__pct_encoded() + or self.__sub_delims() + or self.__take(":") + ): pass if self.__peek("@"): @@ -1103,7 +1134,13 @@ def __ipv6_addrz(self) -> bool: """ start = self._index - if self.__ipv6_address() and self.__take("%") and self.__take("2") and self.__take("5") and self.__zone_id(): + if ( + self.__ipv6_address() + and self.__take("%") + and self.__take("2") + and self.__take("5") + and self.__zone_id() + ): return True self._index = start @@ -1186,7 +1223,11 @@ def __is_path_end(self) -> bool: > number sign ("#") character, or by the end of the URI. """ - return self._index >= len(self._string) or self._string[self._index] == "?" or self._string[self._index] == "#" + return ( + self._index >= len(self._string) + or self._string[self._index] == "?" + or self._string[self._index] == "#" + ) def __path_abempty(self) -> bool: """Determine whether the current position is a path-abempty. @@ -1329,7 +1370,12 @@ def __segment_nz_nc(self) -> bool: """ start = self._index - while self.__unreserved() or self.__pct_encoded() or self.__sub_delims() or self.__take("@"): + while ( + self.__unreserved() + or self.__pct_encoded() + or self.__sub_delims() + or self.__take("@") + ): pass if self._index - start > 0: @@ -1347,7 +1393,11 @@ def __pchar(self) -> bool: """ return ( - self.__unreserved() or self.__pct_encoded() or self.__sub_delims() or self.__take(":") or self.__take("@") + self.__unreserved() + or self.__pct_encoded() + or self.__sub_delims() + or self.__take(":") + or self.__take("@") ) def __query(self) -> bool: diff --git a/protovalidate/internal/legacy.py b/protovalidate/_legacy.py similarity index 85% rename from protovalidate/internal/legacy.py rename to protovalidate/_legacy.py index 34d5a12..9454e28 100644 --- a/protovalidate/internal/legacy.py +++ b/protovalidate/_legacy.py @@ -12,13 +12,19 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import typing -from google.protobuf import descriptor as google_descriptor -from google.protobuf import message as google_message from protobuf import Message from protobuf.wkt import FileDescriptorProto, FileDescriptorSet +if typing.TYPE_CHECKING: + from google.protobuf import ( + descriptor as google_descriptor, + message as google_message, + ) + class LegacyMessageConverter: """Copies google.protobuf messages into protobuf-py dynamic messages.""" @@ -35,7 +41,7 @@ def normalize(self, msg: Message | google_message.Message) -> Message: if isinstance(msg, Message): return msg # Normalize upb descriptor type - desc = typing.cast(google_descriptor.Descriptor, msg.DESCRIPTOR) + desc = typing.cast("google_descriptor.Descriptor", msg.DESCRIPTOR) cls = self._types.get(desc) if cls is None: cls = _message_class(desc) @@ -53,7 +59,9 @@ def _message_class(desc: google_descriptor.Descriptor) -> type[Message]: return result.type -def _collect_files(file: google_descriptor.FileDescriptor, out: dict[str, FileDescriptorProto]) -> None: +def _collect_files( + file: google_descriptor.FileDescriptor, out: dict[str, FileDescriptorProto] +) -> None: if file.name in out: return for dep in file.dependencies: diff --git a/protovalidate/internal/rules.py b/protovalidate/_rules.py similarity index 81% rename from protovalidate/internal/rules.py rename to protovalidate/_rules.py index 1dcfbe1..7b324e9 100644 --- a/protovalidate/internal/rules.py +++ b/protovalidate/_rules.py @@ -12,10 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import dataclasses import datetime import typing -from collections.abc import Callable, Iterable import celpy from celpy import celtypes @@ -39,9 +40,12 @@ ) from protobuf.wkt import Duration, FieldDescriptorProto, Timestamp +from protovalidate._cel_field_presence import InterpretedRunner, in_has +from protovalidate._core import CompilationError, RuleContext, Rules, Violation from protovalidate._gen.buf.validate import validate_pb -from protovalidate.internal._core import CompilationError, RuleContext, Rules, Violation -from protovalidate.internal.cel_field_presence import InterpretedRunner, in_has + +if typing.TYPE_CHECKING: + from collections.abc import Callable, Iterable _TYPE_CTORS: dict[ScalarType, Callable[..., celtypes.Value]] = { ScalarType.BOOL: celtypes.BoolType, @@ -76,7 +80,9 @@ def _scalar_zero(scalar: ScalarType) -> str | bytes | bool | float | int: return 0 -def _wire_type(field: DescField | DescExtension | ScalarType | DescMessage | DescEnum) -> FieldDescriptorProto.Type: +def _wire_type( + field: DescField | DescExtension | ScalarType | DescMessage | DescEnum, +) -> FieldDescriptorProto.Type: """The FieldDescriptorProto.Type for a field path element, reported in validation messages.""" match field: case ScalarType(): @@ -87,7 +93,10 @@ def _wire_type(field: DescField | DescExtension | ScalarType | DescMessage | Des return FieldDescriptorProto.Type.ENUM case _: match field.value: - case DescFieldValueMessage(delimited_encoding=True) | DescFieldValueList(delimited_encoding=True): + case ( + DescFieldValueMessage(delimited_encoding=True) + | DescFieldValueList(delimited_encoding=True) + ): return FieldDescriptorProto.Type.GROUP case _: return field.proto.type @@ -102,7 +111,9 @@ def _read_key(field: DescField | DescExtension) -> DescField | Extension: return field.type if isinstance(field, DescExtension) else field -def _scalar_of(subject: DescField | ScalarType | DescMessage | DescEnum) -> ScalarType | None: +def _scalar_of( + subject: DescField | ScalarType | DescMessage | DescEnum, +) -> ScalarType | None: match subject: case ScalarType(): return subject @@ -112,7 +123,9 @@ def _scalar_of(subject: DescField | ScalarType | DescMessage | DescEnum) -> Scal return None -def _message_of(subject: DescField | ScalarType | DescMessage | DescEnum) -> DescMessage | None: +def _message_of( + subject: DescField | ScalarType | DescMessage | DescEnum, +) -> DescMessage | None: match subject: case DescMessage(): return subject @@ -122,7 +135,9 @@ def _message_of(subject: DescField | ScalarType | DescMessage | DescEnum) -> Des return None -def _enum_of(subject: DescField | ScalarType | DescMessage | DescEnum) -> DescEnum | None: +def _enum_of( + subject: DescField | ScalarType | DescMessage | DescEnum, +) -> DescEnum | None: match subject: case DescEnum(): return subject @@ -137,13 +152,15 @@ def make_duration(msg: Duration) -> celtypes.DurationType: def make_timestamp(msg: Timestamp) -> celtypes.TimestampType: - return celtypes.TimestampType(1970, 1, 1) + celtypes.DurationType(seconds=msg.seconds, nanos=msg.nanos) + return celtypes.TimestampType(1970, 1, 1) + celtypes.DurationType( + seconds=msg.seconds, nanos=msg.nanos + ) class MessageConverter: """Converts protobuf-py values into celpy celtypes for CEL evaluation.""" - def __init__(self): + def __init__(self) -> None: self._fields_by_name: dict[DescMessage, dict[str, DescField]] = {} # Well-known types convert to native CEL values rather than maps. self._wkt: dict[str, Callable[..., celtypes.Value]] = { @@ -195,7 +212,9 @@ def field_value(self, val: typing.Any, kind: DescFieldValue) -> celtypes.Value: case DescFieldValueScalar(scalar=scalar): return self.scalar(val, scalar) - def scalar(self, val: typing.Any, kind: ScalarType | DescMessage | DescEnum) -> celtypes.Value: + def scalar( + self, val: typing.Any, kind: ScalarType | DescMessage | DescEnum + ) -> celtypes.Value: match kind: case DescMessage(): return self.message(val) @@ -230,7 +249,7 @@ class MessageType(celtypes.MapType): msg: Message - def __init__(self, msg: Message, conv: MessageConverter): + def __init__(self, msg: Message, conv: MessageConverter) -> None: super().__init__() self.msg = msg self._conv = conv @@ -245,12 +264,13 @@ def convert_field(self, field: DescField) -> celtypes.Value: """Convert one of this message's fields to its CEL value.""" return self._conv.field(self.msg, field) - def __getitem__(self, key): - field = self._fields[key] - if field not in self.msg: - if in_has(): - raise KeyError - return self._conv.zero(field) + def __getitem__(self, key: str | celtypes.Value) -> celtypes.Value: + if isinstance(key, str): + field = self._fields[key] + if field not in self.msg: + if in_has(): + raise KeyError + return self._conv.zero(field) return super().__getitem__(key) @@ -262,7 +282,9 @@ def _field_to_element(field: DescField | DescExtension) -> validate_pb.FieldPath ) -def _indexed_field_element(field: DescField, index: int) -> validate_pb.FieldPathElement: +def _indexed_field_element( + field: DescField, index: int +) -> validate_pb.FieldPathElement: return validate_pb.FieldPathElement( field_number=field.number, field_name=field.name, @@ -291,7 +313,12 @@ def _map_key_element(field: DescField, key: typing.Any) -> validate_pb.FieldPath | ScalarType.SINT64 ): subscript = Oneof(field="int_key", value=key) - case ScalarType.UINT32 | ScalarType.FIXED32 | ScalarType.UINT64 | ScalarType.FIXED64: + case ( + ScalarType.UINT32 + | ScalarType.FIXED32 + | ScalarType.UINT64 + | ScalarType.FIXED64 + ): subscript = Oneof(field="uint_key", value=key) case ScalarType.STRING: subscript = Oneof(field="string_key", value=key) @@ -340,7 +367,7 @@ def __init__( env: celpy.Environment, funcs: dict[str, celpy.CELFunction], conv: MessageConverter, - ): + ) -> None: self._env = env self._funcs = funcs self._conv = conv @@ -356,7 +383,7 @@ def _validate_cel( this_value: typing.Any | None = None, this_cel: celtypes.Value | None = None, for_key: bool = False, - ): + ) -> None: if not self._cel: return activation: dict[str, celtypes.Value] = {} @@ -364,7 +391,9 @@ def _validate_cel( activation["this"] = this_cel activation["rules"] = self._rules_cel if self._uses_now: - activation["now"] = celtypes.TimestampType(datetime.datetime.now(tz=datetime.timezone.utc)) + activation["now"] = celtypes.TimestampType( + datetime.datetime.now(tz=datetime.timezone.utc) + ) for cel in self._cel: activation["rule"] = cel.rule_cel result = cel.runner.evaluate(activation) @@ -381,7 +410,7 @@ def _validate_cel( rule_id=cel.rule.id, message=message, for_key=for_key, - ), + ) ) case celtypes.StringType() if result: ctx.add( @@ -392,7 +421,7 @@ def _validate_cel( rule_id=cel.rule.id, message=result, for_key=for_key, - ), + ) ) case Exception(): raise result @@ -403,7 +432,7 @@ def add_rule( *, rule_field: DescField | DescExtension | None = None, rule_path: validate_pb.FieldPath | None = None, - ): + ) -> None: if isinstance(rules, str): expression = rules rules = validate_pb.Rule(id=expression, expression=expression) @@ -430,11 +459,11 @@ def add_rule( class MessageOneofRule(Rules): """Validates a single buf.validate.MessageOneofRule given via the (buf.validate.message).oneof option.""" - def __init__(self, fields: list[DescField], *, required: bool): + def __init__(self, fields: list[DescField], *, required: bool) -> None: self._fields = fields self._required = required - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: num_set_fields = sum(1 for field in self._fields if field in message) if num_set_fields > 1: ctx.add( @@ -465,12 +494,12 @@ def __init__( env: celpy.Environment, funcs: dict[str, celpy.CELFunction], conv: MessageConverter, - ): + ) -> None: super().__init__(rules, env=env, funcs=funcs, conv=conv) self._oneofs = [] self._desc = desc - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: if self._cel: self._validate_cel(ctx, this_cel=self._conv.message(message)) if ctx.done: @@ -480,7 +509,7 @@ def validate(self, ctx: RuleContext, message: Message): if ctx.done: return - def add_oneof(self, rule: validate_pb.MessageOneofRule): + def add_oneof(self, rule: validate_pb.MessageOneofRule) -> None: fields = [] seen = set() if len(rule.fields) == 0: @@ -525,7 +554,9 @@ def add_oneof(self, rule: validate_pb.MessageOneofRule): } -def _type_mismatch(subject: DescField | ScalarType | DescMessage | DescEnum, expected: str) -> CompilationError: +def _type_mismatch( + subject: DescField | ScalarType | DescMessage | DescEnum, expected: str +) -> CompilationError: actual = _wire_type(subject).name.lower() name = subject.name if isinstance(subject, DescField) else actual return CompilationError(f"field {name} has type {actual} but expected {expected}") @@ -537,10 +568,14 @@ def _check_field_type( wrapper_name: str | None = None, ) -> None: if (expected is not None and _scalar_of(subject) == expected) or ( - (message := _message_of(subject)) is not None and message.type_name == wrapper_name + (message := _message_of(subject)) is not None + and message.type_name == wrapper_name ): return - raise _type_mismatch(subject, (wrapper_name or "message") if expected is None else expected.name.lower()) + raise _type_mismatch( + subject, + (wrapper_name or "message") if expected is None else expected.name.lower(), + ) class FieldRules(CelRules): @@ -564,7 +599,7 @@ def __init__( force_ignore_empty: bool = False, registry: Registry | None = None, conv: MessageConverter, - ): + ) -> None: type_oneof = field_level.type type_case = type_oneof.field if type_oneof is not None else None rules_pb = type_oneof.value if type_oneof is not None else None @@ -574,7 +609,11 @@ def __init__( field_level.ignore == validate_pb.Ignore.IF_ZERO_VALUE or force_ignore_empty # A presence-tracking field (not a map/list element) ignores empty by default. - or (not for_items and isinstance(field, DescField) and field.presence.name != "IMPLICIT") + or ( + not for_items + and isinstance(field, DescField) + and field.presence.name != "IMPLICIT" + ) ) self._required = field_level.required if rules_pb is not None: @@ -595,7 +634,10 @@ def __init__( cel, rule_field=rule_field, rule_path=validate_pb.FieldPath( - elements=[_field_to_element(rule_field), _field_to_element(type_field)] + elements=[ + _field_to_element(rule_field), + _field_to_element(type_field), + ] ), ) # Custom predefined rules are extensions on the rules message, @@ -617,18 +659,28 @@ def __init__( cel, rule_field=ext_field, rule_path=validate_pb.FieldPath( - elements=[_field_to_element(ext_field), _field_to_element(type_field)] + elements=[ + _field_to_element(ext_field), + _field_to_element(type_field), + ] ), ) cel_expression_field = _spec_field(validate_pb.FieldRules, "cel_expression") for i, cel in enumerate(field_level.cel_expression): self.add_rule( cel, - rule_path=validate_pb.FieldPath(elements=[_indexed_field_element(cel_expression_field, i)]), + rule_path=validate_pb.FieldPath( + elements=[_indexed_field_element(cel_expression_field, i)] + ), ) cel_field = _spec_field(validate_pb.FieldRules, "cel") for i, cel in enumerate(field_level.cel): - self.add_rule(cel, rule_path=validate_pb.FieldPath(elements=[_indexed_field_element(cel_field, i)])) + self.add_rule( + cel, + rule_path=validate_pb.FieldPath( + elements=[_indexed_field_element(cel_field, i)] + ), + ) @property def _read_field(self) -> DescField: @@ -637,18 +689,20 @@ def _read_field(self) -> DescField: assert isinstance(self._field, DescField) # noqa: S101 return self._field - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: field = self._read_field if field not in message: if self._required: ctx.add( Violation( - field=validate_pb.FieldPath(elements=[_field_to_element(field)]), + field=validate_pb.FieldPath( + elements=[_field_to_element(field)] + ), rule=FieldRules._required_rule_path, rule_value=self._required, rule_id="required", message="value is required", - ), + ) ) return if self._ignore_empty: @@ -669,11 +723,18 @@ def validate_item( item_field: ScalarType | DescMessage | DescEnum, *, for_key: bool = False, - ): + ) -> None: self._validate_value(ctx, value, for_key=for_key) - self._validate_cel(ctx, this_value=value, this_cel=self._conv.scalar(value, item_field), for_key=for_key) + self._validate_cel( + ctx, + this_value=value, + this_cel=self._conv.scalar(value, item_field), + for_key=for_key, + ) - def _validate_value(self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False): + def _validate_value( + self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False + ) -> None: pass @@ -684,14 +745,14 @@ class AnyRules(FieldRules): elements=[ _field_to_element(_spec_field(validate_pb.AnyRules, "in")), _field_to_element(_spec_field(validate_pb.FieldRules, "any")), - ], + ] ) _not_in_rule_path: typing.ClassVar[validate_pb.FieldPath] = validate_pb.FieldPath( elements=[ _field_to_element(_spec_field(validate_pb.AnyRules, "not_in")), _field_to_element(_spec_field(validate_pb.FieldRules, "any")), - ], + ] ) def __init__( @@ -703,15 +764,18 @@ def __init__( *, registry: Registry | None = None, conv: MessageConverter, - ): + ) -> None: super().__init__(env, funcs, field, field_level, registry=registry, conv=conv) type_oneof = field_level.type - assert type_oneof is not None and type_oneof.field == "any" # noqa: S101 + assert type_oneof is not None # noqa: S101 + assert type_oneof.field == "any" # noqa: S101 any_rules = type_oneof.value self._in: list[str] = list(any_rules.in_) self._not_in: list[str] = list(any_rules.not_in) - def _validate_value(self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False): + def _validate_value( + self, ctx: RuleContext, value: typing.Any, *, for_key: bool = False + ) -> None: if len(self._in) > 0 and value.type_url not in self._in: ctx.add( Violation( @@ -739,11 +803,13 @@ class EnumRules(FieldRules): _defined_only = False - _defined_only_rule_path: typing.ClassVar[validate_pb.FieldPath] = validate_pb.FieldPath( - elements=[ - _field_to_element(_spec_field(validate_pb.EnumRules, "defined_only")), - _field_to_element(_spec_field(validate_pb.FieldRules, "enum")), - ], + _defined_only_rule_path: typing.ClassVar[validate_pb.FieldPath] = ( + validate_pb.FieldPath( + elements=[ + _field_to_element(_spec_field(validate_pb.EnumRules, "defined_only")), + _field_to_element(_spec_field(validate_pb.FieldRules, "enum")), + ] + ) ) def __init__( @@ -757,7 +823,7 @@ def __init__( force_ignore_empty: bool = False, registry: Registry | None = None, conv: MessageConverter, - ): + ) -> None: super().__init__( env, funcs, @@ -769,7 +835,8 @@ def __init__( conv=conv, ) type_oneof = field_level.type - assert type_oneof is not None and type_oneof.field == "enum" # noqa: S101 + assert type_oneof is not None # noqa: S101 + assert type_oneof.field == "enum" # noqa: S101 if type_oneof.value.defined_only: self._defined_only = True enum = _enum_of(field) @@ -777,7 +844,7 @@ def __init__( assert enum is not None # noqa: S101 self._defined_numbers = {v.number for v in enum.values} - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: super().validate(ctx, message) if ctx.done: return @@ -790,7 +857,7 @@ def validate(self, ctx: RuleContext, message: Message): rule_value=self._defined_only, rule_id="enum.defined_only", message="value must be one of the defined enum values", - ), + ) ) @@ -814,12 +881,12 @@ def __init__( *, registry: Registry | None = None, conv: MessageConverter, - ): + ) -> None: super().__init__(env, funcs, field, field_level, registry=registry, conv=conv) if item_rules is not None: self._item_rules = item_rules - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: super().validate(ctx, message) if ctx.done: return @@ -868,14 +935,14 @@ def __init__( *, registry: Registry | None = None, conv: MessageConverter, - ): + ) -> None: super().__init__(env, funcs, field, field_level, registry=registry, conv=conv) if key_rules is not None: self._key_rules = key_rules if value_rules is not None: self._value_rules = value_rules - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: super().validate(ctx, message) if ctx.done: return @@ -889,7 +956,9 @@ def validate(self, ctx: RuleContext, message: Message): if key_ctx.has_errors(): key_ctx.add_rule_path_elements(MapRules._key_rules_suffix) map_ctx = ctx.sub_context() - if self._value_rules is not None and (not self._value_rules._ignore_empty or v): + if self._value_rules is not None and ( + not self._value_rules._ignore_empty or v + ): self._value_rules.validate_item(map_ctx, v, value.value) if map_ctx.has_errors(): map_ctx.add_rule_path_elements(MapRules._value_rules_suffix) @@ -904,17 +973,19 @@ class OneofRules(Rules): required = True - def __init__(self, oneof: DescOneof, rules: validate_pb.OneofRules): + def __init__(self, oneof: DescOneof, rules: validate_pb.OneofRules) -> None: self._oneof = oneof if not rules.required: self.required = False - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: if getattr(message, self._oneof.local_name) is None: if self.required: ctx.add( Violation( - field=validate_pb.FieldPath(elements=[_oneof_to_element(self._oneof)]), + field=validate_pb.FieldPath( + elements=[_oneof_to_element(self._oneof)] + ), rule_id="required", message="exactly one field is required in oneof", ) @@ -923,8 +994,7 @@ def validate(self, ctx: RuleContext, message: Message): def _message_child(field: DescField) -> DescMessage | None: - """The sub-message a field recurses into: a map value, list element, or - singular message, if message-typed.""" + """The sub-message a field recurses into.""" match field.value: case DescFieldValueMap(value=DescMessage() as message): return message @@ -942,7 +1012,9 @@ class RuleFactory: _env: celpy.Environment _funcs: dict[str, celpy.CELFunction] - def __init__(self, funcs: dict[str, celpy.CELFunction], registry: Registry | None = None): + def __init__( + self, funcs: dict[str, celpy.CELFunction], registry: Registry | None = None + ) -> None: self._env = celpy.Environment(runner_class=InterpretedRunner) self._funcs = funcs self._registry = registry @@ -953,15 +1025,19 @@ def get(self, desc: DescMessage) -> list[Rules]: if desc not in self._cache: try: self._cache[desc] = self._new_rules(desc) - except Exception as e: + except Exception as e: # noqa: BLE001 self._cache[desc] = e result = self._cache[desc] if isinstance(result, Exception): raise result return result - def _new_message_rule(self, rules: validate_pb.MessageRules, desc: DescMessage) -> MessageRules: - result = MessageRules(rules, desc, env=self._env, funcs=self._funcs, conv=self._conv) + def _new_message_rule( + self, rules: validate_pb.MessageRules, desc: DescMessage + ) -> MessageRules: + result = MessageRules( + rules, desc, env=self._env, funcs=self._funcs, conv=self._conv + ) for oneof in rules.oneof: result.add_oneof(oneof) for expr in rules.cel_expression: @@ -996,7 +1072,14 @@ def _new_scalar_field_rule( return EnumRules(self._env, self._funcs, field, field_level, **kw) case "any": _check_field_type(field, None, "google.protobuf.Any") - return AnyRules(self._env, self._funcs, field, field_level, registry=self._registry, conv=self._conv) + return AnyRules( + self._env, + self._funcs, + field, + field_level, + registry=self._registry, + conv=self._conv, + ) case _ if type_case in _RULE_FIELD_TYPES: expected, wrapper = _RULE_FIELD_TYPES[type_case] _check_field_type(field, expected, wrapper) @@ -1006,18 +1089,30 @@ def _new_scalar_field_rule( raise CompilationError(msg) def _new_field_rule( - self, field: DescField, rules: validate_pb.FieldRules, *, force_ignore_empty: bool = False + self, + field: DescField, + rules: validate_pb.FieldRules, + *, + force_ignore_empty: bool = False, ) -> FieldRules | None: type_oneof = rules.type match field.value: case DescFieldValueMap() as value: - map_rules = type_oneof.value if type_oneof is not None and type_oneof.field == "map" else None + map_rules = ( + type_oneof.value + if type_oneof is not None and type_oneof.field == "map" + else None + ) key_rules = None value_rules = None if map_rules is not None and map_rules.keys is not None: - key_rules = self._new_scalar_field_rule(value.key, map_rules.keys, for_items=True) + key_rules = self._new_scalar_field_rule( + value.key, map_rules.keys, for_items=True + ) if map_rules is not None and map_rules.values is not None: - value_rules = self._new_scalar_field_rule(value.value, map_rules.values, for_items=True) + value_rules = self._new_scalar_field_rule( + value.value, map_rules.values, for_items=True + ) return MapRules( self._env, self._funcs, @@ -1030,14 +1125,28 @@ def _new_field_rule( ) case DescFieldValueList() as value: item_rule = None - rep_rules = type_oneof.value if type_oneof is not None and type_oneof.field == "repeated" else None + rep_rules = ( + type_oneof.value + if type_oneof is not None and type_oneof.field == "repeated" + else None + ) if rep_rules is not None and rep_rules.items is not None: - item_rule = self._new_scalar_field_rule(value.element, rep_rules.items) + item_rule = self._new_scalar_field_rule( + value.element, rep_rules.items + ) return RepeatedRules( - self._env, self._funcs, field, rules, item_rule, registry=self._registry, conv=self._conv + self._env, + self._funcs, + field, + rules, + item_rule, + registry=self._registry, + conv=self._conv, ) case _: - return self._new_scalar_field_rule(field, rules, force_ignore_empty=force_ignore_empty) + return self._new_scalar_field_rule( + field, rules, force_ignore_empty=force_ignore_empty + ) def _new_rules(self, desc: DescMessage) -> list[Rules]: result: list[Rules] = [] @@ -1063,15 +1172,23 @@ def _new_rules(self, desc: DescMessage) -> list[Rules]: if field_opts is not None and validate_pb.ext_field in field_opts: field_level = field_opts[validate_pb.ext_field] if field_level is not None: - force_ignore_empty = ignore_field not in field_level and field.name in all_msg_oneof_fields + force_ignore_empty = ( + ignore_field not in field_level + and field.name in all_msg_oneof_fields + ) if field_level.ignore == validate_pb.Ignore.ALWAYS: continue - if field_rule := self._new_field_rule(field, field_level, force_ignore_empty=force_ignore_empty): + if field_rule := self._new_field_rule( + field, field_level, force_ignore_empty=force_ignore_empty + ): result.append(field_rule) type_oneof = field_level.type if type_oneof is not None and type_oneof.field == "repeated": rep = type_oneof.value - if rep.items is not None and rep.items.ignore == validate_pb.Ignore.ALWAYS: + if ( + rep.items is not None + and rep.items.ignore == validate_pb.Ignore.ALWAYS + ): continue sub_desc = _message_child(field) if sub_desc is None: @@ -1087,16 +1204,24 @@ def _new_rules(self, desc: DescMessage) -> list[Rules]: class _SubMessageRule(Rules): - """Recurses into a message-typed field's own rules. Subclasses supply how to - enumerate the sub-messages to validate (singular, map values, list items) - as (sub_message, field_path_element) pairs.""" + """Recurses into a message-typed field's own rules. + + Subclasses supply how to enumerate the sub-messages to validate (singular, map values, list items) + as (sub_message, field_path_element) pairs. + """ - def __init__(self, factory: RuleFactory, field: DescField, sub_desc: DescMessage): + def __init__( + self, factory: RuleFactory, field: DescField, sub_desc: DescMessage + ) -> None: self._factory = factory self._field = field self._sub_desc = sub_desc - def _validate_each(self, ctx: RuleContext, items: Iterable[tuple[Message, validate_pb.FieldPathElement]]): + def _validate_each( + self, + ctx: RuleContext, + items: Iterable[tuple[Message, validate_pb.FieldPathElement]], + ) -> None: rules = self._factory.get(self._sub_desc) if not rules: return @@ -1110,18 +1235,28 @@ def _validate_each(self, ctx: RuleContext, items: Iterable[tuple[Message, valida class SubMsgRule(_SubMessageRule): - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: if self._field in message: - self._validate_each(ctx, [(message[self._field], _field_to_element(self._field))]) + self._validate_each( + ctx, [(message[self._field], _field_to_element(self._field))] + ) class MapValMsgRule(_SubMessageRule): - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: val = message[self._field] - self._validate_each(ctx, ((v, _map_key_element(self._field, k)) for k, v in val.items())) + self._validate_each( + ctx, ((v, _map_key_element(self._field, k)) for k, v in val.items()) + ) class RepeatedMsgRule(_SubMessageRule): - def validate(self, ctx: RuleContext, message: Message): + def validate(self, ctx: RuleContext, message: Message) -> None: val = message[self._field] - self._validate_each(ctx, ((item, _indexed_field_element(self._field, i)) for i, item in enumerate(val))) + self._validate_each( + ctx, + ( + (item, _indexed_field_element(self._field, i)) + for i, item in enumerate(val) + ), + ) diff --git a/protovalidate/internal/string_format.py b/protovalidate/_string_format.py similarity index 94% rename from protovalidate/internal/string_format.py rename to protovalidate/_string_format.py index a5d952c..e40eb2a 100644 --- a/protovalidate/internal/string_format.py +++ b/protovalidate/_string_format.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import math import re from decimal import Decimal @@ -23,12 +25,14 @@ class StringFormat: """An implementation of string.format() in CEL.""" - def __init__(self): + def __init__(self) -> None: self.fmt = None def format(self, fmt: celtypes.Value, args: celtypes.Value) -> celpy.Result: if not isinstance(fmt, celtypes.StringType): - return celpy.CELEvalError("format() requires a string as the first argument") + return celpy.CELEvalError( + "format() requires a string as the first argument" + ) if not isinstance(args, celtypes.ListType): return celpy.CELEvalError("format() requires a list as the second argument") # printf style formatting @@ -87,7 +91,9 @@ def format(self, fmt: celtypes.Value, args: celtypes.Value) -> celpy.Result: return celtypes.StringType(result) - def __validate_number(self, arg: celtypes.DoubleType | celtypes.IntType | celtypes.UintType) -> str | None: + def __validate_number( + self, arg: celtypes.DoubleType | celtypes.IntType | celtypes.UintType + ) -> str | None: if math.isnan(arg): return "NaN" if math.isinf(arg): @@ -183,7 +189,7 @@ def __format_string(self, arg: celtypes.Value) -> str: return f"{arg:g}" if isinstance(arg, celtypes.DurationType): return self.__format_duration(arg) - if isinstance(arg, celtypes.IntType) or isinstance(arg, celtypes.UintType): + if isinstance(arg, (celtypes.IntType, celtypes.UintType)): result = self.__validate_number(arg) if result is not None: return result @@ -205,7 +211,10 @@ def __format_list(self, arg: celtypes.ListType) -> str: return "[" + ", ".join(self.__format_string(val) for val in arg) + "]" def __format_map(self, arg: celtypes.MapType) -> str: - m = {self.__format_string(cel_key): self.__format_string(cel_val) for cel_key, cel_val in arg.items()} + m = { + self.__format_string(cel_key): self.__format_string(cel_val) + for cel_key, cel_val in arg.items() + } return "{" + ", ".join(key + ": " + val for key, val in sorted(m.items())) + "}" def __format_duration(self, arg: celtypes.DurationType) -> str: diff --git a/protovalidate/validator.py b/protovalidate/_validator.py similarity index 67% rename from protovalidate/validator.py rename to protovalidate/_validator.py index c51a76d..ed01f2c 100644 --- a/protovalidate/validator.py +++ b/protovalidate/_validator.py @@ -19,32 +19,41 @@ from protobuf import Message, Registry +from protovalidate import _backend, _extra_func, _rules +from protovalidate._core import CompilationError, RuleContext, Violation from protovalidate._gen.buf.validate import validate_pb -from protovalidate.internal import backend, extra_func -from protovalidate.internal import rules as _rules -from protovalidate.internal._core import CompilationError, RuleContext, Violation if TYPE_CHECKING: from google.protobuf import message as google_message -__all__ = ["CompilationError", "ValidationError", "Validator", "Violation", "Violations"] +__all__ = [ + "CompilationError", + "ValidationError", + "Validator", + "Violation", + "Violations", +] Violations = validate_pb.Violations class _Engine(typing.Protocol): """A CEL backend: turns a protobuf-py message into a list of violations.""" - def collect_violations(self, message: Message, *, fail_fast: bool) -> list[Violation]: ... + def collect_violations( + self, message: Message, *, fail_fast: bool + ) -> list[Violation]: ... class _CelpyEngine: """The pure-Python celpy engine, evaluating directly over protobuf-py.""" - def __init__(self, registry: Registry | None): - self._factory = _rules.RuleFactory(extra_func.make_extra_funcs(), registry) + def __init__(self, registry: Registry | None) -> None: + self._factory = _rules.RuleFactory(_extra_func.make_extra_funcs(), registry) - def collect_violations(self, message: Message, *, fail_fast: bool) -> list[Violation]: + def collect_violations( + self, message: Message, *, fail_fast: bool + ) -> list[Violation]: ctx = RuleContext(fail_fast=fail_fast) for rule in self._factory.get(type(message).desc()): rule.validate(ctx, message) @@ -65,13 +74,15 @@ class _CelExprEngine: global google pool the bridge mirrors into. """ - def __init__(self, registry: Registry | None): # noqa: ARG002 - accepted for API symmetry - from protovalidate.internal import celexpr # noqa: PLC0415 - optional dependency + def __init__(self, registry: Registry | None) -> None: # noqa: ARG002 - accepted for API symmetry + from protovalidate import _celexpr # noqa: PLC0415 - self._bridge = celexpr.GoogleBridge() - self._factory = celexpr.RuleFactory(celexpr.make_extension(), self._bridge) + self._bridge = _celexpr.GoogleBridge() + self._factory = _celexpr.RuleFactory(_celexpr.make_extension(), self._bridge) - def collect_violations(self, message: Message, *, fail_fast: bool) -> list[Violation]: + def collect_violations( + self, message: Message, *, fail_fast: bool + ) -> list[Violation]: bridged = self._bridge.to_google(message) ctx = RuleContext(fail_fast=fail_fast) for rule in self._factory.get(type(message).desc()): @@ -84,8 +95,7 @@ def collect_violations(self, message: Message, *, fail_fast: bool) -> list[Viola class Validator: - """ - Validates Protobuf messages against static rules. + """Validates Protobuf messages against static rules. Both protobuf-py messages and legacy google.protobuf messages are accepted. A google.protobuf message is validated by copying it into a @@ -99,8 +109,9 @@ class Validator: _engine: _Engine - def __init__(self, registry: Registry | None = None): - """ + def __init__(self, registry: Registry | None = None) -> None: + """Creates a new validator. + Parameters: registry: An optional Registry used to resolve custom predefined-rule extensions. If omitted, only standard rules are applied. @@ -108,24 +119,25 @@ def __init__(self, registry: Registry | None = None): try: import google.protobuf.message # noqa: F401, PLC0415 - from protovalidate.internal.legacy import LegacyMessageConverter # noqa: PLC0415 + from protovalidate._legacy import LegacyMessageConverter # noqa: PLC0415 self._legacy = LegacyMessageConverter() except ImportError: self._legacy = None - if backend.CEL_EXPR_AVAILABLE: + if _backend.CEL_EXPR_AVAILABLE: self._engine = _CelExprEngine(registry) else: self._engine = _CelpyEngine(registry) - def validate(self, message: Message | google_message.Message, *, fail_fast: bool = False): - """ - Validates the given message against the static rules defined in - the message's descriptor. + def validate( + self, message: Message | google_message.Message, *, fail_fast: bool = False + ) -> None: + """Validates the given message against the static rules defined in the message's descriptor. Parameters: message: The message to validate. fail_fast: If true, validation will stop after the first iteration. + Raises: CompilationError: If the static rules could not be compiled. ValidationError: If the message is invalid. The violations raised as part of this error should @@ -138,16 +150,12 @@ def validate(self, message: Message | google_message.Message, *, fail_fast: bool raise ValidationError(msg, violations) def collect_violations( - self, - message: Message | google_message.Message, - *, - fail_fast: bool = False, + self, message: Message | google_message.Message, *, fail_fast: bool = False ) -> list[Violation]: - """ - Validates the given message against the static rules defined in - the message's descriptor. Compared to `validate`, `collect_violations` simply - returns the violations as a list and puts the burden of raising an appropriate - exception on the caller. + """Validates the given message against the static rules defined in the message's descriptor. + + Compared to `validate`, `collect_violations` simply returns the violations as a list and puts + the burden of raising an appropriate exception on the caller. The violations returned from this method should always be equal to the violations raised as part of the ValidationError in the call to `validate`. @@ -155,6 +163,7 @@ def collect_violations( Parameters: message: The message to validate. fail_fast: If true, validation will stop after the first iteration. + Raises: CompilationError: If the static rules could not be compiled. """ @@ -169,26 +178,21 @@ def _coerce(self, message: Message | google_message.Message) -> Message: class ValidationError(ValueError): - """ - An error raised when a message fails to validate. - """ + """An error raised when a message fails to validate.""" _violations: list[Violation] - def __init__(self, msg: str, violations: list[Violation]): + def __init__(self, msg: str, violations: list[Violation]) -> None: super().__init__(msg) self._violations = violations def to_proto(self) -> validate_pb.Violations: - """ - Provides the Protobuf form of the validation errors. - """ - return validate_pb.Violations(violations=[violation.proto for violation in self._violations]) + """Provides the Protobuf form of the validation errors.""" + return validate_pb.Violations( + violations=[violation.proto for violation in self._violations] + ) @property def violations(self) -> list[Violation]: - """ - Provides the validation errors as a simple Python list, rather than the - Protobuf-specific collection type used by Violations. - """ + """Returns the violation errors.""" return self._violations diff --git a/pyproject.toml b/pyproject.toml index eb1d2f5..9af31e2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,52 +97,71 @@ strict = true testpaths = ["test"] [tool.ruff] -line-length = 120 -exclude = [ - # Protobuf generated code - "test/gen/**", -] +target-version = "py310" +exclude = ["*_pb2.py", "*_pb2.pyi"] + +[tool.ruff.format] +skip-magic-trailing-comma = true +docstring-code-format = true -lint.select = [ - "A", - "ARG", - "B", - "C", - "DTZ", - "E", - "EM", - "F", - "FBT", - "I", - "ICN", - "N", - "PLC", - "PLE", - "PLR", - "PLW", +[tool.ruff.lint] +typing-extensions = false +select = ["ALL"] +ignore = [ + # Rules recommended to ignore when using ruff formatting + # https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules + "W191", + "E111", + "E114", + "E117", + "E501", + "D206", + "D300", "Q", - "RUF", - "S", - "T", - "TID", - "UP", - "W", - "YTT", + "COM812", + "COM819", + + "TRY003", # Avoid specifying long messages outside the exception class + "SLF001", # Private member accessed + "ANN401", # Dynamically typed expressions (typing.Any) are disallowed + "PLR2004", # Magic value used in comparison - not helpful to name obvious constants + # Complexity rules: humans reviewing code can judge whether complexity is justified. + "C901", # McCabe complexity + "PLR0911", # Too many return statements + "PLR0912", # Too many branches + "PLR0913", # Too many arguments in function definition + "PLR0914", # Too many local variables + "PLR0915", # Too many statements + "PLR0916", # Too many boolean expressions + "PLR1702", # Too many nested blocks + + # TODO rules + "TD", + "FIX002", # Line contains TODO, consider resolving the issue ] -lint.ignore = [ - # Ignore complexity - "C901", - "PLR0911", - "PLR0912", - "PLR0913", - "PLR0915", - # Ignore magic values - in this library, most are obvious in context. - "PLR2004", + +[tool.ruff.lint.per-file-ignores] +"**/test/**/*.py" = [ + # missing docstrings; unnecessary in tests. + "D1", + + "INP001", # File is part of an implicit namespace package + + "FBT001", # boolean positional args are fine in tests + "FBT002", # boolean default positional args are fine in tests + "FBT003", # boolean positional values are fine in tests + "S101", # assert is expected in tests +] +"**/scripts/*.py" = [ + # missing docstrings; unnecessary in scripts. + "D1", ] [tool.ruff.lint.isort] -known-first-party = ["protovalidate", "buf"] +required-imports = ["from __future__ import annotations"] +combine-as-imports = true +known-third-party = [] +split-on-trailing-comma = false -[tool.ruff.lint.per-file-ignores] -# Tests can use assertions. -"test/**/*" = ["S101"] +[tool.ruff.lint.pydocstyle] +convention = "google" diff --git a/protovalidate/internal/__init__.py b/scripts/__init__.py similarity index 100% rename from protovalidate/internal/__init__.py rename to scripts/__init__.py diff --git a/scripts/generate_cel.py b/scripts/generate_cel.py index b4ea016..b18313e 100644 --- a/scripts/generate_cel.py +++ b/scripts/generate_cel.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import shutil import subprocess from pathlib import Path @@ -25,8 +27,7 @@ def main() -> None: - """Generates CEL conformance protos and fetches CEL testdata""" - + """Generates CEL conformance protos and fetches CEL testdata.""" shutil.rmtree(test_dir / "gen" / "cel", ignore_errors=True) subprocess.run( # noqa: S603 @@ -48,7 +49,6 @@ def main() -> None: ) as res: out_path = test_dir / "testdata" / f"string_ext_{CEL_SPEC_VERSION}.textproto" out_path.parent.mkdir(parents=True, exist_ok=True) - with open(out_path, "wb") as f: - f.write(res.read()) + out_path.write_bytes(res.read()) fix_protobuf_imports.callback(test_dir / "gen", dry=False) diff --git a/scripts/generate_protovalidate.py b/scripts/generate_protovalidate.py index b69d1cf..0091d05 100644 --- a/scripts/generate_protovalidate.py +++ b/scripts/generate_protovalidate.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import re import shutil import subprocess @@ -21,15 +23,16 @@ def main() -> None: + """Generate protovalidate and protovalidate-testing proto stubs.""" if re.match(r"^v\d+\.\d+\.\d+(\-.+)?$", PROTOVALIDATE_VERSION): # Version tag, fetch from BSR protovalidate_path = f"buf.build/bufbuild/protovalidate:{PROTOVALIDATE_VERSION}" - protovalidate_testing_path = f"buf.build/bufbuild/protovalidate-testing:{PROTOVALIDATE_VERSION}" + protovalidate_testing_path = ( + f"buf.build/bufbuild/protovalidate-testing:{PROTOVALIDATE_VERSION}" + ) else: # Not a tag, generally an unreleased commit, fetch directly from git - protovalidate_path = ( - f"https://github.com/bufbuild/protovalidate.git#subdir=proto/protovalidate,ref={PROTOVALIDATE_VERSION}" - ) + protovalidate_path = f"https://github.com/bufbuild/protovalidate.git#subdir=proto/protovalidate,ref={PROTOVALIDATE_VERSION}" protovalidate_testing_path = f"https://github.com/bufbuild/protovalidate.git#subdir=proto/protovalidate-testing,ref={PROTOVALIDATE_VERSION}" repo = Path(__file__).parent.parent diff --git a/test/conformance/runner.py b/test/conformance/runner.py index 81dcf8e..76cc7c7 100644 --- a/test/conformance/runner.py +++ b/test/conformance/runner.py @@ -12,20 +12,23 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import os import sys import celpy import protobuf -from google.protobuf import descriptor_pb2 as google_descriptor_pb2 -from google.protobuf import descriptor_pool as google_descriptor_pool -from google.protobuf import message as google_message -from google.protobuf import message_factory as google_message_factory -from protobuf import Oneof, Registry -from protobuf import wkt as pb_wkt +from google.protobuf import ( + descriptor_pb2 as google_descriptor_pb2, + descriptor_pool as google_descriptor_pool, + message as google_message, + message_factory as google_message_factory, +) +from protobuf import Oneof, Registry, wkt as pb_wkt import protovalidate -from protovalidate.internal import backend +from protovalidate import _backend from ..gen.buf.validate import validate_pb # noqa: TID252 from ..gen.buf.validate.conformance.harness.harness_pb import ( # noqa: TID252 @@ -38,10 +41,12 @@ _LEGACY = os.environ.get("PROTOVALIDATE_CONFORMANCE_LEGACY") == "1" if os.environ.get("PROTOVALIDATE_CONFORMANCE_BACKEND") == "celpy": - backend.CEL_EXPR_AVAILABLE = False + _backend.CEL_EXPR_AVAILABLE = False -def build_google_pool(fdset: pb_wkt.FileDescriptorSet) -> google_descriptor_pool.DescriptorPool: +def build_google_pool( + fdset: pb_wkt.FileDescriptorSet, +) -> google_descriptor_pool.DescriptorPool: pool = google_descriptor_pool.DescriptorPool() by_name = {file.name: file for file in fdset.file} added: set[str] = set() @@ -53,28 +58,33 @@ def add(name: str) -> None: added.add(name) for dep in proto.dependency: add(dep) - pool.Add(google_descriptor_pb2.FileDescriptorProto.FromString(proto.to_binary())) + pool.Add( + google_descriptor_pb2.FileDescriptorProto.FromString(proto.to_binary()) + ) for file in fdset.file: add(file.name) return pool -def run_test_case(validator: protovalidate.Validator, tc: protobuf.Message | google_message.Message) -> TestResult: +def run_test_case( + validator: protovalidate.Validator, tc: protobuf.Message | google_message.Message +) -> TestResult: # Run the validator try: violations = validator.collect_violations(tc) if len(violations) > 0: # Convert from protovalidate bundled proto to test harness's. - pv_violations = protovalidate.Violations(violations=[violation.proto for violation in violations]) + pv_violations = protovalidate.Violations( + violations=[violation.proto for violation in violations] + ) return TestResult( result=Oneof( field="validation_error", value=validate_pb.Violations.from_binary(pv_violations.to_binary()), ) ) - else: - return TestResult(result=Oneof(field="success", value=True)) + return TestResult(result=Oneof(field="success", value=True)) except protovalidate.CompilationError as e: return TestResult(result=Oneof(field="compilation_error", value=str(e))) except celpy.CELEvalError as e: @@ -83,7 +93,7 @@ def run_test_case(validator: protovalidate.Validator, tc: protobuf.Message | goo return TestResult(result=Oneof(field="runtime_error", value=str(e))) except RuntimeError as e: return TestResult(result=Oneof(field="runtime_error", value=str(e))) - except Exception as e: + except Exception as e: # noqa: BLE001 return TestResult(result=Oneof(field="unexpected_error", value=str(e))) @@ -97,24 +107,34 @@ def run_any_test_case( if isinstance(registry, Registry): desc = registry.message(type_name) if desc is None: - return TestResult(result=Oneof(field="unexpected_error", value=f"unknown type: {type_name}")) + return TestResult( + result=Oneof( + field="unexpected_error", value=f"unknown type: {type_name}" + ) + ) unpacked = tc.unpack(desc) if unpacked is None: - return TestResult(result=Oneof(field="unexpected_error", value=f"cannot unpack {tc.type_url}")) + return TestResult( + result=Oneof( + field="unexpected_error", value=f"cannot unpack {tc.type_url}" + ) + ) msg = unpacked else: try: google_desc = registry.FindMessageTypeByName(type_name) except KeyError: - return TestResult(result=Oneof(field="unexpected_error", value=f"unknown type: {type_name}")) + return TestResult( + result=Oneof( + field="unexpected_error", value=f"unknown type: {type_name}" + ) + ) msg = google_message_factory.GetMessageClass(google_desc)() msg.ParseFromString(tc.value) return run_test_case(validator, msg) -def run_conformance_test( - request: TestConformanceRequest, -) -> TestConformanceResponse: +def run_conformance_test(request: TestConformanceRequest) -> TestConformanceResponse: registry = request.fdset.to_registry() # The registry resolves the conformance suite's custom predefined-rule extensions. validator = protovalidate.Validator(registry=registry) diff --git a/test/conformance/test_conformance.py b/test/conformance/test_conformance.py index fd8a125..0c48fda 100644 --- a/test/conformance/test_conformance.py +++ b/test/conformance/test_conformance.py @@ -41,7 +41,7 @@ def maybe_patch_args_with_debug(args: list[str]) -> list[str]: ) return _pydev_bundle.pydev_monkey.patch_args(args) - except Exception: + except Exception: # noqa: BLE001 return args diff --git a/test/conftest.py b/test/conftest.py index 4424b43..4bdc3ba 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -14,20 +14,22 @@ from __future__ import annotations +from typing import Any + import protovalidate -from protovalidate.internal import backend +from protovalidate import _backend -BACKENDS: list[str] = ["celpy", *(["cel-expr"] if backend.CEL_EXPR_AVAILABLE else [])] +BACKENDS: list[str] = ["celpy", *(["cel-expr"] if _backend.CEL_EXPR_AVAILABLE else [])] -def make_validator(cel_backend: str, **kwargs) -> protovalidate.Validator: - original = backend.CEL_EXPR_AVAILABLE - backend.CEL_EXPR_AVAILABLE = cel_backend == "cel-expr" +def make_validator(cel_backend: str, **kwargs: Any) -> protovalidate.Validator: + original = _backend.CEL_EXPR_AVAILABLE + _backend.CEL_EXPR_AVAILABLE = cel_backend == "cel-expr" try: return protovalidate.Validator(**kwargs) finally: - backend.CEL_EXPR_AVAILABLE = original + _backend.CEL_EXPR_AVAILABLE = original -def backend_validators(**kwargs) -> list[protovalidate.Validator]: +def backend_validators(**kwargs: Any) -> list[protovalidate.Validator]: return [make_validator(name, **kwargs) for name in BACKENDS] diff --git a/test/test_benchmark.py b/test/test_benchmark.py index 0ce496b..f65447f 100644 --- a/test/test_benchmark.py +++ b/test/test_benchmark.py @@ -12,8 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import random -from collections.abc import Callable, Iterator +from typing import TYPE_CHECKING, Any import pytest from protobuf import Message, Oneof @@ -28,9 +30,6 @@ UInt32Value, UInt64Value, ) -from pytest_benchmark.fixture import BenchmarkFixture - -import protovalidate from .conftest import BACKENDS, make_validator from .gen.bench.v1.bench_pb import ( @@ -43,8 +42,20 @@ BenchRepeatedScalarUnique, BenchScalar, ) -from .gen.bench.v1.native_pb import BenchGT, MultiRule, StringMatching, WrapperTesting -from .gen.bench.v1.native_pb import TestByteMatching as ByteMatching +from .gen.bench.v1.native_pb import ( + BenchGT, + MultiRule, + StringMatching, + TestByteMatching as ByteMatching, + WrapperTesting, +) + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from pytest_benchmark.fixture import BenchmarkFixture + + import protovalidate def gen_bytes(n: int, salt: int) -> bytes: @@ -100,8 +111,14 @@ def gen_complex(depth: int) -> BenchComplexSchema: map_i32_i64={1: 10, 2: 20, 3: 30}, map_u64_bool={1: True, 2: False}, map_str_bytes={"k": gen_bytes(2, 0)}, - map_str_msg={"a": BenchScalar(x=random.randint(1, 100)), "b": BenchScalar(x=random.randint(1, 100))}, - map_i64_msg={1: BenchScalar(x=random.randint(1, 100)), 2: BenchScalar(x=random.randint(1, 100))}, + map_str_msg={ + "a": BenchScalar(x=random.randint(1, 100)), + "b": BenchScalar(x=random.randint(1, 100)), + }, + map_i64_msg={ + 1: BenchScalar(x=random.randint(1, 100)), + 2: BenchScalar(x=random.randint(1, 100)), + }, enum_field=BenchEnum.ONE, choice=Oneof(field="oneof_str", value=random.choice(words)), self_ref=gen_complex(depth - 1) if depth > 0 else None, @@ -113,21 +130,41 @@ def validator(request: pytest.FixtureRequest) -> protovalidate.Validator: return make_validator(request.param) -def param(*args, id: str) -> pytest.param: # noqa: A002 - """Copies pytest id to a fixture parameter since pytest-benchmark doesn't allow - grouping on the former.""" +def param(*args: Any, id: str) -> pytest.param: # noqa: A002 return pytest.param(id, *args, id=id) # Use lambda factories to allow random seed fixture to apply before computing cases = [ param(lambda: BenchScalar(x=42), id="scalar"), - param(lambda: BenchRepeatedScalar(x=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), id="repeated_scalar"), - param(lambda: BenchRepeatedMessage(x=[BenchScalar(x=i + 1) for i in range(10)]), id="repeated_message"), - param(lambda: BenchRepeatedScalarUnique(x=[1.1, 2.2, 3.3, 4.4, 5.5, 6.6, 7.7, 8.8]), id="repeated_unique_scalar"), - param(lambda: BenchRepeatedBytesUnique(x=[gen_bytes(4, i + 1) for i in range(8)]), id="repeated_unique_bytes"), param( - lambda: BenchMap(entries={"k1": "v1", "k2": "v2", "k3": "v3", "k4": "v4", "k5": "v5", "k6": "v6", "k7": "v7"}), + lambda: BenchRepeatedScalar(x=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), + id="repeated_scalar", + ), + param( + lambda: BenchRepeatedMessage(x=[BenchScalar(x=i + 1) for i in range(10)]), + id="repeated_message", + ), + param( + lambda: BenchRepeatedScalarUnique(x=[1.1, 2.2, 3.3, 4.4, 5.5, 6.6, 7.7, 8.8]), + id="repeated_unique_scalar", + ), + param( + lambda: BenchRepeatedBytesUnique(x=[gen_bytes(4, i + 1) for i in range(8)]), + id="repeated_unique_bytes", + ), + param( + lambda: BenchMap( + entries={ + "k1": "v1", + "k2": "v2", + "k3": "v3", + "k4": "v4", + "k5": "v5", + "k6": "v6", + "k7": "v7", + } + ), id="map", ), param(lambda: gen_complex(1), id="complex_schema"), @@ -197,6 +234,6 @@ def test_benchmark( message_factory: Callable[[], Message], benchmark: BenchmarkFixture, validator: protovalidate.Validator, -): +) -> None: message = message_factory() benchmark(validator.collect_violations, message) diff --git a/test/test_format.py b/test/test_format.py index e558286..13b8685 100644 --- a/test/test_format.py +++ b/test/test_format.py @@ -12,23 +12,28 @@ # See the License for the specific language governing permissions and # limitations under the License. -from collections.abc import Iterable, MutableMapping +from __future__ import annotations + from itertools import chain from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any import celpy import pytest from celpy import celtypes from google.protobuf import text_format as google_text_format -from protovalidate.internal import extra_func -from protovalidate.internal.cel_field_presence import InterpretedRunner +from protovalidate import _extra_func +from protovalidate._cel_field_presence import InterpretedRunner -from .gen.cel.expr import eval_pb2 from .gen.cel.expr.conformance.test import simple_pb2 from .versions import CEL_SPEC_VERSION +if TYPE_CHECKING: + from collections.abc import Iterable, MutableMapping + + from .gen.cel.expr import eval_pb2 + skipped_tests = [ # cel-python seems to have a bug with ints and booleans in the same map which evaluate to the same value # which the test data for this test has. For example: {1: 'value1', true: 'value2'}]). @@ -36,7 +41,7 @@ # "no such overload: IntType(0) != # BoolType(False) ",)) # TODO: Check if this bug is fixed in newer versions of cel-python. - "map support (all key types)", + "map support (all key types)" ] skipped_error_tests = [ # cel-python does not support Protobuf messages at the moment and these tests use a MessageType @@ -47,15 +52,15 @@ ] -def load_test_data(file_name: str) -> simple_pb2.SimpleTestFile: +def load_test_data(file_name: Path) -> simple_pb2.SimpleTestFile: msg = simple_pb2.SimpleTestFile() - with open(file_name) as file: - text_data = file.read() - google_text_format.Parse(text_data, msg) + google_text_format.Parse(file_name.read_bytes(), msg) return msg -def build_variables(bindings: MutableMapping[str, eval_pb2.ExprValue]) -> dict[Any, Any]: +def build_variables( + bindings: MutableMapping[str, eval_pb2.ExprValue], +) -> dict[Any, Any]: binder = {} for key, value in bindings.items(): if value.HasField("value"): @@ -83,16 +88,22 @@ def get_eval_error_message(test: simple_pb2.SimpleTest) -> str | None: # The test data from the cel-spec conformance tests testdata_dir = Path(__file__).parent / "testdata" -cel_test_data = load_test_data(testdata_dir / f"string_ext_{CEL_SPEC_VERSION}.textproto") +cel_test_data = load_test_data( + testdata_dir / f"string_ext_{CEL_SPEC_VERSION}.textproto" +) # Our supplemental tests of functionality not in the cel conformance file, but defined in the spec. -supplemental_test_data = load_test_data(testdata_dir / "string_ext_supplemental.textproto") +supplemental_test_data = load_test_data( + testdata_dir / "string_ext_supplemental.textproto" +) # Combine the test data from both files into one sections = cel_test_data.section sections.extend(supplemental_test_data.section) # Find the format tests which test successful formatting -_format_tests: Iterable[simple_pb2.SimpleTest] = chain.from_iterable(x.test for x in sections if x.name == "format") +_format_tests: Iterable[simple_pb2.SimpleTest] = chain.from_iterable( + x.test for x in sections if x.name == "format" +) # Find the format error tests which test errors during formatting _format_error_tests: Iterable[simple_pb2.SimpleTest] = chain.from_iterable( x.test for x in sections if x.name == "format_errors" @@ -101,36 +112,39 @@ def get_eval_error_message(test: simple_pb2.SimpleTest) -> str | None: env = celpy.Environment(runner_class=InterpretedRunner) -def test_format_successes(subtests: pytest.Subtests): - """Tests success scenarios for string.format""" +def test_format_successes(subtests: pytest.Subtests) -> None: + """Tests success scenarios for string.format.""" for format_test in _format_tests: with subtests.test(msg=format_test.name): if format_test.name in skipped_tests: pytest.skip(f"skipped test: {format_test.name}") ast = env.compile(format_test.expr) - prog = env.program(ast, functions=extra_func.make_extra_funcs()) + prog = env.program(ast, functions=_extra_func.make_extra_funcs()) bindings = build_variables(format_test.bindings) result = prog.evaluate(bindings) expected = get_expected_result(format_test) - assert expected is not None, f"[{format_test.name}]: expected a success result to be defined" + assert expected is not None, ( + f"[{format_test.name}]: expected a success result to be defined" + ) assert result == expected -def test_format_errors(subtests: pytest.Subtests): - """Tests error scenarios for string.format""" +def test_format_errors(subtests: pytest.Subtests) -> None: + """Tests error scenarios for string.format.""" for format_error_test in _format_error_tests: with subtests.test(msg=format_error_test.name): if format_error_test.name in skipped_error_tests: pytest.skip(f"skipped test: {format_error_test.name}") ast = env.compile(format_error_test.expr) - prog = env.program(ast, functions=extra_func.make_extra_funcs()) + prog = env.program(ast, functions=_extra_func.make_extra_funcs()) bindings = build_variables(format_error_test.bindings) - try: + with pytest.raises(celpy.CELEvalError) as exc_info: prog.evaluate(bindings) - pytest.fail(f"[{format_error_test.name}]: expected an error to be raised during evaluation") - except celpy.CELEvalError as e: - msg = get_eval_error_message(format_error_test) - assert msg is not None, f"[{format_error_test.name}]: expected an eval error to be defined" - assert str(e) == msg + + msg = get_eval_error_message(format_error_test) + assert msg is not None, ( + f"[{format_error_test.name}]: expected an eval error to be defined" + ) + assert str(exc_info.value) == msg diff --git a/test/test_matches.py b/test/test_matches.py index f0bdff4..b0cea0f 100644 --- a/test/test_matches.py +++ b/test/test_matches.py @@ -12,13 +12,15 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import celpy from celpy import celtypes -from protovalidate.internal.extra_func import cel_matches +from protovalidate._extra_func import cel_matches -def test_function_matches_re2(): +def test_function_matches_re2() -> None: empty_string = celtypes.StringType("") # \z is valid re2 syntax for end of text assert cel_matches(empty_string, "^\\z") diff --git a/test/test_validate.py b/test/test_validate.py index ebfa6f2..6c5b8ec 100644 --- a/test/test_validate.py +++ b/test/test_validate.py @@ -12,28 +12,52 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol + import protobuf import pytest from protobuf import Oneof import protovalidate -from protovalidate.internal import rules +from protovalidate import _rules from .conftest import backend_validators from .gen.tests.example.v1 import validations_pb, validations_pb2 -validators: list[protovalidate.Validator] = [ +if TYPE_CHECKING: + from google.protobuf import message as google_message + + +class ValidatorProtocol(Protocol): + def validate( + self, + message: protobuf.Message | google_message.Message, + *, + fail_fast: bool = False, + ) -> None: ... + + def collect_violations( + self, + message: protobuf.Message | google_message.Message, + *, + fail_fast: bool = False, + ) -> list[_rules.Violation]: ... + + +validators: list[ValidatorProtocol] = [ protovalidate, # global module singleton *backend_validators(), ] @pytest.mark.parametrize("validator", validators) -def test_ninf(validator): +def test_ninf(validator: ValidatorProtocol) -> None: msg = validations_pb.DoubleFinite() msg.val = float("-inf") - expected_violation = rules.Violation( + expected_violation = _rules.Violation( message="must be finite", rule_id="double.finite", field_value=msg.val, @@ -44,11 +68,11 @@ def test_ninf(validator): @pytest.mark.parametrize("validator", validators) -def test_map_key(validator): +def test_map_key(validator: ValidatorProtocol) -> None: msg = validations_pb.MapKeys() msg.val[1] = "a" - expected_violation = rules.Violation( + expected_violation = _rules.Violation( message="must be less than 0", rule_id="sint64.lt", for_key=True, @@ -60,21 +84,21 @@ def test_map_key(validator): @pytest.mark.parametrize("validator", validators) -def test_sfixed64_valid(validator): +def test_sfixed64_valid(validator: ValidatorProtocol) -> None: msg = validations_pb.SFixed64ExLTGT(val=11) check_valid(validator, msg) @pytest.mark.parametrize("validator", validators) -def test_oneofs(validator): +def test_oneofs(validator: ValidatorProtocol) -> None: msg = validations_pb.Oneof(o=Oneof(field="y", value=123)) check_valid(validator, msg) @pytest.mark.parametrize("validator", validators) -def test_protovalidate_oneof_valid(validator): +def test_protovalidate_oneof_valid(validator: ValidatorProtocol) -> None: msg = validations_pb.ProtovalidateOneof() msg.a = "A" @@ -82,47 +106,53 @@ def test_protovalidate_oneof_valid(validator): @pytest.mark.parametrize("validator", validators) -def test_protovalidate_oneof_violation(validator): +def test_protovalidate_oneof_violation(validator: ValidatorProtocol) -> None: msg = validations_pb.ProtovalidateOneof() msg.a = "A" msg.b = "B" - expected_violation = rules.Violation(message="only one of a, b can be set", rule_id="message.oneof") + expected_violation = _rules.Violation( + message="only one of a, b can be set", rule_id="message.oneof" + ) check_invalid(validator, msg, [expected_violation]) @pytest.mark.parametrize("validator", validators) -def test_protovalidate_oneof_required_violation(validator): +def test_protovalidate_oneof_required_violation(validator: ValidatorProtocol) -> None: msg = validations_pb.ProtovalidateOneofRequired() - expected_violation = rules.Violation(message="one of a, b must be set", rule_id="message.oneof") + expected_violation = _rules.Violation( + message="one of a, b must be set", rule_id="message.oneof" + ) check_invalid(validator, msg, [expected_violation]) @pytest.mark.parametrize("validator", validators) -def test_protovalidate_oneof_unknown_field_name(validator): - """Tests that a compilation error is thrown when specifying a oneof rule with an invalid field name""" +def test_protovalidate_oneof_unknown_field_name(validator: ValidatorProtocol) -> None: + """Tests that a compilation error is thrown when specifying a oneof rule with an invalid field name.""" msg = validations_pb.ProtovalidateOneofUnknownFieldName() check_compilation_errors( - validator, msg, 'field "xxx" not found in message tests.example.v1.ProtovalidateOneofUnknownFieldName' + validator, + msg, + 'field "xxx" not found in message tests.example.v1.ProtovalidateOneofUnknownFieldName', ) @pytest.mark.parametrize("validator", validators) -def test_repeated(validator): +def test_repeated(validator: ValidatorProtocol) -> None: msg = validations_pb.RepeatedEmbedSkip(val=[validations_pb.Embed(val=-1)]) check_valid(validator, msg) @pytest.mark.parametrize("validator", validators) -def test_maps(validator): +def test_maps(validator: ValidatorProtocol) -> None: msg = validations_pb.MapMinMax() - expected_violation = rules.Violation( + expected_violation = _rules.Violation( message="map must be at least 2 entries", rule_id="map.min_pairs", field_value={}, @@ -133,27 +163,27 @@ def test_maps(validator): @pytest.mark.parametrize("validator", validators) -def test_timestamp(validator): +def test_timestamp(validator: ValidatorProtocol) -> None: msg = validations_pb.TimestampGTNow() check_valid(validator, msg) @pytest.mark.parametrize("validator", validators) -def test_multiple_validations(validator): +def test_multiple_validations(validator: ValidatorProtocol) -> None: """Test that a message with multiple violations correctly returns all of them.""" msg = validations_pb.MultipleValidations() msg.title = "bar" msg.name = "blah" - expected_violation1 = rules.Violation( + expected_violation1 = _rules.Violation( message="does not have prefix `foo`", rule_id="string.prefix", field_value=msg.title, rule_value="foo", ) - expected_violation2 = rules.Violation( + expected_violation2 = _rules.Violation( message="must be at least 5 characters", rule_id="string.min_len", field_value=msg.name, @@ -164,23 +194,20 @@ def test_multiple_validations(validator): @pytest.mark.parametrize("validator", validators) -def test_concatenated_values(validator): - msg = validations_pb.ConcatenatedValues( - bar=["a", "b", "c"], - baz=["d", "e", "f"], - ) +def test_concatenated_values(validator: ValidatorProtocol) -> None: + msg = validations_pb.ConcatenatedValues(bar=["a", "b", "c"], baz=["d", "e", "f"]) check_valid(validator, msg) @pytest.mark.parametrize("validator", validators) -def test_fail_fast(validator): - """Test that fail fast correctly fails on first violation""" +def test_fail_fast(validator: ValidatorProtocol) -> None: + """Test that fail fast correctly fails on first violation.""" msg = validations_pb.MultipleValidations() msg.title = "bar" msg.name = "blah" - expected_violation = rules.Violation( + expected_violation = _rules.Violation( message="does not have prefix `foo`", rule_id="string.prefix", field_value=msg.title, @@ -200,7 +227,7 @@ def test_fail_fast(validator): @pytest.mark.parametrize("validator", validators) -def test_legacy_message_valid(validator): +def test_legacy_message_valid(validator: ValidatorProtocol) -> None: """A google.protobuf message validates through the legacy conversion path.""" msg = validations_pb2.DoubleFinite() msg.val = 1.0 @@ -209,11 +236,11 @@ def test_legacy_message_valid(validator): @pytest.mark.parametrize("validator", validators) -def test_legacy_message_invalid(validator): +def test_legacy_message_invalid(validator: ValidatorProtocol) -> None: msg = validations_pb2.DoubleFinite() msg.val = float("-inf") - expected_violation = rules.Violation( + expected_violation = _rules.Violation( message="must be finite", rule_id="double.finite", field_value=msg.val, @@ -231,11 +258,11 @@ def test_legacy_message_invalid(validator): @pytest.mark.parametrize("validator", validators) -def test_legacy_message_map_key(validator): +def test_legacy_message_map_key(validator: ValidatorProtocol) -> None: msg = validations_pb2.MapKeys() msg.val[1] = "a" - expected_violation = rules.Violation( + expected_violation = _rules.Violation( message="must be less than 0", rule_id="sint64.lt", for_key=True, @@ -247,7 +274,9 @@ def test_legacy_message_map_key(validator): _compare_violations(violations, [expected_violation]) -def check_valid(validator: protovalidate.Validator, msg: protobuf.Message): +def check_valid( + validator: ValidatorProtocol, msg: protobuf.Message | google_message.Message +) -> None: # Test validate validator.validate(msg) @@ -256,12 +285,19 @@ def check_valid(validator: protovalidate.Validator, msg: protobuf.Message): assert len(violations) == 0 -def check_invalid(validator: protovalidate.Validator, msg: protobuf.Message, expected: list[rules.Violation]): +def check_invalid( + validator: ValidatorProtocol, + msg: protobuf.Message | google_message.Message, + expected: list[_rules.Violation], +) -> None: # Test validate with pytest.raises(protovalidate.ValidationError) as exc_info: validator.validate(msg) e = exc_info.value - assert str(e) == f"invalid {type(msg).desc().name}" + if isinstance(msg, protobuf.Message): + assert str(e) == f"invalid {type(msg).desc().name}" + else: + assert str(e) == f"invalid {msg.DESCRIPTOR.name}" _compare_violations(e.violations, expected) # ty: ignore # Test collect_violations @@ -269,7 +305,11 @@ def check_invalid(validator: protovalidate.Validator, msg: protobuf.Message, exp _compare_violations(violations, expected) -def check_compilation_errors(validator: protovalidate.Validator, msg: protobuf.Message, expected: str): +def check_compilation_errors( + validator: ValidatorProtocol, + msg: protobuf.Message | google_message.Message, + expected: str, +) -> None: """A helper function for testing compilation errors when validating. The tests are run using validators created via all possible methods and @@ -286,7 +326,9 @@ def check_compilation_errors(validator: protovalidate.Validator, msg: protobuf.M assert str(cvce.value) == expected -def _compare_violations(actual: list[rules.Violation], expected: list[rules.Violation]): +def _compare_violations( + actual: list[_rules.Violation], expected: list[_rules.Violation] +) -> None: """Compares two lists of violations. The violations are expected to be in the expected order also.""" assert len(actual) == len(expected) for a, e in zip(actual, expected, strict=True): diff --git a/test/versions.py b/test/versions.py index 11b5bb3..69648ec 100644 --- a/test/versions.py +++ b/test/versions.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from __future__ import annotations + import os # Version of the cel-spec that this implementation is conformant with.