Source code for ironic.common.args

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, 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.

import functools
import inspect

import jsonschema
from oslo_utils import netutils
from oslo_utils import strutils
from oslo_utils import uuidutils

from ironic.common import exception
from ironic.common.i18n import _
from ironic.common import utils


[docs] def string(name, value): """Validate that the value is a string :param name: Name of the argument :param value: A string value :returns: The string value, or None if value is None :raises: InvalidParameterValue if the value is not a string """ if value is None: return if not isinstance(value, str): raise exception.InvalidParameterValue( _('Expected string for %s: %s') % (name, value)) return value
[docs] def boolean(name, value): """Validate that the value is a string representing a boolean :param name: Name of the argument :param value: A string value :returns: The boolean representation of the value, or None if value is None :raises: InvalidParameterValue if the value cannot be converted to a boolean """ if value is None: return try: return strutils.bool_from_string(value, strict=True) except ValueError as e: raise exception.InvalidParameterValue( _('Invalid %s: %s') % (name, e))
[docs] def uuid(name, value): """Validate that the value is a UUID :param name: Name of the argument :param value: A UUID string value :returns: The value, or None if value is None :raises: InvalidParameterValue if the value is not a valid UUID """ if value is None: return if not uuidutils.is_uuid_like(value): raise exception.InvalidParameterValue( _('Expected UUID for %s: %s') % (name, value)) return value
[docs] def name(name, value): """Validate that the value is a logical name :param name: Name of the argument :param value: A logical name string value :returns: The value, or None if value is None :raises: InvalidParameterValue if the value is not a valid logical name """ if value is None: return if not utils.is_valid_logical_name(value): raise exception.InvalidParameterValue( _('Expected name for %s: %s') % (name, value)) return value
[docs] def host_port(name, value): if value is None: return try: host, port = netutils.parse_host_port(value) except (ValueError, TypeError) as exc: raise exception.InvalidParameterValue(f'{name}: {exc}') if not host: raise exception.InvalidParameterValue(_('Missing host in %s') % name) return value
[docs] def uuid_or_name(name, value): """Validate that the value is a UUID or logical name :param name: Name of the argument :param value: A UUID or logical name string value :returns: The value, or None if value is None :raises: InvalidParameterValue if the value is not a valid UUID or logical name """ if value is None: return if (not utils.is_valid_logical_name(value) and not uuidutils.is_uuid_like(value)): raise exception.InvalidParameterValue( _('Expected UUID or name for %s: %s') % (name, value)) return value
[docs] def string_list(name, value): """Validate and convert comma delimited string to a list. :param name: Name of the argument :param value: A comma separated string of values :returns: A list of unique values (lower-cased), maintaining the same order, or None if value is None :raises: InvalidParameterValue if the value is not a string """ if value is None: return if isinstance(value, list): return [string(name, item) for item in value] value = string(name, value) items = [] for v in str(value).split(','): v_norm = v.strip().lower() if v_norm and v_norm not in items: items.append(v_norm) return items
[docs] def integer(name, value): """Validate that the value represents an integer :param name: Name of the argument :param value: A value representing an integer :returns: The value as an int, or None if value is None :raises: InvalidParameterValue if the value does not represent an integer """ if value is None: return try: return int(value) except (ValueError, TypeError): raise exception.InvalidParameterValue( _('Expected an integer for %s: %s') % (name, value))
[docs] def mac_address(name, value): """Validate that the value represents a MAC address :param name: Name of the argument :param value: A string value representing a MAC address :returns: The value as a normalized MAC address, or None if value is None :raises: InvalidParameterValue if the value is not a valid MAC address """ if value is None: return try: return utils.validate_and_normalize_mac(value) except exception.InvalidMAC: raise exception.InvalidParameterValue( _('Expected valid MAC address for %s: %s') % (name, value))
def _or(name, value, validators): last_error = None for v in validators: try: return v(name=name, value=value) except exception.Invalid as e: last_error = e if last_error: raise last_error
[docs] def or_valid(*validators): """Validates if at least one supplied validator passes :param name: Name of the argument :param value: A value :returns: The value returned from the first successful validator :raises: The error from the last validator when every validation fails """ assert validators, 'No validators specified for or_valid' return functools.partial(_or, validators=validators)
def _and(name, value, validators): for v in validators: value = v(name=name, value=value) return value
[docs] def and_valid(*validators): """Validates that every supplied validator passes The value returned from each validator is passed as the value to the next one. :param name: Name of the argument :param value: A value :returns: The value transformed through every supplied validator :raises: The error from the first failed validator """ assert validators, 'No validators specified for or_valid' return functools.partial(_and, validators=validators)
def _validate_schema(name, value, schema): if value is None: return try: jsonschema.validate(value, schema) except jsonschema.exceptions.ValidationError as e: error_msg = _('Schema error for %s: %s') % (name, e.message) # Sometimes the root message is too generic, try to find a possible # root cause: cause = None current = e while current.context: current = jsonschema.exceptions.best_match(current.context) cause = current.message if cause is not None: error_msg += _('. Possible root cause: %s') % cause raise exception.InvalidParameterValue(error_msg) return value
[docs] def schema(schema): """Return a validator function which validates the value with jsonschema :param: schema dict representing jsonschema to validate with :returns: validator function which takes name and value arguments """ jsonschema.Draft4Validator.check_schema(schema) return functools.partial(_validate_schema, schema=schema)
def _validate_dict(name, value, validators): if value is None: return _validate_types(name, value, (dict, )) for k, v in validators.items(): if k in value: value[k] = v(name=k, value=value[k]) return value
[docs] def dict_valid(**validators): """Return a validator function which validates dict fields Validators will replace the value with the validation result. Any dict item which has no validator is ignored. When a key is missing in the value then the corresponding validator will not be run. :param: validators dict where the key is a dict key to validate and the value is a validator function to run on that value :returns: validator function which takes name and value arguments """ return functools.partial(_validate_dict, validators=validators)
def _validate_types(name, value, types): if not isinstance(value, types): str_types = ', '.join([str(t) for t in types]) raise exception.InvalidParameterValue( _('Expected types %s for %s: %s') % (str_types, name, value)) return value
[docs] def types(*types): """Return a validator function which checks the value is one of the types :param: types one or more types to use for the isinstance test :returns: validator function which takes name and value arguments """ # Replace None with the None type types = tuple((type(None) if tp is None else tp) for tp in types) return functools.partial(_validate_types, types=types)
def _apply_validator(name, value, val_functions): if callable(val_functions): return val_functions(name, value) for v in val_functions: value = v(name, value) return value def _inspect(function): sig = inspect.signature(function) param_keyword = None # **kwargs parameter param_positional = None # *args parameter params = [] for param in sig.parameters.values(): if param.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD: params.append(param) elif param.kind == inspect.Parameter.VAR_KEYWORD: param_keyword = param elif param.kind == inspect.Parameter.VAR_POSITIONAL: param_positional = param else: assert False, 'Unsupported parameter kind %s %s' % ( param.name, param.kind ) return params, param_positional, param_keyword
[docs] def validate(*args, **kwargs): """Decorator which validates and transforms function arguments """ assert not args, 'Validators must be specified by argument name' assert kwargs, 'No validators specified' validators = kwargs def inner_function(function): params, param_positional, param_keyword = _inspect(function) @functools.wraps(function) def inner_check_args(*args, **kwargs): args = list(args) args_len = len(args) kwargs_next = {} next_arg_index = 0 if not param_keyword: # ensure each named argument belongs to a param kwarg_keys = set(kwargs) param_names = set(p.name for p in params) extra_args = kwarg_keys.difference(param_names) if extra_args: raise exception.InvalidParameterValue( _('Unexpected arguments: %s') % ', '.join(extra_args)) for i, param in enumerate(params): if i == 0 and param.name == 'self': # skip validating self continue val_function = validators.get(param.name) if not val_function: continue if i < args_len: # validate positional argument args[i] = val_function(param.name, args[i]) next_arg_index = i + 1 elif param.name in kwargs: # validate keyword argument kwargs_next[param.name] = val_function( param.name, kwargs.pop(param.name)) elif param.default == inspect.Parameter.empty: # no argument was provided, and there is no default # in the parameter, so this is a mandatory argument raise exception.MissingParameterValue( _('Missing mandatory parameter: %s') % param.name) if param_positional: # handle validating *args val_function = validators.get(param_positional.name) remaining = args[next_arg_index:] if val_function and remaining: args = args[:next_arg_index] args.extend(val_function(param_positional.name, remaining)) # handle validating remaining **kwargs if kwargs: val_function = (param_keyword and validators.get(param_keyword.name)) if val_function: kwargs_next.update( val_function(param_keyword.name, kwargs)) else: # make sure unvalidated keyword arguments are kept kwargs_next.update(kwargs) return function(*args, **kwargs_next) return inner_check_args return inner_function
patch = schema({ 'type': 'array', 'items': { 'type': 'object', 'properties': { 'path': {'type': 'string', 'pattern': '^(/[\\w-]+)+$'}, 'op': {'type': 'string', 'enum': ['add', 'replace', 'remove']}, 'value': {} }, 'additionalProperties': False, 'required': ['op', 'path'] } }) """Validate a patch API operation"""