87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
import shutil
|
|
from pathlib import Path
|
|
|
|
from loguru import logger
|
|
|
|
from src.dirmanager import DirManager
|
|
from src.env_manager import EnvFile, EnvManager
|
|
from src.html_helper import merge_html_files
|
|
from src.runner import RUNNER_DICT, Runner
|
|
from src.utils import rmtree
|
|
|
|
|
|
class Coordinator:
|
|
def __init__(self, env_paths_list: list[Path], output_dir: Path, session_id: str) -> 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.DIR = DirManager(output_dir=output_dir, session_id=session_id)
|
|
self.ENV = EnvManager(env_paths_list)
|
|
|
|
def setup_test(self) -> None:
|
|
logger.info("calling setup_test()")
|
|
self.DIR.create_all_dirs()
|
|
self._copy_env_files()
|
|
|
|
def _copy_env_files(self) -> None:
|
|
"""Copies all env files to STATES/env_files. Files will be renamed to their own TYPE value."""
|
|
env_files_dir = self.DIR.STATES / "env_files"
|
|
env_files_dir.mkdir(exist_ok=True)
|
|
for env_file in self.ENV.env_files:
|
|
shutil.copy(env_file.env_path, env_files_dir / env_file.env_type)
|
|
|
|
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 = []
|
|
for env_file in env_files:
|
|
RunnerClass = RUNNER_DICT[env_file.config["TYPE"]]
|
|
runners.append(
|
|
RunnerClass(
|
|
dotenv_path=env_file.env_path, output_dir=self.DIR.output_dir, session_id=self.DIR.session_id
|
|
)
|
|
)
|
|
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)
|