Source code for heat.engine.template_common

#
#    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 functools
import weakref

from heat.common import exception
from heat.common.i18n import _
from heat.engine import conditions
from heat.engine import function
from heat.engine import output
from heat.engine import template


[docs]class CommonTemplate(template.Template):
    """A class of the common implementation for HOT and CFN templates.

    This is *not* a stable interface, and any third-parties who create derived
    classes from it do so at their own risk.
    """

    def __init__(self, template, template_id=None, files=None, env=None):
        super(CommonTemplate, self).__init__(template, template_id=template_id,
                                             files=files, env=env)
        self._conditions_cache = None, None

    @classmethod
    def _parse_resource_field(cls, key, valid_types, typename,
                              rsrc_name, rsrc_data, parse_func):
        """Parse a field in a resource definition.

        :param key: The name of the key
        :param valid_types: Valid types for the parsed output
        :param typename: Description of valid type to include in error output
        :param rsrc_name: The resource name
        :param rsrc_data: The unparsed resource definition data
        :param parse_func: A function to parse the data, which takes the
            contents of the field and its path in the template as arguments.
        """
        if key in rsrc_data:
            data = parse_func(rsrc_data[key], '.'.join([cls.RESOURCES,
                                                        rsrc_name,
                                                        key]))
            if not isinstance(data, valid_types):
                args = {'name': rsrc_name, 'key': key,
                        'typename': typename}
                message = _('Resource %(name)s %(key)s type '
                            'must be %(typename)s') % args
                raise TypeError(message)
            return data
        else:
            return None

    def _rsrc_defn_args(self, stack, name, data):
        if self.RES_TYPE not in data:
            args = {'name': name, 'type_key': self.RES_TYPE}
            msg = _('Resource %(name)s is missing "%(type_key)s"') % args
            raise KeyError(msg)

        parse = functools.partial(self.parse, stack)

        def no_parse(field, path):
            return field

        yield ('resource_type',
               self._parse_resource_field(self.RES_TYPE,
                                          str, 'string',
                                          name, data, parse))

        yield ('properties',
               self._parse_resource_field(self.RES_PROPERTIES,
                                          (collections.abc.Mapping,
                                           function.Function), 'object',
                                          name, data, parse))

        yield ('metadata',
               self._parse_resource_field(self.RES_METADATA,
                                          (collections.abc.Mapping,
                                           function.Function), 'object',
                                          name, data, parse))

        depends = self._parse_resource_field(self.RES_DEPENDS_ON,
                                             collections.abc.Sequence,
                                             'list or string',
                                             name, data, no_parse)
        if isinstance(depends, str):
            depends = [depends]
        elif depends:
            for dep in depends:
                if not isinstance(dep, str):
                    msg = _('Resource %(name)s %(key)s '
                            'must be a list of strings') % {
                                'name': name, 'key': self.RES_DEPENDS_ON}
                    raise exception.StackValidationFailed(message=msg)

        yield 'depends', depends

        del_policy = self._parse_resource_field(self.RES_DELETION_POLICY,
                                                (str,
                                                 function.Function),
                                                'string',
                                                name, data, parse)
        deletion_policy = function.resolve(del_policy)
        if deletion_policy is not None:
            if deletion_policy not in self.deletion_policies:
                msg = _('Invalid deletion policy "%s"') % deletion_policy
                raise exception.StackValidationFailed(message=msg)
            else:
                deletion_policy = self.deletion_policies[deletion_policy]
        yield 'deletion_policy', deletion_policy

        yield ('update_policy',
               self._parse_resource_field(self.RES_UPDATE_POLICY,
                                          (collections.abc.Mapping,
                                           function.Function), 'object',
                                          name, data, parse))

        yield ('description',
               self._parse_resource_field(self.RES_DESCRIPTION,
                                          str, 'string',
                                          name, data, no_parse))

    def _get_condition_definitions(self):
        """Return the condition definitions of template."""
        return {}

[docs]    def conditions(self, stack):
        get_cache_stack, cached_conds = self._conditions_cache
        if (cached_conds is not None and
                get_cache_stack is not None and
                get_cache_stack() is stack):
            return cached_conds

        raw_defs = self._get_condition_definitions()
        if not isinstance(raw_defs, collections.abc.Mapping):
            message = _('Condition definitions must be a map. Found a '
                        '%s instead') % type(raw_defs).__name__
            raise exception.StackValidationFailed(
                error='Conditions validation error',
                message=message)

        parsed = {n: self.parse_condition(stack, c,
                                          '.'.join([self.CONDITIONS, n]))
                  for n, c in raw_defs.items()}
        conds = conditions.Conditions(parsed)

        get_cache_stack = weakref.ref(stack) if stack is not None else None
        self._conditions_cache = get_cache_stack, conds
        return conds

[docs]    def outputs(self, stack):
        conds = self.conditions(stack)

        outputs = self.t.get(self.OUTPUTS) or {}

        def get_outputs():
            for key, val in outputs.items():
                if not isinstance(val, collections.abc.Mapping):
                    message = _('Output definitions must be a map. Found a '
                                '%s instead') % type(val).__name__
                    raise exception.StackValidationFailed(
                        error='Output validation error',
                        path=[self.OUTPUTS, key],
                        message=message)

                if self.OUTPUT_VALUE not in val:
                    message = _('Each output definition must contain '
                                'a %s key.') % self.OUTPUT_VALUE
                    raise exception.StackValidationFailed(
                        error='Output validation error',
                        path=[self.OUTPUTS, key],
                        message=message)

                description = val.get(self.OUTPUT_DESCRIPTION)

                if hasattr(self, 'OUTPUT_CONDITION'):
                    path = [self.OUTPUTS, key, self.OUTPUT_CONDITION]
                    cond = self.parse_condition(stack,
                                                val.get(self.OUTPUT_CONDITION),
                                                '.'.join(path))
                    try:
                        enabled = conds.is_enabled(function.resolve(cond))
                    except ValueError as exc:
                        path = [self.OUTPUTS, key, self.OUTPUT_CONDITION]
                        message = str(exc)
                        raise exception.StackValidationFailed(path=path,
                                                              message=message)

                    if not enabled:
                        yield key, output.OutputDefinition(key, None,
                                                           description)
                        continue

                value_def = self.parse(stack, val[self.OUTPUT_VALUE],
                                       path='.'.join([self.OUTPUTS, key,
                                                      self.OUTPUT_VALUE]))

                yield key, output.OutputDefinition(key, value_def, description)

        return dict(get_outputs())