implement env manager (#6)

* add EnvManager class

* holds all functions that are env file related

* integrates runner dependency resolution

* add integration and unit tests for EnvManager

Reviewed-on: local-it-infrastructure/e2e_tests#6
Co-authored-by: Daniel <d.brummerloh@gmail.com>
Co-committed-by: Daniel <d.brummerloh@gmail.com>
This commit is contained in:
Daniel 2023-12-04 17:09:01 +01:00 committed by dan
parent d3dc0f942a
commit 3fa10aaa69
18 changed files with 264 additions and 198 deletions

2
conftest.py Normal file
View file

@ -0,0 +1,2 @@
# this file exists so that tests inside /tests always find /src imports,
# because this will cause the root (/) to be added to sys.path

View file

@ -5,7 +5,7 @@ from pathlib import Path
from loguru import logger
from src.coordinator import Coordinator
from src.dirmanager import DirManager
from src.dir_manager import DirManager
from src.utils import get_session_id
# ----------------------------- lookup env files ----------------------------- #

View file

@ -1,6 +1,6 @@
from playwright.sync_api import BrowserContext, expect
from src.dirmanager import DirManager
from src.dir_manager import DirManager
def test_wordpress(admin_session: BrowserContext, dotenv_config: dict[str, str], DIR: DirManager):

View file

@ -14,7 +14,7 @@ from dotenv import dotenv_values
from playwright.sync_api import BrowserContext, expect
from pytest import Parser
from src.dirmanager import DirManager
from src.dir_manager import DirManager
# global timeout and LOCALE
LOCALE = {"Accept-Language": "de_DE"}

View file

@ -1,78 +1,34 @@
import shutil
from pathlib import Path
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.dir_manager import DirManager
from src.env_manager import EnvFile, EnvManager
from src.html_helper import merge_html_files
from src.runner import Runner
from src.tests_authentik.runner_authentik import RunnerAuthentik
from src.tests_nextcloud.runner_nextcloud import RunnerNextcloud
from src.tests_wordpress.runner_wordpress import RunnerWordpress
from src.runner_dict import RUNNER_DICT
from src.utils import rmtree
# Register all runners here. Each .env file with TYPE=authentik will be run with RunnerAuthentik
RUNNER_DICT: dict[str, type[Runner]] = {
"authentik": RunnerAuthentik,
"wordpress": RunnerWordpress,
"nextcloud": RunnerNextcloud,
}
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.output_dir = output_dir
self.session_id = session_id
# parse env files
self.env_files: list[EnvFile] = self._getn_env_files_list(env_paths_list)
self.dependency_rules: list[DependencyRule] = self._get_dependency_rules(self.env_files)
@staticmethod
def _getn_env_files_list(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, config=config, env_type=env_type))
return env_files
@staticmethod
def _get_dependency_rules(env_files: list[EnvFile]) -> 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.name, dependency=dependency.name)
dependency_rules.append(dependency_rule)
return dependency_rules
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_files:
shutil.copy(env_file.env_path, env_files_dir / env_file.env_type)
self.ENV.copy_env_files(self.DIR)
def run_test(self) -> None:
logger.info("calling run_test()")
self.runners: list[Runner] = self._load_runners(self.env_files)
self.runners: list[Runner] = self._load_runners(self.ENV.env_files)
for runner in self.runners:
runner.run_setups()
for runner in self.runners:
@ -87,7 +43,9 @@ class Coordinator:
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.output_dir, session_id=self.session_id)
RunnerClass(
dotenv_path=env_file.env_path, output_dir=self.DIR.output_dir, session_id=self.DIR.session_id
)
)
return runners

View file

@ -20,18 +20,25 @@ class DirManager:
# root test dir
if isinstance(output_dir, str):
output_dir = Path(output_dir)
self._output_dir = output_dir.resolve()
self.output_dir = output_dir.resolve()
self.session_id = session_id
def create_all_dirs(self):
self.create_dirs(self._output_dir, exist_ok=True)
self.create_dirs(
[self.SESSION, self.RECORDS, self.HTML, self.STATES, self.ENV_FILES, self.RESULTS], exist_ok=True
)
def create_all_dirs(self) -> None:
dirs: list[Path] = [
self.OUTPUT_DIR,
self.SESSION,
self.RECORDS,
self.HTML,
self.STATES,
self.ENV_FILES,
self.RESULTS,
]
for d in dirs:
d.mkdir(exist_ok=True)
@property
def OUTPUT_DIR(self):
return self._output_dir
return self.output_dir
@property
def SESSION(self):
@ -56,15 +63,3 @@ class DirManager:
@property
def RESULTS(self):
return self.SESSION / "results"
@staticmethod
def create_dirs(dirs: Path | list[Path] | dict[str, Path], exist_ok=False):
match dirs:
case Path():
dirs.mkdir(exist_ok=exist_ok)
case list():
for d in dirs:
d.mkdir(exist_ok=exist_ok)
case dict():
for d in dirs.values():
d.mkdir(exist_ok=exist_ok)

