improved code documentation
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
Mathieu Broillet 2023-06-12 16:54:44 +02:00
parent 6a6cddbc5f
commit 9cc09aafa1
No known key found for this signature in database
GPG Key ID: 7D4F25BC50A0AA32
3 changed files with 212 additions and 28 deletions

View File

@ -6,13 +6,21 @@ from src.utils import lxc_commands_utils
def are_all_conditions_met(lxc): def are_all_conditions_met(lxc):
""" """
Check conditions for running the creations steps for an LXC Check conditions for running the creation steps for an LXC.
The conditions are for checking if the LXC has already been configured or not
If all conditions are met, the deploy/updates steps are run This function checks if the LXC with the given name has already been configured or not. If all conditions are met,
Otherwise, we run the creation steps 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
:param lxc:
:return: are conditions met
""" """
creation = lxc.get_creation() creation = lxc.get_creation()
@ -36,6 +44,20 @@ def are_all_conditions_met(lxc):
def optional(var, placeholder): def optional(var, placeholder):
"""Return a placeholder if the given variable is None, otherwise return the variable
Parameters
----------
var
Variable to check if it is None
placeholder
Placeholder to return if the variable is None
Returns
-------
var or placeholder
"""
if var is None: if var is None:
return placeholder return placeholder
else: else:
@ -43,12 +65,14 @@ def optional(var, placeholder):
def run_steps(lxc): def run_steps(lxc):
""" """Run the creation steps for the given LXC
Run creation steps for an LXC
:param lxc: lxc Parameters
:return: None ----------
lxc : LXC
The LXC object used to run the creation steps
""" """
creation = lxc.get_creation() creation = lxc.get_creation()
creation_steps = creation["creation_steps"] creation_steps = creation["creation_steps"]

View File

