import shutil from pathlib import Path from typing import TYPE_CHECKING, NamedTuple from dotenv import dotenv_values if TYPE_CHECKING: from pytest_abra.dir_manager import DirManager from pytest_abra.runner import Runner class EnvFile(NamedTuple): env_path: Path env_config: dict[str, str] env_type: str def __repr__(self) -> str: return f"EnvFile(type={self.env_type})" class DependencyRule(NamedTuple): child: str dependency: str class EnvManager: def __init__(self, env_paths: list[Path], RUNNER_DICT: dict[str, type["Runner"]]): self.env_files: list[EnvFile] = self._get_env_files(env_paths) self.dependency_rules: list[DependencyRule] = self._get_dependency_rules(self.env_files, RUNNER_DICT) self.env_files = self.sort_env_files_by_rule(self.env_files, self.dependency_rules) @staticmethod def _get_env_files(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"] env_files.append(EnvFile(env_path=env_path, env_config=config, env_type=env_type)) return env_files @staticmethod def _get_dependency_rules(env_files: list[EnvFile], RUNNER_DICT: dict[str, type["Runner"]]) -> list[DependencyRule]: dependency_rules: list[DependencyRule] = [] for env_file in env_files: child_runner_class = RUNNER_DICT[env_file.env_type] for dependency in child_runner_class.dependencies: dependency_rule = DependencyRule(child=child_runner_class.env_type, dependency=dependency) dependency_rules.append(dependency_rule) return dependency_rules @staticmethod def _get_indices_by_string(in_list: list[EnvFile], string: str) -> list[int]: """returns all indices of items in in_list, where item.env_type matches string""" return [index for index, element in enumerate(in_list) if element.env_type == string] @staticmethod def _swap_item_with_previous(in_list: list[EnvFile], index: int): """swaps item at index N with item at index N-1""" assert index > 0, "cannot swap with negative index" in_list[index], in_list[index - 1] = in_list[index - 1], in_list[index] @classmethod def is_rule_satisfied(cls, env_list: list[EnvFile], rule: DependencyRule, swap=False) -> bool: """returns if the ordering in in_list is compliant with the given rule if swap=True, some reordering will happen in case of a violated rule""" child_indices = cls._get_indices_by_string(env_list, rule.child) parent_indices = cls._get_indices_by_string(env_list, rule.dependency) for child_index in child_indices: for parent_index in parent_indices: if not parent_index < child_index: if swap: cls._swap_item_with_previous(env_list, parent_index) return False return True @classmethod def sort_env_files_by_rule(cls, env_list: list[EnvFile], rules: list[DependencyRule]) -> list[EnvFile]: out_list = env_list.copy() for _ in range(10_000): rule_satisfied: list[bool] = [] for rule in rules: rule_satisfied.append(cls.is_rule_satisfied(out_list, rule, swap=True)) if all(rule_satisfied): return out_list raise ValueError( "Could not resolve test order. This is possibly due to a circular dependency (a on b, b on c, c on a)" ) def copy_env_files(self, DIR: "DirManager") -> None: """Copies all env files to STATES/env_files. Files will be renamed to -- 00-authentik-login.test.dev.local-it.cloud.env""" for index, env_file in enumerate(self.env_files): file_name = "-".join([str(index).zfill(2), env_file.env_type, env_file.env_path.name]) shutil.copy(env_file.env_path, DIR.ENV_FILES / file_name)