Daniel
0b4e0a0c16
Before, recipes_dir had to be present in the importable paths of the python interpreter. This was solved by adding it to the PYTHONPATH env var. Now, abratest handles this by itself. Reviewed-on: local-it-infrastructure/e2e_tests#11 Co-authored-by: Daniel <d.brummerloh@gmail.com> Co-committed-by: Daniel <d.brummerloh@gmail.com>
113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
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_files
|
|
from pytest_abra.runner import Runner
|
|
from pytest_abra.utils import rmtree
|
|
|
|
|
|
class Coordinator:
|
|
def __init__(
|
|
self, env_paths_list: 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_list])
|
|
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_list, self.RUNNER_DICT)
|
|
self.TIMEOUT = timeout
|
|
|
|
def setup_test(self) -> None:
|
|
logger.info("calling setup_test()")
|
|
self.DIR.create_all_dirs()
|
|
self.ENV.copy_env_files(self.DIR)
|
|
|
|
def run_test(self) -> None:
|
|
logger.info("calling run_test()")
|
|
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_test() 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.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_path = str(self.DIR.RECORDS / "html")
|
|
out_path = str(self.DIR.RECORDS / "full-report.html")
|
|
title = "combined.html"
|
|
merge_html_files(in_path, out_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
|