Source code for catcher.core.runner

import os
import traceback
from os.path import join

from catcher.core.var_holder import VariablesHolder
from catcher.core.parser import Parser
from catcher.core.step_factory import StepFactory
from catcher.core.test import Test
from catcher.core.filters_factory import FiltersFactory
from catcher.modules.log_storage import LogStorage
from catcher.steps.step import SkipException
from catcher.utils import logger
from catcher.utils.file_utils import cut_path
from catcher.utils.logger import warning, info, debug, OptionalOutput
from catcher.core.mod_factory import ModulesFactory

[docs]class Runner: def __init__(self, path: str, tests_path: str, inventory: str or None, modules=None, cmd_env=None, system_environment=None, resources=None, output_format=None, filter_list=None) -> None: # singletons init should be done before services (like vars holder), as singletons maybe used there # modules should be done before filters, as requirements module installs dependencies for custom bifs ModulesFactory(resources_dir=resources or os.path.join(path, 'resources')) FiltersFactory(custom_modules=filter_list) StepFactory(modules) self.tests_path = tests_path self.path = path self.parser = Parser(path, inventory) self.var_holder = VariablesHolder(path, system_environment=system_environment, inventory_vars=self.parser.read_inventory(), cmd_env=cmd_env, resources=resources) if output_format: logger.log_storage = LogStorage(output_format)
[docs] def run_tests(self, output: str = 'full') -> bool: """ Run the testcase :param output: 'full' - all output possible. 'limited' - only test end report & summary. 'final' - only summary. """ try: [mod.before() for mod in ModulesFactory().modules.values()] results = [] for parse_result in self.parser.read_tests(self.tests_path): if parse_result.should_run: # parse successful variables = self.var_holder.variables # each test has it's own copy of global variables variables['TEST_NAME'] = parse_result.test.file # variables are shared between test and includes with OptionalOutput(output == 'final'): for include in parse_result.run_on_include: # run all includes before the main test. self._run_test(include, variables, output=output, test_type='include') result = self._run_test(parse_result.test, variables, output=output) results.append(result) self._run_finally(parse_result.test, result) else: # parse failed (dependency/parsing problem) warning('Test ' + cut_path(self.tests_path, parse_result.test) +' failed: ') + str(parse_result.parse_error)) logger.log_storage.test_parse_fail(parse_result.test, parse_result.parse_error) results.append(False) return all(results) finally: logger.log_storage.write_report(join(self.path, 'reports')) logger.log_storage.print_summary(self.tests_path) [mod.after() for mod in ModulesFactory().modules.values()]
def _run_test(self, test: Test, global_variables: dict, output: str = 'full', test_type='test') -> bool: try: self.var_holder.prepare_variables(test, global_variables) logger.log_storage.test_start(test.file, test_type=test_type) test.check_ignored() with OptionalOutput(output == 'limited'): info(test_type.capitalize() + ' ' + cut_path(self.tests_path, test.file) +' passed.')) logger.log_storage.test_end(test.file, True, test_type=test_type) return True except SkipException: info(test_type.capitalize() + ' ' + cut_path(self.tests_path, test.file) + logger.yellow(' skipped.')) logger.log_storage.test_end(test.file, True, end_comment='Skipped', test_type=test_type) return True except Exception as e: warning(test_type.capitalize() + ' ' + cut_path(self.tests_path, test.file) +' failed: ') + str(e)) debug(traceback.format_exc()) logger.log_storage.test_end(test.file, False, str(e), test_type=test_type) return False def _run_finally(self, test, result: bool): if test and logger.log_storage.test_start(test.file, test_type='{}_cleanup'.format(test.file)) try: test.run_finally(result) info('Test ' + cut_path(self.tests_path, test.file) + ' [cleanup] ' +' passed.')) logger.log_storage.test_end(test.file, True, test_type='{} [cleanup]'.format(test.file)) except Exception as e: warning('Test ' + cut_path(self.tests_path, test.file) + ' [cleanup] ' +' failed: ') + str(e)) debug(traceback.format_exc()) logger.log_storage.test_end(test.file, False, test_type='{} [cleanup]'.format(test.file))