diff --git a/src/env_manager.py b/src/env_manager.py index 6857293..cfc81fb 100644 --- a/src/env_manager.py +++ b/src/env_manager.py @@ -3,10 +3,9 @@ from pathlib import Path from typing import NamedTuple from dotenv import dotenv_values -from loguru import logger from src.dir_manager import DirManager -from src.runner import RUNNER_DICT +from src.runner_dict import RUNNER_DICT class EnvFile(NamedTuple): @@ -23,55 +22,14 @@ class DependencyRule(NamedTuple): dependency: str -def _get_indices_with_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] - - -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] - - -def is_rule_satisfied(in_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 = _get_indices_with_string(in_list, rule.child) - parent_indices = _get_indices_with_string(in_list, rule.dependency) - for child_index in child_indices: - for parent_index in parent_indices: - if not parent_index < child_index: - if swap: - _swap_item_with_previous(in_list, parent_index) - return False - return True - - -def sort_env_files_by_rule(env_list: list[EnvFile], rules: list[DependencyRule]) -> list[EnvFile]: - in_list = env_list.copy() - - for _ in range(10_000): - rule_satisfied: list[bool] = [] - for rule in rules: - rule_satisfied.append(is_rule_satisfied(in_list, rule, swap=True)) - - if all(rule_satisfied): - return in_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)" - ) - - class EnvManager: def __init__(self, env_paths_list: list[Path]): - self.env_files: list[EnvFile] = self._getn_env_files_list(env_paths_list) + self.env_files: list[EnvFile] = self._get_env_files(env_paths_list) self.dependency_rules: list[DependencyRule] = self._get_dependency_rules(self.env_files) + self.env_files = self.sort_env_files_by_rule(self.env_files, self.dependency_rules) @staticmethod - def _getn_env_files_list(env_paths: list[Path]) -> list[EnvFile]: + 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: @@ -92,6 +50,48 @@ class EnvManager: 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 their own TYPE value.""" env_files_dir = DIR.STATES / "env_files"