#
# 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 collections
import json
import numbers
import re
from oslo_cache import core
from oslo_config import cfg
from oslo_log import log
from oslo_utils import reflection
from oslo_utils import strutils
from heat.common import cache
from heat.common import exception
from heat.common.i18n import _
from heat.engine import resources
# decorator that allows to cache the value
# of the function based on input arguments
MEMOIZE = core.get_memoization_decorator(conf=cfg.CONF,
region=cache.get_cache_region(),
group="constraint_validation_cache")
LOG = log.getLogger(__name__)
[docs]
class Schema(collections.abc.Mapping):
"""Schema base class for validating properties or parameters.
Schema objects are serializable to dictionaries following a superset of
the HOT input Parameter schema using dict().
Serialises to JSON in the form::
{
'type': 'list',
'required': False
'constraints': [
{
'length': {'min': 1},
'description': 'List must not be empty'
}
],
'schema': {
'*': {
'type': 'string'
}
},
'description': 'An example list property.'
}
"""
KEYS = (
TYPE, DESCRIPTION, DEFAULT, SCHEMA, REQUIRED, CONSTRAINTS,
IMMUTABLE,
) = (
'type', 'description', 'default', 'schema', 'required', 'constraints',
'immutable',
)
# Keywords for data types; each Schema subclass can define its respective
# type name used in templates
TYPE_KEYS = (
INTEGER_TYPE, STRING_TYPE, NUMBER_TYPE, BOOLEAN_TYPE, MAP_TYPE,
LIST_TYPE,
) = (
'INTEGER', 'STRING', 'NUMBER', 'BOOLEAN', 'MAP',
'LIST',
)
# Default type names for data types used in templates; can be overridden by
# subclasses
TYPES = (
INTEGER, STRING, NUMBER, BOOLEAN, MAP, LIST, ANY,
) = (
'Integer', 'String', 'Number', 'Boolean', 'Map', 'List', 'Any',
)
def __init__(self, data_type, description=None,
default=None, schema=None,
required=False, constraints=None, label=None,
immutable=False):
self._len = None
self.label = label
self.type = data_type
if self.type not in self.TYPES:
raise exception.InvalidSchemaError(
message=_('Invalid type (%s)') % self.type)
if required and default is not None:
LOG.warning("Option 'required=True' should not be used with "
"any 'default' value (%s)", default)
self.description = description
self.required = required
self.immutable = immutable
if isinstance(schema, type(self)):
if self.type != self.LIST:
msg = _('Single schema valid only for '
'%(ltype)s, not %(utype)s') % dict(ltype=self.LIST,
utype=self.type)
raise exception.InvalidSchemaError(message=msg)
self.schema = AnyIndexDict(schema)
else:
self.schema = schema
if self.schema is not None and self.type not in (self.LIST,
self.MAP):
msg = _('Schema valid only for %(ltype)s or '
'%(mtype)s, not %(utype)s') % dict(ltype=self.LIST,
mtype=self.MAP,
utype=self.type)
raise exception.InvalidSchemaError(message=msg)
self.constraints = constraints or []
self.default = default
[docs]
def validate(self, context=None):
"""Validates the schema.
This method checks if the schema itself is valid, and if the
default value - if present - complies to the schema's constraints.
"""
for c in self.constraints:
if not self._is_valid_constraint(c):
err_msg = _('%(name)s constraint '
'invalid for %(utype)s') % dict(
name=type(c).__name__,
utype=self.type)
raise exception.InvalidSchemaError(message=err_msg)
self._validate_default(context)
# validated nested schema(ta)
if self.schema:
if isinstance(self.schema, AnyIndexDict):
self.schema.value.validate(context)
else:
for nested_schema in self.schema.values():
nested_schema.validate(context)
def _validate_default(self, context):
if self.default is not None:
try:
self.validate_constraints(self.default, context,
[CustomConstraint])
except (ValueError, TypeError) as exc:
raise exception.InvalidSchemaError(
message=_('Invalid default %(default)s (%(exc)s)') %
dict(default=self.default, exc=exc))
[docs]
def set_default(self, default=None):
"""Set the default value for this Schema object."""
self.default = default
def _is_valid_constraint(self, constraint):
valid_types = getattr(constraint, 'valid_types', [])
return any(self.type == getattr(self, t, None) for t in valid_types)
[docs]
@staticmethod
def str_to_num(value):
"""Convert a string representation of a number into a numeric type."""
if isinstance(value, numbers.Number):
return value
try:
return int(value)
except ValueError:
return float(value)
[docs]
def to_schema_type(self, value):
"""Returns the value in the schema's data type."""
try:
# We have to be backwards-compatible for Integer and Number
# Schema types and try to convert string representations of
# number into "real" number types, therefore calling
# str_to_num below.
if self.type == self.INTEGER:
num = Schema.str_to_num(value)
if isinstance(num, float):
raise ValueError(_('%s is not an integer.') % num)
return num
elif self.type == self.NUMBER:
return Schema.str_to_num(value)
elif self.type == self.STRING:
return str(value)
elif self.type == self.BOOLEAN:
return strutils.bool_from_string(str(value),
strict=True)
except ValueError:
raise ValueError(_('Value "%(val)s" is invalid for data type '
'"%(type)s".')
% {'val': value, 'type': self.type})
return value
[docs]
def validate_constraints(self, value, context=None, skipped=None):
if not skipped:
skipped = []
try:
for constraint in self.constraints:
if type(constraint) not in skipped:
constraint.validate(value, self, context)
except ValueError as ex:
raise exception.StackValidationFailed(message=str(ex))
def __getitem__(self, key):
if key == self.TYPE:
return self.type.lower()
elif key == self.DESCRIPTION:
if self.description is not None:
return self.description
elif key == self.DEFAULT:
if self.default is not None:
return self.default
elif key == self.SCHEMA:
if self.schema is not None:
return dict((n, dict(s)) for n, s in self.schema.items())
elif key == self.REQUIRED:
return self.required
elif key == self.CONSTRAINTS:
if self.constraints:
return [dict(c) for c in self.constraints]
raise KeyError(key)
def __iter__(self):
for k in self.KEYS:
try:
self[k]
except KeyError:
pass
else:
yield k
def __len__(self):
if self._len is None:
self._len = len(list(iter(self)))
return self._len
[docs]
class AnyIndexDict(collections.abc.Mapping):
"""A Mapping that returns the same value for any integer index.
Used for storing the schema for a list. When converted to a dictionary,
it contains a single item with the key '*'.
"""
ANYTHING = '*'
def __init__(self, value):
self.value = value
def __getitem__(self, key):
if key != self.ANYTHING and not isinstance(key, int):
raise KeyError(_('Invalid key %s') % key)
return self.value
def __iter__(self):
yield self.ANYTHING
def __len__(self):
return 1
[docs]
class Constraint(collections.abc.Mapping):
"""Parent class for constraints on allowable values for a Property.
Constraints are serializable to dictionaries following the HOT input
Parameter constraints schema using dict().
"""
(DESCRIPTION,) = ('description',)
def __init__(self, description=None):
self.description = description
def __str__(self):
def desc():
if self.description:
yield self.description
yield self._str()
return '\n'.join(desc())
[docs]
def validate(self, value, schema=None, context=None):
if not self._is_valid(value, schema, context):
if self.description:
err_msg = self.description
else:
err_msg = self._err_msg(value)
raise ValueError(err_msg)
@classmethod
def _name(cls):
return '_'.join(w.lower() for w in re.findall('[A-Z]?[a-z]+',
cls.__name__))
def __getitem__(self, key):
if key == self.DESCRIPTION:
if self.description is None:
raise KeyError(key)
return self.description
if key == self._name():
return self._constraint()
raise KeyError(key)
def __iter__(self):
if self.description is not None:
yield self.DESCRIPTION
yield self._name()
def __len__(self):
return 2 if self.description is not None else 1
[docs]
class Range(Constraint):
"""Constrain values within a range.
Serializes to JSON as::
{
'range': {'min': <min>, 'max': <max>},
'description': <description>
}
"""
(MIN, MAX) = ('min', 'max')
valid_types = (Schema.INTEGER_TYPE, Schema.NUMBER_TYPE,)
def __init__(self, min=None, max=None, description=None):
super(Range, self).__init__(description)
self.min = min
self.max = max
for param in (min, max):