remade the conditions checker

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

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)
+                                        exception_on_empty_stdout=exception_on_empty_stdout)

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}\"")