import importlib import re import sys from pathlib import Path from loguru import logger from pytest_abra.dir_manager import DirManager from pytest_abra.env_manager import EnvFile, EnvManager from pytest_abra.html_helper import merge_html_reports from pytest_abra.runner import Runner from pytest_abra.utils import rmtree class Coordinator: def __init__( self, env_paths: list[Path], output_dir: Path, session_id: str, recipes_dir: Path, timeout: int, ) -> None: # logging out_string = "".join([e.name + "\n" for e in env_paths]) out_string += f"output_dir = {output_dir}\n" out_string += f"session_id = {session_id}" logger.info(f"initialize Coordinator instance with\nenv_paths_list =\n{out_string}") self.RUNNER_DICT = self.create_runner_dict(recipes_dir) self.DIR = DirManager(output_dir=output_dir, session_id=session_id, recipes_dir=recipes_dir) self.ENV = EnvManager(env_paths=env_paths, RUNNER_DICT=self.RUNNER_DICT) self.TIMEOUT = timeout def prepare_tests(self) -> None: logger.info("calling prepare_tests()") self.DIR.create_all_dirs() self.ENV.copy_env_files(self.DIR) # todo: check that tests are unique # todo: create random testuser creds and load them def run_tests(self) -> None: logger.info("calling run_tests()") self.runners: list[Runner] = self._load_runners(self.ENV.env_files) for runner in self.runners: runner.run_setups() for runner in self.runners: runner.run_tests() for runner in self.runners: runner.run_cleanups() logger.info("run_tests() finished") def _load_runners(self, env_files: list[EnvFile]) -> list[Runner]: """Creates an instance of the correct Runner class for each given env file""" runners: list[Runner] = [] for index, env_file in enumerate(env_files): RunnerClass = self.RUNNER_DICT[env_file.env_config["TYPE"]] runners.append(RunnerClass(coordinator=self, runner_index=index)) return runners def combine_html(self) -> None: """combines all generated pytest html reports into one""" in_dir_path = str(self.DIR.RECORDS / "html") out_file_path = str(self.DIR.RECORDS / "full-report.html") title = "combined.html" merge_html_reports(in_dir_path, out_file_path, title) def collect_traces(self): """moves all traces into SESSION/RECORDS dir if tests are rerun and generate another trace, the new trace will get a unique name such as tracename-0 tracename-1 ... """ def get_new_path(root_dir: Path, base_name: str, index=0) -> Path: new_name_alt = base_name + f"-{index}" if not (root_dir / new_name_alt).is_dir(): return root_dir / new_name_alt else: index += 1 return get_new_path(root_dir, base_name, index=index) trace_root_dir = self.DIR.RECORDS / "traces" for f in trace_root_dir.rglob("*/trace.zip"): new_path = get_new_path(self.DIR.RECORDS, f.parent.name) f.parent.rename(new_path) rmtree(trace_root_dir) @staticmethod def create_runner_dict(recipes_dir: Path) -> dict[str, type["Runner"]]: """Creates a dictionary holding all the RunnerClasses that can be discovered in recipes_dir example: RUNNER_DICT: dict[str, type["Runner"]] = { "authentik": RunnerAuthentik, "wordpress": RunnerWordpress, "nextcloud": RunnerNextcloud, } The Runner classes are automatically imported with importlib. The imports are successful because recipes_dir is added to sys.path. """ RUNNER_DICT: dict[str, type["Runner"]] = dict() runner_discovery_pattern = re.compile("Runner.+") # make it possible to import modules from recipes_dir sys.path.append(recipes_dir.as_posix()) for module_path in recipes_dir.rglob("*/runner*.py"): rel_path = module_path.relative_to(recipes_dir).as_posix().replace("/", ".").replace(".py", "") module = importlib.import_module(rel_path) runner_class_names = [name for name in dir(module) if runner_discovery_pattern.match(name)] assert len(runner_class_names) == 1 runner_class_name = runner_class_names[0] RunnerClass: type[Runner] = getattr(module, runner_class_name) RUNNER_DICT[RunnerClass.env_type] = RunnerClass return RUNNER_DICT