def test_formatter_returns_informative_error(self):
     with self.assertRaisesRegexp(ValueError, '{a}'):
         traverse(
             self.data,
             formatter_context={'b': 13},
             keywords={'generator': self.TestGenerator.evaluate}
         )

    def get_attributes(cls, instance, all_plugins_versions=False):
        """Get attributes for current Cluster instance.

        :param instance: Cluster instance
        :param all_plugins_versions: Get attributes of all versions of plugins
        :returns: dict
        """
        try:
            attrs = db().query(models.Attributes).filter(
                models.Attributes.cluster_id == instance.id
            ).one()
        except MultipleResultsFound:
            raise errors.InvalidData(
                u"Multiple rows with attributes were found for cluster '{0}'"
                .format(instance.name)
            )
        except NoResultFound:
            raise errors.InvalidData(
                u"No attributes were found for cluster '{0}'"
                .format(instance.name)
            )
        attrs = dict(attrs)

        # Merge plugins attributes into editable ones
        plugin_attrs = PluginManager.get_plugins_attributes(
            instance, all_versions=all_plugins_versions)
        plugin_attrs = traverse(plugin_attrs, AttributesGenerator, {
            'cluster': instance,
            'settings': settings,
        })
        attrs['editable'].update(plugin_attrs)

        return attrs

 def generate_fields(cls, instance):
     instance.generated = traverse(
         instance.generated,
         AttributesGenerator
     )
     db().add(instance)
     db().flush()

    def serialize(self, node_id, formatter_context=None):
        """Serialize task in expected by orchestrator format

        If serialization is performed on the remote worker
        we should pass formatter_context parameter with values
        from the master node settings

        :param formatter_context: formatter context
        :param node_id: the target node_id
        """

        logger.debug(
            "serialize task %s for node %s",
            self.task_template['id'], node_id
        )
        formatter_context = formatter_context \
            or self.context.get_formatter_context(node_id)
        task = utils.traverse(
            self.task_template,
            utils.text_format_safe,
            formatter_context,
            {
                'yaql_exp': self.context.get_yaql_interpreter(
                    node_id, self.task_template['id'])
            }
        )
        return self.normalize(self.finalize(task, node_id))

 def get_default_vmware_attributes(cls, instance):
     """Get metadata from release with empty value section."""
     editable = instance.release.vmware_attributes_metadata.get("editable")
     editable = traverse(editable, AttributesGenerator, {
         'cluster': instance,
         'settings': settings,
     })
     return editable

    def generate_fields(cls, instance):
        """Generate field values for Cluster attributes using
        generators.

        :param instance: Attributes instance
        :returns: None
        """
        instance.generated = traverse(
            instance.generated or {}, AttributesGenerator, {"cluster": instance.cluster, "settings": settings}
        )

    def generate_fields(cls, instance):
        """Generate field values for Cluster attributes using
        generators.

        :param instance: Attributes instance
        :returns: None
        """
        instance.generated = traverse(
            instance.generated,
            AttributesGenerator
        )

    def get_default_editable_attributes(cls, instance):
        """Get editable attributes from release metadata

        :param instance: Cluster instance
        :returns: Dict object
        """
        editable = instance.release.attributes_metadata.get("editable")
        editable = traverse(editable, AttributesGenerator, {"cluster": instance, "settings": settings})
        # when attributes created we need to understand whether should plugin
        # be applied for created cluster
        plugin_attrs = PluginManager.get_plugin_attributes(instance)
        editable = dict(plugin_attrs, **editable)
        return editable

 def test_custom_keywords(self):
     data = {
         'key1': {'exp1': 'name1', 'exp1_arg': 'arg1'},
         'key2': {'exp2': 'name2'},
     }
     generator1 = mock.MagicMock(return_value='val1')
     generator2 = mock.MagicMock(return_value='val2')
     result = traverse(data, keywords={
         'exp1': generator1,
         'exp2': generator2
     })
     self.assertEqual({'key1': 'val1', 'key2': 'val2'}, result)
     generator1.assert_called_once_with('name1', 'arg1')
     generator2.assert_called_once_with('name2')

    def get_default_editable_attributes(cls, instance):
        """Get editable attributes from release metadata

        :param instance: Cluster instance
        :returns: Dict object
        """
        editable = instance.release.attributes_metadata.get("editable")
        # Add default attributes of connected plugins
        plugin_attrs = PluginManager.get_plugins_attributes(
            instance, all_versions=True, default=True)
        editable = dict(plugin_attrs, **editable)
        editable = traverse(editable, AttributesGenerator, {
            'cluster': instance,
            'settings': settings,
        })

        return editable

    def test_w_formatting_context(self):
        result = traverse(self.data, self.TestGenerator, {'a': 13})

        self.assertEqual(result, {
            'foo': 'testvalue',
            'bar': 'test 13 string',
            'baz': 42,
            'regex': {
                'source': 'test {a} string',
                'error': 'an {a} error'
            },
            'list': [
                {
                    'x': 'a 13 a',
                },
                {
                    'y': 'b 42 b',
                }
            ]})

    def serialize(self, node_id):
        """Serialize task in expected by orchestrator format.

        :param node_id: the target node_id
        """

        logger.debug(
            "serialize task %s for node %s",
            self.task_template['id'], node_id
        )
        task = utils.traverse(
            self.task_template,
            utils.text_format_safe,
            self.context.get_formatter_context(node_id),
            {
                'yaql_exp': self.context.get_yaql_interpreter(
                    node_id, self.task_template['id'])
            }
        )
        return self.normalize(self.finalize(task, node_id))

    def serialize(self, node_id):
        logger.debug(
            "serialize task %s for node %s",
            self.task_template['id'], node_id
        )
        task = utils.traverse(
            self.task_template,
            utils.text_format_safe,
            self.context.get_formatter_context(node_id),
            {
                'yaql_exp': self.context.get_yaql_interpreter(node_id)
            }
        )
        if not self.should_execute(task, node_id):
            logger.debug(
                "Task %s is skipped by condition.", self.task_template['id']
            )
            return super(DefaultTaskSerializer, self).serialize(node_id)

        task.setdefault('parameters', {}).setdefault('cwd', '/')
        task.setdefault('fail_on_error', True)
        return self.get_required_fields(task)

    def test_wo_formatting_context(self):
        result = traverse(
            self.data,
            keywords={'generator': self.TestGenerator.evaluate}
        )

        self.assertEqual(result, {
            'foo': 'testvalue',
            'bar': 'test {a} string',
            'baz': 42,
            'regex': {
                'source': 'test {a} string',
                'error': 'an {a} error'
            },
            'list': [
                {
                    'x': 'a {a} a',
                },
                {
                    'y': 'b 42 b',
                }
            ]})

    def test_w_safe_formatting_context(self):
        data = self.data.copy()
        data['bar'] = 'test {b} value'
        result = traverse(
            data, text_format_safe, {'a': 13},
            keywords={'generator': self.TestGenerator.evaluate}
        )

        self.assertEqual(result, {
            'foo': 'testvalue',
            'bar': 'test {b} value',
            'baz': 42,
            'regex': {
                'source': 'test {a} string',
                'error': 'an {a} error'
            },
            'list': [
                {
                    'x': 'a 13 a',
                },
                {
                    'y': 'b 42 b',
                }
            ]})