merged commands_utils into Machine class

src/utils/commands_utils.py

@@ -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, "<container-name>", "<command>")
-    """
-
-    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)

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, "<container-name>", "<command>")
+        """
+
+        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)