@ -5,6 +5,18 @@ from src.utils.resources_utils import get_path
def run_script(lxc, step): def run_script(lxc, step):
"""Method use to dispatch the script to the correct method depending on the type of script, being
local (on the machine running this program), remote (url) or in the LXC.
Parameters
----------
lxc : LXC
The LXC object used to run the script
step: dict
Dictionary containing the step information about the script to run
Typically read from the "creation/steps/<step>" in JSON file
"""
# Run local script # Run local script
if "path" in step: if "path" in step:
_run_local_script_on_lxc(lxc, step["path"]) _run_local_script_on_lxc(lxc, step["path"])
@ -18,23 +30,31 @@ def run_script(lxc, step):
_run_script_on_lxc(lxc, step["lxc_path"]) _run_script_on_lxc(lxc, step["lxc_path"])
def _run_script_on_lxc(lxc, path): def _run_script_on_lxc(lxc, path: "Path inside the LXC"):
"""Run a script in the LXC"""
lxc.run_command(f"bash {path}") lxc.run_command(f"bash {path}")
def _run_local_script_on_lxc(lxc, path): def _run_local_script_on_lxc(lxc, path: "Path to the script on the machine running this program"):
"""Run a local script in the LXC"""
path = get_path(lxc, path) path = get_path(lxc, path)
with open(path, "r") as file: with open(path, "r") as file:
script = file.read() script = file.read()
lxc.run_command("'bash -s' <<'ENDSSH'\n" + script) lxc.run_command("'bash -s' <<'ENDSSH'\n" + script)
def _run_remote_script_on_lxc(lxc, url): def _run_remote_script_on_lxc(lxc, url: "URL to the script"):
"""Run a remote script in the LXC"""
if lxc.has_program("curl"): if lxc.has_program("curl"):
lxc.run_command(f"curl -sSL {url} | bash") lxc.run_command(f"curl -sSL {url} | bash")
def create_folder_in_lxc(lxc, path, permission=755): def create_folder_in_lxc(lxc, path, permission=755):
"""Create a folder in the LXC"""
lxc.run_command(f"mkdir -p {path}") lxc.run_command(f"mkdir -p {path}")
if permission != 755: if permission != 755:
@ -42,24 +62,76 @@ def create_folder_in_lxc(lxc, path, permission=755):
def create_file_in_lxc(lxc, path, permission=644): def create_file_in_lxc(lxc, path, permission=644):
"""Create a file in the LXC"""
lxc.run_command(f"touch {path}") lxc.run_command(f"touch {path}")
if permission != 644: if permission != 644:
lxc.run_command(f"chmod {permission} {path}") lxc.run_command(f"chmod {permission} {path}")
def copy_local_file_to_lxc(lxc, path, destination): def copy_local_file_to_lxc(lxc, path, destination):
"""Copy a local file to the LXC
Parameters
----------
lxc : LXC
The LXC object of the LXC to copy the file to
path: str
Path to the file on the machine running this program
destination: str
Path to the destination in the LXC
"""
proxmox_utils.run_command_locally(f"scp {path} {lxc.get_ssh_string()}:{destination}") proxmox_utils.run_command_locally(f"scp {path} {lxc.get_ssh_string()}:{destination}")
def copy_local_folder_to_lxc(lxc, path, destination): def copy_local_folder_to_lxc(lxc, path, destination):
"""Copy a local folder to the LXC
Parameters
----------
lxc : LXC
The LXC object of the LXC to copy the folder to
path: str
Path to the folder on the machine running this program
destination: str
Path to the destination in the LXC
"""
proxmox_utils.run_command_locally(f"scp -r {path} {lxc.get_ssh_string()}:{destination}") proxmox_utils.run_command_locally(f"scp -r {path} {lxc.get_ssh_string()}:{destination}")
def run_docker_command(lxc, container, command): def run_docker_command(lxc, container, command):
"""Run a command inside a docker container on the LXC
Parameters
----------
lxc : LXC
The LXC object of the LXC to access the docker container on
container : str
Name of the docker container to run the command in
command : str
Command to run in the docker container
"""
lxc.run_command(f"docker exec -it {container} {command}") lxc.run_command(f"docker exec -it {container} {command}")
def run_docker_compose_command(lxc, command, working_directory=None): def run_docker_compose_command(lxc, command, working_directory=None):
"""Run a docker-compose command in the LXC
Parameters
----------
lxc: LXC
The LXC object of the LXC to run the docker-compose command in
command: str
The docker-compose command to run
working_directory: str
The working directory to run the command in
Examples
--------
>>> run_docker_compose_command(lxc, "up -d", "/home/user/traefik")
"""
docker_compose_exec = "docker-compose" docker_compose_exec = "docker-compose"
if not lxc.has_program(docker_compose_exec): if not lxc.has_program(docker_compose_exec):
docker_compose_exec = "docker compose" docker_compose_exec = "docker compose"
@ -71,6 +143,7 @@ def run_docker_compose_command(lxc, command, working_directory=None):
def download_file(lxc, url, destination): def download_file(lxc, url, destination):
"""Download a file from a URL to the LXC"""
if type(url) is list: if type(url) is list:
for u in url: for u in url:
lxc.run_command(f"wget {u} -O {destination}") lxc.run_command(f"wget {u} -O {destination}")
@ -79,6 +152,23 @@ def download_file(lxc, url, destination):
def unzip_file(lxc, path, destination=None): def unzip_file(lxc, path, destination=None):
"""Unzip a file in the LXC
Parameters
----------
lxc: LXC
The LXC object of the LXC to unzip the file in
path: str
Path to the file to unzip
destination: str
Path to the destination to unzip the file to
Examples
--------
>>> unzip_file(lxc, "/home/user/file.zip", "/home/user/destination")
>>> unzip_file(lxc, "/home/user/file.tar.gz", "/home/user/destination")
"""
if destination is None: if destination is None:
destination = Path(path).parent destination = Path(path).parent
@ -89,6 +179,21 @@ def unzip_file(lxc, path, destination=None):
def install_package(lxc, package): def install_package(lxc, package):
"""Install a package in the LXC
Parameters
----------
lxc: LXC
The LXC object of the LXC to install the package in
package: str or list
Name of the package to install
Examples
--------
>>> install_package(lxc, "nginx")
>>> install_package(lxc, ["nginx", "apache2"])
"""
if type(package) is list: if type(package) is list:
for p in package: for p in package:
lxc.run_command(f"{proxmox_utils.get_install_package_command(lxc.get_os_name())} {p} -y") lxc.run_command(f"{proxmox_utils.get_install_package_command(lxc.get_os_name())} {p} -y")
@ -97,6 +202,21 @@ def install_package(lxc, package):
def remove_package(lxc, package): def remove_package(lxc, package):
"""Remove a package in the LXC
Parameters
----------
lxc: LXC
The LXC object of the LXC to remove the package in
package: str or list
Name of the package to remove
Examples
--------
>>> remove_package(lxc, "nginx")
>>> remove_package(lxc, ["nginx", "apache2"])
"""
if type(package) is list: if type(package) is list:
packages = [] packages = []
for p in package: for p in package:
@ -108,6 +228,29 @@ def remove_package(lxc, package):
def replace_in_files(lxc, path, search, replace, case_sensitive=False): def replace_in_files(lxc, path, search, replace, case_sensitive=False):
"""Replace a string in one or multiples files in the LXC
Parameters
----------
lxc : LXC
The LXC object of the LXC to replace the string in
path : str or list
Path to the file(s) to replace the string in
search : str
String to search for
replace : str
String to replace the search string with
case_sensitive : bool
Whether the search should be case sensitive or not
Examples
--------
>>> replace_in_files(lxc, "/home/user/file.txt", "username=root", "username=administrator"
>>> replace_in_files(lxc, ["/home/user/file1.txt", "/home/user/file2.txt"], \
"username=root", "username=administrator", case_sensitive=True)
"""
if type(path) is list: if type(path) is list:
for p in path: for p in path:
lxc.run_command(f"sed -i {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {p}") lxc.run_command(f"sed -i {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {p}")

