140 lines
5.4 KiB
Python
140 lines
5.4 KiB
Python
import importlib
|
|
import json
|
|
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 generate_random_string, load_json_to_environ, 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)
|
|
self.load_test_credentials()
|
|
|
|
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.RESULTS / "html")
|
|
out_file_path = str(self.DIR.RESULTS / "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/RESULTS 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.RESULTS / "traces"
|
|
for f in trace_root_dir.rglob("*/trace.zip"):
|
|
new_path = get_new_path(self.DIR.RESULTS, f.parent.name)
|
|
f.parent.rename(new_path)
|
|
rmtree(trace_root_dir)
|
|
|
|
def load_test_credentials(self):
|
|
"""Load test user credentials. If not available, create them randomly.
|
|
|
|
Test users are created during testing but should be deleted after the test. In case test
|
|
users are not deleted after tests by accident, the user credentials are not known to an
|
|
attacker."""
|
|
|
|
test_credentials_path = self.DIR.STATES / "credentials_test.json"
|
|
if not test_credentials_path.is_file():
|
|
test_credentials = {
|
|
"TEST_USER": "test-" + generate_random_string(6),
|
|
"TEST_PASS": generate_random_string(12, punctuation=True),
|
|
}
|
|
|
|
with open(test_credentials_path, "w") as json_file:
|
|
json.dump(test_credentials, json_file)
|
|
|
|
load_json_to_environ(test_credentials_path)
|
|
|
|
@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)
|
|
RunnerClass._tests_path = module_path.parent
|
|
RUNNER_DICT[RunnerClass.env_type] = RunnerClass
|
|
return RUNNER_DICT
|