View file

@ -1,52 +0,0 @@
from pathlib import Path
from typing import NamedTuple
from loguru import logger
class EnvFile(NamedTuple):
env_path: Path
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
def _is_rule_satisfied(in_list: list, rule: DependencyRule) -> tuple[bool, int]:
child_indices = [index for index, element in enumerate(in_list) if element.env_type == rule.child]
child_index = min(child_indices)
# child_index = in_list.index(rule.child)
parent_indices = [index for index, element in enumerate(in_list) if element.env_type == rule.dependency]
parent_index = max(parent_indices)
# parent_index = in_list.index(rule.dependency)
return parent_index < child_index, parent_index
def sort_env_files_by_rule(env_list: list[EnvFile], rules: list[DependencyRule]) -> list:
in_list = env_list.copy()
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]
for _ in range(10_000):
rule_satisfied: list[bool] = []
for rule in rules:
is_rule_satisfied, parent_index = _is_rule_satisfied(in_list, rule)
if is_rule_satisfied:
rule_satisfied.append(True)
else:
rule_satisfied.append(False)
# parent_index = in_list.index(rule.dependency)
swap_item_with_previous(in_list, parent_index)
if all(rule_satisfied):
return in_list
logger.error("could not find order that satisfys all rules")
raise ValueError

100
src/env_manager.py Normal file
View file

@ -0,0 +1,100 @@
import shutil
from pathlib import Path
from typing import NamedTuple
from dotenv import dotenv_values
from src.dir_manager import DirManager
from src.runner_dict import RUNNER_DICT
class EnvFile(NamedTuple):
env_path: Path
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: list[Path]):
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 _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, config=config, env_type=env_type))
return env_files
@staticmethod
def _get_dependency_rules(env_files: list[EnvFile]) -> 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.name, dependency=dependency.name)
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"
env_files_dir.mkdir(exist_ok=True)
for env_file in self.env_files:
shutil.copy(env_file.env_path, env_files_dir / env_file.env_type)

View file

@ -6,7 +6,7 @@ import pytest
from dotenv import dotenv_values
from loguru import logger
from src.dirmanager import DirManager
from src.dir_manager import DirManager
@dataclass

16
src/runner_dict.py Normal file
View file

@ -0,0 +1,16 @@
from typing import TYPE_CHECKING
from src.tests_authentik.runner_authentik import RunnerAuthentik
from src.tests_nextcloud.runner_nextcloud import RunnerNextcloud
from src.tests_wordpress.runner_wordpress import RunnerWordpress
if TYPE_CHECKING:
from src.runner import Runner
# Register all runners here. Each .env file with TYPE=authentik will be run with RunnerAuthentik
RUNNER_DICT: dict[str, type["Runner"]] = {
"authentik": RunnerAuthentik,
"wordpress": RunnerWordpress,
"nextcloud": RunnerNextcloud,
}

View file

@ -4,7 +4,7 @@ import pytest
from dotenv import dotenv_values
from playwright.sync_api import BrowserContext, Page
from src.dirmanager import DirManager
from src.dir_manager import DirManager
@pytest.fixture

View file

@ -4,7 +4,7 @@ import re
from playwright.sync_api import BrowserContext, expect
from src.dirmanager import DirManager
from src.dir_manager import DirManager
ADMIN_USER = os.environ["ADMIN_USER"]
ADMIN_PASS = os.environ["ADMIN_PASS"]

View file

