def test_evaluate_recursively_arbitrary_dict(self):
        context = {
            "auth_token": "123",
            "project_id": "mistral"
        }
        data = {
            "parameters": {
                "parameter1": {
                    "name1": "$.auth_token",
                    "name2": "val_name2"
                },
                "param2": [
                    "var1",
                    "var2",
                    "/servers/{$.project_id}/bla"
                ]
            },
            "token": "$.auth_token"
        }

        applied = expr.evaluate_recursively(data, context)

        self.assertDictEqual(
            {
                "parameters": {
                    "parameter1": {
                        "name1": "123",
                        "name2": "val_name2"
                    },
                    "param2": ["var1", "var2", "/servers/mistral/bla"]
                },
                "token": "123"
            },
            applied)

    def _convert_params(self, params):
        base_params_spec = self.action_spec.base_parameters

        if not base_params_spec:
            return {}

        return expr.evaluate_recursively(base_params_spec, params)

def _get_adhoc_action_input(action_def, input_dict,
                            wf_name=None, wf_spec=None):
    action_spec = spec_parser.get_action_spec(action_def.spec)

    base_name = action_spec.get_base()

    action_def = resolve_action_definition(
        base_name,
        wf_name if wf_name else None,
        wf_spec.get_name() if wf_spec else None
    )

    _inject_action_ctx_for_validating(action_def, input_dict)
    e_utils.validate_input(action_def, input_dict, action_spec)

    base_input = action_spec.get_base_input()

    if base_input:
        input_dict = expr.evaluate_recursively(
            base_input,
            input_dict
        )
    else:
        input_dict = {}

    return input_dict

def _get_environment(params):
    env = params.get('env', {})

    if not env:
        return {}

    if isinstance(env, dict):
        env_dict = env
    elif isinstance(env, six.string_types):
        env_db = db_api.load_environment(env)

        if not env_db:
            raise exc.InputException(
                'Environment is not found: %s' % env
            )

        env_dict = env_db.variables
    else:
        raise exc.InputException(
            'Unexpected value type for environment [env=%s, type=%s]'
            % (env, type(env))
        )

    if ('evaluate_env' in params and
            not params['evaluate_env']):
        return env_dict
    else:
        return expr.evaluate_recursively(env_dict, {'__env': env_dict})

    def test_evaluate_recursively(self):
        task_spec_dict = {
            'parameters': {
                'p1': 'My string',
                'p2': '<% $.param2 %>',
                'p3': ''
            },
            'publish': {
                'new_key11': 'new_key1'
            }
        }

        modified_task = expr.evaluate_recursively(
            task_spec_dict,
            {'param2': 'val32'}
        )

        self.assertDictEqual(
            {
                'parameters': {
                    'p1': 'My string',
                    'p2': 'val32',
                    'p3': ''
                },
                'publish': {
                    'new_key11': 'new_key1'
                }
            },
            modified_task
        )

def add_workflow_variables_to_context(wf_ex, wf_spec):
    wf_ex.context = wf_ex.context or {}

    return utils.merge_dicts(
        wf_ex.context,
        expr.evaluate_recursively(wf_spec.get_vars(), wf_ex.context)
    )

def _schedule_run_action(task_ex, task_spec, action_input, index):
    wf_ex = task_ex.workflow_execution
    wf_spec = spec_parser.get_workflow_spec(wf_ex.spec)

    action_spec_name = task_spec.get_action_name()

    action_def = action_handler.resolve_definition(
        action_spec_name,
        task_ex,
        wf_spec
    )

    action_ex = action_handler.create_action_execution(
        action_def, action_input, task_ex, index
    )

    target = expr.evaluate_recursively(
        task_spec.get_target(),
        utils.merge_dicts(
            copy.deepcopy(action_input),
            copy.copy(task_ex.in_context)
        )
    )

    scheduler.schedule_call(
        None,
        'mistral.engine.action_handler.run_existing_action',
        0,
        action_ex_id=action_ex.id,
        target=target
    )

    def _get_with_items_values(self):
        """Returns all values evaluated from 'with-items' expression.

        Example:
          DSL:
            with-items:
              - var1 in <% $.arrayI %>
              - var2 in <% $.arrayJ %>

        where arrayI = [1,2,3] and arrayJ = [a,b,c]

        The result of the method in this case will be:
            {
                'var1': [1,2,3],
                'var2': [a,b,c]
            }

        :return: Evaluated 'with-items' expression values.
        """
        ctx_view = data_flow.ContextView(
            self.ctx,
            self.wf_ex.context,
            self.wf_ex.input
        )

        return expr.evaluate_recursively(
            self.task_spec.get_with_items(),
            ctx_view
        )

