Merge "Migrate backup test cases under test_delete_subcloud_backup.py (1/2)"

keywords/cloud_platform/dcmanager/dcmanager_subcloud_backup_keywords.py

@@ -0,0 +1,190 @@
+from typing import Optional
+
+from framework.ssh.ssh_connection import SSHConnection
+from framework.validation.validation import validate_equals_with_retry
+from keywords.base_keyword import BaseKeyword
+from keywords.cloud_platform.command_wrappers import source_openrc
+from keywords.files.file_keywords import FileKeywords
+
+
+class DcManagerSubcloudBackupKeywords(BaseKeyword):
+    """
+    This class contains all the keywords related to the 'dcmanager subcloud-backup <create/delete>' command.
+    """
+
+    def __init__(self, ssh_connection: SSHConnection):
+        """
+        Constructor.
+
+        Args:
+            ssh_connection (SSHConnection): SSH connection to the target system.
+        """
+        self.ssh_connection = ssh_connection
+
+    def create_subcloud_backup(
+        self,
+        sysadmin_password: str,
+        con_ssh: SSHConnection,
+        path: str,
+        subcloud: Optional[str] = None,
+        local_only: bool = False,
+        backup_yaml: Optional[str] = None,
+        group: Optional[str] = None,
+        registry: bool = False,
+    ) -> None:
+        """
+        Creates a backup of the specified subcloud.
+
+        Args:
+            sysadmin_password (str): Subcloud sysadmin password needed for backup creation.
+            con_ssh (SSHConnection): SSH connection to execute the command (central_ssh or subcloud_ssh).
+            path (str): The directory path where the backup file will be checked.
+            subcloud (Optional[str]): The name of the subcloud to backup. Defaults to None.
+            local_only (bool): If True, backup will be stored only in the subcloud. Defaults to False.
+            backup_yaml (Optional[str]): path to use the yaml file. Defaults to None.
+            group (Optional[str]): Subcloud group name to create backup. Defaults to None.
+            registry (bool): Option to add the registry backup in the same task. Defaults to False.
+
+        Returns:
+            None:
+        """
+        # Command construction
+        cmd = (
+            f"dcmanager subcloud-backup create --sysadmin-password {sysadmin_password}"
+        )
+        if subcloud:
+            cmd += f" --subcloud {subcloud}"
+        if local_only:
+            cmd += " --local-only"
+        if backup_yaml:
+            cmd += f" --backup-values {backup_yaml}"
+        if group:
+            cmd += f" --group {group}"
+        if registry:
+            cmd += " --registry-images"
+
+        self.ssh_connection.send(source_openrc(cmd))
+        self.validate_success_return_code(self.ssh_connection)
+
+        # Use wait_for_backup_creation to ensure the file is created
+        self.wait_for_backup_creation(con_ssh, path, subcloud)
+
+    def wait_for_backup_creation(
+        self,
+        con_ssh: SSHConnection,
+        path: str,
+        subcloud: Optional[str],
+        check_interval: int = 30,
+        timeout: int = 600,
+    ) -> None:
+        """
+        Waits for the backup file to be created in the specified path.
+
+        Args:
+            con_ssh (SSHConnection): SSH connection to execute the command (central_ssh or subcloud_ssh).
+            path (str): The path where the backup file is expected.
+            subcloud (Optional[str]): The name of the subcloud to check.
+            check_interval (int): Time interval (in seconds) to check for file creation. Defaults to 30.
+            timeout (int): Maximum time (in seconds) to wait for file creation. Defaults to 600.
+
+        Returns:
+            None:
+        """
+
+        def check_backup_created() -> str:
+            """
+            Checks if the backup has been created.
+
+            Returns:
+                str: A message indicating whether the backup has been successfully created or not.
+            """
+            check_file = FileKeywords(con_ssh).validate_file_exists_with_sudo(path)
+            if check_file:
+                return f"Backup should be created at {path}"
+            else:
+                return "Backup not created yet."
+
+        validate_equals_with_retry(
+            function_to_execute=check_backup_created,
+            expected_value=f"Backup should be created at {path}",
+            validation_description=f"Backup creation for subcloud {subcloud} completed.",
+            timeout=timeout,
+            polling_sleep_time=check_interval,
+        )
+
+    def delete_subcloud_backup(
+        self,
+        con_ssh: SSHConnection,
+        path: str,
+        release: str,
+        subcloud: Optional[str] = None,
+        local_only: bool = False,
+        group: Optional[str] = None,
+        sysadmin_password: str = None,
+    ) -> None:
+        """
+        Sends the command to delete the backup of the specified subcloud and waits for confirmation of its deletion.
+
+        Args:
+            con_ssh (SSHConnection): SSH connection to execute the command (central_ssh or subcloud_ssh).
+            path (str): The path where the backup file is located.
+            release (str): Required to delete a release backup.
+            subcloud (Optional[str]): The name of the subcloud to delete the backup. Defaults to None.
+            local_only (bool): If True, only deletes the local backup in the subcloud. Defaults to False.
+            group (Optional[str]): Subcloud group name to delete backup. Defaults to None.
+            sysadmin_password (str): Subcloud sysadmin password needed for deletion on local_path. Defaults to None.
+
+        Returns:
+            None:
+        """
+        # Command construction for backup deletion
+        cmd = f"dcmanager subcloud-backup delete {release}"
+        if subcloud:
+            cmd += f" --subcloud {subcloud}"
+        if local_only:
+            cmd += " --local-only"
+        if group:
+            cmd += f" --group {group}"
+        if sysadmin_password:
+            cmd += f" --sysadmin-password {sysadmin_password}"
+
+        self.ssh_connection.send(source_openrc(cmd))
+        self.validate_success_return_code(self.ssh_connection)
+
+        # Call wait_for_backup_deletion method to wait and verify the backup deletion.
+        self.wait_for_backup_deletion(con_ssh, path, subcloud)
+
+    def wait_for_backup_deletion(
+        self, con_ssh: SSHConnection, path: str, subcloud: str
+    ) -> None:
+        """
+        Waits for the backup to be deleted by checking for the absence of the backup file.
+
+        Args:
+            con_ssh (SSHConnection): SSH connection object to execute the command.
+            path (str): The path where the backup file was located.
+            subcloud (str): The name of the subcloud to delete the backup.
+
+        Returns:
+            None:
+        """
+
+        def check_backup_deleted() -> str:
+            """
+            Checks if the backup has been deleted.
+
+            Returns:
+                str: Confirmation message if the backup is deleted, otherwise an error message.
+            """
+            check_file = FileKeywords(con_ssh).validate_file_exists_with_sudo(path)
+            if not check_file:
+                return f"Backup successfully deleted from {path} for {subcloud}"
+            else:
+                return f"Backup still exists at {path}."
+
+        # Using validate_equals_with_retry to ensure the backup is deleted.
+        validate_equals_with_retry(
+            function_to_execute=check_backup_deleted,
+            expected_value=f"Backup successfully deleted from {path} for {subcloud}",
+            validation_description=f"Backup deletion for subcloud {subcloud} completed.",
+        )

