From df75e0afd36a6f95cf6e50ccc8052e037f001b00 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 2 Dec 2023 15:21:03 +0100 Subject: [PATCH] rework env file parsing logic with EnvFile class, preparing for dependency resolution --- src/coordinator.py | 28 +++++++++++++++------------- src/env_file_helper.py | 13 ++++++++++--- 2 files changed, 25 insertions(+), 16 deletions(-) diff --git a/src/coordinator.py b/src/coordinator.py index 885bb9d..b0721c9 100644 --- a/src/coordinator.py +++ b/src/coordinator.py @@ -5,6 +5,7 @@ from dotenv import dotenv_values from loguru import logger from src.dirmanager import DirManager +from src.env_file_helper import DependencyRule, EnvFile, sort_env_files_by_rule from src.html_helper import merge_html_files from src.runner import Runner from src.tests_authentik.runner_authentik import RunnerAuthentik @@ -31,18 +32,18 @@ class Coordinator: self.output_dir = output_dir self.session_id = session_id - self.env_paths: dict[str, Path] = dict() - self.env_configs: dict[str, dict[str, str]] = dict() # todo: needed? - self._parse_env_files(env_paths_list) + self.env_files: list[EnvFile] = self._parse_env_files(env_paths_list) - def _parse_env_files(self, env_paths: list[Path]): + def _parse_env_files(self, env_paths: list[Path]) -> list[EnvFile]: + """Returns a list of EnvFile objects created from the given env files""" + env_files: list[EnvFile] = [] for env_path in env_paths: assert env_path.is_file(), f"the env file {env_path} does not exist" config: dict[str, str] = dotenv_values(env_path) # type: ignore assert "TYPE" in config, f"the env file {env_path} does not specify the required TYPE key." env_type = config["TYPE"] - self.env_paths[env_type] = env_path - self.env_configs[env_type] = config # todo: needed? + env_files.append(EnvFile(env_path=env_path, config=config, env_type=env_type)) + return env_files def setup_test(self): logger.info("calling setup_test()") @@ -53,24 +54,25 @@ class Coordinator: """Copies all env filesto 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 type_key, env_path in self.env_paths.items(): - shutil.copy(env_path, env_files_dir / type_key) + for env_file in self.env_files: + shutil.copy(env_file.env_path, env_files_dir / env_file.env_type) def run_test(self): logger.info("calling run_test()") - self.runners: list[Runner] = self._load_runners(self.env_paths.values()) + self.runners: list[Runner] = self._load_runners(self.env_files) for runner in self.runners: runner.run_tests() for runner in self.runners: runner.run_cleanup() logger.info("run_test() finished") - def _load_runners(self, env_files: list[Path]) -> list[Runner]: + def _load_runners(self, env_files: list[EnvFile]) -> list[Runner]: runners = [] for env_file in env_files: - config: dict[str, str] = dotenv_values(env_file) # type: ignore - RunnerClass = RUNNER_DICT[config["TYPE"]] - runners.append(RunnerClass(dotenv_path=env_file, output_dir=self.output_dir, session_id=self.session_id)) + RunnerClass = RUNNER_DICT[env_file.config["TYPE"]] + runners.append( + RunnerClass(dotenv_path=env_file.env_path, output_dir=self.output_dir, session_id=self.session_id) + ) return runners def combine_html(self): diff --git a/src/env_file_helper.py b/src/env_file_helper.py index 20ca508..361dca0 100644 --- a/src/env_file_helper.py +++ b/src/env_file_helper.py @@ -1,20 +1,27 @@ +from pathlib import Path from typing import NamedTuple from loguru import logger -class Rule(NamedTuple): +class EnvFile(NamedTuple): + env_path: Path + config: dict[str, str] + env_type: str + + +class DependencyRule(NamedTuple): child: str parent: str -def _is_rule_satisfied(in_list: list, rule: Rule) -> bool: +def _is_rule_satisfied(in_list: list, rule: DependencyRule) -> bool: child_index = in_list.index(rule.child) parent_index = in_list.index(rule.parent) return parent_index < child_index -def sort_env_files_by_rule(env_list: list, rules: list[Rule]) -> list: +def sort_env_files_by_rule(env_list: list, rules: list[DependencyRule]) -> list: in_list = env_list.copy() def swap_item_with_previous(in_list: list, index: int):