def evaluate_object_fields(obj, context):
    fields = inspect_utils.get_public_fields(obj)

    evaluated_fields = expr.evaluate_recursively(fields, context)

    for k, v in evaluated_fields.items():
        setattr(obj, k, v)

def add_environment_to_context(wf_ex):
    wf_ex.context = wf_ex.context or {}

    # If env variables are provided, add an evaluated copy into the context.
    if 'env' in wf_ex.params:
        env = copy.deepcopy(wf_ex.params['env'])
        # An env variable can be an expression of other env variables.
        wf_ex.context['__env'] = expr.evaluate_recursively(env, {'__env': env})

    def _convert_result(self, result):
        transformer = self.action_spec.output

        if not transformer:
            return result

        # Use base action result as a context for evaluating expressions.
        return expr.evaluate_recursively(transformer, result)

 def _get_target(self, input_dict):
     return expr.evaluate_recursively(
         self.task_spec.get_target(),
         utils.merge_dicts(
             copy.deepcopy(input_dict),
             copy.deepcopy(self.ctx)
         )
     )

def get_task_output(task, result):
    publish_transformer = task['task_spec'].get('publish')

    output = expr.evaluate_recursively(publish_transformer, result) or {}

    if result:
        output['task'] = {task['name']: result}

    return output

    def _evaluate_expression(self, expression, ctx=None):
        ctx_view = data_flow.ContextView(
            data_flow.get_current_task_dict(self.task_ex),
            data_flow.get_workflow_environment_dict(self.wf_ex),
            ctx or self.ctx,
            self.wf_ex.context,
            self.wf_ex.input
        )

        return expr.evaluate_recursively(expression, ctx_view)

    def convert(self, event_type, event):
        edef = self.get_event_definition(event_type)

        if edef is None:
            LOG.debug('No event definition found for type: %s, use default '
                      'settings instead.', event_type)

            return expressions.evaluate_recursively(DEFAULT_PROPERTIES, event)

        return edef.convert(event)

    def _get_action_input(self, ctx=None):
        ctx = ctx or self.ctx

        input_dict = expr.evaluate_recursively(self.task_spec.get_input(), ctx)

        return utils.merge_dicts(
            input_dict,
            self._get_action_defaults(),
            overwrite=False
        )

def _get_action_input(wf_spec, task_ex, task_spec, ctx):
    input_dict = expr.evaluate_recursively(task_spec.get_input(), ctx)

    action_spec_name = task_spec.get_action_name()

    action_def = e_utils.resolve_action_definition(
        task_ex.workflow_name,
        wf_spec.get_name(),
        action_spec_name
    )

    input_dict = utils.merge_dicts(
        input_dict,
        _get_action_defaults(task_ex, task_spec),
        overwrite=False
    )

    if action_def.spec:
        # Ad-hoc action.
        action_spec = spec_parser.get_action_spec(action_def.spec)

        base_name = action_spec.get_base()

        action_def = e_utils.resolve_action_definition(
            task_ex.workflow_name,
            wf_spec.get_name(),
            base_name
        )

        e_utils.validate_input(action_def, action_spec, input_dict)

        base_input = action_spec.get_base_input()

        if base_input:
            input_dict = expr.evaluate_recursively(
                base_input,
                input_dict
            )
        else:
            input_dict = {}

    return input_dict

def transform_action_result(action_spec_name, result, wf_name=None, wf_spec_name=None):
    action_def = resolve_action_definition(action_spec_name, wf_name, wf_spec_name)

    if not action_def.spec:
        return result

    transformer = spec_parser.get_action_spec(action_def.spec).get_output()

    if transformer is None:
        return result

    return wf_utils.Result(data=expr.evaluate_recursively(transformer, result.data), error=result.error)

    def test_evaluate_recursively(self):
        task_spec_dict = {
            "parameters": {"p1": "My string", "p2": "<% $.param2 %>", "p3": ""},
            "publish": {"new_key11": "new_key1"},
        }

        modified_task = expr.evaluate_recursively(task_spec_dict, {"param2": "val32"})

        self.assertDictEqual(
            {"parameters": {"p1": "My string", "p2": "val32", "p3": ""}, "publish": {"new_key11": "new_key1"}},
            modified_task,
        )

    def _prepare_input(self, input_dict):
        base_input_expr = self.action_spec.get_base_input()

        if base_input_expr:
            base_input_dict = expr.evaluate_recursively(
                base_input_expr,
                input_dict
            )
        else:
            base_input_dict = {}

        return super(AdHocAction, self)._prepare_input(base_input_dict)