move functions to envmanager
This commit is contained in:
parent
e903360767
commit
eb0714da9b
1 changed files with 46 additions and 46 deletions
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue