From 25b208094dc0385fb95251999aa64b9dc424e6af Mon Sep 17 00:00:00 2001 From: Mathieu Broillet Date: Thu, 15 Jun 2023 17:10:16 +0200 Subject: [PATCH] merged commands_utils into Machine class --- src/utils/commands_utils.py | 195 ------------------------------------ src/utils/machine.py | 185 +++++++++++++++++++++++++++++++++- 2 files changed, 183 insertions(+), 197 deletions(-) delete mode 100644 src/utils/commands_utils.py diff --git a/src/utils/commands_utils.py b/src/utils/commands_utils.py deleted file mode 100644 index 0024773..0000000 --- a/src/utils/commands_utils.py +++ /dev/null @@ -1,195 +0,0 @@ -from pathlib import PurePosixPath - -from src.utils import utils -from src.utils.machine import LinuxMachine - - -def run_docker_command(linux_machine: LinuxMachine, container: str, command: str): - """Run a command inside a docker container on a linux host - - Parameters - ---------- - linux_machine : LinuxMachine - The LinuxMachine object of the linux host to run the command in - container : str - Name of the docker container to run the command in - command : str - Command to run in the docker container - - Examples - -------- - >>> run_docker_command(linux_machine, "", "") - """ - - linux_machine.run_command(f"docker exec -it {container} {command}", exception_on_empty_stdout=False) - - -def run_docker_compose_command(linux_machine: LinuxMachine, command: str, working_directory: str = None): - """Run a docker-compose command on a linux host - - Parameters - ---------- - linux_machine: LinuxMachine - The LinuxMachine object of the linux host to run the command in - command: str - The docker-compose command to run - working_directory: str, optional - The working directory to run the command in - - Examples - -------- - >>> run_docker_compose_command(linux_machine, "up -d", "/home/user/traefik") - """ - - docker_compose_exec = "docker-compose" - if not linux_machine.has_program(docker_compose_exec): - docker_compose_exec = "docker compose" - - if working_directory is not None: - linux_machine.run_command(f"cd {working_directory} && {docker_compose_exec} {command}", return_status_code=True) - else: - linux_machine.run_command(f"{docker_compose_exec} {command}", return_status_code=True) - - -def download_file(linux_machine: LinuxMachine, url: str, destination: str): - """Download a file from a URL to the Linux Machine and save it to the destination - - Parameters - ---------- - linux_machine: LinuxMachine - The LinuxMachine object of the linux host to download the file to - url: str - URL of the file to download - destination: str - Path to the destination to save the file to - Needs to end with a trailing slash - - Examples - -------- - >>> download_file(linux_machine, "https://example.com/file.zip", "/home/user/") - - """ - if type(url) is list: - for u in url: - linux_machine.run_command(f"wget {u} --directory-prefix={destination}", return_status_code=True) - else: - linux_machine.run_command(f"wget {url} --directory-prefix={destination}", return_status_code=True) - - -def unzip_file(linux_machine: LinuxMachine, path: str, destination: str = None): - """Unzip a file - - Parameters - ---------- - linux_machine: LinuxMachine - The LinuxMachine object of the linux host to unzip the file in - path: str - Path to the file to unzip - destination: str, optional - Path to the destination to unzip the file to - If not specified, it will unzip the file in the same directory as the file - Needs to end with a trailing slash - - Examples - -------- - >>> unzip_file(linux_machine, "/home/user/file.zip", "/home/user/extracted_files/") - >>> unzip_file(linux_machine, "/home/user/file.tar.gz", "/home/user/extracted_files/") - """ - - if destination is None: - destination = PurePosixPath(path).parent - - if ".zip" in path: - linux_machine.run_command(f"unzip {path} -d {destination}", return_status_code=True) - elif ".tar.gz" in path: - linux_machine.run_command(f"mkdir -p {destination} && tar -xzf {path} --directory {destination}", - return_status_code=True) - - -def install_package(linux_machine: LinuxMachine, package: str or list): - """Install a package in the Linux Machine - - Parameters - ---------- - linux_machine: LinuxMachine - The LinuxMachine object of the host to install the package in - package: str or list - Name(s) of the package(s) to install - - Examples - -------- - >>> install_package(linux_machine, "nginx") - >>> install_package(linux_machine, ["nginx", "apache2"]) - """ - - if type(package) is list: - for p in package: - linux_machine.run_command(f"{utils.get_install_package_command(linux_machine.get_os_name())} {package}", - return_status_code=True) - else: - linux_machine.run_command(f"{utils.get_install_package_command(linux_machine.get_os_name())} {package}", - return_status_code=True) - - -def remove_package(linux_machine: LinuxMachine, package: str or list): - """Remove a package in the Linux Machine - - Parameters - ---------- - linux_machine: LinuxMachine - The LinuxMachine object of the host to remove the package in - package: str or list - Name(s) of the package(s) to remove - - Examples - -------- - >>> remove_package(linux_machine, "nginx") - >>> remove_package(linux_machine, ["nginx", "apache2"]) - """ - - if type(package) is list: - packages = [] - for p in package: - packages.append(p) - - linux_machine.run_command( - f"{utils.get_remove_package_command(linux_machine.get_os_name())} {' '.join(packages)}", - return_status_code=True) - - else: - linux_machine.run_command(f"{utils.get_remove_package_command(linux_machine.get_os_name())} {package}", - return_status_code=True) - - -def replace_in_files(linux_machine: LinuxMachine, path: str or list, search: str, replace: str, - case_sensitive: bool = False): - """Replace a string in one or multiples files in a LinuxMachine - - Parameters - ---------- - linux_machine : LinuxMachine - The LinuxMachine object of the host to replace the string in - path : str or list of str - 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, optional - Whether the search should be case sensitive or not - - Examples - -------- - >>> replace_in_files(linux_machine, "/home/user/file.txt", "username=root", "username=administrator" - >>> replace_in_files(linux_machine, ["/home/user/file1.txt", "/home/user/file2.txt"], \ - "username=root", "username=administrator", case_sensitive=True) - - """ - - if type(path) is list: - for p in path: - linux_machine.run_command(f"sed {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {p}", - return_status_code=True) - else: - linux_machine.run_command(f"sed {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {path}", - return_status_code=True) diff --git a/src/utils/machine.py b/src/utils/machine.py index 0c9efdd..62bbc7b 100644 --- a/src/utils/machine.py +++ b/src/utils/machine.py @@ -1,4 +1,6 @@ -from pathlib import Path +from pathlib import Path, PurePosixPath + +from src.utils import utils class LinuxMachine(): @@ -13,7 +15,13 @@ class LinuxMachine(): return self.run_command("uptime -p") def get_os_name(self): - """Get OS name""" + """Get OS name + + Returns + ------- + str + OS name + """ return self.run_command("""cat /etc/os-release | grep -E '^NAME=' | cut -d '=' -f 2 | tr -d '"'""") def get_memory(self): @@ -116,3 +124,176 @@ class LinuxMachine(): directory = str(directory.as_posix()) self.run_command(f"rm -rf {directory}", return_status_code=True) + + def run_docker_command(self, container, command): + """Run a command inside a docker container on a linux host + + Parameters + ---------- + container : str + Name of the docker container to run the command in + command : str + Command to run in the docker container + + Examples + -------- + >>> self.run_docker_command(linux_machine, "", "") + """ + + if not self.has_program("docker"): + raise Exception(f"Docker is not installed on this machine {self.get_hostname()}") + + self.run_command(f"docker exec -it {container} {command}", exception_on_empty_stdout=False) + + def run_docker_compose_command(self, command: str, working_directory: str = None): + """Run a docker-compose command on a linux host + + Parameters + ---------- + command: str + The docker-compose command to run + working_directory: str, optional + The working directory to run the command in + + Examples + -------- + >>> self.run_docker_compose_command(linux_machine, "up -d", "/home/user/traefik") + """ + + docker_compose_exec = "docker-compose" + if not self.has_program(docker_compose_exec): + docker_compose_exec = "docker compose" + + if working_directory is not None: + self.run_command(f"cd {working_directory} && {docker_compose_exec} {command}", return_status_code=True) + else: + self.run_command(f"{docker_compose_exec} {command}", return_status_code=True) + + def download_file(self, url: str, destination: str): + """Download a file from a URL to the Linux Machine and save it to the destination + + Parameters + ---------- + url: str + URL of the file to download + destination: str + Path to the destination to save the file to + Needs to end with a trailing slash + + Examples + -------- + >>> self.download_file("https://example.com/file.zip", "/home/user/") + """ + + if type(url) is list: + for u in url: + self.run_command(f"wget {u} --directory-prefix={destination}", return_status_code=True) + else: + self.run_command(f"wget {url} --directory-prefix={destination}", return_status_code=True) + + def unzip_file(self, path: str, destination: str = None): + """Unzip a file + + Parameters + ---------- + path: str + Path to the file to unzip + destination: str, optional + Path to the destination to unzip the file to + If not specified, it will unzip the file in the same directory as the file + Needs to end with a trailing slash + + Examples + -------- + >>> self.unzip_file("/home/user/file.zip", "/home/user/extracted_files/") + >>> self.unzip_file("/home/user/file.tar.gz", "/home/user/extracted_files/") + """ + + if destination is None: + destination = PurePosixPath(path).parent + + if ".zip" in path: + self.run_command(f"unzip {path} -d {destination}", return_status_code=True) + elif ".tar.gz" in path: + self.run_command(f"mkdir -p {destination} && tar -xzf {path} --directory {destination}", + return_status_code=True) + + def install_package(self, package: str or list): + """Install a package in the Linux Machine + + Parameters + ---------- + package: str or list + Name(s) of the package(s) to install + + Examples + -------- + >>> self.install_package("nginx") + >>> self.install_package(["nginx", "apache2"]) + """ + + if type(package) is list: + for p in package: + self.run_command(f"{utils.get_install_package_command(self.get_os_name())} {package}", + return_status_code=True) + else: + self.run_command(f"{utils.get_install_package_command(self.get_os_name())} {package}", + return_status_code=True) + + def remove_package(self, package: str or list): + """Remove a package in the Linux Machine + + Parameters + ---------- + package: str or list + Name(s) of the package(s) to remove + + Examples + -------- + >>> self.remove_package("nginx") + >>> self.remove_package(["nginx", "apache2"]) + """ + + if type(package) is list: + packages = [] + for p in package: + packages.append(p) + + self.run_command( + f"{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}", + return_status_code=True) + + def replace_in_files(self, path: str or list, search: str, replace: str, + case_sensitive: bool = False): + """Replace a string in one or multiples files in a LinuxMachine + + Parameters + ---------- + path : str or list of str + 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, optional + Whether the search should be case sensitive or not + + Examples + -------- + >>> self.replace_in_files("/home/user/file.txt", "username=root", "username=administrator" + >>> self.replace_in_files(["/home/user/file1.txt", "/home/user/file2.txt"], \ + "username=root", "username=administrator", case_sensitive=True) + + """ + + if type(path) is list: + for p in path: + self.run_command(f"sed {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {p}", + return_status_code=True) + else: + self.run_command(f"sed {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {path}", + return_status_code=True)