userkeywordrunner.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  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.
 
from itertools import chain
 
from robotide.lib.robot.errors import (ExecutionFailed, ExecutionPassed, ExecutionStatus,
                          ExitForLoop, ContinueForLoop, DataError,
                          PassExecution, ReturnFromKeyword,
                          UserKeywordExecutionFailed, VariableError)
from robotide.lib.robot.result import Keyword as KeywordResult
from robotide.lib.robot.utils import getshortdoc, DotDict, prepr, split_tags_from_doc
from robotide.lib.robot.variables import is_list_var, VariableAssignment
 
from .arguments import DefaultValue
from .statusreporter import StatusReporter
from .steprunner import StepRunner
from .timeouts import KeywordTimeout
 
 
class UserKeywordRunner():
 
    def __init__(self, handler, name=None):
        self._handler_handler = handler
        self.namename = name or handler.name
 
    @property
    longname = property
    
    def longname(self):
        libname = self._handler_handler.libname
        return '%s.%s' % (libname, self.namename) if libname else self.namename
 
    @property
    libname = property
    
    def libname(self):
        return self._handler_handler.libname
 
    @property
    arguments = property
    
    def arguments(self):
        return self._handler_handler.arguments
 
    def run(self, kw, context):
        assignment = VariableAssignment(kw.assign)
        result = self._get_result_get_result(kw, assignment, context.variables)
        with StatusReporter(context, result):
            with assignment.assigner(context) as assigner:
                return_value = self._run_run(context, kw.args, result)
                assigner.assign(return_value)
                return return_value
 
    def _get_result(self, kw, assignment, variables):
        handler = self._handler_handler
        doc = variables.replace_string(handler.doc, ignore_errors=True)
        doc, tags = split_tags_from_doc(doc)
        tags = variables.replace_list(handler.tags, ignore_errors=True) + tags
        return KeywordResult(kwname=self.namename,
                             libname=handler.libname,
                             doc=getshortdoc(doc),
                             args=kw.args,
                             assign=tuple(assignment),
                             tags=tags,
                             type=kw.type)
 
    def _run(self, context, args, result):
        variables = context.variables
        args = self._resolve_arguments_resolve_arguments(args, variables)
        with context.user_keyword:
            self._set_arguments_set_arguments(args, context)
            timeout = self._get_timeout_get_timeout(variables)
            if timeout is not None:
                result.timeout = str(timeout)
            with context.timeout(timeout):
                exception, return_ = self._execute_execute(context)
                if exception and not exception.can_continue(context.in_teardown):
                    raise exception
                return_value = self._get_return_value_get_return_value(variables, return_)
                if exception:
                    exception.return_value = return_value
                    raise exception
                return return_value
 
    def _get_timeout(self, variables=None):
        timeout = self._handler_handler.timeout
        if not timeout:
            return None
        timeout = KeywordTimeout(timeout.value, timeout.message, variables)
        return timeout
 
    def _resolve_arguments(self, arguments, variables=None):
        return self.argumentsargumentsarguments.resolve(arguments, variables)
 
    def _set_arguments(self, arguments, context):
        positional, named = arguments
        variables = context.variables
        args, kwargs = self.argumentsargumentsarguments.map(positional, named,
                                          replace_defaults=False)
        self._set_variables_set_variables(args, kwargs, variables)
        context.output.trace(lambda: self._trace_log_args_message_trace_log_args_message(variables))
 
    def _set_variables(self, positional, kwargs, variables):
        spec = self.argumentsargumentsarguments
        args, varargs = self._split_args_and_varargs_split_args_and_varargs(positional)
        kwonly, kwargs = self._split_kwonly_and_kwargs_split_kwonly_and_kwargs(kwargs)
        for name, value in chain(zip(spec.positional, args), kwonly):
            if isinstance(value, DefaultValue):
                value = value.resolve(variables)
            variables['${%s}' % name] = value
        if spec.varargs:
            variables['@{%s}' % spec.varargs] = varargs
        if spec.kwargs:
            variables['&{%s}' % spec.kwargs] = DotDict(kwargs)
 
    def _split_args_and_varargs(self, args):
        if not self.argumentsargumentsarguments.varargs:
            return args, []
        positional = len(self.argumentsargumentsarguments.positional)
        return args[:positional], args[positional:]
 
    def _split_kwonly_and_kwargs(self, all_kwargs):
        kwonly = []
        kwargs = []
        for name, value in all_kwargs:
            target = kwonly if name in self.argumentsargumentsarguments.kwonlyargs else kwargs
            target.append((name, value))
        return kwonly, kwargs
 
    def _trace_log_args_message(self, variables):
        args = ['${%s}' % arg for arg in self.argumentsargumentsarguments.positional]
        if self.argumentsargumentsarguments.varargs:
            args.append('@{%s}' % self.argumentsargumentsarguments.varargs)
        if self.argumentsargumentsarguments.kwargs:
            args.append('&{%s}' % self.argumentsargumentsarguments.kwargs)
        return self._format_trace_log_args_message_format_trace_log_args_message(args, variables)
 
    def _format_trace_log_args_message(self, args, variables):
        args = ['%s=%s' % (name, prepr(variables[name])) for name in args]
        return 'Arguments: [ %s ]' % ' | '.join(args)
 
    def _execute(self, context):
        handler = self._handler_handler
        if not (handler.keywords or handler.return_value):
            raise DataError("User keyword '%s' contains no keywords." % self.namename)
        if context.dry_run and 'robot:no-dry-run' in handler.tags:
            return None, None
        error = return_ = pass_ = None
        try:
            StepRunner(context).run_steps(handler.keywords)
        except ReturnFromKeyword as exception:
            return_ = exception
            error = exception.earlier_failures
        except (ExitForLoop, ContinueForLoop) as exception:
            pass_ = exception
        except ExecutionPassed as exception:
            pass_ = exception
            error = exception.earlier_failures
            if error:
                error.continue_on_failure = False
        except ExecutionFailed as exception:
            error = exception
        with context.keyword_teardown(error):
            td_error = self._run_teardown_run_teardown(context)
        if error or td_error:
            error = UserKeywordExecutionFailed(error, td_error)
        return error or pass_, return_
 
    def _get_return_value(self, variables, return_):
        ret = self._handler_handler.return_value if not return_ else return_.return_value
        if not ret:
            return None
        contains_list_var = any(is_list_var(item) for item in ret)
        try:
            ret = variables.replace_list(ret)
        except DataError as err:
            raise VariableError('Replacing variables from keyword return value '
                                'failed: %s' % err.message)
        if len(ret) != 1 or contains_list_var:
            return ret
        return ret[0]
 
    def _run_teardown(self, context):
        if not self._handler_handler.teardown:
            return None
        try:
            name = context.variables.replace_string(self._handler_handler.teardown.name)
        except DataError as err:
            return ExecutionFailed(err.message, syntax=True)
        if name.upper() in ('', 'NONE'):
            return None
        try:
            StepRunner(context).run_step(self._handler_handler.teardown, name)
        except PassExecution:
            return None
        except ExecutionStatus as err:
            return err
        return None
 
    def dry_run(self, kw, context):
        assignment = VariableAssignment(kw.assign)
        result = self._get_result_get_result(kw, assignment, context.variables)
        with StatusReporter(context, result):
            assignment.validate_assignment()
            self._dry_run_dry_run(context, kw.args, result)
 
    def _dry_run(self, context, args, result):
        self._resolve_arguments_resolve_arguments(args)
        with context.user_keyword:
            timeout = self._get_timeout_get_timeout()
            if timeout:
                result.timeout = str(timeout)
            error, _ = self._execute_execute(context)
            if error:
                raise error
 
 
class EmbeddedArgumentsRunner(UserKeywordRunner):
 
    def __init__(self, handler, name):
        UserKeywordRunner.__init__(self, handler, name)
        match = handler.embedded_name.match(name)
        if not match:
            raise ValueError('Does not match given name')
        self.embedded_argsembedded_args = list(zip(handler.embedded_args, match.groups()))
 
    def _resolve_arguments(self, args, variables=None):
        # Validates that no arguments given.
        self.argumentsargumentsarguments.resolve(args, variables)
        if not variables:
            return []
        return [(n, variables.replace_scalar(v)) for n, v in self.embedded_argsembedded_args]
 
    def _set_arguments(self, embedded_args, context):
        variables = context.variables
        for name, value in embedded_args:
            variables['${%s}' % name] = value
        context.output.trace(lambda: self._trace_log_args_message_trace_log_args_message_trace_log_args_message(variables))
 
    def _trace_log_args_message(self, variables):
        args = ['${%s}' % arg for arg, _ in self.embedded_argsembedded_args]
        return self._format_trace_log_args_message_format_trace_log_args_message(args, variables)