mathieu/ProxmoxDeploy
mathieu/ProxmoxDeploy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

refactor lots of code (remove useless getters, move to separates packages, etc)
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Mathieu Broillet 2023-06-22 11:10:22 +02:00
parent 08be432453
commit f20b2bbd38
No known key found for this signature in database
GPG Key ID: 7D4F25BC50A0AA32
11 changed files with 220 additions and 395 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/lxc/creation_utils.py
View File

@ -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"]:

230
src/lxc/lxc.py
View File

@ -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)

135
src/lxc/lxc_utils.py
View File

@ -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

19
src/utils/machine.py → src/machine/machine.py
View File

@ -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,

90
src/machine/machine_utils.py Normal file
View File

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

4
src/main.py
View File

@ -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()

8
src/utils/proxmox.py → src/proxmox/proxmox.py
View File

@ -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."""

0
src/utils/__init__.py
View File

2
src/utils/git_utils.py
View File

@ -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):

2
src/utils/resources_utils.py
View File

@ -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)

96
src/utils/utils.py
View File

@ -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())