runner.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 robotide.lib.robot.errors import ExecutionFailed, ExecutionStatus, DataError, PassExecution
from robotide.lib.robot.model import SuiteVisitor
from robotide.lib.robot.result import TestSuite, Result
from robotide.lib.robot.utils import get_timestamp, is_list_like, NormalizedDict, unic
from robotide.lib.robot.variables import VariableScopes
 
from .context import EXECUTION_CONTEXTS
from .steprunner import StepRunner
from .namespace import Namespace
from .status import SuiteStatus, TestStatus
from .timeouts import TestTimeout
 
 
# Some 'extract method' love needed here. Perhaps even 'extract class'.
 
class Runner(SuiteVisitor):
 
    def __init__(self, output, settings):
        self.resultresult = None
        self._output_output = output
        self._settings_settings = settings
        self._variables_variables = VariableScopes(settings)
        self._suite_suite = None
        self._suite_status_suite_status = None
        self._executed_tests_executed_tests = None
 
    @property
    _context = property
    
    def _context(self):
        return EXECUTION_CONTEXTS.current
 
    def start_suite(self, suite):
        self._output_output.library_listeners.new_suite_scope()
        result = TestSuite(source=suite.source,
                           name=suite.name,
                           doc=suite.doc,
                           metadata=suite.metadata,
                           starttime=get_timestamp(),
                           rpa=self._settings_settings.rpa)
        if not self.resultresult:
            result.set_criticality(self._settings_settings.critical_tags,
                                   self._settings_settings.non_critical_tags)
            self.resultresult = Result(root_suite=result, rpa=self._settings_settings.rpa)
            self.resultresult.configure(status_rc=self._settings_settings.status_rc,
                                  stat_config=self._settings_settings.statistics_config)
        else:
            self._suite_suite.suites.append(result)
        self._suite_suite = result
        self._suite_status_suite_status = SuiteStatus(self._suite_status_suite_status,
                                         self._settings_settings.exit_on_failure,
                                         self._settings_settings.exit_on_error,
                                         self._settings_settings.skip_teardown_on_exit)
        ns = Namespace(self._variables_variables, result, suite.resource)
        ns.start_suite()
        ns.variables.set_from_variable_table(suite.resource.variables)
        EXECUTION_CONTEXTS.start_suite(result, ns, self._output_output,
                                       self._settings_settings.dry_run)
        self._context_context_context.set_suite_variables(result)
        if not self._suite_status_suite_status.failures:
            ns.handle_imports()
            ns.variables.resolve_delayed()
        result.doc = self._resolve_setting_resolve_setting(result.doc)
        result.metadata = [(self._resolve_setting_resolve_setting(n), self._resolve_setting_resolve_setting(v))
                           for n, v in result.metadata.items()]
        self._context_context_context.set_suite_variables(result)
        self._output_output.start_suite(ModelCombiner(suite, result,
                                               tests=suite.tests,
                                               suites=suite.suites,
                                               test_count=suite.test_count))
        self._output_output.register_error_listener(self._suite_status_suite_status.error_occurred)
        self._run_setup_run_setup(suite.keywords.setup, self._suite_status_suite_status)
        self._executed_tests_executed_tests = NormalizedDict(ignore='_')
 
    def _resolve_setting(self, value):
        if is_list_like(value):
            return self._variables_variables.replace_list(value, ignore_errors=True)
        return self._variables_variables.replace_string(value, ignore_errors=True)
 
    def end_suite(self, suite):
        self._suite_suite.message = self._suite_status_suite_status.message
        self._context_context_context.report_suite_status(self._suite_suite.status,
                                          self._suite_suite.full_message)
        with self._context_context_context.suite_teardown():
            failure = self._run_teardown_run_teardown(suite.keywords.teardown, self._suite_status_suite_status)
            if failure:
                self._suite_suite.suite_teardown_failed(unic(failure))
                if self._suite_suite.statistics.critical.failed:
                    self._suite_status_suite_status.critical_failure_occurred()
        self._suite_suite.endtime = get_timestamp()
        self._suite_suite.message = self._suite_status_suite_status.message
        self._context_context_context.end_suite(ModelCombiner(suite, self._suite_suite))
        self._suite_suite = self._suite_suite.parent
        self._suite_status_suite_status = self._suite_status_suite_status.parent
        self._output_output.library_listeners.discard_suite_scope()
 
    def visit_test(self, test):
        if test.name in self._executed_tests_executed_tests:
            self._output_output.warn("Multiple test cases with name '%s' executed in "
                              "test suite '%s'." % (test.name, self._suite_suite.longname))
        self._executed_tests_executed_tests[test.name] = True
        result = self._suite_suite.tests.create(name=test.name,
                                          doc=self._resolve_setting_resolve_setting(test.doc),
                                          tags=self._resolve_setting_resolve_setting(test.tags),
                                          starttime=get_timestamp(),
                                          timeout=self._get_timeout_get_timeout(test))
        self._context_context_context.start_test(result)
        self._output_output.start_test(ModelCombiner(test, result))
        status = TestStatus(self._suite_status_suite_status, result.critical)
        if status.exit:
            self._add_exit_combine_add_exit_combine()
            result.tags.add('robot:exit')
        if not status.failures and not test.name:
            status.test_failed('Test case name cannot be empty.')
        if not status.failures and not test.keywords.normal:
            status.test_failed('Test case contains no keywords.')
        self._run_setup_run_setup(test.keywords.setup, status, result)
        try:
            if not status.failures:
                StepRunner(self._context_context_context,
                           test.template).run_steps(test.keywords.normal)
            else:
                status.test_failed(status.message)
        except PassExecution as exception:
            err = exception.earlier_failures
            if err:
                status.test_failed(err)
            else:
                result.message = exception.message
        except ExecutionStatus as err:
            status.test_failed(err)
        result.status = status.status
        result.message = status.message or result.message
        if status.teardown_allowed:
            with self._context_context_context.test_teardown(result):
                failure = self._run_teardown_run_teardown(test.keywords.teardown, status,
                                             result)
                if failure and result.critical:
                    status.critical_failure_occurred()
        if not status.failures and result.timeout and result.timeout.timed_out():
            status.test_failed(result.timeout.get_message())
            result.message = status.message
        result.status = status.status
        result.endtime = get_timestamp()
        self._output_output.end_test(ModelCombiner(test, result))
        self._context_context_context.end_test(result)
 
    def _add_exit_combine(self):
        exit_combine = ('NOT robot:exit', '')
        if exit_combine not in self._settings_settings['TagStatCombine']:
            self._settings_settings['TagStatCombine'].append(exit_combine)
 
    def _get_timeout(self, test):
        if not test.timeout:
            return None
        return TestTimeout(test.timeout.value, test.timeout.message,
                           self._variables_variables, rpa=test.parent.rpa)
 
    def _run_setup(self, setup, status, result=None):
        if not status.failures:
            exception = self._run_setup_or_teardown_run_setup_or_teardown(setup)
            status.setup_executed(exception)
            if result and isinstance(exception, PassExecution):
                result.message = exception.message
 
    def _run_teardown(self, teardown, status, result=None):
        if status.teardown_allowed:
            exception = self._run_setup_or_teardown_run_setup_or_teardown(teardown)
            status.teardown_executed(exception)
            failed = not isinstance(exception, PassExecution)
            if result and exception:
                result.message = status.message if failed else exception.message
            return exception if failed else None
 
    def _run_setup_or_teardown(self, data):
        if not data:
            return None
        try:
            name = self._variables_variables.replace_string(data.name)
        except DataError as err:
            if self._settings_settings.dry_run:
                return None
            return err
        if name.upper() in ('', 'NONE'):
            return None
        try:
            StepRunner(self._context_context_context).run_step(data, name=name)
        except ExecutionStatus as err:
            return err
 
 
class ModelCombiner():
 
    def __init__(self, data, result, **priority):
        self.datadata = data
        self.resultresult = result
        self.prioritypriority = priority
 
    def __getattr__(self, name):
        if name in self.prioritypriority:
            return self.prioritypriority[name]
        if hasattr(self.resultresult, name):
            return getattr(self.resultresult, name)
        if hasattr(self.datadata, name):
            return getattr(self.datadata, name)
        raise AttributeError(name)