diff --git a/src/lxc/creation_utils.py b/src/lxc/creation_utils.py index 451552d..09fc401 100644 --- a/src/lxc/creation_utils.py +++ b/src/lxc/creation_utils.py @@ -1,120 +1,16 @@ from __future__ import annotations import logging -from pathlib import Path from typing import TYPE_CHECKING from . import lxc_utils +from ..utils import conditions_utils from ..utils.resources_utils import get_path if TYPE_CHECKING: from .lxc import LXC -def are_all_conditions_met(lxc: LXC): - """ - Check conditions for running the creation steps for an LXC. - - This function checks if the LXC with the given name has already been configured or not. If all conditions are met, - the deployment/updates steps are run. Otherwise, the creation steps are run. - - Parameters - ---------- - lxc : LXC - The LXC object used to check the conditions - - Returns - ------- - bool - True if all conditions are met, False otherwise - - """ - - creation = lxc.creation - steps = creation["steps"] - result = [] - - if 'conditions' in creation: - global_conditions = creation["conditions"] - - # Check the global conditions - for condition_type in global_conditions: - result = check_conditions(c_type=condition_type, parent=global_conditions, lxc=lxc, result=result) - - # Check the specific conditions for each step - for step in steps: - if "conditions" in step: - for condition_type in step["conditions"]: - result = check_conditions(c_type=condition_type, parent=step["conditions"], lxc=lxc, result=result) - - if len(result) == 0: - return False - - return all(result) - - -def check_conditions(c_type: str, parent: dict, lxc: LXC, result: list = None): - """Check the conditions for the given LXC - - Parameters - ---------- - c_type : str - The type of condition to check - parent : dict - The parent dictionary of the condition - lxc : LXC - The LXC object used to check the conditions - result : list, optional - The list of results for the conditions to append to, by default None (creates a new list) - - Returns - ------- - list - The list of results for the conditions - """ - - if not result: - result = [] - - if c_type == "files" or c_type == "file": - for file in parent[c_type]: - result.append(lxc.has_file(Path(file))) - elif c_type == "folders" or c_type == "folder": - for folder in parent[c_type]: - result.append(lxc.has_directory(Path(folder))) - elif c_type == "programs" or c_type == "program": - for program in parent[c_type]: - result.append(lxc.has_program(program)) - - return result - - -def check_step_conditions(step: dict, lxc: LXC): - """Check the conditions for a given step - - Parameters - ---------- - step : dict - The step to check the conditions for - lxc : LXC - The LXC object used to check the conditions - - Returns - ------- - bool - True if all conditions are met, False otherwise - """ - - result = [] - if "conditions" in step: - for condition in step['conditions']: - result = check_conditions(condition, parent=step['conditions'], lxc=lxc, result=result) - else: - return False - - return all(result) - - def run_steps(lxc: LXC): """Run the creation steps for the given LXC @@ -129,7 +25,7 @@ def run_steps(lxc: LXC): for index, step in enumerate(creation_steps): - if check_step_conditions(step, lxc): + if conditions_utils.verify_step_conditions(lxc, step): logging.info(f"Conditions already met for step {index + 1}/{len(creation_steps)} for LXC {lxc.id}, " f"skipping...") continue diff --git a/src/lxc/lxc.py b/src/lxc/lxc.py index b000cba..ec3088f 100644 --- a/src/lxc/lxc.py +++ b/src/lxc/lxc.py @@ -5,6 +5,7 @@ from typing import TYPE_CHECKING from . import creation_utils, lxc_utils from ..machine.machine import LinuxMachine +from ..utils import conditions_utils if TYPE_CHECKING: from ..proxmox.proxmox import ProxmoxHost @@ -194,7 +195,7 @@ class LXC(LinuxMachine): logging.info(f"Running creation checks for LXC {self.id}") # Check if all creation conditions are met - if not creation_utils.are_all_conditions_met(self): + if not conditions_utils.are_all_conditions_met(self): logging.info(f"Not all creation conditions met for LXC {self.id}, running creation steps...") # Run creation steps @@ -247,4 +248,4 @@ class LXC(LinuxMachine): return self.pve.run_command(command=f"pct exec {self.id} -- {command}", return_status_code=return_status_code, exception_on_exit=exception_on_exit, - exception_on_empty_stdout=exception_on_empty_stdout) \ No newline at end of file + exception_on_empty_stdout=exception_on_empty_stdout) diff --git a/src/utils/conditions_utils.py b/src/utils/conditions_utils.py new file mode 100644 index 0000000..2022fa0 --- /dev/null +++ b/src/utils/conditions_utils.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from ..lxc.lxc import LXC + from ..machine.machine import LinuxMachine + + +def are_all_conditions_met(lxc: LXC): + """ + Check conditions for running the creation steps for an LXC. + + Parameters + ---------- + lxc + + Returns + ------- + + """ + + results = [] + + creation_dict = lxc.creation + steps_dict = lxc.creation["steps"] + + # Check the global conditions + if 'conditions' in creation_dict: + global_conditions = creation_dict["conditions"] + + for condition in global_conditions: + results.append(verify_condition(lxc, condition_type=condition, data=global_conditions[condition])) + + # Check the specific conditions for each step + for step in steps_dict: + if "conditions" in step: + for condition in step["conditions"]: + results.append(verify_condition(lxc, condition_type=condition, data=step["conditions"][condition])) + + if len(results) == 0: + return False + + return all(results) + + +def verify_condition(linux_machine: LinuxMachine, condition_type: str, data: list or str): + results = [] + + if type(condition_type) is str: + data = [data] + + for condition in data: + print(condition_type, condition) + if condition_type == "file" or condition_type == "files": + results.append(_check_file_condition(linux_machine, condition)) + elif condition_type == "folder" or condition_type == "folders": + results.append(_check_folder_condition(linux_machine, condition)) + elif condition_type == "program" or condition_type == "programs": + results.append(_check_program_condition(linux_machine, condition)) + elif condition_type == "command" or condition_type == "commands": + results.append(_check_command_condition(linux_machine, condition)) + elif condition_type == "docker": + results.append(_check_docker_condition(linux_machine, condition)) + else: + print("Unknown condition type: " + condition_type) + return False + + return all(results) + + +def verify_step_conditions(linux_machine: LinuxMachine, step: dict): + results = [] + + if "conditions" in step: + for condition in step["conditions"]: + results.append( + verify_condition(linux_machine, condition_type=condition, data=step["conditions"][condition])) + + +def _check_file_condition(linux_machine: LinuxMachine, file: list): + return all([linux_machine.has_file(Path(f)) for f in file]) + + +def _check_folder_condition(linux_machine: LinuxMachine, folder: list): + return all([linux_machine.has_directory(Path(f)) for f in folder]) + + +def _check_program_condition(linux_machine: LinuxMachine, program: list): + return all([linux_machine.has_program(p) for p in program]) + + +def _check_command_condition(linux_machine: LinuxMachine, command: list): + result = [] + + for c in command: + cmd = c.split(" ")[0] + value = c.split(" ")[1] + result = linux_machine.run_command(cmd) + if result != value: + result.append(False) + + return all(result) + + +def _check_docker_condition(linux_machine: LinuxMachine, container: list): + return container in linux_machine.run_command(f"docker ps -q -f name=\"{container}\"")