Source code for catcher.core.test

import time
import traceback
from typing import Union

from catcher.steps.step import Step, SkipException
from catcher.steps.stop import StopException
from catcher.utils import logger
from catcher.utils.logger import debug, info
from catcher.utils.misc import fill_template_str
from catcher.steps.check import Operator
from catcher.core.step_factory import StepFactory

[docs]class Test: """ Testcase. Contains variables, includes, steps and final actions. :ignore: if true will ignore this test. A condition based on :meth:`catcher.steps.check.Check` can be used to compute ignore condition dynamically. :include: other test to run. See :meth:`catcher.core.include.Include`. Can be a string in case of single include or a list of include in case of multiple. Each of them will be run before the test passing variables to each other and, finally, to the test. In case of `run_on_include` include's property is false or `as` alias is set include won't be run before the test. You can run such include via :meth:`` step later. See :doc:`includes` for more info. :variables: test local variables which override inventory variables. These variables will be available only in this test or in test which includes this test. Variables itself can contain templates. See :doc:`variables` for more info. :steps: A list of test actions which will be run one by one. They can use variables and support templates. Each step can register it's output as a new variable. See :meth:`catcher.steps.step.Step` for available options. See :doc:`internal_modules` and `external modules <>`_ for more info. :finally: A list of clean up actions which will be run after test finishes execution. Condition for every clean up action run can be set (by default they run in any case). See :doc:`tests` for more info and examples. """ def __init__(self, path: str, file: str, includes: dict = None, variables: dict = None, config: dict = None, steps: list = None, final: list = None, ignore: Union[bool, str, dict] = None) -> None: self.path = path = final self.file = file self.includes = includes self.config = config self.variables = variables self.steps = steps self.ignore = ignore self.include = None # if this test is include it will refer to `Include` class
[docs] def check_ignored(self): if self.ignore: if not isinstance(self.ignore, bool): self.ignore = Operator.find_operator(self.ignore).operation(self.variables) if self.ignore: raise SkipException('Test ignored')
[docs] def run(self, tag=None, raise_stop=False) -> dict: for step in self.steps: if not self._run_step(step, tag, raise_stop): break return self.variables
[docs] def run_finally(self, result: bool): for step in if not self._run_step(step, result=result): return
def _run_step(self, step, tag=None, raise_stop=False, result=None) -> bool: [action] = step.keys() ignore_errors = get_or_default('ignore_errors', step[action], False) if tag is not None: # skip if tag specified step_tag = get_or_default('tag', step[action], None) if step_tag != tag: return True if result is not None: # skip if result is specified (step is final) run_type = get_or_default('run_if', step[action], 'always') if (not result and run_type == 'pass') or (result and run_type == 'fail'): debug('Skip final action') return True actions = StepFactory().get_actions(self.path, step) for action_object in actions: # override all variables with cmd variables if not self._run_actions(step, action, action_object, self.variables, raise_stop, ignore_errors): return False return True def _run_actions(self, step, action, action_object, variables, raise_stop, ignore_errors) -> bool: action_name = get_action_name(action, action_object, variables) start = time.process_time() try: logger.log_storage.new_step(step, variables) action_object.check_skip(variables) self.variables = action_object.action(self.includes, variables) # repeat for run (variables were computed after name) action_name = get_action_name(action, action_object, self.variables) info('Step ' + action_name + get_timing(start) +' OK')) logger.log_storage.step_end(step, self.variables) return True except StopException as e: # stop a test without error if raise_stop: # or raise error if configured logger.log_storage.step_end(step, variables, success=False, output=str(e)) raise e debug('Skip ' + action_name + ' due to ' + str(e)) info('Step ' + action_name + get_timing(start) +' OK')) logger.log_storage.step_end(step, self.variables, success=True, output=str(e)) return False # stop current test except SkipException as e: # skip this step info('Step ' + action_name + logger.yellow(' skipped')) logger.log_storage.step_end(step, self.variables, success=True, output=str(e)) return True except Exception as e: if ignore_errors: # continue actions & steps execution debug('Step ' + action_name + get_timing(start) +' failed') + ', but we ignore it') logger.log_storage.step_end(step, variables, success=True) return True else: info('Step ' + action_name + get_timing(start) +' failed: ') + str(e)) debug(traceback.format_exc()) logger.log_storage.step_end(step, variables, success=False, output=str(e)) raise e def __repr__(self) -> str: return str(self.steps)
[docs]def get_or_default(key: str, body: dict or str, default: any) -> any: if isinstance(body, dict): return body.get(key, default) else: return default
[docs]def get_action_name(action_type: str, action: Step, variables: dict): if is not None: return fill_template_str(, variables) return action_type
[docs]def get_timing(start_time) -> str: delta = time.process_time() - start_time if delta > 60: return ' [{:.0f}m {:.0f}s]'.format(delta / 60, delta - 60 * int(delta / 60)) return ' [{:.0f}s]'.format(delta)