[FIX] cetmix_tower_server: SSH connection test

- cx.tower.server: Refactor SSH connection retrieval function.
  Use better naming, allow to pass timeout.
- cx.tower.server: Refactor SSH connection test function.
- cetmix.tower: Use SSH connection test function from cx.tower.server

cetmix_tower_server/models/cetmix_tower.py

@@ -1,10 +1,11 @@
 # Copyright (C) 2024 Cetmix OÜ
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+import time
+
 from odoo import _, api, models
 from odoo.exceptions import ValidationError
 
-from .constants import SSH_CONNECTION_ERROR, SSH_CONNECTION_TIMEOUT
-from .cx_tower_server import SSH
+from .constants import SSH_CONNECTION_ERROR
 
 
 class CetmixTower(models.AbstractModel):
@@ -105,84 +106,65 @@ def server_get_variable_value(
         return value
 
     @api.model
-    def server_check_ssh_connection(self, server_reference, attempts=5, timeout=15):
+    def server_check_ssh_connection(
+        self,
+        server_reference,
+        attempts=5,
+        wait_time=10,
+        try_command=True,
+        try_file=True,
+    ):
         """Check if SSH connection to the server is available.
+        This method only checks if the connection is available,
+        it does not execute any commands to check if they are working.
 
         Args:
             server_reference (Char): Server reference.
             attempts (int): Number of attempts to try the connection.
                 Default is 5.
-            timeout (int): Timeout in seconds for each connection attempt.
-                Default is 15 seconds.
+            wait_time (int): Wait time in seconds between connection attempts.
+                Default is 10 seconds.
+            try_command (bool): Try to execute a command.
+                Default is True.
+            try_file (bool): Try file operations.
+                Default is True.
         Raises:
             ValidationError:
                 If the provided server reference is invalid or
                 the server cannot be found.
         Returns:
             dict: {
                 "code": int,
-                    # 0 for success,
-                    # 408 if the SSH connection timed out after all attempts,
-                    # 503 if there was a generic SSH connection error.
+                    0 for success,
+                    error code for failure
                 "message": str  # Description of the result
             }
         """
         server = self.env["cx.tower.server"].get_by_reference(server_reference)
         if not server:
             raise ValidationError(_("No server found for the provided reference."))
 
-        # Prepare SSH connection parameters
-        ssh_params = {
-            "host": server.ip_v4_address or server.ip_v6_address,
-            "username": server.ssh_username,
-            "port": int(server.ssh_port),
-            "timeout": timeout,
-            "mode": server.ssh_auth_mode,
-        }
-
-        if server.ssh_auth_mode == "p":
-            ssh_params["password"] = server.ssh_password
-        elif server.ssh_auth_mode == "k":
-            ssh_params["ssh_key"] = server.ssh_key_id.sudo().secret_value
-
-        # Initialize SSH connection instance
-        ssh_connection = SSH(**ssh_params)
-
         # Try connecting multiple times
         for attempt in range(1, attempts + 1):
-            try:
-                ssh_connection.connection()
+            result = server.test_ssh_connection(
+                raise_on_error=False,
+                return_notification=False,
+                try_command=try_command,
+                try_file=try_file,
+            )
+            if result.get("status") == 0:
                 return {
-                    "code": 0,
+                    "exit_code": 0,
                     "message": _("Connection successful."),
                 }
-            except TimeoutError as e:
-                if attempt == attempts:
-                    return {
-                        "code": SSH_CONNECTION_TIMEOUT,
-                        "message": _(
-                            "Connection timed out after %(attempts)s attempts. "
-                            "Error: %(err)s",
-                            attempts=attempts,
-                            err=str(e),
-                        ),
-                    }
-            except Exception as e:
-                if attempt == attempts:
-                    return {
-                        "code": SSH_CONNECTION_ERROR,
-                        "message": _(
-                            "Failed to connect after %(attempts)s attempts. "
-                            "Error: %(err)s",
-                            attempts=attempts,
-                            err=str(e),
-                        ),
-                    }
-            finally:
-                ssh_connection.disconnect()
-
-        # If all attempts fail
-        return {
-            "code": SSH_CONNECTION_ERROR,
-            "message": _("All connection connection attempts have failed."),
-        }
+            if attempt == attempts:
+                return {
+                    "exit_code": SSH_CONNECTION_ERROR,
+                    "message": _(
+                        "Failed to connect after %(attempts)s attempts. "
+                        "Error: %(err)s",
+                        attempts=attempts,
+                        err=result.get("error", ""),
+                    ),
+                }
+            time.sleep(wait_time)

cetmix_tower_server/models/constants.py

@@ -35,6 +35,3 @@
 
 # Returned when an SSH connection error occurs
 SSH_CONNECTION_ERROR = 503
-
-# Returned when the connection times out
-SSH_CONNECTION_TIMEOUT = 408