context.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 contextlib import contextmanager
 
from robot.errors import DataError
 
 
class ExecutionContexts:
 
    def __init__(self):
        self._contexts_contexts = []
 
    @property
    current = property
    
    def current(self):
        return self._contexts_contexts[-1] if self._contexts_contexts else None
 
    @property
    top = property
    
    def top(self):
        return self._contexts_contexts[0] if self._contexts_contexts else None
 
    def __iter__(self):
        return iter(self._contexts_contexts)
 
    @property
    namespaces = property
    
    def namespaces(self):
        return (context.namespace for context in self)
 
    def start_suite(self, suite, namespace, output, dry_run=False):
        ctx = _ExecutionContext(suite, namespace, output, dry_run)
        self._contexts_contexts.append(ctx)
        return ctx
 
    def end_suite(self):
        self._contexts_contexts.pop()
 
 
# This is ugly but currently needed e.g. by BuiltIn
EXECUTION_CONTEXTS = ExecutionContexts()
 
 
class _ExecutionContext:
    
    _started_keywords_threshold = 100
 
    def __init__(self, suite, namespace, output, dry_run=False):
        self.suitesuite = suite
        self.testtest = None
        self.timeoutstimeouts = set()
        self.namespacenamespace = namespace
        self.outputoutput = output
        self.dry_rundry_run = dry_run
        self.in_suite_teardownin_suite_teardown = False
        self.in_test_teardownin_test_teardown = False
        self.in_keyword_teardownin_keyword_teardown = 0
        self._started_keywords_started_keywords = 0
        self.timeout_occurredtimeout_occurred = False
        self.user_keywordsuser_keywords = []
        self.step_typesstep_types = []
 
    @contextmanager
    def suite_teardown(self):
        self.in_suite_teardownin_suite_teardown = True
        try:
            yield
        finally:
            self.in_suite_teardownin_suite_teardown = False
 
    @contextmanager
    def test_teardown(self, test):
        self.variablesvariablesvariables.set_test('${TEST_STATUS}', test.status)
        self.variablesvariablesvariables.set_test('${TEST_MESSAGE}', test.message)
        self.in_test_teardownin_test_teardown = True
        self._remove_timeout_remove_timeout(test.timeout)
        try:
            yield
        finally:
            self.in_test_teardownin_test_teardown = False
 
    @contextmanager
    def keyword_teardown(self, error):
        self.variablesvariablesvariables.set_keyword('${KEYWORD_STATUS}', 'FAIL' if error else 'PASS')
        self.variablesvariablesvariables.set_keyword('${KEYWORD_MESSAGE}', str(error or ''))
        self.in_keyword_teardownin_keyword_teardown += 1
        try:
            yield
        finally:
            self.in_keyword_teardownin_keyword_teardown -= 1
 
    @contextmanager
    def user_keyword(self, handler):
        self.user_keywordsuser_keywords.append(handler)
        self.namespacenamespace.start_user_keyword()
        try:
            yield
        finally:
            self.namespacenamespace.end_user_keyword()
            self.user_keywordsuser_keywords.pop()
 
    def warn_on_invalid_private_call(self, handler):
        parent = self.user_keywordsuser_keywords[-1] if self.user_keywordsuser_keywords else None
        if not parent or parent.source != handler.source:
            self.warnwarn(f"Keyword '{handler.longname}' is private and should only "
                      f"be called by keywords in the same file.")
 
    @contextmanager
    def timeout(self, timeout):
        self._add_timeout_add_timeout(timeout)
        try:
            yield
        finally:
            self._remove_timeout_remove_timeout(timeout)
 
    @property
    in_teardown = property
    
    def in_teardown(self):
        return bool(self.in_suite_teardownin_suite_teardown or
                    self.in_test_teardownin_test_teardown or
                    self.in_keyword_teardownin_keyword_teardown)
 
    @property
    variables = property
    
    def variables(self):
        return self.namespacenamespace.variables
 
    def continue_on_failure(self, default=False):
        parents = ([self.testtest] if self.testtest else []) + self.user_keywordsuser_keywords
        for index, parent in enumerate(reversed(parents)):
            if (parent.tags.robot('recursive-stop-on-failure')
                    or index == 0 and parent.tags.robot('stop-on-failure')):
                return False
            if (parent.tags.robot('recursive-continue-on-failure')
                    or index == 0 and parent.tags.robot('continue-on-failure')):
                return True
        return default or self.in_teardownin_teardownin_teardown
 
    @property
    allow_loop_control = property
    
    def allow_loop_control(self):
        for typ in reversed(self.step_typesstep_types):
            if typ == 'ITERATION':
                return True
            if typ == 'KEYWORD':
                return False
        return False
 
    def end_suite(self, suite):
        for name in ['${PREV_TEST_NAME}',
                     '${PREV_TEST_STATUS}',
                     '${PREV_TEST_MESSAGE}']:
            self.variablesvariablesvariables.set_global(name, self.variablesvariablesvariables[name])
        self.outputoutput.end_suite(suite)
        self.namespacenamespace.end_suite(suite)
        EXECUTION_CONTEXTS.end_suite()
 
    def set_suite_variables(self, suite):
        self.variablesvariablesvariables['${SUITE_NAME}'] = suite.longname
        self.variablesvariablesvariables['${SUITE_SOURCE}'] = suite.source or ''
        self.variablesvariablesvariables['${SUITE_DOCUMENTATION}'] = suite.doc
        self.variablesvariablesvariables['${SUITE_METADATA}'] = suite.metadata.copy()
 
    def report_suite_status(self, status, message):
        self.variablesvariablesvariables['${SUITE_STATUS}'] = status
        self.variablesvariablesvariables['${SUITE_MESSAGE}'] = message
 
    def start_test(self, test):
        self.testtest = test
        self._add_timeout_add_timeout(test.timeout)
        self.namespacenamespace.start_test()
        self.variablesvariablesvariables.set_test('${TEST_NAME}', test.name)
        self.variablesvariablesvariables.set_test('${TEST_DOCUMENTATION}', test.doc)
        self.variablesvariablesvariables.set_test('@{TEST_TAGS}', list(test.tags))
 
    def _add_timeout(self, timeout):
        if timeout:
            timeout.start()
            self.timeoutstimeouts.add(timeout)
 
    def _remove_timeout(self, timeout):
        if timeout in self.timeoutstimeouts:
            self.timeoutstimeouts.remove(timeout)
 
    def end_test(self, test):
        self.testtest = None
        self._remove_timeout_remove_timeout(test.timeout)
        self.namespacenamespace.end_test()
        self.variablesvariablesvariables.set_suite('${PREV_TEST_NAME}', test.name)
        self.variablesvariablesvariables.set_suite('${PREV_TEST_STATUS}', test.status)
        self.variablesvariablesvariables.set_suite('${PREV_TEST_MESSAGE}', test.message)
        self.timeout_occurredtimeout_occurred = False
 
    def start_keyword(self, keyword):
        self._started_keywords_started_keywords += 1
        if self._started_keywords_started_keywords > self._started_keywords_threshold_started_keywords_threshold:
            raise DataError('Maximum limit of started keywords and control '
                            'structures exceeded.')
        self.outputoutput.start_keyword(keyword)
        if keyword.libname != 'BuiltIn':
            self.step_typesstep_types.append(keyword.type)
 
    def end_keyword(self, keyword):
        self.outputoutput.end_keyword(keyword)
        self._started_keywords_started_keywords -= 1
        if keyword.libname != 'BuiltIn':
            self.step_typesstep_types.pop()
 
    def get_runner(self, name):
        return self.namespacenamespace.get_runner(name)
 
    def trace(self, message):
        self.outputoutput.trace(message)
 
    def debug(self, message):
        self.outputoutput.debug(message)
 
    def info(self, message):
        self.outputoutput.info(message)
 
    def warn(self, message):
        self.outputoutput.warn(message)
 
    def fail(self, message):
        self.outputoutput.fail(message)
 
    def skip(self, message):
        self.outputoutput.skip(message)