Formatting: Format all python code with black.
This CL is probably not what you're looking for, it's only
automated formatting. Ignore it with
`git blame --ignore-rev <revision>` for this commit.
BUG=b:233893248
TEST=CQ
Change-Id: I66591d7a738d241aed3290138c0f68065ab10a6d
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/3879174
Reviewed-by: Mike Frysinger <vapier@chromium.org>
Tested-by: Alex Klein <saklein@chromium.org>
diff --git a/api/validate.py b/api/validate.py
index 7501c0e..5ba7f06 100644
--- a/api/validate.py
+++ b/api/validate.py
@@ -23,304 +23,327 @@
def _value(
field: str, message: protobuf_message.Message
) -> Union[bool, int, str, None, List, protobuf_message.Message]:
- """Helper function to fetch the value of the field.
+ """Helper function to fetch the value of the field.
- Args:
- field: The field name. Can be nested via . separation.
- message: The protobuf message it is being fetched from.
+ Args:
+ field: The field name. Can be nested via . separation.
+ message: The protobuf message it is being fetched from.
- Returns:
- The value of the field.
- """
- if not field:
- return message
+ Returns:
+ The value of the field.
+ """
+ if not field:
+ return message
- value = message
- for part in field.split('.'):
- if not isinstance(value, protobuf_message.Message):
- value = None
- break
+ value = message
+ for part in field.split("."):
+ if not isinstance(value, protobuf_message.Message):
+ value = None
+ break
- try:
- value = getattr(value, part)
- except AttributeError as e:
- cros_build_lib.Die('Invalid field: %s', e)
+ try:
+ value = getattr(value, part)
+ except AttributeError as e:
+ cros_build_lib.Die("Invalid field: %s", e)
- return value
+ return value
# pylint: disable=docstring-misnamed-args
def exists(*fields: str):
- """Validate that the paths in |fields| exist.
+ """Validate that the paths in |fields| exist.
- Args:
- fields (str): The fields being checked. Can be . separated nested
- fields.
- """
- assert fields
+ Args:
+ fields (str): The fields being checked. Can be . separated nested
+ fields.
+ """
+ assert fields
- def decorator(func):
- @functools.wraps(func)
- def _exists(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- for field in fields:
- logging.debug('Validating %s exists.', field)
+ def decorator(func):
+ @functools.wraps(func)
+ def _exists(input_proto, output_proto, config, *args, **kwargs):
+ if config.do_validation:
+ for field in fields:
+ logging.debug("Validating %s exists.", field)
- value = _value(field, input_proto)
- if not value or not os.path.exists(value):
- cros_build_lib.Die('%s path does not exist: %s' % (field, value))
+ value = _value(field, input_proto)
+ if not value or not os.path.exists(value):
+ cros_build_lib.Die(
+ "%s path does not exist: %s" % (field, value)
+ )
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _exists
+ return _exists
- return decorator
+ return decorator
def is_in(field: str, values: Iterable):
- """Validate |field| is an element of |values|.
+ """Validate |field| is an element of |values|.
- Args:
- field: The field being checked. May be . separated nested fields.
- values: The possible values field may take.
- """
- assert field
- assert values
+ Args:
+ field: The field being checked. May be . separated nested fields.
+ values: The possible values field may take.
+ """
+ assert field
+ assert values
- def decorator(func):
- @functools.wraps(func)
- def _is_in(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- logging.debug('Validating %s is in %r', field, values)
- value = _value(field, input_proto)
+ def decorator(func):
+ @functools.wraps(func)
+ def _is_in(input_proto, output_proto, config, *args, **kwargs):
+ if config.do_validation:
+ logging.debug("Validating %s is in %r", field, values)
+ value = _value(field, input_proto)
- if value not in values:
- cros_build_lib.Die('%s (%r) must be in %r', field, value, values)
+ if value not in values:
+ cros_build_lib.Die(
+ "%s (%r) must be in %r", field, value, values
+ )
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _is_in
+ return _is_in
- return decorator
+ return decorator
-def each_in(field: str,
- subfield: Optional[str],
- values: Iterable,
- optional: bool = False):
- """Validate each |subfield| of the repeated |field| is in |values|.
+def each_in(
+ field: str,
+ subfield: Optional[str],
+ values: Iterable,
+ optional: bool = False,
+):
+ """Validate each |subfield| of the repeated |field| is in |values|.
- Args:
- field: The field being checked. May be . separated nested fields.
- subfield: The field in the repeated |field| to validate, or None
- when |field| is not a repeated message, e.g. enum, scalars.
- values: The possible values field may take.
- optional: Also allow the field to be empty when True.
- """
- assert field
- assert values
+ Args:
+ field: The field being checked. May be . separated nested fields.
+ subfield: The field in the repeated |field| to validate, or None
+ when |field| is not a repeated message, e.g. enum, scalars.
+ values: The possible values field may take.
+ optional: Also allow the field to be empty when True.
+ """
+ assert field
+ assert values
- def decorator(func):
- @functools.wraps(func)
- def _is_in(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- members = _value(field, input_proto) or []
- if not optional and not members:
- cros_build_lib.Die('The %s field is empty.', field)
- for member in members:
- logging.debug('Validating %s.[each].%s is in %r.', field, subfield,
- values)
- value = _value(subfield, member)
- if value not in values:
- cros_build_lib.Die('%s.[each].%s (%r) must be in %r is required.',
- field, subfield, value, values)
+ def decorator(func):
+ @functools.wraps(func)
+ def _is_in(input_proto, output_proto, config, *args, **kwargs):
+ if config.do_validation:
+ members = _value(field, input_proto) or []
+ if not optional and not members:
+ cros_build_lib.Die("The %s field is empty.", field)
+ for member in members:
+ logging.debug(
+ "Validating %s.[each].%s is in %r.",
+ field,
+ subfield,
+ values,
+ )
+ value = _value(subfield, member)
+ if value not in values:
+ cros_build_lib.Die(
+ "%s.[each].%s (%r) must be in %r is required.",
+ field,
+ subfield,
+ value,
+ values,
+ )
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _is_in
+ return _is_in
- return decorator
+ return decorator
def constraint(description):
- """Define a function to be used as a constraint check.
+ """Define a function to be used as a constraint check.
- A constraint is a function that checks the value of a field and either
- does nothing (returns None) or returns a string indicating why the value
- isn't valid.
+ A constraint is a function that checks the value of a field and either
+ does nothing (returns None) or returns a string indicating why the value
+ isn't valid.
- We bind a human readable description to the constraint for error reporting
- and logging.
+ We bind a human readable description to the constraint for error reporting
+ and logging.
- Args:
- description: Human readable description of the constraint
- """
+ Args:
+ description: Human readable description of the constraint
+ """
- def decorator(func):
- @functools.wraps(func)
- def _func(*args, **kwargs):
- func(*args, **kwargs)
+ def decorator(func):
+ @functools.wraps(func)
+ def _func(*args, **kwargs):
+ func(*args, **kwargs)
- setattr(_func, '__constraint_description__', description)
- return _func
+ setattr(_func, "__constraint_description__", description)
+ return _func
- return decorator
+ return decorator
def check_constraint(field: str, checkfunc: Callable):
- """Validate all values of |field| pass a constraint.
+ """Validate all values of |field| pass a constraint.
- Args:
- field: The field being checked. May be . separated nested fields.
- checkfunc: A constraint function to check on each value
- """
- assert field
- assert constraint
+ Args:
+ field: The field being checked. May be . separated nested fields.
+ checkfunc: A constraint function to check on each value
+ """
+ assert field
+ assert constraint
- # Get description for the constraint if it's set
- constraint_description = getattr(
- checkfunc,
- '__constraint_description__',
- checkfunc.__name__,
- )
+ # Get description for the constraint if it's set
+ constraint_description = getattr(
+ checkfunc,
+ "__constraint_description__",
+ checkfunc.__name__,
+ )
- def decorator(func):
- @functools.wraps(func)
- def _check_constraint(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- values = _value(field, input_proto) or []
+ def decorator(func):
+ @functools.wraps(func)
+ def _check_constraint(
+ input_proto, output_proto, config, *args, **kwargs
+ ):
+ if config.do_validation:
+ values = _value(field, input_proto) or []
- failed = []
- for val in values:
- msg = checkfunc(val)
- if msg is not None:
- failed.append((val, msg))
+ failed = []
+ for val in values:
+ msg = checkfunc(val)
+ if msg is not None:
+ failed.append((val, msg))
- if failed:
- msg = f'{field}.[all] one or more values failed check ' \
- f'"{constraint_description}"\n'
+ if failed:
+ msg = (
+ f"{field}.[all] one or more values failed check "
+ f'"{constraint_description}"\n'
+ )
- for value, msg in failed:
- msg += ' %s: %s\n' % (value, msg)
- cros_build_lib.Die(msg)
+ for value, msg in failed:
+ msg += " %s: %s\n" % (value, msg)
+ cros_build_lib.Die(msg)
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _check_constraint
+ return _check_constraint
- return decorator
+ return decorator
# pylint: disable=docstring-misnamed-args
def require(*fields: str):
- """Verify |fields| have all been set to truthy values.
+ """Verify |fields| have all been set to truthy values.
- Args:
- fields: The fields being checked. May be . separated nested fields.
- """
- assert fields
+ Args:
+ fields: The fields being checked. May be . separated nested fields.
+ """
+ assert fields
- def decorator(func):
- @functools.wraps(func)
- def _require(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- for field in fields:
- logging.debug('Validating %s is set.', field)
+ def decorator(func):
+ @functools.wraps(func)
+ def _require(input_proto, output_proto, config, *args, **kwargs):
+ if config.do_validation:
+ for field in fields:
+ logging.debug("Validating %s is set.", field)
- value = _value(field, input_proto)
- if not value:
- cros_build_lib.Die('%s is required.', field)
+ value = _value(field, input_proto)
+ if not value:
+ cros_build_lib.Die("%s is required.", field)
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _require
+ return _require
- return decorator
+ return decorator
# pylint: disable=docstring-misnamed-args
def require_any(*fields: str):
- """Verify at least one of |fields| have been set.
+ """Verify at least one of |fields| have been set.
- Args:
- fields: The fields being checked. May be . separated nested fields.
- """
- assert fields
+ Args:
+ fields: The fields being checked. May be . separated nested fields.
+ """
+ assert fields
- def decorator(func):
- @functools.wraps(func)
- def _require(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- for field in fields:
- logging.debug('Validating %s is set.', field)
- value = _value(field, input_proto)
- if value:
- break
- else:
- cros_build_lib.Die('At least one of the following must be set: %s',
- ', '.join(fields))
+ def decorator(func):
+ @functools.wraps(func)
+ def _require(input_proto, output_proto, config, *args, **kwargs):
+ if config.do_validation:
+ for field in fields:
+ logging.debug("Validating %s is set.", field)
+ value = _value(field, input_proto)
+ if value:
+ break
+ else:
+ cros_build_lib.Die(
+ "At least one of the following must be set: %s",
+ ", ".join(fields),
+ )
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _require
+ return _require
- return decorator
+ return decorator
-def require_each(field: str,
- subfields: Iterable[str],
- allow_empty: bool = True):
- """Verify |field| each have all of the |subfields| set.
+def require_each(
+ field: str, subfields: Iterable[str], allow_empty: bool = True
+):
+ """Verify |field| each have all of the |subfields| set.
- When |allow_empty| is True, |field| may be empty, and |subfields| are only
- validated when it is not empty. When |allow_empty| is False, |field| must
- also have at least one entry.
+ When |allow_empty| is True, |field| may be empty, and |subfields| are only
+ validated when it is not empty. When |allow_empty| is False, |field| must
+ also have at least one entry.
- Args:
- field: The repeated field being checked. May be . separated nested
- fields.
- subfields: The fields of the repeated message to validate.
- allow_empty: Also require at least one entry in the repeated field.
- """
- assert field
- assert subfields
- assert not isinstance(subfields, str)
+ Args:
+ field: The repeated field being checked. May be . separated nested
+ fields.
+ subfields: The fields of the repeated message to validate.
+ allow_empty: Also require at least one entry in the repeated field.
+ """
+ assert field
+ assert subfields
+ assert not isinstance(subfields, str)
- def decorator(func):
- @functools.wraps(func)
- def _require_each(input_proto, output_proto, config, *args, **kwargs):
- if config.do_validation:
- members = _value(field, input_proto) or []
- if not allow_empty and not members:
- cros_build_lib.Die('The %s field is empty.', field)
- for member in members:
- for subfield in subfields:
- logging.debug('Validating %s.[each].%s is set.', field, subfield)
- value = _value(subfield, member)
- if not value:
- cros_build_lib.Die('%s is required.', field)
+ def decorator(func):
+ @functools.wraps(func)
+ def _require_each(input_proto, output_proto, config, *args, **kwargs):
+ if config.do_validation:
+ members = _value(field, input_proto) or []
+ if not allow_empty and not members:
+ cros_build_lib.Die("The %s field is empty.", field)
+ for member in members:
+ for subfield in subfields:
+ logging.debug(
+ "Validating %s.[each].%s is set.", field, subfield
+ )
+ value = _value(subfield, member)
+ if not value:
+ cros_build_lib.Die("%s is required.", field)
- return func(input_proto, output_proto, config, *args, **kwargs)
+ return func(input_proto, output_proto, config, *args, **kwargs)
- return _require_each
+ return _require_each
- return decorator
+ return decorator
def validation_complete(func: Callable):
- """Automatically skip the endpoint when called after all other validators.
+ """Automatically skip the endpoint when called after all other validators.
- This decorator MUST be applied after all other validate decorators.
- The config can be checked manually if there is non-decorator validation, but
- this is much cleaner if it is all done in decorators.
- """
+ This decorator MUST be applied after all other validate decorators.
+ The config can be checked manually if there is non-decorator validation, but
+ this is much cleaner if it is all done in decorators.
+ """
- @functools.wraps(func)
- def _validate_only(request, response, configs, *args, **kwargs):
- if configs.validate_only:
- # Avoid calling the endpoint.
- return 0
- else:
- return func(request, response, configs, *args, **kwargs)
+ @functools.wraps(func)
+ def _validate_only(request, response, configs, *args, **kwargs):
+ if configs.validate_only:
+ # Avoid calling the endpoint.
+ return 0
+ else:
+ return func(request, response, configs, *args, **kwargs)
- return _validate_only
+ return _validate_only