Source code for robot.running.librarykeywordrunner

#  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
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  See the License for the specific language governing permissions and
#  limitations under the License.

from robot.errors import DataError
from robot.output import LOGGER
from robot.result import Keyword as KeywordResult
from robot.utils import prepr, safe_str
from robot.variables import contains_variable, is_list_variable, VariableAssignment

from .bodyrunner import BodyRunner
from .model import Keyword
from .outputcapture import OutputCapturer
from .signalhandler import STOP_SIGNAL_MONITOR
from .statusreporter import StatusReporter

[docs]class LibraryKeywordRunner: def __init__(self, handler, name=None): self._handler = handler = name or self.pre_run_messages = None @property def library(self): return self._handler.library @property def libname(self): return @property def longname(self): return '%s.%s' % (,
[docs] def run(self, kw, context, run=True): assignment = VariableAssignment(kw.assign) result = self._get_result(kw, assignment) with StatusReporter(kw, result, context, run): if run: with assignment.assigner(context) as assigner: return_value = self._run(context, kw.args) assigner.assign(return_value) return return_value
def _get_result(self, kw, assignment): handler = self._handler return KeywordResult(, libname=handler.libname, doc=handler.shortdoc, args=kw.args, assign=tuple(assignment), tags=handler.tags, type=kw.type) def _run(self, context, args): if self.pre_run_messages: for message in self.pre_run_messages: context.output.message(message) variables = context.variables if not context.dry_run else None positional, named = self._handler.resolve_arguments(args, variables) context.output.trace(lambda: self._trace_log_args(positional, named)) runner = self._runner_for(context, self._handler.current_handler(), positional, dict(named)) return self._run_with_output_captured_and_signal_monitor(runner, context) def _trace_log_args(self, positional, named): args = [prepr(arg) for arg in positional] args += ['%s=%s' % (safe_str(n), prepr(v)) for n, v in named] return 'Arguments: [ %s ]' % ' | '.join(args) def _runner_for(self, context, handler, positional, named): timeout = self._get_timeout(context) if timeout and def runner(): with LOGGER.delayed_logging: context.output.debug(timeout.get_message) return, args=positional, kwargs=named) return runner return lambda: handler(*positional, **named) def _get_timeout(self, context): return min(context.timeouts) if context.timeouts else None def _run_with_output_captured_and_signal_monitor(self, runner, context): with OutputCapturer(): return self._run_with_signal_monitoring(runner, context) def _run_with_signal_monitoring(self, runner, context): try: STOP_SIGNAL_MONITOR.start_running_keyword(context.in_teardown) return runner() finally: STOP_SIGNAL_MONITOR.stop_running_keyword()
[docs] def dry_run(self, kw, context): assignment = VariableAssignment(kw.assign) result = self._get_result(kw, assignment) with StatusReporter(kw, result, context, run=False): assignment.validate_assignment() self._dry_run(context, kw.args)
def _dry_run(self, context, args): if self._executed_in_dry_run(self._handler): self._run(context, args) else: self._handler.resolve_arguments(args) def _executed_in_dry_run(self, handler): keywords_to_execute = ('BuiltIn.Import Library', 'BuiltIn.Set Library Search Order', 'BuiltIn.Set Tags', 'BuiltIn.Remove Tags') return (handler.libname == 'Reserved' or handler.longname in keywords_to_execute)
[docs]class EmbeddedArgumentsRunner(LibraryKeywordRunner): def __init__(self, handler, name): super().__init__(handler, name) self._embedded_args = handler.name_regexp.match(name).groups() def _run(self, context, args): if args: raise DataError("Positional arguments are not allowed when using " "embedded arguments.") return super()._run(context, self._embedded_args) def _dry_run(self, context, args): return super()._dry_run(context, self._embedded_args) def _get_result(self, kw, assignment): result = super()._get_result(kw, assignment) result.sourcename = return result
[docs]class RunKeywordRunner(LibraryKeywordRunner): def __init__(self, handler, default_dry_run_keywords=False): super().__init__(handler) self._default_dry_run_keywords = default_dry_run_keywords def _get_timeout(self, context): return None def _run_with_output_captured_and_signal_monitor(self, runner, context): return self._run_with_signal_monitoring(runner, context) def _dry_run(self, context, args): super()._dry_run(context, args) keywords = [kw for kw in self._get_dry_run_keywords(args) if not contains_variable(] BodyRunner(context).run(keywords) def _get_dry_run_keywords(self, args): name = if name == 'Run Keyword If': return self._get_run_kw_if_keywords(args) if name == 'Run Keywords': return self._get_run_kws_keywords(args) if self._default_dry_run_keywords: return self._get_default_run_kw_keywords(args) return [] def _get_run_kw_if_keywords(self, given_args): for kw_call in self._get_run_kw_if_calls(given_args): if kw_call: yield Keyword(name=kw_call[0], args=kw_call[1:]) def _get_run_kw_if_calls(self, given_args): while 'ELSE IF' in given_args: kw_call, given_args = self._split_run_kw_if_args(given_args, 'ELSE IF', 2) yield kw_call if 'ELSE' in given_args: kw_call, else_call = self._split_run_kw_if_args(given_args, 'ELSE', 1) yield kw_call yield else_call elif self._validate_kw_call(given_args): expr, kw_call = given_args[0], given_args[1:] if not is_list_variable(expr): yield kw_call def _split_run_kw_if_args(self, given_args, control_word, required_after): index = list(given_args).index(control_word) expr_and_call = given_args[:index] remaining = given_args[index+1:] if not (self._validate_kw_call(expr_and_call) and self._validate_kw_call(remaining, required_after)): raise DataError("Invalid 'Run Keyword If' usage.") if is_list_variable(expr_and_call[0]): return (), remaining return expr_and_call[1:], remaining def _validate_kw_call(self, kw_call, min_length=2): if len(kw_call) >= min_length: return True return any(is_list_variable(item) for item in kw_call) def _get_run_kws_keywords(self, given_args): for kw_call in self._get_run_kws_calls(given_args): yield Keyword(name=kw_call[0], args=kw_call[1:]) def _get_run_kws_calls(self, given_args): if 'AND' not in given_args: for kw_call in given_args: yield [kw_call,] else: while 'AND' in given_args: index = list(given_args).index('AND') kw_call, given_args = given_args[:index], given_args[index + 1:] yield kw_call if given_args: yield given_args def _get_default_run_kw_keywords(self, given_args): index = list(self._handler.arguments.positional).index('name') return [Keyword(name=given_args[index], args=given_args[index+1:])]