@ -1,25 +1,17 @@
import pytest
from playwright.sync_api import BrowserContext, expect
import os
# todo: what is this test for, why is it a fixture? -> ignore for now
from playwright.sync_api import Page
@pytest.fixture(scope="session", autouse=True)
def delete_nextcloud_user(admin_context: BrowserContext):
def delete_nextcloud_user(authentik_admin_page: Page):
"""Delete Nextcloud User"""
yield
context = setup_context(browser, f"{STATES}/admin_state.json")
page = context.new_page()
page.goto(CONFIG["domain"])
with page.expect_popup() as nextcloud_info:
page.get_by_role("link", name="Nextcloud").click()
with authentik_admin_page.expect_popup() as nextcloud_info:
authentik_admin_page.get_by_role("link", name="Nextcloud").click()
nextcloud = nextcloud_info.value
nextcloud.get_by_role("link", name="Open settings menu").click()
nextcloud.get_by_role("link", name="Users").click()
nextcloud.locator("#app-content div").filter(has_text=testuser["username"]).get_by_role(
nextcloud.locator("#app-content div").filter(has_text=os.environ["NEXTCLOUD_USER"]).get_by_role(
"button", name="Toggle user actions menu"
).click()
nextcloud.get_by_role("button", name="Delete user").click()
nextcloud.get_by_role("button", name=f"Delete authentik-{testuser['username']}'s account").click()
context.tracing.stop(path=f"{RECORDS}/nextcloud_delete_user.zip")
context.close()
nextcloud.get_by_role("button", name=f"Delete authentik-{os.environ["NEXTCLOUD_USER"]}'s account").click()

View file

@ -1 +1,16 @@
from src.tests_authentik.fixtures_authentik import admin_context, authentik_admin_page, user_context
import os
from src.tests_authentik.fixtures_authentik import (
authentik_admin_context,
authentik_admin_page,
authentik_user_context,
authentik_user_page,
)
NEXTCLOUD_DEMO_USER = {
"NEXTCLOUD_USER": "next_demo_user",
"NEXTCLOUD_PASS": "P@ss.123",
}
for key, value in NEXTCLOUD_DEMO_USER.items():
os.environ[key] = value

View file

@ -4,7 +4,7 @@ import pytest
from dotenv import dotenv_values
from playwright.sync_api import BrowserContext, Page
from src.dirmanager import DirManager
from src.dir_manager import DirManager
# from src.tests_authentik.fixtures_authentik import (
# authentik_admin_context,

View file

@ -1,17 +1,15 @@
import pytest
from playwright.sync_api import BrowserContext, Page, expect
from src.dirmanager import DirManager
from src.dir_manager import DirManager
@pytest.mark.xfail(reason="wordpress sso login has not been generated")
def test_visit_from_domain(authentik_admin_context: BrowserContext, dotenv_config: dict[str, str]):
"""visit wordpress directly with admin_session, expect not to be logged in"""
page = authentik_admin_context.new_page()
url = "https://" + dotenv_config["DOMAIN"]
page.goto(url)
# look for content wrapper
expect(page.locator("#wpcontent")).to_be_visible(timeout=3_000)
with pytest.raises(AssertionError):
# look for admin bar
expect(page.locator("#wpadminbar")).to_be_visible(timeout=3_000)

View file

@ -2,7 +2,7 @@
from playwright.sync_api import BrowserContext, expect
from src.dirmanager import DirManager
from src.dir_manager import DirManager
def test_welcome_message(context: BrowserContext, dotenv_config: dict[str, str], DIR: DirManager):

View file

@ -1,59 +1,101 @@
import sys
from pathlib import Path
from icecream import ic
import pytest
sys.path.append(Path(__file__).parent.parent.resolve().__str__())
# import pytest
# from prototyping.sorting_algo import Rule, is_rule_satisfied, sort_by_rules
from src.coordinator import Coordinator
from src.env_file_helper import DependencyRule, EnvFile, sort_env_files_by_rule
# @pytest.fixture
# def in_list():
# return ["a", "b", "c", "d", "e", "f", "g"]
# from src.env_file_helper import DependencyRule, EnvFile, sort_env_files_by_rule
from src.env_manager import DependencyRule, EnvFile, EnvManager
# @pytest.fixture
# def rules() -> list[Rule]:
# return [ # X depends on Y
# Rule("a", "e"),
# Rule("b", "e"),
# Rule("b", "f"),
# Rule("c", "e"),
# Rule("d", "e"),
# Rule("f", "e"),
# ]
def test_complex_sorting() -> None:
demo_rules = [ # X depends on Y
DependencyRule("a", "e"),
DependencyRule("b", "e"),
DependencyRule("b", "f"),
DependencyRule("c", "e"),
DependencyRule("d", "e"),
DependencyRule("f", "e"),
]
demo_types = ["a", "b", "c", "d", "e", "f", "g"]
env_files = [EnvFile(env_type=t, env_path=Path(), config=dict()) for t in demo_types]
EnvManager.sort_env_files_by_rule
sorted_env_files = EnvManager.sort_env_files_by_rule(env_files, demo_rules)
assert sorted_env_files[0].env_type == "e"
# def has_rules_satisfied(in_list, rules):
# rule_satisfied: list[bool] = []
# for rule in rules:
# if is_rule_satisfied(in_list, rule):
# rule_satisfied.append(True)
# else:
# rule_satisfied.append(False)
# return all(rule_satisfied)
def test_circular_import() -> None:
"""This test will raise ValueError because the example input cannot be correctly ordered"""
demo_rules = [
DependencyRule("a", "b"),
DependencyRule("b", "c"),
DependencyRule("c", "a"),
]
demo_types = ["a", "b", "c"]
env_files = [EnvFile(env_type=t, env_path=Path(), config=dict()) for t in demo_types]
with pytest.raises(ValueError):
EnvManager.sort_env_files_by_rule(env_files, demo_rules)
# def test_stuff(in_list, rules):
# sort_by_rules(in_list, rules)
# assert has_unique_elements(in_list)
# assert has_rules_satisfied(in_list, rules)
def test_real_env_files() -> None:
"""authentik should be first"""
ENV_FILES = [
ENV_FILES = [
Path("envfiles/blog.test.dev.local-it.cloud.env"), # wordpress
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
]
]
env_files: list[EnvFile] = EnvManager._get_env_files(ENV_FILES)
dependency_rules: list[DependencyRule] = EnvManager._get_dependency_rules(env_files)
sorted_env_files = EnvManager.sort_env_files_by_rule(env_files, dependency_rules)
assert sorted_env_files[0].env_type == "authentik"
env_files: list[EnvFile] = Coordinator._getn_env_files_list(ENV_FILES)
dependency_rules: list[DependencyRule] = Coordinator._get_dependency_rules(env_files)
def test_real_env_files_duplicate() -> None:
"""authentik should be first"""
ic(env_files)
sorted_env_files = sort_env_files_by_rule(env_files, dependency_rules)
ic(env_files)
ic(sorted_env_files)
ENV_FILES = [
Path("envfiles/blog.test.dev.local-it.cloud.env"), # wordpress
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
]
env_files: list[EnvFile] = EnvManager._get_env_files(ENV_FILES)
dependency_rules: list[DependencyRule] = EnvManager._get_dependency_rules(env_files)
sorted_env_files = EnvManager.sort_env_files_by_rule(env_files, dependency_rules)
assert sorted_env_files[0].env_type == "authentik"
assert sorted_env_files[1].env_type == "authentik"
assert sorted_env_files[2].env_type == "wordpress"
def test_real_env_files_duplicate_six() -> None:
"""authentik should be first"""
ENV_FILES = [
Path("envfiles/blog.test.dev.local-it.cloud.env"), # wordpress
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
Path("envfiles/blog.test.dev.local-it.cloud.env"), # wordpress
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
Path("envfiles/blog.test.dev.local-it.cloud.env"), # wordpress
]
env_files: list[EnvFile] = EnvManager._get_env_files(ENV_FILES)
dependency_rules: list[DependencyRule] = EnvManager._get_dependency_rules(env_files)
sorted_env_files = EnvManager.sort_env_files_by_rule(env_files, dependency_rules)
assert sorted_env_files[0].env_type == "authentik"
assert sorted_env_files[1].env_type == "authentik"
assert sorted_env_files[2].env_type == "authentik"
assert sorted_env_files[3].env_type == "wordpress"
assert sorted_env_files[4].env_type == "wordpress"
assert sorted_env_files[5].env_type == "wordpress"
def test_env_manager() -> None:
env_paths_list = [
Path("envfiles/blog.test.dev.local-it.cloud.env"), # wordpress
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
Path("envfiles/login.test.dev.local-it.cloud.env"), # authentik
]
ENV = EnvManager(env_paths_list)
assert ENV.env_files[0].env_type == "authentik"
assert ENV.env_files[1].env_type == "authentik"
assert ENV.env_files[2].env_type == "wordpress"