keywords/files/file_keywords.py

@@ -33,10 +33,10 @@ class FileKeywords(BaseKeyword):
             sftp_client.get(remote_file_path, local_file_path)
         except Exception as e:
             get_logger().log_error(
-                f'Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}'
+                f"Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}"
             )
             raise KeywordException(
-                f'Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}'
+                f"Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}"
             )
         return True
 
@@ -66,10 +66,10 @@ class FileKeywords(BaseKeyword):
                 sftp_client.put(local_file_path, remote_file_path)
         except Exception as e:
             get_logger().log_error(
-                f'Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}'
+                f"Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}"
             )
             raise KeywordException(
-                f'Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}'
+                f"Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}"
             )
         return True
 
@@ -102,7 +102,7 @@ class FileKeywords(BaseKeyword):
         Returns:
             bool: True if delete successful, False otherwise.
         """
-        self.ssh_connection.send_as_sudo(f'rm {file_name}')
+        self.ssh_connection.send_as_sudo(f"rm {file_name}")
         return self.file_exists(file_name)
 
     def get_files_in_dir(self, file_dir: str) -> list[str]:
@@ -137,7 +137,7 @@ class FileKeywords(BaseKeyword):
         end_line = 10000  # we can handle 10000 lines without issue
         end_time = time.time() + 300
 
-        grep_arg = ''
+        grep_arg = ""
         if grep_pattern:
             grep_arg = f"| grep {grep_pattern}"
 
@@ -152,3 +152,31 @@ class FileKeywords(BaseKeyword):
             end_line = end_line + 10000
 
         return total_output
+
+    def validate_file_exists_with_sudo(self, path: str) -> bool:
+        """
+        Validates whether a file or directory exists at the specified path using sudo.
+
+        Args:
+            path (str): The path to the file or directory.
+
+        Returns:
+            bool: True if the file/directory exists, False otherwise.
+
+        Raises:
+            KeywordException: If there is an error executing the SSH command.
+        """
+        try:
+            cmd = f"find {path} -mtime 0"
+            output = self.ssh_connection.send_as_sudo(cmd)
+
+            # Handle encoding issues
+            output = "".join(
+                [line.replace("‘", "").replace("’", "") for line in output]
+            )
+
+            return "No such file or directory" not in output
+
+        except Exception as e:
+            get_logger().log_error(f"Failed to check file existence at {path}: {e}")
+            raise KeywordException(f"Failed to check file existence at {path}: {e}")

testcases/cloud_platform/regression/dc/backup_restore/test_delete_subcloud_backup.py

@@ -0,0 +1,150 @@
+from pytest import mark
+
+from config.configuration_manager import ConfigurationManager
+from framework.logging.automation_logger import get_logger
+from keywords.cloud_platform.dcmanager.dcmanager_subcloud_backup_keywords import (
+    DcManagerSubcloudBackupKeywords,
+)
+from keywords.cloud_platform.dcmanager.dcmanager_subcloud_list_keywords import (
+    DcManagerSubcloudListKeywords,
+)
+from keywords.cloud_platform.rest.bare_metal.hosts.get_hosts_keywords import (
+    GetHostsKeywords,
+)
+from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
+from keywords.cloud_platform.system.host.system_host_list_keywords import (
+    SystemHostListKeywords,
+)
+
+
+@mark.p2
+@mark.lab_has_subcloud
+def test_delete_backup_central(request):
+    """
+    Verify delete centralized subcloud backup
+
+    Test Steps:
+        - Create a Subcloud backup and check it on central path
+        - Delete the backup created and the backup is deleted
+    Teardown:
+        - Remove files created while the Tc was running.
+
+    """
+    central_ssh = LabConnectionKeywords().get_active_controller_ssh()
+    host = SystemHostListKeywords(central_ssh).get_active_controller().get_host_name()
+    host_show_output = GetHostsKeywords().get_hosts().get_system_host_show_object(host)
+
+    # Gets the lowest subcloud (the subcloud with the lowest id).
+    dcmanager_subcloud_list_keywords = DcManagerSubcloudListKeywords(central_ssh)
+    lowest_subcloud = (
+        dcmanager_subcloud_list_keywords.get_dcmanager_subcloud_list().get_healthy_subcloud_with_lowest_id()
+    )
+    subcloud_name = lowest_subcloud.get_name()
+    subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(subcloud_name)
+
+    # Gets the lowest subcloud sysadmin password needed for backup creation.
+    lab_config = ConfigurationManager.get_lab_config().get_subcloud(subcloud_name)
+    subcloud_password = lab_config.get_admin_credentials().get_password()
+
+    # Get the sw_version if available (used in vbox environments).
+    release = host_show_output.get_sw_version()
+    # If sw_version is not available, fall back to software_load (used in physical labs).
+    if not release:
+        release = host_show_output.get_software_load()
+
+    dc_manager_backup = DcManagerSubcloudBackupKeywords(central_ssh)
+
+    # Path to where the backup file will store.
+    central_path = f"/opt/dc-vault/backups/{subcloud_name}/{release}"
+
+    def teardown():
+        get_logger().log_info("Removing test files during teardown")
+        central_ssh.send_as_sudo("rm -r -f /opt/dc-vault/backups/")
+        subcloud_ssh.send_as_sudo("rm -r -f /opt/platform-backup/backups/")
+
+    request.addfinalizer(teardown)
+
+    # Create a sbcloud backup
+    get_logger().log_info(f"Create {subcloud_name} backup on Central Cloud")
+    dc_manager_backup.create_subcloud_backup(
+        subcloud_password, central_ssh, central_path, subcloud=subcloud_name
+    )
+
+    # Delete the backup created
+    get_logger().log_info(f"Delete {subcloud_name} backup on Central Cloud")
+    dc_manager_backup.delete_subcloud_backup(
+        central_ssh, central_path, release, subcloud=subcloud_name
+    )
+
+
+@mark.p2
+@mark.lab_has_subcloud
+def test_delete_backup_local(request):
+    """
+    Verify delete subcloud backup on local path
+
+    Test Steps:
+        - Create a Subcloud backup and check it on local path
+        - Delete the backup created and verify the backup is deleted
+    Teardown:
+        - Remove files created while the Tc was running.
+
+    """
+    central_ssh = LabConnectionKeywords().get_active_controller_ssh()
+    host = SystemHostListKeywords(central_ssh).get_active_controller().get_host_name()
+    host_show_output = GetHostsKeywords().get_hosts().get_system_host_show_object(host)
+
+    # Gets the lowest subcloud (the subcloud with the lowest id).
+    dcmanager_subcloud_list_keywords = DcManagerSubcloudListKeywords(central_ssh)
+    lowest_subcloud = (
+        dcmanager_subcloud_list_keywords.get_dcmanager_subcloud_list().get_healthy_subcloud_with_lowest_id()
+    )
+    subcloud_name = lowest_subcloud.get_name()
+    subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(subcloud_name)
+
+    # Gets the lowest subcloud sysadmin password needed for backup backup creation and deletion on local_path.
+    lab_config = ConfigurationManager.get_lab_config().get_subcloud(subcloud_name)
+    subcloud_password = lab_config.get_admin_credentials().get_password()
+
+    # Get the sw_version if available (used in vbox environments).
+    release = host_show_output.get_sw_version()
+    # If sw_version is not available, fall back to software_load (used in physical labs).
+    if not release:
+        release = host_show_output.get_software_load()
+
+    dc_manager_backup = DcManagerSubcloudBackupKeywords(central_ssh)
+
+    # Path to where the backup file will store.
+    local_path = (
+        f"/opt/platform-backup/backups/{release}/{subcloud_name}_platform_backup_*.tgz"
+    )
+
+    def teardown():
+        get_logger().log_info("Removing test files during teardown")
+        subcloud_ssh.send_as_sudo("rm -r -f /opt/platform-backup/backups/")
+
+    request.addfinalizer(teardown)
+
+    # Create a subcloud backup on local
+    get_logger().log_info(f"Create {subcloud_name} backup on local")
+    dc_manager_backup.create_subcloud_backup(
+        subcloud_password,
+        subcloud_ssh,
+        local_path,
+        subcloud=subcloud_name,
+        local_only=True,
+    )
+
+    # path where the backup directory should be checked for deletion.
+    path = f"/opt/platform-backup/backups/{release}/"
+
+    # Delete the backup created on subcloud
+    get_logger().log_info(f"Delete {subcloud_name} backup on Central Cloud")
+    dc_manager_backup.delete_subcloud_backup(
+        subcloud_ssh,
+        path,
+        release,
+        subcloud=subcloud_name,
+        local_only=True,
+        sysadmin_password=subcloud_password,
+    )