diff --git a/src/lxc/creation_utils.py b/src/lxc/creation_utils.py index 7e2af18..451552d 100644 --- a/src/lxc/creation_utils.py +++ b/src/lxc/creation_utils.py @@ -6,7 +6,6 @@ from typing import TYPE_CHECKING from . import lxc_utils from ..utils.resources_utils import get_path -from ..utils.utils import optional if TYPE_CHECKING: from .lxc import LXC @@ -31,19 +30,7 @@ def are_all_conditions_met(lxc: LXC): """ - # creation = lxc.get_creation() - # conditions = creation["conditions"] - # - # result = [] - # - # for file in conditions['files']: - # result.append(Path(file).is_file()) - # for folder in conditions['folders']: - # result.append(Path(folder).is_dir()) - # for program in conditions['programs']: - # result.append(lxc.run_command("which " + program, return_status_code=True) == 0) - - creation = lxc.get_creation() + creation = lxc.creation steps = creation["steps"] result = [] @@ -89,7 +76,7 @@ def check_conditions(c_type: str, parent: dict, lxc: LXC, result: list = None): if not result: result = [] - if c_type == "files" or c_type == "file": + 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": @@ -102,8 +89,8 @@ def check_conditions(c_type: str, parent: dict, lxc: LXC, result: list = None): return result -def check_conditions_for_step(step: dict, lxc: LXC): - """Check the conditions for the given step +def check_step_conditions(step: dict, lxc: LXC): + """Check the conditions for a given step Parameters ---------- @@ -137,17 +124,17 @@ def run_steps(lxc: LXC): The LXC object used to run the creation steps """ - creation = lxc.get_creation() + creation = lxc.creation creation_steps = creation["steps"] for index, step in enumerate(creation_steps): - if check_conditions_for_step(step, lxc): - logging.info(f"Conditions already met for step {index + 1}/{len(creation_steps)} for LXC {lxc.get_id()}, " + if check_step_conditions(step, lxc): + logging.info(f"Conditions already met for step {index + 1}/{len(creation_steps)} for LXC {lxc.id}, " f"skipping...") continue - logging.info(f"Running step {index + 1}/{len(creation_steps)} for LXC {lxc.get_id()}...") + logging.info(f"Running step {index + 1}/{len(creation_steps)} for LXC {lxc.id}...") match step["type"]: diff --git a/src/lxc/lxc.py b/src/lxc/lxc.py index 927faf9..b000cba 100644 --- a/src/lxc/lxc.py +++ b/src/lxc/lxc.py @@ -1,19 +1,24 @@ +from __future__ import annotations + import logging +from typing import TYPE_CHECKING from . import creation_utils, lxc_utils -from ..utils.machine import LinuxMachine -from ..utils.proxmox import ProxmoxHost +from ..machine.machine import LinuxMachine + +if TYPE_CHECKING: + from ..proxmox.proxmox import ProxmoxHost class LXC(LinuxMachine): """LXC object""" - def __init__(self, lxc_id: int, lxc_hostname: str, os: dict, resources: dict, network: dict, options: dict, + def __init__(self, id: int, hostname: str, os: dict, resources: dict, network: dict, options: dict, creation: dict, deploy: dict, features: dict, proxmox_host: ProxmoxHost): super().__init__() - self.lxc_id = lxc_id - self.lxc_hostname = lxc_hostname + self.id = id + self.hostname = hostname self.os = os self.os_name = os.get("name") @@ -49,54 +54,16 @@ class LXC(LinuxMachine): self.pve = proxmox_host def __str__(self): - return f"LXC {self.lxc_id} ({self.lxc_hostname})" + return f"LXC {self.id} ({self.hostname})" def __repr__(self): - return f"LXC {self.lxc_id} ({self.lxc_hostname})" + return f"LXC {self.id} ({self.hostname})" def __eq__(self, other): - return self.lxc_id == other.lxc_id + return self.id == other.id def __hash__(self): - return hash(self.lxc_id) - - def get_id(self): - """Get LXC ID - - Returns - ------- - int - LXC ID - """ - return self.lxc_id - - def get_hostname(self): - """ - Get LXC hostname - :return: lxc hostname - """ - return self.lxc_hostname - - def get_os(self): - """ - Get OS - :return: os - """ - return self.os - - def get_os_name(self): - """ - Get OS name - :return: os name - """ - return self.os_name - - def get_os_release(self): - """ - Get OS release - :return: os release - """ - return self.os_release + return hash(self.id) def get_os_template(self): """ @@ -117,63 +84,7 @@ class LXC(LinuxMachine): return f"local:vztmpl/{template_name}" - def get_resources(self): - """ - Get resources - :return: resources - """ - return self.resources - - def get_cpu(self): - """ - Get CPU - :return: cpu - """ - return self.cpu - - def get_memory(self): - """ - Get memory - :return: memory - """ - return self.memory - - def get_swap(self): - """ - Get swap - :return: swap - """ - return self.swap - - def get_disk(self): - """ - Get disk - :return: disk - """ - return self.disk - - def get_storage(self): - """ - Get storage - :return: storage - """ - return self.storage - - def get_network(self): - """ - Get network - :return: network - """ - return self.network - - def get_bridge(self): - """ - Get bridge - :return: bridge - """ - return self.bridge - - def get_ipv4(self, netmask: bool = False, use_ssh: bool = False): + def retrieve_ipv4(self, netmask: bool = False, use_ssh: bool = False): """ Get IPv4 :return: ipv4 @@ -200,136 +111,59 @@ class LXC(LinuxMachine): else: return self.ipv4 - def get_ipv6(self): + def retrieve_ipv6(self): """ Get IPv6 :return: ipv6 """ return self.ipv6 - def get_mac(self): - """ - Get MAC - :return: mac - """ - return self.mac - - def get_gateway4(self): - """ - Get gateway IPV4 - :return: gateway - """ - return self.gateway4 - - def get_gateway6(self): - """ - Get gateway IPV6 - :return: gateway - """ - return self.gateway6 - - def get_vlan(self): - """ - Get VLAN - :return: vlan - """ - return self.vlan - - def get_options(self): - """ - Get options - :return: options - """ - return self.options - - def is_privileged(self): - """ - Is privileged - :return: privileged - """ - return self.privileged - - def is_start_on_boot(self): - """ - Is start on boot - :return: start on boot - """ - return self.start_on_boot - - def get_startup_order(self): - """ - Get startup order - :return: startup order - """ - return self.startup_order - - def get_password(self): - """ - Get password - :return: password - """ - return self.password - def get_ssh_string(self): # TODO: implement hostname and maybe ipv6? - return "root@" + self.get_ipv4() - - def get_deploy(self): - """ - Get deployments - :return: deployments - """ - return self.deploy - - def get_creation(self): - """ - Get creation - :return: creation - """ - return self.creation + return "root@" + self.retrieve_ipv4() def is_running(self): """ Is running :return: is lxc running? (boolean) """ - return self.pve.is_lxc_running(self.lxc_id) + return self.pve.is_lxc_running(self.id) def does_exist(self): """ Does exist :return: does lxc exist? (boolean) """ - return self.pve.does_lxc_exist(self.lxc_id) + return self.pve.does_lxc_exist(self.id) def start(self): """ Start LXC """ - self.pve.start_lxc(self.lxc_id) + self.pve.start_lxc(self.id) def stop(self): """ Stop LXC """ - self.pve.stop_lxc(self.lxc_id) + self.pve.stop_lxc(self.id) def reboot(self): """ Reboot LXC """ - logging.info(f"Rebooting LXC {self.lxc_id}") - self.pve.reboot_lxc(self.lxc_id) + logging.info(f"Rebooting LXC {self.id}") + self.pve.reboot_lxc(self.id) def create(self): """ Create LXC """ if self.does_exist(): - logging.info(f"LXC {self.lxc_id} already exists, skipping creation") + logging.info(f"LXC {self.id} already exists, skipping creation") self.pve.run_command(command=lxc_utils.generate_pct_command_for_lxc(self, create=False)) else: - logging.info(f"Creating LXC {self.lxc_id}") + logging.info(f"Creating LXC {self.id}") self.pve.run_command(command=lxc_utils.generate_pct_command_for_lxc(self, create=True)) self.start() @@ -341,7 +175,7 @@ class LXC(LinuxMachine): # Install and configure OpenSSH on LXC logging.info("Setting up SSH for LXC") lxc_utils.run_protected_script(lxc=self, script_path="protected/scripts/install-config-ssh.sh") - self.pve.run_command(f"ssh-keygen -f '/root/.ssh/known_hosts' -R {self.get_ipv4(use_ssh=False)}") + self.pve.run_command(f"ssh-keygen -f '/root/.ssh/known_hosts' -R {self.retrieve_ipv4(use_ssh=False)}") # Add main, community and testing repo for Alpine # if "alpine" in self.os_name: @@ -358,13 +192,13 @@ class LXC(LinuxMachine): if not self.is_running(): self.start() - logging.info(f"Running creation checks for LXC {self.lxc_id}") + 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): - logging.info(f"Not all creation conditions met for LXC {self.lxc_id}, running creation steps...") + logging.info(f"Not all creation conditions met for LXC {self.id}, running creation steps...") # Run creation steps - logging.info(f"Running creation steps for LXC {self.lxc_id}") + logging.info(f"Running creation steps for LXC {self.id}") creation_utils.run_steps(self) def deploy(self): @@ -405,12 +239,12 @@ class LXC(LinuxMachine): # Using pct exec works every time but is 8x slower than using ssh if use_ssh: return self.pve.run_command( - command=f"ssh -o StrictHostKeyChecking=no root@{self.get_ipv4()} -- \"{command}\"", + command=f"ssh -o StrictHostKeyChecking=no root@{self.retrieve_ipv4()} -- \"{command}\"", return_status_code=return_status_code, exception_on_exit=exception_on_exit, exception_on_empty_stdout=exception_on_empty_stdout) else: - return self.pve.run_command(command=f"pct exec {self.lxc_id} -- {command}", + 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) \ No newline at end of file diff --git a/src/lxc/lxc_utils.py b/src/lxc/lxc_utils.py index f45b8b0..bb1c822 100644 --- a/src/lxc/lxc_utils.py +++ b/src/lxc/lxc_utils.py @@ -1,10 +1,17 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + import yaml from .lxc import LXC +from ..proxmox.proxmox import ProxmoxHost from ..utils import utils -from ..utils.proxmox import ProxmoxHost from ..utils.resources_utils import get_path +if TYPE_CHECKING: + from ..proxmox.proxmox import ProxmoxHost + lxcs = [] @@ -37,8 +44,7 @@ def load_lxc(lxc_id: int, content: str or bytes, pve: ProxmoxHost): data = yaml.safe_load(content) # Extract values from JSON - lxc_id = lxc_id - lxc_hostname = data.get("lxc_hostname") + hostname = data.get("lxc_hostname") os = data.get("os") resources = data.get("resources") network = data.get("network") @@ -48,41 +54,41 @@ def load_lxc(lxc_id: int, content: str or bytes, pve: ProxmoxHost): features = data.get("features") # Create LXC object - lxc = LXC(lxc_id, lxc_hostname, os, resources, network, options, creation, deploy, features, pve) + lxc = LXC(lxc_id, hostname, os, resources, network, options, creation, deploy, features, pve) lxcs.append(lxc) -def get_tteck_env_variables(self): - """ - Get TTECK environment variables to run scripts silently - ! Deprecated for now ! - - :return: environment variables - """ - - env_variables = { - "CT_TYPE": "1", - "PW": self.password, - "CT_ID": self.lxc_id, - "HN": self.lxc_hostname, - "DISK_SIZE": self.disk, - "CORE_COUNT": self.cpu, - "RAM_SIZE": self.memory, - "BRG": self.bridge, - "NET": self.ipv4, - "GATE": self.gateway4, - "DISABLEIP6": "no", - "MTU": "", - "SD": "", - "NS": "", - "MAC": self.mac, - "VLAN": self.vlan, - # "SSH": self.ssh, - "VERB": "no" - } - - env_command = " && ".join([f"export {name}=\"{value}\"" for name, value in env_variables.items()]) - return env_command +# def get_tteck_env_variables(self): +# """ +# Get TTECK environment variables to run scripts silently +# ! Deprecated for now ! +# +# :return: environment variables +# """ +# +# env_variables = { +# "CT_TYPE": "1", +# "PW": self.password, +# "CT_ID": self.id, +# "HN": self.hostname, +# "DISK_SIZE": self.disk, +# "CORE_COUNT": self.cpu, +# "RAM_SIZE": self.memory, +# "BRG": self.bridge, +# "NET": self.ipv4, +# "GATE": self.gateway4, +# "DISABLEIP6": "no", +# "MTU": "", +# "SD": "", +# "NS": "", +# "MAC": self.mac, +# "VLAN": self.vlan, +# # "SSH": self.ssh, +# "VERB": "no" +# } +# +# env_command = " && ".join([f"export {name}=\"{value}\"" for name, value in env_variables.items()]) +# return env_command def generate_pct_command_for_lxc(lxc: LXC, create: bool = True): @@ -93,12 +99,12 @@ def generate_pct_command_for_lxc(lxc: LXC, create: bool = True): # Common parameters for both create and update commands common_params = [ - f"--hostname {lxc.get_hostname()}", - f"--cores {lxc.get_cpu()}", - f"--memory {lxc.get_memory()}", - f"--swap {lxc.get_memory()}", - f"--onboot {int(lxc.is_start_on_boot())}", - f"--ostype {lxc.get_os_name()}", + f"--hostname {lxc.hostname}", + f"--cores {lxc.cpu}", + f"--memory {lxc.memory}", + f"--swap {lxc.swap}", + f"--onboot {int(lxc.start_on_boot)}", + f"--ostype {lxc.os_name}", ] # Check and include specific net0 parameters @@ -115,9 +121,9 @@ def generate_pct_command_for_lxc(lxc: LXC, create: bool = True): if lxc.network.get("hwaddr") and lxc.network.get("hwaddr") != "": net0_params.append(f"hwaddr={lxc.network['hwaddr']}") if lxc.network.get("ip") and lxc.network.get("ip") != "": - net0_params.append(f"ip={lxc.get_ipv4(netmask=True)}") + net0_params.append(f"ip={lxc.retrieve_ipv4(netmask=True)}") if lxc.network.get("ip6") and lxc.network.get("ip6") != "": - net0_params.append(f"ip6={lxc.get_ipv6()}") + net0_params.append(f"ip6={lxc.retrieve_ipv6()}") if lxc.network.get("link_down") and lxc.network.get("link_down") != "": net0_params.append(f"link_down={lxc.network['link_down']}") if lxc.network.get("mtu") and lxc.network.get("mtu") != "": @@ -137,40 +143,41 @@ def generate_pct_command_for_lxc(lxc: LXC, create: bool = True): if create: # Additional parameters for create command create_params = [ - f"--password {lxc.get_password()}", - f"--storage {lxc.get_storage()}", - f"--unprivileged {not lxc.is_privileged()}", - f"--rootfs volume={lxc.get_storage()}:{lxc.get_disk()},size={lxc.get_disk()}", + f"--password {lxc.password}", + f"--storage {lxc.storage}", + f"--unprivileged {not lxc.privileged}", + f"--rootfs volume={lxc.storage}:{lxc.disk},size={lxc.disk}", "--ssh-public-keys /root/.ssh/id_rsa.pub", ] # Check and include specific features based on their values - features = [] - if lxc.features.get("force_rw_sys") and lxc.features.get("force_rw_sys") != "": - features.append(f"force_rw_sys={lxc.features['force_rw_sys']}") - if lxc.features.get("fuse") and lxc.features.get("fuse") != "": - features.append(f"fuse={lxc.features['fuse']}") - if lxc.features.get("keyctl") and lxc.features.get("keyctl") != "": - features.append(f"keyctl={lxc.features['keyctl']}") - if lxc.features.get("mknod") and lxc.features.get("mknod") != "": - features.append(f"mknod={lxc.features['mknod']}") - if lxc.features.get("mount") and lxc.features.get("mount") != "": - features.append(f"mount={';'.join(lxc.features['mount'])}") - if lxc.features.get("nesting") and lxc.features.get("nesting") != "": - features.append(f"nesting={lxc.features['nesting']}") + features_params = [] + if lxc.features is not None and type(lxc.features) is dict: + if lxc.features.get("force_rw_sys") and lxc.features.get("force_rw_sys") != "": + features_params.append(f"force_rw_sys={lxc.features['force_rw_sys']}") + if lxc.features.get("fuse") and lxc.features.get("fuse") != "": + features_params.append(f"fuse={lxc.features['fuse']}") + if lxc.features.get("keyctl") and lxc.features.get("keyctl") != "": + features_params.append(f"keyctl={lxc.features['keyctl']}") + if lxc.features.get("mknod") and lxc.features.get("mknod") != "": + features_params.append(f"mknod={lxc.features['mknod']}") + if lxc.features.get("mount") and lxc.features.get("mount") != "": + features_params.append(f"mount={';'.join(lxc.features['mount'])}") + if lxc.features.get("nesting") and lxc.features.get("nesting") != "": + features_params.append(f"nesting={lxc.features['nesting']}") - if features: - create_params.append(f"--features {','.join(features)}") + if features_params: + create_params.append(f"--features {','.join(features_params)}") # Combine common and create-specific parameters command_params = common_params + create_params # Create command - pct_command = f"pct create {lxc.get_id()} {lxc.get_os_template()} {' '.join(command_params)}" + pct_command = f"pct create {lxc.id} {lxc.get_os_template()} {' '.join(command_params)}" else: # Update command # Combine common parameters only command_params = common_params - pct_command = f"pct set {lxc.get_id()} {' '.join(command_params)}" + pct_command = f"pct set {lxc.id} {' '.join(command_params)}" # TODO: add gateway4 # f"ip6={self.ipv6},gw6={self.gateway6},trunks={self.vlan} " \ # TODO diff --git a/src/utils/machine.py b/src/machine/machine.py similarity index 94% rename from src/utils/machine.py rename to src/machine/machine.py index 5a97885..3ecbe85 100644 --- a/src/utils/machine.py +++ b/src/machine/machine.py @@ -1,9 +1,9 @@ from pathlib import Path, PurePosixPath -from src.utils import utils +from . import machine_utils -class LinuxMachine(): +class LinuxMachine: def __init__(self): self.known_programs = [] @@ -26,20 +26,23 @@ class LinuxMachine(): str OS name """ + if hasattr(self, "os_name"): + return self.os_name + return self.run_command("""cat /etc/os-release | grep -E '^NAME=' | cut -d '=' -f 2 | tr -d '"'""") def get_memory(self): """Get memory""" return self.run_command("free -m | grep Mem | awk '{print $2}'") - def get_ipv4(self): + def retrieve_ipv4(self): """Get IPv4 address""" if self.has_program("ip"): return self.run_command("""ip addr | grep 'state UP' -A2 | tail -n1 | awk '{print $2}' | cut -f1 -d'/'""") elif self.has_program("ifconfig"): return self.run_command(command="ifconfig eth0 | awk '/inet addr/{print substr($2,6)}'") - def get_ipv6(self): + def retrieve_ipv6(self): pass def get_mac(self): @@ -254,10 +257,10 @@ class LinuxMachine(): packages.append(p) self.run_command( - f"{utils.get_install_package_command(self.get_os_name())} {' '.join(packages)}", + f"{machine_utils.get_install_package_command(self.get_os_name())} {' '.join(packages)}", return_status_code=True) else: - self.run_command(f"{utils.get_install_package_command(self.get_os_name())} {package}", + self.run_command(f"{machine_utils.get_install_package_command(self.get_os_name())} {package}", return_status_code=True, use_ssh=use_ssh) def remove_package(self, package: str or list): @@ -280,11 +283,11 @@ class LinuxMachine(): packages.append(p) self.run_command( - f"{utils.get_remove_package_command(self.get_os_name())} {' '.join(packages)}", + f"{machine_utils.get_remove_package_command(self.get_os_name())} {' '.join(packages)}", return_status_code=True) else: - self.run_command(f"{utils.get_remove_package_command(self.get_os_name())} {package}", + self.run_command(f"{machine_utils.get_remove_package_command(self.get_os_name())} {package}", return_status_code=True) def replace_in_files(self, path: str or list, search: str, replace: str, diff --git a/src/machine/machine_utils.py b/src/machine/machine_utils.py new file mode 100644 index 0000000..e41c96e --- /dev/null +++ b/src/machine/machine_utils.py @@ -0,0 +1,90 @@ +def get_install_package_command(distribution: str): + """Retrieve the correct command to install a package based on the distribution specified + It supports all the distribution supported by Proxmox VE (from the pct command documentation). + Debian, Ubuntu, CentOS, Fedora, Gentoo, Alpine, ArchLinux, Devuan, NixOS, OpenSUSE + + Parameters + ---------- + distribution: str + Name of the distribution as specific in the proxmox pct command documentation + + See Also + -------- + https://pve.proxmox.com/pve-docs/pct.1.html + + Returns + ------- + str + Beginning of the command to install a package, the package name should be appended to it + """ + + distribution = distribution.lower() + + if "debian" in distribution: + return "apt-get install -y" + elif "ubuntu" in distribution: + return "apt-get install -y" + elif "centos" in distribution: + return "yum install -y" + elif "fedora" in distribution: + return "yum install -y" + elif "gentoo" in distribution: + return "emerge -a" + elif "alpine" in distribution: + return "apk add" + elif "archlinux" in distribution: + return "pacman -S --noconfirm" + elif "devuan" in distribution: + return "apt-get install -y" + elif "nixos" in distribution: + return "nix-env -i" + elif "opensuse" in distribution: + return "zypper install -y" + else: + raise Exception(f"Unsupported distribution: {distribution}") + + +def get_remove_package_command(distribution: str): + """Retrieve the correct command to uninstall a package based on the distribution specified + It supports all the distribution supported by Proxmox VE (from the pct command documentation). + Debian, Ubuntu, CentOS, Fedora, Gentoo, Alpine, ArchLinux, Devuan, NixOS, OpenSUSE + + Parameters + ---------- + distribution: str + Name of the distribution as specific in the proxmox pct command documentation + + See Also + -------- + https://pve.proxmox.com/pve-docs/pct.1.html + + Returns + ------- + str + Beginning of the command to uninstall a package, the package name should be appended to it + """ + + distribution = distribution.lower() + + if "debian" in distribution: + return "apt-get remove -y" + elif "ubuntu" in distribution: + return "apt-get remove -y" + elif "centos" in distribution: + return "yum remove -y" + elif "fedora" in distribution: + return "yum remove -y" + elif "gentoo" in distribution: + return "emerge -C" + elif "alpine" in distribution: + return "apk del" + elif "archlinux" in distribution: + return "pacman -R --noconfirm" + elif "devuan" in distribution: + return "apt-get remove -y" + elif "nixos" in distribution: + return "nix-env -e" + elif "opensuse" in distribution: + return "zypper remove -y" + else: + raise Exception(f"Unsupported distribution: {distribution}") diff --git a/src/main.py b/src/main.py index 1d62a28..184919b 100644 --- a/src/main.py +++ b/src/main.py @@ -3,7 +3,7 @@ from pathlib import Path from .lxc.lxc_utils import load_lxc, get_all_lxcs from .utils import git_utils -from .utils.proxmox import ProxmoxHost +from .proxmox.proxmox import ProxmoxHost def run(args): @@ -44,7 +44,7 @@ def run(args): load_lxc(content=lxc_file_content, lxc_id=int(lxc_folder), pve=pve) for lxc in get_all_lxcs(): - logging.info(f"Loading LXC {lxc.lxc_id}") + logging.info(f"Loading LXC {lxc.id}") lxc.create() lxc.start() lxc.run_creation() diff --git a/src/utils/proxmox.py b/src/proxmox/proxmox.py similarity index 97% rename from src/utils/proxmox.py rename to src/proxmox/proxmox.py index 8baea8c..8420353 100644 --- a/src/utils/proxmox.py +++ b/src/proxmox/proxmox.py @@ -9,7 +9,7 @@ from typing import TYPE_CHECKING from fabric import Connection -from src.utils.machine import LinuxMachine +from ..machine.machine import LinuxMachine if TYPE_CHECKING: from ..lxc.lxc import LXC @@ -77,10 +77,6 @@ class ProxmoxHost(LinuxMachine): if self.host is not None: self.connection = Connection(host=self.host, user=self.user, port=self.port) - def get_connection(self): - """Get the connection to the Proxmox host.""" - return self.connection - def run_command(self, command: str, return_status_code: bool = False, exception_on_exit: bool = True, exception_on_empty_stdout: bool = False, **kwargs): """Run a command on the Proxmox VE host @@ -254,7 +250,7 @@ class ProxmoxHost(LinuxMachine): destination = Path(destination) lxc.create_directory(destination.parent.as_posix(), use_ssh=use_ssh) - self.run_command(f"pct push {lxc.get_id()} {source} {destination.as_posix()}") + self.run_command(f"pct push {lxc.id} {source} {destination.as_posix()}") def copy_folder_to_lxc(self, lxc: LXC, source: str or Path, destination: str or Path): """Copy the given folder to the given LXC.""" diff --git a/src/utils/__init__.py b/src/utils/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/utils/git_utils.py b/src/utils/git_utils.py index 3e1ec81..e45283d 100644 --- a/src/utils/git_utils.py +++ b/src/utils/git_utils.py @@ -1,6 +1,6 @@ import logging -from src.utils.machine import LinuxMachine +from ..machine.machine import LinuxMachine def clone_repo(url, path, linux_machine: LinuxMachine): diff --git a/src/utils/resources_utils.py b/src/utils/resources_utils.py index a16aed7..12dc2db 100644 --- a/src/utils/resources_utils.py +++ b/src/utils/resources_utils.py @@ -45,5 +45,5 @@ def get_path(lxc: LXC, path: str): return app_path.joinpath("protected_resources", path[len("protected/"):]) else: # Use VM/LXC folder - lxc_id = lxc.get_id() + lxc_id = lxc.id return parent.joinpath("lxc", str(lxc_id), path) diff --git a/src/utils/utils.py b/src/utils/utils.py index 55525f1..b72ce61 100644 --- a/src/utils/utils.py +++ b/src/utils/utils.py @@ -1,98 +1,6 @@ from pathlib import Path -from src.utils.proxmox import ProxmoxHost - - -def get_install_package_command(distribution: str): - """Retrieve the correct command to install a package based on the distribution specified - It supports all the distribution supported by Proxmox VE (from the pct command documentation). - Debian, Ubuntu, CentOS, Fedora, Gentoo, Alpine, ArchLinux, Devuan, NixOS, OpenSUSE - - Parameters - ---------- - distribution: str - Name of the distribution as specific in the proxmox pct command documentation - - See Also - -------- - https://pve.proxmox.com/pve-docs/pct.1.html - - Returns - ------- - str - Beginning of the command to install a package, the package name should be appended to it - """ - - distribution = distribution.lower() - - if "debian" in distribution: - return "apt-get install -y" - elif "ubuntu" in distribution: - return "apt-get install -y" - elif "centos" in distribution: - return "yum install -y" - elif "fedora" in distribution: - return "yum install -y" - elif "gentoo" in distribution: - return "emerge -a" - elif "alpine" in distribution: - return "apk add" - elif "archlinux" in distribution: - return "pacman -S --noconfirm" - elif "devuan" in distribution: - return "apt-get install -y" - elif "nixos" in distribution: - return "nix-env -i" - elif "opensuse" in distribution: - return "zypper install -y" - else: - raise Exception(f"Unsupported distribution: {distribution}") - - -def get_remove_package_command(distribution: str): - """Retrieve the correct command to uninstall a package based on the distribution specified - It supports all the distribution supported by Proxmox VE (from the pct command documentation). - Debian, Ubuntu, CentOS, Fedora, Gentoo, Alpine, ArchLinux, Devuan, NixOS, OpenSUSE - - Parameters - ---------- - distribution: str - Name of the distribution as specific in the proxmox pct command documentation - - See Also - -------- - https://pve.proxmox.com/pve-docs/pct.1.html - - Returns - ------- - str - Beginning of the command to uninstall a package, the package name should be appended to it - """ - - distribution = distribution.lower() - - if "debian" in distribution: - return "apt-get remove -y" - elif "ubuntu" in distribution: - return "apt-get remove -y" - elif "centos" in distribution: - return "yum remove -y" - elif "fedora" in distribution: - return "yum remove -y" - elif "gentoo" in distribution: - return "emerge -C" - elif "alpine" in distribution: - return "apk del" - elif "archlinux" in distribution: - return "pacman -R --noconfirm" - elif "devuan" in distribution: - return "apt-get remove -y" - elif "nixos" in distribution: - return "nix-env -e" - elif "opensuse" in distribution: - return "zypper remove -y" - else: - raise Exception(f"Unsupported distribution: {distribution}") +from ..proxmox.proxmox import ProxmoxHost def optional(var: any, placeholder: any): @@ -137,4 +45,4 @@ def copy_local_file_to_pve(pve: ProxmoxHost, local_source: str or Path, destinat destination = Path(destination) pve.run_command(f"mkdir -p {destination.parent.as_posix()}") - pve.get_connection().put(local_source, destination.as_posix()) + pve.connection.put(local_source, destination.as_posix())