View File

@ -7,19 +7,31 @@ lxcs = []
def get_all_lxcs(): def get_all_lxcs():
"""Get all LXC objects
Returns
-------
list
List of all loaded LXC objects
""" """
Get all LXCs
:return: list of all lxcs
"""
return lxcs return lxcs
def get_lxc(lxc_id): def get_lxc(lxc_id):
"""Get LXC by ID
Parameters
----------
lxc_id : str
ID of the LXC to get
Returns
-------
LXC
LXC object
""" """
Get LXC by ID
:param lxc_id: lxc id
:return: lxc object
"""
for lxc in lxcs: for lxc in lxcs:
if lxc.get_id() == lxc_id: if lxc.get_id() == lxc_id:
return lxc return lxc
@ -28,13 +40,20 @@ def get_lxc(lxc_id):
def load_lxc(file, lxc_id): def load_lxc(file, lxc_id):
""" """Load LXC from JSON file
Load LXC from JSON file
:param file: json config file Parameters
:param lxc_id: id of the lxc ----------
:return: file : str
Path to the JSON file of the LXC to load
lxc_id : str
ID of the LXC to load
Examples
--------
>>> load_lxc("./resources/lxc/100/config.json", "100")
""" """
# Load JSON data # Load JSON data
data = json.loads(file) data = json.loads(file)
@ -54,9 +73,7 @@ def load_lxc(file, lxc_id):
class LXC: class LXC:
""" """LXC object"""
LXC class
"""
def __init__(self, lxc_id, lxc_hostname, os, resources, network, options, creation, deploy): def __init__(self, lxc_id, lxc_hostname, os, resources, network, options, creation, deploy):
self.lxc_id = lxc_id self.lxc_id = lxc_id